feat(lsm): require recorded index catch-up, as an explicit activation (#3911)

> Stacked on #3780. Blocked only on #3922 (`lance` → `v11.0.0-beta.6`),
so CI
> stays red until that lands.

## Missing coverage must mean "not known to be covered"

#3780 caps the SSTable exclusion watermark at an index's recorded
catch-up when
there is one, and silently ignores the case where there is none. On a
table that
requires catch-up, an absent entry means the index is *not* known to
hold the
compacted rows — and the LSM base arm reads base through the index
(`fast_search`, no brute-force tail), so dropping that SSTable loses
those rows
for that query.

```rust
Some(caught_up) => watermark = watermark.min(caught_up),
None if catchup_required => watermark = 0,   // retain everything
None => {}
```

`catchup_required` reads the manifest feature bit directly, and requires
both
words: a half-set manifest is treated as legacy, which is the
conservative side.
Without the bit the field is not maintained at all, so absence carries
no
information and behaviour is unchanged.

## Activation, as a table-level entry point

`Table::require_mem_wal_index_catchup()` performs the one-way switch,
separate
from `set_lsm_write_spec`: a table carrying the bit retains every
generation
until something records catch-up, so it has to follow the deployment of
whatever
repairs coverage, not the creation of the table.

This is a convenience, not the only path — a writer holding the dataset
calls
the equivalent on `DatasetMemWalExt`, which is what the WAL pod does.
Lance
enforces the preconditions either way: the MemWAL index must exist, and
the
table must not already carry `compacted_sstables` from before this
protocol,
since those numbers cannot be validated.

## Still correct after the Lance rework

lance-format/lance#8481 replaced the transmitted `IndexCatchupAdvance`
with a
position derived at commit time from the version a transaction read.
That
changed how a writer earns coverage; it did not change what a reader may
conclude from its absence. The rule here, and the field it reads, are
unchanged.

## Tests

Existing `exclusion_watermarks` unit tests carry the new argument.
Coverage
against a real dataset follows once #3922 lands and this can build.

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
     }

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
 // =============================================================================

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<Vec<String>> {
 fn exclusion_watermarks(
     details: &MemWalIndexDetails,
     index_names: &[String],
+    catchup_required: bool,
 ) -> HashMap<Uuid, u64> {
     let mut exclude: HashMap<Uuid, u64> = 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<ShardSnapshot>, HashMap<Uuid, InMemoryMemTables>)> {
-    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]