diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 9228b4baf..e1dd942db 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -643,15 +643,6 @@ 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. @@ -1793,20 +1784,6 @@ 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. /// @@ -3354,10 +3331,6 @@ 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/computed_columns.rs b/rust/lancedb/src/table/computed_columns.rs index 841b13856..24cc48658 100644 --- a/rust/lancedb/src/table/computed_columns.rs +++ b/rust/lancedb/src/table/computed_columns.rs @@ -2151,7 +2151,6 @@ mod tests { .set_lsm_write_spec(LsmWriteSpec::unsharded()) .await .unwrap(); - table.require_mem_wal_index_catchup().await.unwrap(); let mut merge = table.merge_insert(&["id"]); merge diff --git a/rust/lancedb/src/table/merge/lsm.rs b/rust/lancedb/src/table/merge/lsm.rs index 5751cd916..44a2874de 100644 --- a/rust/lancedb/src/table/merge/lsm.rs +++ b/rust/lancedb/src/table/merge/lsm.rs @@ -192,36 +192,6 @@ 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 6155ec095..255d649b1 100644 --- a/rust/lancedb/src/table/query/lsm.rs +++ b/rust/lancedb/src/table/query/lsm.rs @@ -36,7 +36,6 @@ 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; @@ -249,7 +248,6 @@ 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 { @@ -262,13 +260,10 @@ fn exclusion_watermarks( .and_then(|icp| icp.caught_up_generation_for_shard(&entry.shard_id)) { 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 => {} + // No entry 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. + None => watermark = 0, } } exclude.entry(entry.shard_id).or_insert(watermark); @@ -283,26 +278,13 @@ 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 catchup_required = requires_index_catchup(dataset); - let exclude = exclusion_watermarks(details, index_names, catchup_required); + let exclude = exclusion_watermarks(details, index_names); let shard_ids = dataset.list_mem_wal_latest_shard_ids().await?; // Use the dataset's own object store (not `ObjectStore::from_uri`, which @@ -789,50 +771,29 @@ mod tests { }; // Plain scan: drop every compacted generation (through 5). - assert_eq!( - exclusion_watermarks(&details, &[], false).get(&shard), - Some(&5) - ); + assert_eq!(exclusion_watermarks(&details, &[]).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()], false).get(&shard), + exclusion_watermarks(&details, &["fts_idx".to_string()]).get(&shard), Some(&2) ); - // A caught-up index — or one untracked in index_catchup — falls back to the - // compaction watermark. + // An index absent from index_catchup is *not* known to hold these rows, + // so nothing may be excluded and every generation stays readable from + // its SSTable. An indexed query against an index that has not caught up + // must not silently lose rows. assert_eq!( - 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), + exclusion_watermarks(&details, &["untracked_idx".to_string()]).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) - ); + assert_eq!(exclusion_watermarks(&details, &mixed).get(&shard), Some(&0)); } /// A hybrid search reads a vector and a full-text index, and either may lag. @@ -859,31 +820,29 @@ mod tests { // Each index alone stops at its own catch-up. assert_eq!( - exclusion_watermarks(&details, &["vec_idx".to_string()], false).get(&shard), + exclusion_watermarks(&details, &["vec_idx".to_string()]).get(&shard), Some(&7) ); assert_eq!( - exclusion_watermarks(&details, &["fts_idx".to_string()], false).get(&shard), + exclusion_watermarks(&details, &["fts_idx".to_string()]).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, false).get(&shard), - Some(&4) - ); + assert_eq!(exclusion_watermarks(&details, &both).get(&shard), Some(&4)); let reversed = ["fts_idx".to_string(), "vec_idx".to_string()]; assert_eq!( - exclusion_watermarks(&details, &reversed, false).get(&shard), + exclusion_watermarks(&details, &reversed).get(&shard), Some(&4) ); } - /// An index with no catch-up entry contributes no cap today, so a lagging - /// sibling must still govern rather than being widened by the untracked one. + /// An index with no catch-up entry is not known to hold these rows, so it + /// retains everything -- it is never widened by a tracked sibling that has + /// caught up further. #[test] - fn an_untracked_index_does_not_widen_a_lagging_sibling() { + fn an_untracked_index_retains_everything() { let shard = Uuid::from_u128(1); let details = MemWalIndexDetails { compacted_sstables: vec![CompactedSsTable::new(shard, 9)], @@ -896,8 +855,11 @@ mod tests { }; let both = ["fts_idx".to_string(), "untracked_idx".to_string()]; + assert_eq!(exclusion_watermarks(&details, &both).get(&shard), Some(&0)); + + // The tracked sibling still caps on its own. assert_eq!( - exclusion_watermarks(&details, &both, false).get(&shard), + exclusion_watermarks(&details, &["fts_idx".to_string()]).get(&shard), Some(&4) ); } diff --git a/rust/lancedb/src/table/refresh.rs b/rust/lancedb/src/table/refresh.rs index e94f20f2b..a3f3da92f 100644 --- a/rust/lancedb/src/table/refresh.rs +++ b/rust/lancedb/src/table/refresh.rs @@ -141,11 +141,19 @@ pub(crate) async fn execute_refresh_column_async(table: &NativeTable, column: &s /// un-compacted MemWAL tiers it cannot reach -- success would silently omit /// readable rows. async fn ensure_no_lsm_write_spec(table: &NativeTable) -> Result<()> { - // The catch-up flag outlives unset and marks retained SSTable rows. - let catchup = table.dataset.get().await?.manifest().reader_feature_flags - & lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP - != 0; - if catchup || table.get_lsm_write_spec().await?.is_some() { + use lance::dataset::mem_wal::DatasetMemWalExt; + + // Unset drops the MemWAL index, so the spec alone stops describing a table + // whose SSTables still hold rows. The shard directories outlive it and are + // the durable evidence. + let retained_sstables = !table + .dataset + .get() + .await? + .list_mem_wal_latest_shard_ids() + .await? + .is_empty(); + if retained_sstables || table.get_lsm_write_spec().await?.is_some() { return Err(Error::NotSupported { message: "refresh_column is not supported on a table with an LSM write \ spec: rows in un-compacted tiers are invisible to refresh" @@ -921,7 +929,6 @@ mod tests { .set_lsm_write_spec(LsmWriteSpec::unsharded()) .await .unwrap(); - table.require_mem_wal_index_catchup().await.unwrap(); let mut merge = table.merge_insert(&["x"]); merge .when_matched_update_all(None) diff --git a/rust/lancedb/src/table/schema_evolution.rs b/rust/lancedb/src/table/schema_evolution.rs index 4e41f0e85..d10a45eea 100644 --- a/rust/lancedb/src/table/schema_evolution.rs +++ b/rust/lancedb/src/table/schema_evolution.rs @@ -124,18 +124,26 @@ pub(crate) async fn execute_declare( table: &NativeTable, columns: &[(String, String)], ) -> Result { + use lance::dataset::mem_wal::DatasetMemWalExt; + // An LSM write spec keeps visible rows in tiers refresh cannot reach; // checked against latest committed state, not this handle's snapshot. - // The catch-up flag outlives unset and marks retained SSTable rows. table.checkout_latest().await?; computed_columns::ensure_no_function_bindings_for_mutation( table.schema().await?.as_ref(), "schema evolution", )?; - let catchup = table.dataset.get().await?.manifest().reader_feature_flags - & lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP - != 0; - if catchup || table.get_lsm_write_spec().await?.is_some() { + // Unset drops the MemWAL index, so the spec alone stops describing a table + // whose SSTables still hold rows. The shard directories outlive it and are + // the durable evidence. + let retained_sstables = !table + .dataset + .get() + .await? + .list_mem_wal_latest_shard_ids() + .await? + .is_empty(); + if retained_sstables || table.get_lsm_write_spec().await?.is_some() { return Err(Error::NotSupported { message: "computed columns are not supported on a table with an LSM write \ spec: rows in un-compacted tiers are invisible to refresh"