upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-07-23 11:11:00 -05:00
Code Issues Packages Projects Releases Wiki Activity

Enable multivector in model test, repair stale multi-dense vectors length on reload

Browse Source 
The multi-dense appendable storage splits each point across two
independently flushed chunked stores (offsets and vectors). Flushers
snapshot status.len at creation time but msync chunk data at execution
time, so a re-upsert that grows a point's row count in that window can
land its relocated offsets entry and rows durably while the vectors
store's recorded length stays stale. On reload, reads of such entries
are rejected and a WAL-replayed append reuses the referenced rows,
overwriting another point's data.

Fix: reload-time repair when opening the storage. Scan live offset
entries and widen the recorded vectors length to cover the maximum
referenced row end (rows are always msynced before the entries that
reference them, so only the length can lag).

Also enable the "m" multivector in the Collection model test, which is
what surfaces this bug (seed 1 panics at the restart at op 1089 without
the fix; seeds 1-7 and both harness smoke tests pass with it).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Arnaud Gourlay
2026-06-12 15:08:18 +02:00
committed by root
parent dbcfc1c05a
commit 38a1b8a4ee
4 changed files with 150 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
lib/collection/src/model_testing/mod.rs
View File

@@ -58,8 +58,7 @@ pub(super) const ALL_CANDIDATES: &[VectorCandidate] = &[
];
/// Names present in the collection schema at fixture time.
// "m" (multivector) is temporarily disabled while its reload divergence is unresolved.
pub(super) const INITIAL_ACTIVE: &[&str] = &["a", "b", "i", "s"];
pub(super) const INITIAL_ACTIVE: &[&str] = &["a", "b", "i", "s", "m"];
/// Dense vector name configured with HNSW `inline_storage` + scalar quantization in the fixture.
pub(super) const INLINE_STORAGE_VECTOR: &str = "i";

3
lib/collection/src/model_testing/op/mod.rs
View File

@@ -428,12 +428,9 @@ impl Op {
22 => {
// CreateVectorName: pick a candidate currently absent from the active set.
// If all candidates are active, fall back to a regular upsert.
// Multivector (`MultiDense`) candidates are excluded while their reload
// divergence is unresolved — see the note on `INITIAL_ACTIVE`.
let inactive: Vec<&'static super::VectorCandidate> = ALL_CANDIDATES
.iter()
.filter(|c| !active.contains(c.name))
.filter(|c| !matches!(c.kind, VectorKind::MultiDense(_)))
.collect();
if inactive.is_empty() {
return upsert_fallback(rng, active, id_pool);

16
lib/segment/src/vector_storage/chunked_vectors/write.rs
View File

@@ -230,6 +230,22 @@ where
Ok(new_id)
}
/// Widen the recorded length to at least `len` without writing any vector data.
///
/// Reload-time repair hook: flushers snapshot `status.len` at creation time but msync
/// chunk data at execution time, so a crash can leave durable rows (and durable
/// references to them in a sibling store) beyond the recorded length. The data is
/// already on disk; only the length lagged. Clamped to the physically existing chunk
/// space so a corrupt caller-supplied value can't make reads address missing chunks.
pub fn ensure_len_at_least(&mut self, len: usize) {
let physical_capacity = self.inner.chunks.len() * self.inner.config.chunk_size_vectors;
let len = len.min(physical_capacity);
if len > self.status.len {
self.status.len = len;
self.inner.len = len;
}
}
pub fn flusher(&self) -> Flusher {
Box::new({
let status_flusher = self.status.flusher();

134
lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs
View File

@@ -573,8 +573,9 @@ pub fn open_appendable_memmap_multi_vector_storage_impl<T: PrimitiveVectorElemen
let offsets_path = path.join(OFFSETS_DIR_PATH);
let deleted_path = path.join(DELETED_DIR_PATH);
let vectors = ChunkedVectors::open(MmapFs, &vectors_path, dim, madvise, Some(populate))?;
let mut vectors = ChunkedVectors::open(MmapFs, &vectors_path, dim, madvise, Some(populate))?;
let offsets = ChunkedVectors::open(MmapFs, &offsets_path, 1, madvise, Some(populate))?;
repair_vectors_len(&mut vectors, &offsets);
let deleted = BitvecFlags::new(
MmapFs,
@@ -592,6 +593,52 @@ pub fn open_appendable_memmap_multi_vector_storage_impl<T: PrimitiveVectorElemen
})
}
/// Reload-time repair for the two-store flush skew.
///
/// `vectors` and `offsets` are flushed independently, and each flusher snapshots its
/// `status.len` at creation time while msyncing chunk data at execution time. A point
/// re-upserted through the append path in that window rewrites its `offsets` entry in
/// place to a freshly-appended row region; the entry can land durably while `vectors`'
/// recorded length still predates the rows it references. The rows themselves are
/// durable (the storage flusher msyncs `vectors` before `offsets`), so it suffices to
/// widen the recorded length to cover the maximum referenced row-end. Without this,
/// reads of such entries are rejected and, worse, a WAL-replayed append reuses those
/// rows (`new_key = vectors.len()`) and overwrites another point's data.
fn repair_vectors_len<T: PrimitiveVectorElement>(
vectors: &mut ChunkedVectors<T, MmapFile>,
offsets: &ChunkedVectors<MultivectorMmapOffset, MmapFile>,
) {
let mut max_row_end = 0;
for key in 0..offsets.len() {
let Some(entry) = offsets.get::<Sequential>(key as VectorOffsetType) else {
continue;
};
let &[
MultivectorMmapOffset {
offset,
count: _,
// The reserved region is `offset..offset + capacity`; `count` only shrinks
// within it on in-place re-upserts, and the original append recorded
// `offset + capacity` as the length contribution.
capacity,
},
] = entry.as_ref()
else {
unreachable!("multi-vector offsets are stored as vectors of length 1");
};
max_row_end = max_row_end.max(offset as usize + capacity as usize);
}
if max_row_end > vectors.len() {
log::warn!(
"Multi-vector offsets reference rows beyond the recorded vectors length \
({} > {}); widening it (stale length snapshot from an interrupted flush)",
max_row_end,
vectors.len(),
);
vectors.ensure_len_at_least(max_row_end);
}
}
/// Find files related to this dense vector storage
#[cfg(test)]
pub(crate) fn find_storage_files(vector_storage_path: &Path) -> OperationResult<Vec<PathBuf>> {
@@ -675,4 +722,89 @@ mod tests {
"find_storage_files must find same files that storage reports",
);
}
/// Reproduces the on-disk state left by an interrupted flush: rows and the offsets
/// entries referencing them are durable, but the vectors store's recorded length
/// (`status.dat`) is stale. Without `repair_vectors_len`, reopening makes those
/// points unreadable and lets appends overwrite their rows.
#[test]
fn repair_stale_vectors_len_on_reload() {
const POINT_COUNT: PointOffsetType = 100;
const DIM: usize = 4;
let dir = Builder::new().prefix("storage_dir").tempdir().unwrap();
let mut rng = StdRng::seed_from_u64(RAND_SEED);
let hw_counter = HardwareCounterCell::disposable();
let mut expected = Vec::new();
{
let mut storage =
open_appendable_memmap_multi_vector_storage_impl::<VectorElementType>(
dir.path(),
DIM,
Distance::Dot,
MultiVectorConfig::default(),
AdviceSetting::Global,
false,
)
.unwrap();
for internal_id in 0..POINT_COUNT {
let size = rng.random_range(2..=4);
let vectors = std::iter::repeat_with(|| {
std::iter::repeat_with(|| rng.random_range(-1.0..1.0))
.take(DIM)
.collect()
})
.take(size)
.collect::<Vec<Vec<_>>>();
let multivec = MultiDenseVectorInternal::try_from(vectors).unwrap();
storage
.insert_vector(internal_id, VectorRef::from(&multivec), &hw_counter)
.unwrap();
expected.push(multivec);
}
storage.flusher()().unwrap();
}
// Rewind the recorded length to 1 row, leaving the row data and the offsets
// entries untouched: the exact skew an interrupted flush persists.
let status_file = ChunkedVectorsRead::<VectorElementType, MmapFile>::status_file(
&dir.path().join(VECTORS_DIR_PATH),
);
let mut status_bytes = fs::read(&status_file).unwrap();
let true_len = usize::from_ne_bytes(
status_bytes[..std::mem::size_of::<usize>()]
.try_into()
.unwrap(),
);
assert!(true_len > 1, "fixture must have appended multiple rows");
status_bytes[..std::mem::size_of::<usize>()].copy_from_slice(&1usize.to_ne_bytes());
fs::write(&status_file, &status_bytes).unwrap();
let storage = open_appendable_memmap_multi_vector_storage_impl::<VectorElementType>(
dir.path(),
DIM,
Distance::Dot,
MultiVectorConfig::default(),
AdviceSetting::Global,
false,
)
.unwrap();
assert_eq!(
storage.vectors.len(),
true_len,
"repair must restore the full recorded length",
);
for (internal_id, multivec) in expected.iter().enumerate() {
let read = storage
.get_multi_opt::<Random>(internal_id as PointOffsetType)
.unwrap_or_else(|| panic!("point {internal_id} unreadable after reload"));
assert_eq!(
read.as_vec_ref().flattened_vectors,
multivec.flattened_vectors.as_slice(),
"point {internal_id} content diverged after reload",
);
}
}
}