mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-24 04:32:27 +00:00
refactor: replace a spec by unsetting it, not by setting over it
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.
This commit is contained in:
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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::<Vec<_>>()
|
||||
.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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user