From eb9459e3be383761ef9ac4592a7ab5b9c4b6e031 Mon Sep 17 00:00:00 2001 From: XYZhan Date: Thu, 1 Oct 2026 15:01:09 -0400 Subject: [PATCH] refactor: replace a spec by unsetting it, not by setting over it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A spec records how the generations already written were homed, and only its maintained index set could move. Accepting that one repeat means a caller who passes a different sharding — not knowing a spec is installed — gets a success that changed nothing of what they asked for. `set_lsm_write_spec` refuses while a spec is installed, and the refusal names `unset_lsm_write_spec`. Unset drains the open writer first, so the rows it held reach base before the new spec installs. The default intent still reaches a live writer when an index is created, which is the case that needed no spec change at all. --- python/python/tests/test_lsm_write_spec.py | 14 +++--- rust/lancedb/src/table.rs | 17 ++++--- rust/lancedb/src/table/merge.rs | 34 +++++++++---- rust/lancedb/src/table/merge/lsm.rs | 58 ++++++---------------- rust/lancedb/src/table/query/lsm.rs | 17 ++++--- 5 files changed, 66 insertions(+), 74 deletions(-) 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 256584e86..a99f086ef 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -1987,9 +1987,9 @@ impl Table { /// - [`LsmWriteSpec::identity`] — shard by the raw value of a scalar column. /// - [`LsmWriteSpec::unsharded`] — route every write to a single shard. /// - /// Calling this again on a table that already has a spec is how the - /// maintained index set is changed; the sharding cannot move, because the + /// 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 /// @@ -5753,7 +5753,9 @@ mod tests { "an unnamed set is stored as the intent to maintain everything" ); - // Naming the maintainable subset narrows it, with no unset in between. + // 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()]), @@ -5771,6 +5773,7 @@ mod tests { ); // Opting out entirely is distinct from the default. + table.unset_lsm_write_spec().await.unwrap(); table .set_lsm_write_spec(LsmWriteSpec::unsharded().with_maintained_indexes(Vec::new())) .await @@ -5785,16 +5788,16 @@ mod tests { Some([].as_slice()) ); - // The sharding cannot move, so the repeat that changes it is refused - // and the installed spec is left alone. + // 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("only its maintained index set can be changed")), - "expected the sharding change to be refused, got {err:?}" + if message.contains("already set")), + "expected the repeat to be refused, got {err:?}" ); assert_eq!( table diff --git a/rust/lancedb/src/table/merge.rs b/rust/lancedb/src/table/merge.rs index 1a6560034..9fa5a1314 100644 --- a/rust/lancedb/src/table/merge.rs +++ b/rust/lancedb/src/table/merge.rs @@ -1596,13 +1596,13 @@ mod lsm_tests { assert_eq!(found, 1, "the row written after create_index must be found"); } - /// Changing the set on a live table reaches the writer already open on it. + /// A spec is replaced by unsetting it and setting the new one. /// - /// Without that, the commit would land and the writer would keep building - /// MemTables from the old set, so a query needing the newly maintained - /// index would stay refused until something reopened the writer. + /// `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_set_change_reaches_an_already_open_writer() { + async fn lsm_spec_is_replaced_by_unsetting_it_first() { use crate::index::Index; use lance_index::scalar::FullTextSearchQuery; @@ -1623,14 +1623,25 @@ mod lsm_tests { .unwrap(); upsert_text(&table, vec![(99, "zebra")]).await; - // Change the set in place on the live table. + // 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 change is answered by the index the writer - // picked up. + // 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()) @@ -1641,12 +1652,15 @@ mod lsm_tests { .full_text_search(query) .execute() .await - .expect("the set change reached the open writer") + .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 change must be found"); + assert_eq!( + found, 1, + "the row written after the replacement must be found" + ); } /// A writer opened before the index existed keeps a MemTable without it, so diff --git a/rust/lancedb/src/table/merge/lsm.rs b/rust/lancedb/src/table/merge/lsm.rs index 9b2265c48..d2be0e5b1 100644 --- a/rust/lancedb/src/table/merge/lsm.rs +++ b/rust/lancedb/src/table/merge/lsm.rs @@ -73,6 +73,17 @@ const SHARD_NAMESPACE: Uuid = Uuid::from_u128(0x4c53_4d57_5249_5445_5f53_4841_52 pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec) -> Result<()> { table.dataset.ensure_mutable()?; + { + 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 and cannot be changed" + .into(), + }); + } + } + // 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 @@ -91,7 +102,6 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec) }); } } - // Before the builder borrows the dataset clone. let dataset = table.dataset.get().await?; validate_maintained_indexes(&dataset, requested).await?; Some(requested.to_vec()) @@ -99,39 +109,6 @@ pub(crate) async fn set_lsm_write_spec(table: &NativeTable, spec: LsmWriteSpec) None => None, }; - // A table that already has a spec accepts one kind of repeat: the same - // sharding with a different maintained set. The rest of a spec describes - // how the generations already written were homed, so it cannot move. - { - let dataset = table.dataset.get().await?; - if let Some(details) = dataset.mem_wal_index_details().await? { - let installed = lsm_write_spec_from_details(&details, dataset.schema())?; - if !same_except_maintained_indexes(&installed, &spec) { - return Err(Error::InvalidInput { - message: "set_lsm_write_spec: an LSM write spec is already set on this \ - table; only its maintained index set can be changed" - .into(), - }); - } - drop(dataset); - table.checkout_latest().await?; - let mut dataset = (*table.dataset.get().await?).clone(); - dataset - .update_mem_wal_maintained_indexes(maintained_indexes) - .await?; - // The writer this session may already hold was opened with the old - // set, so the change has to be installed on it here to reach a - // table that is being written to. - table - .dataset - .shard_writer() - .refresh_maintained_indexes(&dataset) - .await; - table.dataset.update(dataset); - return Ok(()); - } - } - table.checkout_latest().await?; let mut dataset = (*table.dataset.get().await?).clone(); let schema = arrow_schema::Schema::from(dataset.schema()); @@ -302,11 +279,6 @@ fn lsm_write_spec_from_details( .with_writer_config_defaults(details.writer_config_defaults.clone())) } -/// Whether two specs agree on everything an installed spec cannot change. -fn same_except_maintained_indexes(a: &LsmWriteSpec, b: &LsmWriteSpec) -> bool { - a.clone().with_maintained_indexes(None) == b.clone().with_maintained_indexes(None) -} - /// Resolve the single routing column name from a sharding field's source id. /// /// `set_lsm_write_spec` records the shard column by its Lance field id, so the @@ -468,10 +440,10 @@ impl ShardWriterCache { /// 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: both callers have already committed the - /// change durably, 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. + /// 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 diff --git a/rust/lancedb/src/table/query/lsm.rs b/rust/lancedb/src/table/query/lsm.rs index bdeb2e1f4..f3e40fc7b 100644 --- a/rust/lancedb/src/table/query/lsm.rs +++ b/rust/lancedb/src/table/query/lsm.rs @@ -468,9 +468,11 @@ 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. + // 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() @@ -704,10 +706,11 @@ 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. + // 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()