mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-12 23:02:28 +00:00
feat: maintain every index a table has, including ones created later (#4342)
**The maintained set is an intent, not a snapshot.** An unnamed set is stored as the intent to maintain every index, which lance re-resolves whenever it builds a MemTable, so an index created after the spec installs is maintained too. A named set is checked here, where the caller can be told which names exist; an index of a kind the MemWAL cannot mirror fails a named set and is skipped by an unnamed one. **A newly created index reaches a writer that is already open.** lancedb caches a `ShardWriter` per shard and reuses it, so `create_index` installs the current set on the cached writer. A failure there is reported rather than propagated: the index is committed durably by then, and the writer this process holds may be fenced or poisoned, which is no reason to fail creating an index. **A table carries one spec.** `set_lsm_write_spec` fails while one is installed and names `unset_lsm_write_spec` in the error, because the generations already written were homed under that spec. Accepting a repeat that changed only the maintained set would hand a caller who also passed a different sharding a success that changed nothing they asked for. **The read path asks what is actually carried.** An unnamed set leaves the stored list empty, so the list alone cannot decide whether an index is maintained. The read asks the spec *and* every resident MemTable whether it carries the index, so a MemTable built before the index joined the set refuses the read rather than answering without its rows. Both spec routes report an unnamed set as `null` rather than `[]`: maintain whatever the table gains, and maintain nothing, are different settings. ## Known gap The guard above covers resident MemTables. Once a writer flushes and closes none is resident, so a retained generation written before the index existed still gets an indexed read that omits its rows. This window is **wider than on main**: a set resolved at install time excluded an index created later, so the read was refused; an intent includes it, and the refusal only holds while a MemTable is resident. Closing it needs the generation's own coverage, which this layer cannot see — its snapshot carries a generation and a path — so it belongs in lance's scanner or flush path. ## Testing `cargo nextest run -p lancedb --lib --features remote` — **1479 passed, 3 skipped**. `cargo fmt --all --check`, and `cargo clippy --features remote --lib --tests --examples -- -D warnings`, both clean. Covered: the round trip of an unnamed set, a set over an installed spec refused, a spec replaced through unset, a read refused while a resident MemTable predates the index, and an index created mid-write reaching the open writer. Two Python binding tests move with the semantics: an unnamed set now reads back as `None` rather than the list it resolves to, and the refusal is matched on `already set`. Neither was run locally — no Python build here — so CI is the check on them.
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"
|
||||
|
||||
+64
-38
@@ -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<Vec<String>>,
|
||||
/// Default `ShardWriter` configuration recorded in the MemWAL index.
|
||||
writer_config_defaults: HashMap<String, String>,
|
||||
@@ -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<Vec<String>>,
|
||||
/// Default `ShardWriter` configuration recorded in the MemWAL index.
|
||||
writer_config_defaults: HashMap<String, String>,
|
||||
@@ -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<Vec<String>>,
|
||||
/// Default `ShardWriter` configuration recorded in the MemWAL index.
|
||||
writer_config_defaults: HashMap<String, String>,
|
||||
@@ -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]
|
||||
|
||||
@@ -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(())
|
||||
}
|
||||
|
||||
@@ -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<dyn RecordBatchReader + Send> {
|
||||
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::<Vec<_>>()
|
||||
.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::<Vec<_>>()
|
||||
.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::<Vec<_>>()
|
||||
.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;
|
||||
|
||||
@@ -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<Vec<String>> {
|
||||
let Some(requested) = requested else {
|
||||
let all: Vec<String> = 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
|
||||
|
||||
@@ -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<Uuid, InMemoryMemTables>,
|
||||
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<bool> {
|
||||
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<String>,
|
||||
maintained: &[String],
|
||||
details: &MemWalIndexDetails,
|
||||
arm: &str,
|
||||
column: &str,
|
||||
) -> Result<Option<String>> {
|
||||
@@ -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,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user