diff --git a/python/python/tests/test_lsm_write_spec.py b/python/python/tests/test_lsm_write_spec.py index d43cb7532..ac767ea31 100644 --- a/python/python/tests/test_lsm_write_spec.py +++ b/python/python/tests/test_lsm_write_spec.py @@ -52,9 +52,9 @@ def test_set_lsm_write_spec_validates(tmp_path): with pytest.raises(Exception, match="num_buckets"): table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 1025)) - # Happy path then mutation rejected. + # Happy path, then a set over the installed spec is refused. table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 4)) - with pytest.raises(Exception, match="mutation"): + with pytest.raises(Exception, match="already set"): table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 8)) @@ -176,18 +176,18 @@ def test_get_lsm_write_spec(tmp_path): table.unset_lsm_write_spec() assert table.get_lsm_write_spec() is None - # Identity round-trips (column recovered from the schema). Leaving the - # maintained set to be inferred picks up the index on the table, so the - # spec reads back naming it rather than as "infer". + # Identity round-trips (column recovered from the schema). An unnamed + # maintained set reads back as None: what is stored is the intent to + # maintain every index, not the set it resolves to now. table.set_lsm_write_spec(LsmWriteSpec.identity("id")) spec = table.get_lsm_write_spec() assert spec.spec_type == "identity" assert spec.column == "id" - assert spec.maintained_indexes == [idx_name] + assert spec.maintained_indexes is None table.unset_lsm_write_spec() # Unsharded round-trips (no routing column). Opting out is distinct from - # the inferred default. + # the unnamed default. table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([])) spec = table.get_lsm_write_spec() assert spec.spec_type == "unsharded" diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 251baad99..a99f086ef 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -330,7 +330,8 @@ pub use self::merge::MergeResult; /// date) and [`LsmWriteSpec::with_writer_config_defaults`] (default /// `ShardWriter` configuration recorded in the MemWAL index). /// -/// A fresh spec maintains every index on the table, resolved on install. +/// A fresh spec maintains every index on the table, including ones created +/// after it is installed. /// /// Install a spec with [`Table::set_lsm_write_spec`] and remove it with /// [`Table::unset_lsm_write_spec`]. The actual `merge_insert` dispatch @@ -348,9 +349,8 @@ pub enum LsmWriteSpec { num_buckets: u32, /// Indexes the MemWAL maintains in-memory as rows are appended. /// - /// `None` means every index it can maintain, resolved on install — a - /// snapshot, so indexes created later need the spec unset and re-set. - /// `Some([])` maintains nothing. + /// `None` means every index the table has, including ones created + /// later. `Some([])` maintains nothing. maintained_indexes: Option>, /// Default `ShardWriter` configuration recorded in the MemWAL index. writer_config_defaults: HashMap, @@ -363,9 +363,8 @@ pub enum LsmWriteSpec { column: String, /// Indexes the MemWAL maintains in-memory as rows are appended. /// - /// `None` means every index it can maintain, resolved on install — a - /// snapshot, so indexes created later need the spec unset and re-set. - /// `Some([])` maintains nothing. + /// `None` means every index the table has, including ones created + /// later. `Some([])` maintains nothing. maintained_indexes: Option>, /// Default `ShardWriter` configuration recorded in the MemWAL index. writer_config_defaults: HashMap, @@ -374,9 +373,8 @@ pub enum LsmWriteSpec { Unsharded { /// Indexes the MemWAL maintains in-memory as rows are appended. /// - /// `None` means every index it can maintain, resolved on install — a - /// snapshot, so indexes created later need the spec unset and re-set. - /// `Some([])` maintains nothing. + /// `None` means every index the table has, including ones created + /// later. `Some([])` maintains nothing. maintained_indexes: Option>, /// Default `ShardWriter` configuration recorded in the MemWAL index. writer_config_defaults: HashMap, @@ -422,14 +420,15 @@ impl LsmWriteSpec { /// Set which indexes the MemWAL maintains. /// - /// `None` (the default) resolves to every index on the table at install, - /// failing if one cannot be maintained — name the set to install anyway. A - /// list is verbatim: each name must already exist and be maintainable, and - /// an empty list maintains nothing. + /// `None` (the default) is every index on the table, re-read as the table + /// changes, so an index created later is maintained too; one of a kind the + /// MemWAL cannot mirror is skipped rather than failing the table. A list is + /// verbatim: each name must already exist and be maintainable, and an empty + /// list maintains nothing. /// /// ``` /// # use lancedb::table::LsmWriteSpec; - /// // Every index the table has when the spec is installed: + /// // Every index the table has, now and later: /// LsmWriteSpec::unsharded().with_maintained_indexes(None); /// // Exactly these: /// LsmWriteSpec::unsharded().with_maintained_indexes(vec!["id_idx".to_string()]); @@ -1988,6 +1987,10 @@ impl Table { /// - [`LsmWriteSpec::identity`] — shard by the raw value of a scalar column. /// - [`LsmWriteSpec::unsharded`] — route every write to a single shard. /// + /// A table carries one spec: this fails while one is installed, since the + /// generations already written were homed under it. + /// [`Table::unset_lsm_write_spec`] removes one. + /// /// # Example /// /// ``` @@ -5664,29 +5667,22 @@ mod tests { assert_eq!(table.get_lsm_write_spec().await.unwrap(), None); // Identity sharding round-trips (column recovered from the schema). - // A spec left at its default maintains every index on the table, so it - // reads back naming the one on the table rather than as "infer". + // A spec left at its default round-trips as the default: what is stored + // is the intent to maintain everything, not the set it resolves to now. let spec = LsmWriteSpec::identity("region"); table.set_lsm_write_spec(spec.clone()).await.unwrap(); - assert_eq!( - table.get_lsm_write_spec().await.unwrap(), - Some(spec.with_maintained_indexes(vec![idx_name.clone()])) - ); + assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec)); table.unset_lsm_write_spec().await.unwrap(); // Unsharded round-trips (no routing column). let spec = LsmWriteSpec::unsharded(); table.set_lsm_write_spec(spec.clone()).await.unwrap(); - assert_eq!( - table.get_lsm_write_spec().await.unwrap(), - Some(spec.with_maintained_indexes(vec![idx_name])) - ); + assert_eq!(table.get_lsm_write_spec().await.unwrap(), Some(spec)); } - /// The maintained set defaults to every index on the table, resolved at - /// install. An index the memtable cannot build fails the install rather - /// than being dropped: maintaining it would take the table offline for - /// writes, dropping it would hide that from the caller. + /// The maintained set defaults to every index on the table. A named index + /// the memtable cannot build fails the install, because the caller asked + /// for it; an unnamed one is skipped, because it arrived by existing. #[tokio::test] async fn test_set_lsm_write_spec_infers_maintained_indexes() { let tmp_dir = tempdir().unwrap(); @@ -5739,19 +5735,27 @@ mod tests { ); assert_eq!(table.get_lsm_write_spec().await.unwrap(), None); - // The default covers every index, so the bitmap fails it too. - let err = table + // The default names nothing, so the bitmap is skipped rather than + // refusing the table: it is maintained by existing, not by being asked + // for. The intent is what is stored, not the set it resolves to today. + table .set_lsm_write_spec(LsmWriteSpec::unsharded()) .await - .unwrap_err(); - assert!( - matches!(err, Error::InvalidInput { ref message } - if message.contains("tag_bitmap") && message.contains("maintained_indexes")), - "expected the inferred set to be rejected, got {err:?}" + .unwrap(); + assert_eq!( + table + .get_lsm_write_spec() + .await + .unwrap() + .unwrap() + .maintained_indexes(), + None, + "an unnamed set is stored as the intent to maintain everything" ); - assert_eq!(table.get_lsm_write_spec().await.unwrap(), None); - // Naming the maintainable subset installs. + // Narrowing to a named subset: unset first, since an installed spec + // cannot be set over. + table.unset_lsm_write_spec().await.unwrap(); table .set_lsm_write_spec( LsmWriteSpec::unsharded().with_maintained_indexes(vec!["id_btree".to_string()]), @@ -5783,6 +5787,28 @@ mod tests { .maintained_indexes(), Some([].as_slice()) ); + + // An installed spec is never set over, whatever the new one asks for, + // and the refused call leaves it alone. + let err = table + .set_lsm_write_spec(LsmWriteSpec::bucket("id", 4)) + .await + .unwrap_err(); + assert!( + matches!(err, Error::InvalidInput { ref message } + if message.contains("already set")), + "expected the repeat to be refused, got {err:?}" + ); + assert_eq!( + table + .get_lsm_write_spec() + .await + .unwrap() + .unwrap() + .maintained_indexes(), + Some([].as_slice()), + "a refused call changes nothing" + ); } #[tokio::test] diff --git a/rust/lancedb/src/table/create_index.rs b/rust/lancedb/src/table/create_index.rs index c5d6e072f..916b28ed1 100644 --- a/rust/lancedb/src/table/create_index.rs +++ b/rust/lancedb/src/table/create_index.rs @@ -160,6 +160,12 @@ impl NativeTable { builder = builder.name(name); } builder.await?; + // A writer already open was built before this index existed, so it has + // to pick it up here or a read needing it is refused until it reopens. + self.dataset + .shard_writer() + .refresh_maintained_indexes(&dataset) + .await; self.dataset.update(dataset); Ok(()) } diff --git a/rust/lancedb/src/table/merge.rs b/rust/lancedb/src/table/merge.rs index 982818ff6..354eef5a5 100644 --- a/rust/lancedb/src/table/merge.rs +++ b/rust/lancedb/src/table/merge.rs @@ -1505,6 +1505,30 @@ mod lsm_tests { ); } + /// A table with `id`/`text` and a primary key, ready for an LSM write spec. + async fn lsm_text_table(dir: &tempfile::TempDir) -> crate::Table { + let conn = connect(dir.path().to_str().unwrap()) + .execute() + .await + .unwrap(); + let table = conn + .create_table("t", id_text_reader(vec![(1, "alpha")])) + .execute() + .await + .unwrap(); + table.set_unenforced_primary_key(["id"]).await.unwrap(); + table + } + + /// Upsert `rows` through the LSM write path. + async fn upsert_text(table: &crate::Table, rows: Vec<(i64, &str)>) { + let mut builder = table.merge_insert(&[]); + builder + .when_matched_update_all(None) + .when_not_matched_insert_all(); + builder.execute(id_text_reader(rows)).await.unwrap(); + } + /// A reader of `[id: Int64, text: Utf8]` rows. fn id_text_reader(rows: Vec<(i64, &str)>) -> Box { let schema = Arc::new(Schema::new(vec![ @@ -1524,6 +1548,165 @@ mod lsm_tests { Box::new(RecordBatchIterator::new(vec![Ok(batch)], schema)) } + /// Under the default set, an index created while a writer is open reaches + /// that writer. + /// + /// The default maintains every index the table has, so creating one changes + /// what the table maintains without anyone touching the spec. A writer left + /// on its old configs would keep building MemTables the new index does not + /// cover. + #[tokio::test] + async fn lsm_default_set_picks_up_an_index_created_while_writing() { + use crate::index::Index; + use lance_index::scalar::FullTextSearchQuery; + + let dir = tempdir().unwrap(); + let table = lsm_text_table(&dir).await; + // The default set, on a table with no FTS index yet. + table + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap(); + + // Opens the writer, whose MemTable carries no FTS index. + upsert_text(&table, vec![(99, "zebra")]).await; + + // Created with the writer still open and never reopened. + table + .create_index(&["text"], Index::FTS(Default::default())) + .execute() + .await + .unwrap(); + + upsert_text(&table, vec![(100, "zebra stripes")]).await; + + let query = FullTextSearchQuery::new("stripes".to_string()) + .with_column("text".to_string()) + .unwrap(); + let batches = table + .query() + .full_text_search(query) + .execute() + .await + .expect("the created index reached the open writer") + .try_collect::>() + .await + .unwrap(); + let found: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(found, 1, "the row written after create_index must be found"); + } + + /// A spec is replaced by unsetting it and setting the new one. + /// + /// `unset_lsm_write_spec` drains the open writer, so the rows it held + /// survive the replacement, and the writer the next write opens builds its + /// MemTables from the new set. + #[tokio::test] + async fn lsm_spec_is_replaced_by_unsetting_it_first() { + use crate::index::Index; + use lance_index::scalar::FullTextSearchQuery; + + let dir = tempdir().unwrap(); + let table = lsm_text_table(&dir).await; + table + .create_index(&["text"], Index::FTS(Default::default())) + .execute() + .await + .unwrap(); + let fts_index = table.list_indices().await.unwrap()[0].name.clone(); + + // Installed maintaining nothing, then written to: the writer opens with + // no FTS index. + table + .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(Vec::new())) + .await + .unwrap(); + upsert_text(&table, vec![(99, "zebra")]).await; + + // Setting over an installed spec is refused; unset, then set. + let err = table + .set_lsm_write_spec( + LsmWriteSpec::unsharded().with_maintained_indexes(vec![fts_index.clone()]), + ) + .await + .expect_err("an installed spec cannot be set over"); + assert!( + err.to_string().contains("already set"), + "unexpected error: {err}" + ); + table.unset_lsm_write_spec().await.unwrap(); + table + .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(vec![fts_index])) + .await + .unwrap(); + + // A row written after the replacement is answered by the index the new + // writer maintains. + upsert_text(&table, vec![(100, "zebra stripes")]).await; + + let query = FullTextSearchQuery::new("stripes".to_string()) + .with_column("text".to_string()) + .unwrap(); + let batches = table + .query() + .full_text_search(query) + .execute() + .await + .expect("the replaced spec maintains the index") + .try_collect::>() + .await + .unwrap(); + let found: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!( + found, 1, + "the row written after the replacement must be found" + ); + } + + /// A row written before the index exists is searchable once it does. + /// + /// The MemTable holding it was built without the index. Creating the index + /// replaces the writer's configs, which seals that MemTable and waits for + /// its flush, so no resident MemTable is left that cannot answer the read. + #[tokio::test] + async fn lsm_a_row_written_before_the_index_is_searchable_after_it() { + use crate::index::Index; + use lance_index::scalar::FullTextSearchQuery; + + let dir = tempdir().unwrap(); + let table = lsm_text_table(&dir).await; + // No index yet, and the default set: maintain whatever the table has. + table + .set_lsm_write_spec(LsmWriteSpec::unsharded()) + .await + .unwrap(); + + // Opens the writer, whose MemTable therefore carries no FTS index. + upsert_text(&table, vec![(99, "zebra")]).await; + + // Built afterwards: the spec now resolves to it, the MemTable does not. + table + .create_index(&["text"], Index::FTS(Default::default())) + .execute() + .await + .unwrap(); + + let query = FullTextSearchQuery::new("zebra".to_string()) + .with_column("text".to_string()) + .unwrap(); + let batches = table + .query() + .full_text_search(query) + .execute() + .await + .expect("no resident MemTable is left that cannot answer") + .try_collect::>() + .await + .unwrap(); + let found: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(found, 1, "the row written before the index must be found"); + } + #[tokio::test] async fn lsm_read_full_text_search() { use crate::index::Index; diff --git a/rust/lancedb/src/table/merge/lsm.rs b/rust/lancedb/src/table/merge/lsm.rs index fc925c014..0f86fdcac 100644 --- a/rust/lancedb/src/table/merge/lsm.rs +++ b/rust/lancedb/src/table/merge/lsm.rs @@ -77,21 +77,36 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec) let dataset = table.dataset.get().await?; if dataset.mem_wal_index_details().await?.is_some() { return Err(Error::InvalidInput { - message: "set_lsm_write_spec: an LSM write spec is already set on this table; mutation is not supported".into(), + message: "set_lsm_write_spec: an LSM write spec is already set on this \ + table and cannot be changed" + .into(), }); } } - // Before the builder borrows the dataset clone. `list_indices` merges an - // index's segments into one entry, so the result needs no dedup. - let maintained_indexes = { - let dataset = table.dataset.get().await?; - resolve_maintained_indexes( - &dataset, - &table.list_indices().await?, - spec.maintained_indexes(), - ) - .await? + // A named set is checked against the table here, where the caller can still + // be told which names exist. An unnamed set is not resolved at all: it is an + // intent lance re-reads whenever it builds a MemTable, so an index created + // later is maintained without another call. + let maintained_indexes = match spec.maintained_indexes() { + Some(requested) => { + let indices = table.list_indices().await?; + for name in requested { + if !indices.iter().any(|index| &index.name == name) { + return Err(Error::InvalidInput { + message: format!( + "maintained index '{}' does not exist on this table; it has {}", + name, + index_name_list(&indices), + ), + }); + } + } + let dataset = table.dataset.get().await?; + validate_maintained_indexes(&dataset, requested).await?; + Some(requested.to_vec()) + } + None => None, }; table.checkout_latest().await?; @@ -138,7 +153,9 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec) writer_config_defaults } }; - builder = builder.maintained_indexes(maintained_indexes); + if let Some(maintained_indexes) = maintained_indexes { + builder = builder.maintained_indexes(maintained_indexes); + } for (key, value) in writer_config_defaults { builder = builder.add_writer_config_default(key, value); } @@ -147,48 +164,6 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec) Ok(()) } -/// Resolve a spec's maintained-index selection against `indices`, as reported -/// by [`Table::list_indices`](crate::Table::list_indices). -/// -/// `None` means every index on the table, snapshotted now. Lance validates -/// either selection against its shard-writer rules, so a spec that installs is -/// one the MemWAL can open. -/// -/// An unmaintainable index fails an inferred set rather than being dropped from -/// it — dropping would leave the caller believing it is maintained. -async fn resolve_maintained_indexes( - dataset: &Dataset, - indices: &[IndexConfig], - requested: Option<&[String]>, -) -> Result> { - let Some(requested) = requested else { - let all: Vec = indices.iter().map(|index| index.name.clone()).collect(); - validate_maintained_indexes(dataset, &all) - .await - .map_err(|source| Error::InvalidInput { - message: format!( - "cannot maintain every index on this table: {source}. Set \ - maintained_indexes explicitly to choose from {}", - index_name_list(indices), - ), - })?; - return Ok(all); - }; - for name in requested { - if !indices.iter().any(|index| &index.name == name) { - return Err(Error::InvalidInput { - message: format!( - "maintained index '{}' does not exist on this table; it has {}", - name, - index_name_list(indices), - ), - }); - } - } - validate_maintained_indexes(dataset, requested).await?; - Ok(requested.to_vec()) -} - /// Index names for an error message. fn index_name_list(indices: &[IndexConfig]) -> String { if indices.is_empty() { @@ -298,7 +273,9 @@ fn lsm_write_spec_from_details( }; Ok(base - .with_maintained_indexes(details.maintained_indexes.clone()) + .with_maintained_indexes( + (!details.maintain_all_indexes).then(|| details.maintained_indexes.clone()), + ) .with_writer_config_defaults(details.writer_config_defaults.clone())) } @@ -460,6 +437,55 @@ impl ShardWriterCache { Ok(entry) } + /// Install `dataset`'s maintained index set on the cached writer, if one is + /// open, so a change reaches a table that is already being written to. + /// + /// Reported rather than propagated: the index is committed durably by the + /// time this runs, and the writer this process happens to hold may be fenced + /// or poisoned -- which is not a reason to fail creating an index. Any + /// writer opened afterwards reads the committed set. + /// + /// Holds the entry's read lock across the refresh, so a concurrent close + /// waits rather than retiring the writer mid-seal. Writes take the same + /// lock first, so the order matches theirs. + pub(crate) async fn refresh_maintained_indexes(&self, dataset: &Dataset) { + let entry = { + let guard = self.slot.read().await; + guard.as_ref().map(|(_, entry)| entry.clone()) + }; + let Some(entry) = entry else { + return; + }; + let guard = entry.inner.read().await; + let Some(writer) = guard.as_ref() else { + return; + }; + // Replacing the configs seals the active memtable so the next one carries + // the new indexes. The outgoing one keeps the set it was built with, so + // its flush has to land before a read: while it is still frozen, an + // indexed read sees a resident memtable that cannot answer and is + // refused. + let sealed = match dataset.refresh_mem_wal_index_configs(writer).await { + Ok(sealed) => sealed, + Err(error) => { + log::warn!( + "the maintained index set changed but the open shard writer could \ + not be refreshed, so it keeps the set it opened with: {error}" + ); + return; + } + }; + if let Some(fence) = sealed + && let Err(error) = fence.wait().await + { + log::warn!( + "the shard writer took the new maintained index set, but the memtable \ + it sealed was not flushed, so an indexed read stays refused until \ + it is: {error}" + ); + } + } + /// Snapshot the cached writer's shard for the LSM read path: its shard id, /// authoritative in-memory manifest, and active + frozen memtable refs. /// Returns `None` when no writer is currently cached (e.g. nothing has been diff --git a/rust/lancedb/src/table/query/lsm.rs b/rust/lancedb/src/table/query/lsm.rs index 86c1fe5f2..f3e40fc7b 100644 --- a/rust/lancedb/src/table/query/lsm.rs +++ b/rust/lancedb/src/table/query/lsm.rs @@ -33,7 +33,7 @@ use datafusion_physical_plan::expressions::Column; use datafusion_physical_plan::projection::ProjectionExec; use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr}; use lance::Dataset; -use lance::dataset::mem_wal::scanner::InMemoryMemTables; +use lance::dataset::mem_wal::scanner::{InMemoryMemTableRef, InMemoryMemTables}; use lance::dataset::mem_wal::{ DatasetMemWalExt, LsmScanner, ShardManifestStore, ShardSnapshot, ShardWriterConfig, }; @@ -468,20 +468,19 @@ async fn fts_plan( .to_string(), })?; - // Without a maintained in-memory FTS index for this column, the active memtable - // arm produces an empty plan (`active_source_can_execute_fts` returns false), so - // the search silently omits un-compacted documents. Reject rather than mislead. - if !index_maintained( - dataset, - column, - &details.maintained_indexes, - "InvertedIndexDetails", - ) - .await? + // Two ways the fresh tier cannot answer: the spec maintains no FTS index for + // this column, or a resident MemTable was built before the index joined the + // set. Either leaves the memtable arm an empty plan + // (`active_source_can_execute_fts` returns false), so the search silently + // omits un-compacted documents. Reject rather than mislead. + if !index_maintained(dataset, column, details, "InvertedIndexDetails").await? + || !resident_memtables_carry(&in_memory, |memtable| { + memtable.index_store.get_fts_by_column(column).is_some() + }) { return Err(Error::NotSupported { message: format!( - "the MemWAL LSM scanner full-text search requires the FTS index on '{column}' to be maintained by the write spec (LsmWriteSpec::with_maintained_indexes); otherwise un-compacted documents are omitted. set use_lsm(false) to read the base table only" + "the MemWAL LSM scanner full-text search requires the FTS index on '{column}' to be maintained by the write spec (LsmWriteSpec::with_maintained_indexes) and carried by every resident MemTable; otherwise un-compacted documents are omitted. set use_lsm(false) to read the base table only" ), }); } @@ -493,6 +492,27 @@ async fn fts_plan( Ok(scanner.create_plan().await?) } +/// Whether the spec maintains the index named `name`. +/// +/// An unnamed set names nothing and covers everything, so the stored list alone +/// cannot answer this. +fn spec_maintains(details: &MemWalIndexDetails, name: &str) -> bool { + details.maintain_all_indexes || details.maintained_indexes.iter().any(|n| n == name) +} + +/// Whether every resident MemTable carries the index an arm needs. +/// +/// The spec is intent; a MemTable built before the index joined the set +/// contributes nothing to an indexed read, so its rows would go missing. +fn resident_memtables_carry( + in_memory: &HashMap, + carries: impl Fn(&InMemoryMemTableRef) -> bool, +) -> bool { + in_memory + .values() + .all(|tables| carries(&tables.active) && tables.frozen.iter().all(&carries)) +} + /// Whether an index of `type_url_suffix` covering `column` is in the MemWAL spec's /// maintained set. Only a maintained index has its catch-up tracked (so exclusion is /// gated correctly) and its in-memory arm kept current; an unmaintained base index @@ -502,7 +522,7 @@ async fn fts_plan( async fn index_maintained( dataset: &Dataset, column: &str, - maintained: &[String], + details: &MemWalIndexDetails, type_url_suffix: &str, ) -> Result { use lance::index::DatasetIndexExt; @@ -512,7 +532,7 @@ async fn index_maintained( let indices = dataset.load_indices().await?; Ok(indices.iter().any(|idx| { idx.fields.contains(&field.id) - && maintained.iter().any(|m| m == &idx.name) + && spec_maintains(details, &idx.name) && idx .index_details .as_ref() @@ -583,9 +603,7 @@ async fn arm_maintained_index_names( }) .map(|idx| idx.name.clone()) .collect(); - if let Some(name) = - resolve_single_index(segment_names, &details.maintained_indexes, arm, &column)? - { + if let Some(name) = resolve_single_index(segment_names, details, arm, &column)? { names.push(name); } } @@ -604,7 +622,7 @@ async fn arm_maintained_index_names( /// compaction watermark). fn resolve_single_index( mut names: Vec, - maintained: &[String], + details: &MemWalIndexDetails, arm: &str, column: &str, ) -> Result> { @@ -620,7 +638,7 @@ fn resolve_single_index( Ok(names .into_iter() .next() - .filter(|name| maintained.contains(name))) + .filter(|name| spec_maintains(details, name))) } /// Drop the primary-key columns Lance appends internally for dedup when the user's @@ -688,21 +706,19 @@ async fn vector_plan( None => default_vector_column(&arrow_schema, Some(query_vector.len() as i32))?, }; - // The base arm relies on the column's vector index (`fast_search`). Unless it is - // maintained, its catch-up is untracked and exclusion falls back to the - // compaction watermark — dropping compacted SSTables the (lagging) base index has - // not re-indexed. Reject rather than silently omit rows, mirroring the FTS arm. - if !index_maintained( - dataset, - &column, - &details.maintained_indexes, - "VectorIndexDetails", - ) - .await? + // The base arm relies on the column's vector index (`fast_search`). Unmaintained, + // its catch-up is untracked and exclusion falls back to the compaction watermark, + // dropping compacted SSTables the lagging base index has not re-indexed; and a + // resident MemTable built before the index joined the set carries none either. + // Reject rather than silently omit rows, mirroring the FTS arm. + if !index_maintained(dataset, &column, details, "VectorIndexDetails").await? + || !resident_memtables_carry(&in_memory, |memtable| { + memtable.index_store.get_hnsw_by_column(&column).is_some() + }) { return Err(Error::NotSupported { message: format!( - "the MemWAL LSM scanner requires the vector index on '{column}' to be maintained by the write spec (LsmWriteSpec::with_maintained_indexes); otherwise compacted rows not yet re-indexed are omitted. set use_lsm(false) to read the base table only" + "the MemWAL LSM scanner requires the vector index on '{column}' to be maintained by the write spec (LsmWriteSpec::with_maintained_indexes) and carried by every resident MemTable; otherwise compacted rows not yet re-indexed are omitted. set use_lsm(false) to read the base table only" ), }); } @@ -888,7 +904,10 @@ mod tests { #[test] fn resolve_single_index_dedupes_segments() { - let maintained = vec!["fts_idx".to_string()]; + let maintained = MemWalIndexDetails { + maintained_indexes: vec!["fts_idx".to_string()], + ..Default::default() + }; // Two physical segments of ONE logical index must not count as "multiple". assert_eq!( resolve_single_index( @@ -917,4 +936,33 @@ mod tests { None ); } + + /// A spec that maintains every index names none of them, so the stored list + /// is empty while every index on the table is in fact maintained. Reading + /// the list alone would refuse the read path on the default spec. + #[test] + fn an_unnamed_set_maintains_every_index() { + let all = MemWalIndexDetails { + maintained_indexes: Vec::new(), + maintain_all_indexes: true, + ..Default::default() + }; + assert!(spec_maintains(&all, "fts_idx")); + assert_eq!( + resolve_single_index(vec!["fts_idx".to_string()], &all, "full-text", "text").unwrap(), + Some("fts_idx".to_string()), + ); + + // The same empty list without the flag maintains nothing. + let none = MemWalIndexDetails { + maintained_indexes: Vec::new(), + maintain_all_indexes: false, + ..Default::default() + }; + assert!(!spec_maintains(&none, "fts_idx")); + assert_eq!( + resolve_single_index(vec!["fts_idx".to_string()], &none, "full-text", "text").unwrap(), + None, + ); + } }