mirror of
https://github.com/lancedb/lancedb.git
synced 2026-08-22 05:58:20 +00:00
refactor(lsm): remove the index-catchup activation surface (#3980)
Follows lance-format/lance#8680, which removes `FLAG_MEM_WAL_INDEX_CATCHUP`. With one set of semantics there is no mode to switch into. ## Removed `require_mem_wal_index_catchup` — the activation entry point — from the trait, from `Table`, and from the LSM merge module. ## The read path `exclusion_watermarks` loses its `catchup_required` argument and keeps the conservative branch: an index with no entry is not known to hold these rows, so every generation stays readable from its SSTable. Nothing is excluded until an index records that it covers those generations, so a table that has never recorded catch-up reads every row from its SSTables rather than assuming the base covers them. ## One guard needed a replacement, not deletion `refresh_column` and computed-column declaration refuse a table whose rows sit in un-compacted tiers, since refresh enumerates base fragments and would silently omit them. They keyed on the feature bit because `unset_lsm_write_spec` **drops the MemWAL index** — after an unset the write spec no longer describes such a table, and the bit was the only marker that outlived it. Two tests covered this, so deleting the term would have dropped a tested property. Both guards now check for MemWAL shard directories on storage, which outlive the index. That is strictly wider than the bit ever was: the bit only marked tables where activation had run. ## Two tests conflated two different things An index that is *caught up* and one that is *untracked* both fell back to the compaction watermark, because absence carried no information without the bit. Absence now means "not caught up", so untracked retains everything. `an_untracked_index_does_not_widen_a_lagging_sibling` becomes `an_untracked_index_retains_everything`, with the genuinely-caught-up case asserted separately. ## Testing 933 `lancedb` lib tests. `cargo fmt` clean. (The pre-existing `Error::Http` build failure in `job.rs` without the `remote` feature is unrelated and untouched.)
This commit is contained in:
XY Zhan
GitHub
@@ -643,15 +643,6 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
|
||||
message: "set_lsm_write_spec is not supported on this table type".into(),
|
||||
})
|
||||
}
|
||||
/// Switch this table to required index catch-up, one way.
|
||||
///
|
||||
/// The default implementation returns `NotSupported`. Implementations
|
||||
/// that support the MemWAL LSM write path must override this.
|
||||
async fn require_mem_wal_index_catchup(&self) -> Result<()> {
|
||||
Err(Error::NotSupported {
|
||||
message: "require_mem_wal_index_catchup is not supported on this table type".into(),
|
||||
})
|
||||
}
|
||||
/// Remove the [`LsmWriteSpec`] from this table.
|
||||
///
|
||||
/// This is a no-op if no spec is currently set.
|
||||
@@ -1793,20 +1784,6 @@ impl Table {
|
||||
self.inner.set_lsm_write_spec(spec).await
|
||||
}
|
||||
|
||||
/// Switch this table to required index catch-up, one way.
|
||||
///
|
||||
/// Separate from [`Self::set_lsm_write_spec`] on purpose: a table carrying
|
||||
/// the bit retains its SSTables until an index records that it holds the
|
||||
/// compacted rows, so turn it on only once something can repair coverage.
|
||||
/// A writer that already holds the dataset can call the equivalent on
|
||||
/// `DatasetMemWalExt` instead; this is the table-level entry point.
|
||||
///
|
||||
/// Errors if no spec is set, or if the table already records SSTable
|
||||
/// compaction progress from before this protocol.
|
||||
pub async fn require_mem_wal_index_catchup(&self) -> Result<()> {
|
||||
self.inner.require_mem_wal_index_catchup().await
|
||||
}
|
||||
|
||||
/// Remove the [`LsmWriteSpec`] from this table, reverting to the standard
|
||||
/// `merge_insert` write path.
|
||||
///
|
||||
@@ -3354,10 +3331,6 @@ impl BaseTable for NativeTable {
|
||||
merge::lsm::set_lsm_write_spec(self, spec).await
|
||||
}
|
||||
|
||||
async fn require_mem_wal_index_catchup(&self) -> Result<()> {
|
||||
merge::lsm::require_mem_wal_index_catchup(self).await
|
||||
}
|
||||
|
||||
async fn unset_lsm_write_spec(&self) -> Result<()> {
|
||||
merge::lsm::unset_lsm_write_spec(self).await
|
||||
}
|
||||
|
||||
@@ -2151,7 +2151,6 @@ mod tests {
|
||||
.set_lsm_write_spec(LsmWriteSpec::unsharded())
|
||||
.await
|
||||
.unwrap();
|
||||
table.require_mem_wal_index_catchup().await.unwrap();
|
||||
|
||||
let mut merge = table.merge_insert(&["id"]);
|
||||
merge
|
||||
|
||||
@@ -192,36 +192,6 @@ fn index_name_list(indices: &[IndexConfig]) -> String {
|
||||
format!("[{}]", names.join(", "))
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// require_mem_wal_index_catchup
|
||||
// =============================================================================
|
||||
|
||||
/// Switch this table to required index catch-up, one way.
|
||||
///
|
||||
/// Deliberately **not** part of installing the write spec. Until something can
|
||||
/// actually repair coverage, a table carrying the bit reports every index as
|
||||
/// not known to hold the compacted rows, so its SSTables are retained
|
||||
/// indefinitely -- and the WAL pod trims on the legacy rule meanwhile, leaving
|
||||
/// readers pointed at files that are gone. Turn this on only once remote
|
||||
/// maintenance owns the merge and the repair for the table.
|
||||
///
|
||||
/// Lance refuses the activation if the table already records SSTable
|
||||
/// compaction progress: those numbers predate this protocol and cannot be
|
||||
/// validated, so such a table must be drained rather than activated.
|
||||
#[allow(clippy::redundant_pub_crate)]
|
||||
pub(crate) async fn require_mem_wal_index_catchup(table: &NativeTable) -> Result<()> {
|
||||
table.dataset.ensure_mutable()?;
|
||||
let mut dataset = (*table.dataset.get().await?).clone();
|
||||
if dataset.mem_wal_index_details().await?.is_none() {
|
||||
return Err(Error::InvalidInput {
|
||||
message: "require_mem_wal_index_catchup: no LSM write spec is set on this table".into(),
|
||||
});
|
||||
}
|
||||
dataset.require_mem_wal_index_catchup().await?;
|
||||
table.dataset.update(dataset);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// unset_lsm_write_spec
|
||||
// =============================================================================
|
||||
|
||||
@@ -36,7 +36,6 @@ use lance::dataset::mem_wal::{
|
||||
DatasetMemWalExt, LsmScanner, ShardManifestStore, ShardSnapshot, ShardWriterConfig,
|
||||
};
|
||||
use lance_index::mem_wal::{MemWalIndexDetails, ShardManifest};
|
||||
use lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::NativeTable;
|
||||
@@ -249,7 +248,6 @@ fn pk_columns(dataset: &Dataset) -> Result<Vec<String>> {
|
||||
fn exclusion_watermarks(
|
||||
details: &MemWalIndexDetails,
|
||||
index_names: &[String],
|
||||
catchup_required: bool,
|
||||
) -> HashMap<Uuid, u64> {
|
||||
let mut exclude: HashMap<Uuid, u64> = HashMap::new();
|
||||
for entry in &details.compacted_sstables {
|
||||
@@ -262,13 +260,10 @@ fn exclusion_watermarks(
|
||||
.and_then(|icp| icp.caught_up_generation_for_shard(&entry.shard_id))
|
||||
{
|
||||
Some(caught_up) => watermark = watermark.min(caught_up),
|
||||
// No entry. On a table that requires catch-up this means the
|
||||
// index is *not* known to hold these rows, and the base arm is
|
||||
// index-only -- so every generation stays readable from its
|
||||
// SSTable. Without the bit the field is not maintained at all,
|
||||
// and absence carries no information.
|
||||
None if catchup_required => watermark = 0,
|
||||
None => {}
|
||||
// No entry means the index is *not* known to hold these rows,
|
||||
// and the base arm is index-only -- so every generation stays
|
||||
// readable from its SSTable.
|
||||
None => watermark = 0,
|
||||
}
|
||||
}
|
||||
exclude.entry(entry.shard_id).or_insert(watermark);
|
||||
@@ -283,26 +278,13 @@ fn exclusion_watermarks(
|
||||
/// with a live cached `ShardWriter` (this session's in-flight writes) the
|
||||
/// writer's authoritative in-memory manifest and memtables override the
|
||||
/// on-disk view so a read sees data not yet flushed.
|
||||
/// Whether this table reads a missing `index_catchup` entry as "not caught up".
|
||||
///
|
||||
/// Both words must be set. A reader honouring the bit while a writer does not
|
||||
/// would retain SSTables the writer had already trimmed, and the reverse would
|
||||
/// serve rows from files the writer still expects to be excluded -- so a
|
||||
/// half-set manifest is treated as legacy, which is the conservative side.
|
||||
fn requires_index_catchup(dataset: &Dataset) -> bool {
|
||||
let manifest = dataset.manifest();
|
||||
manifest.reader_feature_flags & FLAG_MEM_WAL_INDEX_CATCHUP != 0
|
||||
&& manifest.writer_feature_flags & FLAG_MEM_WAL_INDEX_CATCHUP != 0
|
||||
}
|
||||
|
||||
async fn build_read_context(
|
||||
table: &NativeTable,
|
||||
dataset: &Dataset,
|
||||
details: &MemWalIndexDetails,
|
||||
index_names: &[String],
|
||||
) -> Result<(Vec<ShardSnapshot>, HashMap<Uuid, InMemoryMemTables>)> {
|
||||
let catchup_required = requires_index_catchup(dataset);
|
||||
let exclude = exclusion_watermarks(details, index_names, catchup_required);
|
||||
let exclude = exclusion_watermarks(details, index_names);
|
||||
|
||||
let shard_ids = dataset.list_mem_wal_latest_shard_ids().await?;
|
||||
// Use the dataset's own object store (not `ObjectStore::from_uri`, which
|
||||
@@ -789,50 +771,29 @@ mod tests {
|
||||
};
|
||||
|
||||
// Plain scan: drop every compacted generation (through 5).
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &[], false).get(&shard),
|
||||
Some(&5)
|
||||
);
|
||||
assert_eq!(exclusion_watermarks(&details, &[]).get(&shard), Some(&5));
|
||||
|
||||
// FTS arm with a lagging index: exclusion is capped at the index catch-up
|
||||
// (2), so SSTable generations 3..=5 are retained until the index covers
|
||||
// them — otherwise those documents would silently vanish from FTS results.
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &["fts_idx".to_string()], false).get(&shard),
|
||||
exclusion_watermarks(&details, &["fts_idx".to_string()]).get(&shard),
|
||||
Some(&2)
|
||||
);
|
||||
|
||||
// A caught-up index — or one untracked in index_catchup — falls back to the
|
||||
// compaction watermark.
|
||||
// An index absent from index_catchup is *not* known to hold these rows,
|
||||
// so nothing may be excluded and every generation stays readable from
|
||||
// its SSTable. An indexed query against an index that has not caught up
|
||||
// must not silently lose rows.
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &["caught_up_idx".to_string()], false).get(&shard),
|
||||
Some(&5)
|
||||
);
|
||||
|
||||
// The same missing entry, once the table requires catch-up: absence now
|
||||
// means "not known to hold these rows", so nothing may be excluded and
|
||||
// every generation stays readable from its SSTable. This is the whole
|
||||
// point of the protocol -- an indexed query against a table whose index
|
||||
// has not caught up must not silently lose rows.
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &["untracked_idx".to_string()], true).get(&shard),
|
||||
exclusion_watermarks(&details, &["untracked_idx".to_string()]).get(&shard),
|
||||
Some(&0)
|
||||
);
|
||||
|
||||
// A tracked index is unaffected by the mode: the recorded position is
|
||||
// information either way, and it still caps the exclusion.
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &["fts_idx".to_string()], true).get(&shard),
|
||||
Some(&2)
|
||||
);
|
||||
|
||||
// One missing entry is enough to hold everything back, even alongside an
|
||||
// index that has caught up.
|
||||
let mixed = vec!["fts_idx".to_string(), "untracked_idx".to_string()];
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &mixed, true).get(&shard),
|
||||
Some(&0)
|
||||
);
|
||||
assert_eq!(exclusion_watermarks(&details, &mixed).get(&shard), Some(&0));
|
||||
}
|
||||
|
||||
/// A hybrid search reads a vector and a full-text index, and either may lag.
|
||||
@@ -859,31 +820,29 @@ mod tests {
|
||||
|
||||
// Each index alone stops at its own catch-up.
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &["vec_idx".to_string()], false).get(&shard),
|
||||
exclusion_watermarks(&details, &["vec_idx".to_string()]).get(&shard),
|
||||
Some(&7)
|
||||
);
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &["fts_idx".to_string()], false).get(&shard),
|
||||
exclusion_watermarks(&details, &["fts_idx".to_string()]).get(&shard),
|
||||
Some(&4)
|
||||
);
|
||||
|
||||
// Used together, the lower one governs regardless of order.
|
||||
let both = ["vec_idx".to_string(), "fts_idx".to_string()];
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &both, false).get(&shard),
|
||||
Some(&4)
|
||||
);
|
||||
assert_eq!(exclusion_watermarks(&details, &both).get(&shard), Some(&4));
|
||||
let reversed = ["fts_idx".to_string(), "vec_idx".to_string()];
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &reversed, false).get(&shard),
|
||||
exclusion_watermarks(&details, &reversed).get(&shard),
|
||||
Some(&4)
|
||||
);
|
||||
}
|
||||
|
||||
/// An index with no catch-up entry contributes no cap today, so a lagging
|
||||
/// sibling must still govern rather than being widened by the untracked one.
|
||||
/// An index with no catch-up entry is not known to hold these rows, so it
|
||||
/// retains everything -- it is never widened by a tracked sibling that has
|
||||
/// caught up further.
|
||||
#[test]
|
||||
fn an_untracked_index_does_not_widen_a_lagging_sibling() {
|
||||
fn an_untracked_index_retains_everything() {
|
||||
let shard = Uuid::from_u128(1);
|
||||
let details = MemWalIndexDetails {
|
||||
compacted_sstables: vec![CompactedSsTable::new(shard, 9)],
|
||||
@@ -896,8 +855,11 @@ mod tests {
|
||||
};
|
||||
|
||||
let both = ["fts_idx".to_string(), "untracked_idx".to_string()];
|
||||
assert_eq!(exclusion_watermarks(&details, &both).get(&shard), Some(&0));
|
||||
|
||||
// The tracked sibling still caps on its own.
|
||||
assert_eq!(
|
||||
exclusion_watermarks(&details, &both, false).get(&shard),
|
||||
exclusion_watermarks(&details, &["fts_idx".to_string()]).get(&shard),
|
||||
Some(&4)
|
||||
);
|
||||
}
|
||||
|
||||
@@ -141,11 +141,19 @@ pub(crate) async fn execute_refresh_column_async(table: &NativeTable, column: &s
|
||||
/// un-compacted MemWAL tiers it cannot reach -- success would silently omit
|
||||
/// readable rows.
|
||||
async fn ensure_no_lsm_write_spec(table: &NativeTable) -> Result<()> {
|
||||
// The catch-up flag outlives unset and marks retained SSTable rows.
|
||||
let catchup = table.dataset.get().await?.manifest().reader_feature_flags
|
||||
& lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP
|
||||
!= 0;
|
||||
if catchup || table.get_lsm_write_spec().await?.is_some() {
|
||||
use lance::dataset::mem_wal::DatasetMemWalExt;
|
||||
|
||||
// Unset drops the MemWAL index, so the spec alone stops describing a table
|
||||
// whose SSTables still hold rows. The shard directories outlive it and are
|
||||
// the durable evidence.
|
||||
let retained_sstables = !table
|
||||
.dataset
|
||||
.get()
|
||||
.await?
|
||||
.list_mem_wal_latest_shard_ids()
|
||||
.await?
|
||||
.is_empty();
|
||||
if retained_sstables || table.get_lsm_write_spec().await?.is_some() {
|
||||
return Err(Error::NotSupported {
|
||||
message: "refresh_column is not supported on a table with an LSM write \
|
||||
spec: rows in un-compacted tiers are invisible to refresh"
|
||||
@@ -921,7 +929,6 @@ mod tests {
|
||||
.set_lsm_write_spec(LsmWriteSpec::unsharded())
|
||||
.await
|
||||
.unwrap();
|
||||
table.require_mem_wal_index_catchup().await.unwrap();
|
||||
let mut merge = table.merge_insert(&["x"]);
|
||||
merge
|
||||
.when_matched_update_all(None)
|
||||
|
||||
@@ -124,18 +124,26 @@ pub(crate) async fn execute_declare(
|
||||
table: &NativeTable,
|
||||
columns: &[(String, String)],
|
||||
) -> Result<AddColumnsResult> {
|
||||
use lance::dataset::mem_wal::DatasetMemWalExt;
|
||||
|
||||
// An LSM write spec keeps visible rows in tiers refresh cannot reach;
|
||||
// checked against latest committed state, not this handle's snapshot.
|
||||
// The catch-up flag outlives unset and marks retained SSTable rows.
|
||||
table.checkout_latest().await?;
|
||||
computed_columns::ensure_no_function_bindings_for_mutation(
|
||||
table.schema().await?.as_ref(),
|
||||
"schema evolution",
|
||||
)?;
|
||||
let catchup = table.dataset.get().await?.manifest().reader_feature_flags
|
||||
& lance_table::feature_flags::FLAG_MEM_WAL_INDEX_CATCHUP
|
||||
!= 0;
|
||||
if catchup || table.get_lsm_write_spec().await?.is_some() {
|
||||
// Unset drops the MemWAL index, so the spec alone stops describing a table
|
||||
// whose SSTables still hold rows. The shard directories outlive it and are
|
||||
// the durable evidence.
|
||||
let retained_sstables = !table
|
||||
.dataset
|
||||
.get()
|
||||
.await?
|
||||
.list_mem_wal_latest_shard_ids()
|
||||
.await?
|
||||
.is_empty();
|
||||
if retained_sstables || table.get_lsm_write_spec().await?.is_some() {
|
||||
return Err(Error::NotSupported {
|
||||
message: "computed columns are not supported on a table with an LSM write \
|
||||
spec: rows in un-compacted tiers are invisible to refresh"
|
||||
|
||||
Reference in New Issue
Block a user