diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 32b6bcebc..10822cafa 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -637,6 +637,15 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { message: "set_lsm_write_spec is not supported on this table type".into(), }) } + /// Switch this table to required index catch-up, one way. + /// + /// The default implementation returns `NotSupported`. Implementations + /// that support the MemWAL LSM write path must override this. + async fn require_mem_wal_index_catchup(&self) -> Result<()> { + Err(Error::NotSupported { + message: "require_mem_wal_index_catchup is not supported on this table type".into(), + }) + } /// Remove the [`LsmWriteSpec`] from this table. /// /// This is a no-op if no spec is currently set. @@ -1693,6 +1702,20 @@ impl Table { self.inner.set_lsm_write_spec(spec).await } + /// Switch this table to required index catch-up, one way. + /// + /// Separate from [`Self::set_lsm_write_spec`] on purpose: a table carrying + /// the bit retains its SSTables until an index records that it holds the + /// compacted rows, so turn it on only once something can repair coverage. + /// A writer that already holds the dataset can call the equivalent on + /// `DatasetMemWalExt` instead; this is the table-level entry point. + /// + /// Errors if no spec is set, or if the table already records SSTable + /// compaction progress from before this protocol. + pub async fn require_mem_wal_index_catchup(&self) -> Result<()> { + self.inner.require_mem_wal_index_catchup().await + } + /// Remove the [`LsmWriteSpec`] from this table, reverting to the standard /// `merge_insert` write path. /// @@ -3226,6 +3249,10 @@ impl BaseTable for NativeTable { merge::lsm::set_lsm_write_spec(self, spec).await } + async fn require_mem_wal_index_catchup(&self) -> Result<()> { + merge::lsm::require_mem_wal_index_catchup(self).await + } + async fn unset_lsm_write_spec(&self) -> Result<()> { merge::lsm::unset_lsm_write_spec(self).await } diff --git a/rust/lancedb/src/table/merge/lsm.rs b/rust/lancedb/src/table/merge/lsm.rs index 87c427b3c..eb2feacbd 100644 --- a/rust/lancedb/src/table/merge/lsm.rs +++ b/rust/lancedb/src/table/merge/lsm.rs @@ -183,6 +183,36 @@ fn index_name_list(indices: &[IndexConfig]) -> String { format!("[{}]", names.join(", ")) } +// ============================================================================= +// require_mem_wal_index_catchup +// ============================================================================= + +/// Switch this table to required index catch-up, one way. +/// +/// Deliberately **not** part of installing the write spec. Until something can +/// actually repair coverage, a table carrying the bit reports every index as +/// not known to hold the compacted rows, so its SSTables are retained +/// indefinitely -- and the WAL pod trims on the legacy rule meanwhile, leaving +/// readers pointed at files that are gone. Turn this on only once remote +/// maintenance owns the merge and the repair for the table. +/// +/// Lance refuses the activation if the table already records SSTable +/// compaction progress: those numbers predate this protocol and cannot be +/// validated, so such a table must be drained rather than activated. +#[allow(clippy::redundant_pub_crate)] +pub(crate) async fn require_mem_wal_index_catchup(table: &NativeTable) -> Result<()> { + table.dataset.ensure_mutable()?; + let mut dataset = (*table.dataset.get().await?).clone(); + if dataset.mem_wal_index_details().await?.is_none() { + return Err(Error::InvalidInput { + message: "require_mem_wal_index_catchup: no LSM write spec is set on this table".into(), + }); + } + dataset.require_mem_wal_index_catchup().await?; + table.dataset.update(dataset); + Ok(()) +} + // ============================================================================= // unset_lsm_write_spec // ============================================================================= diff --git a/rust/lancedb/src/table/query/lsm.rs b/rust/lancedb/src/table/query/lsm.rs index 7ccdedf5a..6155ec095 100644 --- a/rust/lancedb/src/table/query/lsm.rs +++ b/rust/lancedb/src/table/query/lsm.rs @@ -36,6 +36,7 @@ use lance::dataset::mem_wal::{ DatasetMemWalExt, LsmScanner, ShardManifestStore, ShardSnapshot, ShardWriterConfig, }; use lance_index::mem_wal::{MemWalIndexDetails, ShardManifest}; +use lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP; use uuid::Uuid; use super::NativeTable; @@ -248,18 +249,26 @@ fn pk_columns(dataset: &Dataset) -> Result> { fn exclusion_watermarks( details: &MemWalIndexDetails, index_names: &[String], + catchup_required: bool, ) -> HashMap { let mut exclude: HashMap = HashMap::new(); for entry in &details.compacted_sstables { let mut watermark = entry.generation; for name in index_names { - if let Some(caught_up) = details + match details .index_catchup .iter() .find(|icp| icp.index_name == *name) .and_then(|icp| icp.caught_up_generation_for_shard(&entry.shard_id)) { - watermark = watermark.min(caught_up); + Some(caught_up) => watermark = watermark.min(caught_up), + // No entry. On a table that requires catch-up this means the + // index is *not* known to hold these rows, and the base arm is + // index-only -- so every generation stays readable from its + // SSTable. Without the bit the field is not maintained at all, + // and absence carries no information. + None if catchup_required => watermark = 0, + None => {} } } exclude.entry(entry.shard_id).or_insert(watermark); @@ -274,13 +283,26 @@ fn exclusion_watermarks( /// with a live cached `ShardWriter` (this session's in-flight writes) the /// writer's authoritative in-memory manifest and memtables override the /// on-disk view so a read sees data not yet flushed. +/// Whether this table reads a missing `index_catchup` entry as "not caught up". +/// +/// Both words must be set. A reader honouring the bit while a writer does not +/// would retain SSTables the writer had already trimmed, and the reverse would +/// serve rows from files the writer still expects to be excluded -- so a +/// half-set manifest is treated as legacy, which is the conservative side. +fn requires_index_catchup(dataset: &Dataset) -> bool { + let manifest = dataset.manifest(); + manifest.reader_feature_flags & FLAG_MEM_WAL_INDEX_CATCHUP != 0 + && manifest.writer_feature_flags & FLAG_MEM_WAL_INDEX_CATCHUP != 0 +} + async fn build_read_context( table: &NativeTable, dataset: &Dataset, details: &MemWalIndexDetails, index_names: &[String], ) -> Result<(Vec, HashMap)> { - let exclude = exclusion_watermarks(details, index_names); + let catchup_required = requires_index_catchup(dataset); + let exclude = exclusion_watermarks(details, index_names, catchup_required); let shard_ids = dataset.list_mem_wal_latest_shard_ids().await?; // Use the dataset's own object store (not `ObjectStore::from_uri`, which @@ -767,22 +789,50 @@ mod tests { }; // Plain scan: drop every compacted generation (through 5). - assert_eq!(exclusion_watermarks(&details, &[]).get(&shard), Some(&5)); + assert_eq!( + exclusion_watermarks(&details, &[], false).get(&shard), + Some(&5) + ); // FTS arm with a lagging index: exclusion is capped at the index catch-up // (2), so SSTable generations 3..=5 are retained until the index covers // them — otherwise those documents would silently vanish from FTS results. assert_eq!( - exclusion_watermarks(&details, &["fts_idx".to_string()]).get(&shard), + exclusion_watermarks(&details, &["fts_idx".to_string()], false).get(&shard), Some(&2) ); // A caught-up index — or one untracked in index_catchup — falls back to the // compaction watermark. assert_eq!( - exclusion_watermarks(&details, &["caught_up_idx".to_string()]).get(&shard), + exclusion_watermarks(&details, &["caught_up_idx".to_string()], false).get(&shard), Some(&5) ); + + // The same missing entry, once the table requires catch-up: absence now + // means "not known to hold these rows", so nothing may be excluded and + // every generation stays readable from its SSTable. This is the whole + // point of the protocol -- an indexed query against a table whose index + // has not caught up must not silently lose rows. + assert_eq!( + exclusion_watermarks(&details, &["untracked_idx".to_string()], true).get(&shard), + Some(&0) + ); + + // A tracked index is unaffected by the mode: the recorded position is + // information either way, and it still caps the exclusion. + assert_eq!( + exclusion_watermarks(&details, &["fts_idx".to_string()], true).get(&shard), + Some(&2) + ); + + // One missing entry is enough to hold everything back, even alongside an + // index that has caught up. + let mixed = vec!["fts_idx".to_string(), "untracked_idx".to_string()]; + assert_eq!( + exclusion_watermarks(&details, &mixed, true).get(&shard), + Some(&0) + ); } /// A hybrid search reads a vector and a full-text index, and either may lag. @@ -809,20 +859,23 @@ mod tests { // Each index alone stops at its own catch-up. assert_eq!( - exclusion_watermarks(&details, &["vec_idx".to_string()]).get(&shard), + exclusion_watermarks(&details, &["vec_idx".to_string()], false).get(&shard), Some(&7) ); assert_eq!( - exclusion_watermarks(&details, &["fts_idx".to_string()]).get(&shard), + exclusion_watermarks(&details, &["fts_idx".to_string()], false).get(&shard), Some(&4) ); // Used together, the lower one governs regardless of order. let both = ["vec_idx".to_string(), "fts_idx".to_string()]; - assert_eq!(exclusion_watermarks(&details, &both).get(&shard), Some(&4)); + assert_eq!( + exclusion_watermarks(&details, &both, false).get(&shard), + Some(&4) + ); let reversed = ["fts_idx".to_string(), "vec_idx".to_string()]; assert_eq!( - exclusion_watermarks(&details, &reversed).get(&shard), + exclusion_watermarks(&details, &reversed, false).get(&shard), Some(&4) ); } @@ -843,7 +896,10 @@ mod tests { }; let both = ["fts_idx".to_string(), "untracked_idx".to_string()]; - assert_eq!(exclusion_watermarks(&details, &both).get(&shard), Some(&4)); + assert_eq!( + exclusion_watermarks(&details, &both, false).get(&shard), + Some(&4) + ); } #[test]