diff --git a/lib/collection/src/model_testing/mod.rs b/lib/collection/src/model_testing/mod.rs index 1008461ac3..68c851165e 100644 --- a/lib/collection/src/model_testing/mod.rs +++ b/lib/collection/src/model_testing/mod.rs @@ -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"; diff --git a/lib/collection/src/model_testing/op/mod.rs b/lib/collection/src/model_testing/op/mod.rs index 9b6746ef21..6508e48369 100644 --- a/lib/collection/src/model_testing/op/mod.rs +++ b/lib/collection/src/model_testing/op/mod.rs @@ -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); diff --git a/lib/segment/src/vector_storage/chunked_vectors/write.rs b/lib/segment/src/vector_storage/chunked_vectors/write.rs index ae85e2c342..e5acb118a0 100644 --- a/lib/segment/src/vector_storage/chunked_vectors/write.rs +++ b/lib/segment/src/vector_storage/chunked_vectors/write.rs @@ -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(); diff --git a/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs b/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs index 22e586c8d7..2b37d2f7ee 100644 --- a/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs +++ b/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs @@ -573,8 +573,9 @@ pub fn open_appendable_memmap_multi_vector_storage_impl( + vectors: &mut ChunkedVectors, + offsets: &ChunkedVectors, +) { + let mut max_row_end = 0; + for key in 0..offsets.len() { + let Some(entry) = offsets.get::(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> { @@ -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::( + 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::>>(); + 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::::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::()] + .try_into() + .unwrap(), + ); + assert!(true_len > 1, "fixture must have appended multiple rows"); + status_bytes[..std::mem::size_of::()].copy_from_slice(&1usize.to_ne_bytes()); + fs::write(&status_file, &status_bytes).unwrap(); + + let storage = open_appendable_memmap_multi_vector_storage_impl::( + 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::(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", + ); + } + } }