From 02969b33db66f06b13201a0d2164eb98fe6a3231 Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Sat, 3 Oct 2026 22:50:21 +0000 Subject: [PATCH 1/2] feat: configure WAL key indexes with bounded ZoneMap predicate deletes --- .../lance-context-core/src/datagen_store.rs | 3 + .../lance-context-core/src/generic_store.rs | 4 + crates/lance-context-core/src/key_index.rs | 20 ++ crates/lance-context-core/src/lib.rs | 2 + .../lance-context-core/src/rollout_store.rs | 118 +++++++++++- crates/lance-context-core/src/store.rs | 1 + crates/lance-context-core/src/store_base.rs | 178 +++++++++++++----- .../tests/wal_merge_generation_cleanup.rs | 3 + .../src/catchup/executor.rs | 2 + .../src/catchup/kubernetes.rs | 4 + .../lance-context-master/src/catchup/tests.rs | 20 +- crates/lance-context-master/src/config.rs | 22 +++ crates/lance-context-master/src/routes.rs | 1 + crates/lance-context-master/src/scheduler.rs | 8 +- crates/lance-context-master/src/state.rs | 3 + crates/lance-context-master/src/task_store.rs | 1 + crates/lance-context-server/src/config.rs | 25 +++ .../src/routes/datagen.rs | 1 + .../src/routes/generic.rs | 1 + .../src/routes/rollouts.rs | 1 + crates/lance-context-server/src/state.rs | 6 + crates/lance-context/src/unified_datagen.rs | 1 + crates/lance-context/src/unified_generic.rs | 2 + docs/benchmarks/merge-index-policy.md | 106 +++++++++++ docs/benchmarks/merge-index-results.json | 133 +++++++++++++ docs/benchmarks/merge_index.py | 95 ++++++++++ 26 files changed, 703 insertions(+), 58 deletions(-) create mode 100644 crates/lance-context-core/src/key_index.rs create mode 100644 docs/benchmarks/merge-index-policy.md create mode 100644 docs/benchmarks/merge-index-results.json create mode 100644 docs/benchmarks/merge_index.py diff --git a/crates/lance-context-core/src/datagen_store.rs b/crates/lance-context-core/src/datagen_store.rs index 5e66edd..cea5156 100644 --- a/crates/lance-context-core/src/datagen_store.rs +++ b/crates/lance-context-core/src/datagen_store.rs @@ -60,6 +60,8 @@ pub struct DatagenStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Index used for base-table key lookup during WAL merge. + pub key_index_type: crate::KeyIndexType, /// Warn when one MemWAL shard has at least this many flushed generations /// pending merge, sampled on every read. `None` uses the crate default /// (256); `Some(0)` disables the warn. The `rollout_wal_pending_generations` @@ -115,6 +117,7 @@ impl DatagenStore { merge_after_generations: options.merge_after_generations, merge_max_generations: options.merge_max_generations, merge_max_bytes: options.merge_max_bytes, + key_index_type: options.key_index_type, pending_generations_warn: options.pending_generations_warn, pending_generations_max: options.pending_generations_max, merge_budget: options.merge_budget.clone(), diff --git a/crates/lance-context-core/src/generic_store.rs b/crates/lance-context-core/src/generic_store.rs index 19ad0e4..266baea 100644 --- a/crates/lance-context-core/src/generic_store.rs +++ b/crates/lance-context-core/src/generic_store.rs @@ -72,6 +72,8 @@ pub struct GenericStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Index used for base-table key lookup during WAL merge. + pub key_index_type: crate::KeyIndexType, /// Warn when one MemWAL shard has at least this many flushed generations /// pending merge, sampled on every read. Every pending generation is a /// separate dataset a read must open, so this is the read-amplification @@ -187,6 +189,7 @@ impl GenericStore { merge_after_generations: options.merge_after_generations, merge_max_generations: options.merge_max_generations, merge_max_bytes: options.merge_max_bytes, + key_index_type: options.key_index_type, pending_generations_warn: options.pending_generations_warn, pending_generations_max: options.pending_generations_max, merge_budget: options.merge_budget.clone(), @@ -868,6 +871,7 @@ mod tests { seal_on_add: true, shard_id: Some(shard.to_string()), merge_max_bytes: Some(1024 * 1024), + key_index_type: Default::default(), merge_budget: Some(budget.clone()), ..Default::default() }; diff --git a/crates/lance-context-core/src/key_index.rs b/crates/lance-context-core/src/key_index.rs new file mode 100644 index 0000000..c4d02b8 --- /dev/null +++ b/crates/lance-context-core/src/key_index.rs @@ -0,0 +1,20 @@ +/// Base-table key index policy used by WAL merge and scheduled indexing. +/// All writers of a table should use the same policy to avoid rebuilding the +/// index back and forth during maintenance. Opening a store does not rebuild it. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, clap::ValueEnum)] +pub enum KeyIndexType { + /// Exact key lookup; preserves the existing default. + #[default] + Btree, + /// Compact zone summaries, with key-only predicate deletion during merge. + Zonemap, +} + +impl KeyIndexType { + pub fn as_str(self) -> &'static str { + match self { + Self::Btree => "btree", + Self::Zonemap => "zonemap", + } + } +} diff --git a/crates/lance-context-core/src/lib.rs b/crates/lance-context-core/src/lib.rs index 17c3082..7e50561 100644 --- a/crates/lance-context-core/src/lib.rs +++ b/crates/lance-context-core/src/lib.rs @@ -11,6 +11,7 @@ mod export; pub mod generic_codec; mod generic_store; mod id; +mod key_index; pub mod merge_budget; pub mod merge_write_scope; pub mod metrics; @@ -59,6 +60,7 @@ pub use export::{ pub use generic_codec::{batch_to_rows, ids_from_batch, rows_to_batch, Row}; pub use generic_store::{GenericStore, GenericStoreOptions}; pub use id::{generate_id, new_uuid}; +pub use key_index::KeyIndexType; pub use merge_budget::{MergeMemoryBudget, MergeReservation}; pub use namespace::{ContextNamespace, PartitionInfo, PartitionSelector, PartitionSpec}; pub use record::{ diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index f78384d..9d47a00 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -386,6 +386,8 @@ pub struct RolloutStoreOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Index used for base-table key lookup during WAL merge. + pub key_index_type: crate::KeyIndexType, /// Warn when one MemWAL shard has at least this many flushed generations /// pending merge, sampled on every read. `None` uses the crate default /// (256); `Some(0)` disables the warn. The `rollout_wal_pending_generations` @@ -483,6 +485,7 @@ impl RolloutStore { merge_after_generations, merge_max_generations, merge_max_bytes, + key_index_type, pending_generations_warn, pending_generations_max, merge_budget, @@ -496,6 +499,7 @@ impl RolloutStore { merge_after_generations, merge_max_generations, merge_max_bytes, + key_index_type, pending_generations_warn, pending_generations_max, merge_budget, @@ -712,8 +716,12 @@ impl RolloutStore { self.base.repair_missing_fragments().await } - /// Build a ZoneMap scalar index on the base table's `id` column. Idempotent. - /// See `StorageBase::create_key_btree_index`. + /// Build the configured key index on the base table's `id` column. + pub async fn create_id_key_index(&mut self) -> LanceResult<()> { + self.base.create_configured_key_index().await + } + + /// Explicitly create a BTree, independent of the configured merge policy. pub async fn create_id_btree_index(&mut self) -> LanceResult<()> { self.base.create_key_btree_index().await } @@ -2851,6 +2859,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3173,6 +3182,7 @@ mod tests { merge_after_generations: Some(0), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3220,6 +3230,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3272,6 +3283,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3326,6 +3338,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3550,6 +3563,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3615,6 +3629,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3718,6 +3733,7 @@ mod tests { merge_after_generations: None, // no merge → epoch never reclaimed merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3779,6 +3795,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3821,6 +3838,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -3959,6 +3977,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -4048,6 +4067,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -4137,6 +4157,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -4188,6 +4209,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -4275,6 +4297,7 @@ mod tests { merge_after_generations: Some(3), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -4315,6 +4338,7 @@ mod tests { merge_after_generations: Some(3), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -4376,6 +4400,7 @@ mod tests { merge_after_generations: Some(2), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -4536,6 +4561,7 @@ mod tests { merge_after_generations: None, // count trigger off merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -4678,13 +4704,81 @@ mod tests { }); } - /// The whole point of the id index: `merge_insert` must take the indexed - /// probe path, not a full-table hash join. Lance's `explain_plan` only - /// renders the full-scan plan and returns `NotSupported` when the job - /// would use a scalar index, so "explain refuses" is the observable - /// signal that the merge will probe. Without the index (and with a - /// ZoneMap, which cannot answer equality exactly) explain succeeds and - /// shows a HashJoin over a LanceScan of the base table. + /// ZoneMap merges use key predicates rather than the full-target merge + /// join. Exercise multiple bounded delete chunks and unusual literal IDs. + #[test] + fn zonemap_merge_preserves_updates_across_delete_chunks_and_retries() { + use crate::KeyIndexType; + use lance::index::DatasetIndexExt; + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_string_lossy().to_string(); + tokio::runtime::Runtime::new().unwrap().block_on(async { + let mut store = RolloutStore::open_with_options( + &uri, + RolloutStoreOptions { + key_index_type: KeyIndexType::Zonemap, + merge_max_generations: Some(1), + ..Default::default() + }, + ) + .await + .unwrap(); + let mut ids: Vec<_> = (0..2050).map(|i| format!("key-{i}")).collect(); + ids.push("quote' AND false -- \\ unicode-你好".to_string()); + for payload in [1u8, 2, 2] { + let records: Vec<_> = ids + .iter() + .map(|id| artifact_record(id, &[payload; 1024])) + .collect(); + store.add(&records).await.unwrap(); + store.flush().await.unwrap(); + assert_eq!(store.cleanup_own_shard().await.unwrap(), 1); + assert_eq!(flushed_generation_count(&store).await, 0); + assert_eq!( + store.base.dataset.count_rows(None).await.unwrap(), + ids.len() + ); + let actual = store.list(None, None).await.unwrap(); + assert_eq!(actual.len(), ids.len()); + let mut scanner = store.base.dataset.scan(); + scanner.project(&["binary_payload"]).unwrap(); + let mut stream = scanner.try_into_stream().await.unwrap(); + while let Some(batch) = stream.try_next().await.unwrap() { + let blobs = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + assert_eq!(blobs.value(row), &[payload; 1024]); + } + } + } + assert!(!store.has_id_btree_index().await.unwrap()); + let indices = store.base.dataset.load_indices().await.unwrap(); + assert!(indices.iter().any(|i| i.name == ROLLOUT_ID_INDEX_NAME + && i.index_details + .as_ref() + .is_some_and(|d| d.type_url.ends_with("ZoneMapIndexDetails")))); + + // A configured maintenance rebuild must preserve the selected type. + store.create_id_key_index().await.unwrap(); + assert!(!store.has_id_btree_index().await.unwrap()); + // Changing the policy takes effect on maintenance, not on opening. + drop(store); + let mut store = RolloutStore::open(&uri).await.unwrap(); + assert!(!store.has_id_btree_index().await.unwrap()); + store + .add(&[artifact_record(&ids[0], &[3; 1024])]) + .await + .unwrap(); + store.flush().await.unwrap(); + store.cleanup_own_shard().await.unwrap(); + assert!(store.has_id_btree_index().await.unwrap()); + assert_eq!(store.list(None, None).await.unwrap().len(), ids.len()); + }); + } + #[test] fn merge_insert_probes_the_id_btree_instead_of_scanning() { use lance::dataset::MergeInsertBuilder; @@ -4825,6 +4919,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: Some(3), merge_budget: None, @@ -4873,6 +4968,7 @@ mod tests { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: Some(2), merge_budget: None, @@ -5102,6 +5198,7 @@ mod tests { shard_id: Some("budget-contention".to_string()), merge_max_generations: Some(0), merge_max_bytes: Some(MIB), + key_index_type: Default::default(), merge_budget: Some(budget.clone()), ..Default::default() }, @@ -5344,6 +5441,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -5445,6 +5543,7 @@ mod tests { merge_after_generations: None, // disabled merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -5480,6 +5579,7 @@ mod tests { merge_after_generations: Some(1), merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, diff --git a/crates/lance-context-core/src/store.rs b/crates/lance-context-core/src/store.rs index 9acba05..184418a 100644 --- a/crates/lance-context-core/src/store.rs +++ b/crates/lance-context-core/src/store.rs @@ -649,6 +649,7 @@ impl ContextStore { shard_id: options.shard_id.clone(), merge_after_generations: options.merge_after_generations, merge_max_generations: options.merge_max_generations, + key_index_type: Default::default(), merge_max_bytes: options.merge_max_bytes, pending_generations_warn: options.pending_generations_warn, pending_generations_max: options.pending_generations_max, diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index 3568079..9542740 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -256,6 +256,8 @@ pub(crate) struct StorageBaseOptions { /// reaches the budget, or the generation-count cap, whichever comes first. /// A generation is indivisible, so even an oversized one is fully merged. pub merge_max_bytes: Option, + /// Index used for base-table key lookup during WAL merge. + pub key_index_type: crate::KeyIndexType, /// Warn when a single shard has at least this many flushed generations /// pending merge (sampled on every LSM read). `None` uses the crate default /// (256); `Some(0)` disables the warning. The metric is always emitted. @@ -331,6 +333,7 @@ pub(crate) struct StorageBase { merge_after_generations: usize, merge_max_generations: usize, merge_max_bytes: usize, + key_index_type: crate::KeyIndexType, /// Per-shard pending-generation count at which reads warn; `0` disables. pending_generations_warn: usize, /// Refuse reads that would open more flushed generations than this, per @@ -394,6 +397,7 @@ impl StorageBase { merge_after_generations, merge_max_generations, merge_max_bytes, + key_index_type, pending_generations_warn, pending_generations_max, merge_budget, @@ -427,6 +431,7 @@ impl StorageBase { merge_after_generations, merge_max_generations, merge_max_bytes, + key_index_type, pending_generations_warn, pending_generations_max, merge_budget, @@ -454,6 +459,7 @@ impl StorageBase { merge_after_generations, merge_max_generations, merge_max_bytes, + key_index_type, pending_generations_warn, pending_generations_max, merge_budget, @@ -482,6 +488,7 @@ impl StorageBase { merge_after_generations: merge_after_generations.unwrap_or(0), merge_max_generations: merge_max_generations.unwrap_or(DEFAULT_MERGE_MAX_GENERATIONS), merge_max_bytes: merge_max_bytes.unwrap_or(DEFAULT_MERGE_MAX_BYTES), + key_index_type, pending_generations_warn: pending_generations_warn .unwrap_or(DEFAULT_PENDING_GENERATIONS_WARN), pending_generations_max: pending_generations_max @@ -1252,8 +1259,9 @@ impl StorageBase { /// Merge prepared WAL rows into the base table by primary key. /// /// `read_flushed_generations` has already reduced the source to one newest - /// row per key. The merge is two commits: a delete-only `merge_insert` on - /// the key column, then a plain append of the rows. + /// row per key. Delete old keys first, then append the prepared rows. + /// BTree uses a delete-only `merge_insert`; ZoneMap uses bounded predicate + /// deletes so Lance does not include target payloads in its hash join. /// /// Not `WhenMatched::UpdateAll`, deliberately. An upsert takes the matched /// target rows *with every column* to join and rewrite them, and rollout @@ -1265,7 +1273,7 @@ impl StorageBase { /// rows straight to fresh fragments. Neither ever holds a target blob. /// /// Last-write-wins and retry idempotence are preserved: a crash between - /// the two commits leaves the old rows deleted and the new ones still in + /// commits leaves the old rows deleted and the new ones still in /// the WAL, and the retry deletes nothing and appends them once. A crash /// after the append but before the manifest drain retries as delete (of /// what was just appended) + append, the same end state. @@ -1274,54 +1282,54 @@ impl StorageBase { batches: Vec, merge_schema: Arc, ) -> LanceResult<()> { - // Without an exact-answer key index the delete-only merge_insert is - // a hash join over the whole base table. On a 174 GB / 67k-row store - // that no master had ever indexed, that join OOMKilled every worker - // that tried to merge it; building the BTree first took 19 s. The - // master builds it before fan-out (#277), but a worker's own timer - // or a manual merge must not depend on the master having been here. - if self.dataset.count_fragments() > 0 { - if !self.has_key_btree_index().await? { - metrics::counter!("rollout_merge_index_built_on_demand_total").increment(1); - info!( - uri = %self.dataset.uri(), - "base table has no key BTree; building it before the merge" - ); - self.create_key_btree_index().await?; - } else { - // Present is not enough: every merge and compaction appends - // fragments the index does not cover, and the delete-only - // merge_insert hash-joins exactly those. A 2.2 GB store whose - // BTree covered 1 of 65 fragments took a worker from 5 to - // 32 GiB on a 37 MB merge and OOMKilled 19 of them at once; - // after one `optimize_indices` the same merge peaked at 3 GiB. - let covered = self.extend_key_btree_index().await?; - if covered > 0 { - metrics::counter!("rollout_merge_index_extended_on_demand_total").increment(1); - } - } - } + observe_phase!("index", self.ensure_merge_key_index().await)?; let key_index = merge_schema.index_of(&self.key_column)?; let key_schema = Arc::new(merge_schema.project(&[key_index])?); let keys = batches .iter() .map(|batch| batch.project(&[key_index])) .collect::, ArrowError>>()?; - let key_reader = RecordBatchIterator::new( - keys.into_iter().map(Ok::), - key_schema, - ); - let mut builder = MergeInsertBuilder::try_new( - Arc::new(self.dataset.clone()), - vec![self.key_column.clone()], - )?; - builder - .when_matched(WhenMatched::Delete) - .when_not_matched(WhenNotMatched::DoNothing); - // Both Lance futures are large; boxing keeps them off the caller's - // stack (the merge runs inside sweeper and request tasks). - let (dataset, _) = Box::pin(builder.try_build()?.execute_reader(key_reader)).await?; - self.dataset = Arc::unwrap_or_clone(dataset); + if self.key_index_type == crate::KeyIndexType::Zonemap { + // merge_insert's non-exact-index path projects target payloads even + // with an ID-only source in Lance 9.0. Predicate deletion instead + // scans only predicate columns and lets ZoneMap prune zones. + // Bound expression size; commit all deletes before appending rows. + // A cancelled prefix remains recoverable from the undrained WAL. + use datafusion::common::ScalarValue; + use datafusion::logical_expr::Expr; + let mut values = Vec::with_capacity(1024); + for batch in &keys { + for row in 0..batch.num_rows() { + values.push(Expr::Literal( + ScalarValue::try_from_array(batch.column(0), row) + .map_err(|e| LanceError::invalid_input(e.to_string()))?, + None, + )); + if values.len() == 1024 { + self.delete_key_values(std::mem::take(&mut values)).await?; + } + } + } + if !values.is_empty() { + self.delete_key_values(values).await?; + } + } else { + let key_reader = RecordBatchIterator::new( + keys.into_iter().map(Ok::), + key_schema, + ); + let mut builder = MergeInsertBuilder::try_new( + Arc::new(self.dataset.clone()), + vec![self.key_column.clone()], + )?; + builder + .when_matched(WhenMatched::Delete) + .when_not_matched(WhenNotMatched::DoNothing); + // Both Lance futures are large; boxing keeps them off the caller's + // stack (the merge runs inside sweeper and request tasks). + let (dataset, _) = Box::pin(builder.try_build()?.execute_reader(key_reader)).await?; + self.dataset = Arc::unwrap_or_clone(dataset); + } let reader = RecordBatchIterator::new( batches.into_iter().map(Ok::), @@ -1331,6 +1339,86 @@ impl StorageBase { Ok(()) } + async fn delete_key_values( + &mut self, + values: Vec, + ) -> LanceResult<()> { + use datafusion::{common::Column, logical_expr::Expr}; + let predicate = + Expr::Column(Column::from_name(self.key_column.clone())).in_list(values, false); + let result = observe_phase!( + "delete_keys", + Box::pin( + lance::dataset::DeleteBuilder::from_expr(Arc::new(self.dataset.clone()), predicate) + .execute() + ) + .await + )?; + self.dataset = Arc::unwrap_or_clone(result.new_dataset); + crate::merge_write_scope::checkpoint(); + Ok(()) + } + + async fn ensure_merge_key_index(&mut self) -> LanceResult<()> { + if self.dataset.count_fragments() == 0 { + return Ok(()); + } + match self.key_index_type { + crate::KeyIndexType::Btree => { + if !self.has_key_btree_index().await? { + self.create_key_btree_index().await?; + metrics::counter!("rollout_merge_index_built_on_demand_total").increment(1); + } else if self.extend_key_btree_index().await? > 0 { + metrics::counter!("rollout_merge_index_extended_on_demand_total").increment(1); + } + } + crate::KeyIndexType::Zonemap => { + use lance::index::DatasetIndexInternalExt as _; + let indices = self.dataset.load_indices().await?; + let present = indices.iter().filter(|i| i.name == ID_INDEX_NAME).any(|i| { + i.index_details + .as_ref() + .is_some_and(|d| d.type_url.ends_with("ZoneMapIndexDetails")) + }); + if !present { + self.create_configured_key_index().await?; + } else if !self + .dataset + .unindexed_fragments(ID_INDEX_NAME) + .await? + .is_empty() + { + self.dataset + .optimize_indices( + &lance_index::optimize::OptimizeOptions::merge(1) + .index_names(vec![ID_INDEX_NAME.to_string()]), + ) + .await?; + self.reload().await?; + } + } + } + Ok(()) + } + + pub async fn create_configured_key_index(&mut self) -> LanceResult<()> { + self.ensure_writable()?; + let kind = match self.key_index_type { + crate::KeyIndexType::Btree => IndexType::BTree, + crate::KeyIndexType::Zonemap => IndexType::ZoneMap, + }; + self.dataset + .create_index_builder( + &[self.key_column.as_str()], + kind, + &ScalarIndexParams::default(), + ) + .name(ID_INDEX_NAME.to_string()) + .replace(true) + .await?; + self.reload().await + } + /// Evolve an older base table to the store's latest additive schema. /// /// Missing nullable columns are added as all-null arrays. Existing unknown diff --git a/crates/lance-context-core/tests/wal_merge_generation_cleanup.rs b/crates/lance-context-core/tests/wal_merge_generation_cleanup.rs index 9444808..739b36b 100644 --- a/crates/lance-context-core/tests/wal_merge_generation_cleanup.rs +++ b/crates/lance-context-core/tests/wal_merge_generation_cleanup.rs @@ -271,6 +271,7 @@ async fn zero_merge_caps_merge_everything_in_one_pass() { merge_after_generations: None, merge_max_generations: Some(0), merge_max_bytes: Some(0), + key_index_type: Default::default(), ..Default::default() }; let mut store = RolloutStore::open_with_options(&uri, opts).await.unwrap(); @@ -297,6 +298,7 @@ async fn assert_blob_merge_passes(max_generations: usize, max_bytes: usize, pass shard_id: Some("byte-budget".to_string()), merge_max_generations: Some(max_generations), merge_max_bytes: Some(max_bytes), + key_index_type: Default::default(), ..Default::default() }; let mut store = RolloutStore::open_with_options(&uri, opts).await.unwrap(); @@ -366,6 +368,7 @@ async fn oversized_generation_is_merged_in_full_and_makes_progress() { merge_after_generations: Some(1), merge_max_generations: Some(8), merge_max_bytes: Some(1), + key_index_type: Default::default(), ..Default::default() }; let mut store = RolloutStore::open_with_options(&uri, opts).await.unwrap(); diff --git a/crates/lance-context-master/src/catchup/executor.rs b/crates/lance-context-master/src/catchup/executor.rs index 6cf4ca9..dba46fd 100644 --- a/crates/lance-context-master/src/catchup/executor.rs +++ b/crates/lance-context-master/src/catchup/executor.rs @@ -102,6 +102,7 @@ async fn merge_passes(state: &Arc, target: &str) -> Result merge_after_generations: Some(0), merge_max_generations: Some(8), merge_max_bytes: Some(config.merge_max_bytes), + key_index_type: state.config.key_index_type, merge_budget: Some(budget.clone()), session: Some(session.clone()), pending_generations_max: Some(0), @@ -128,6 +129,7 @@ async fn merge_passes(state: &Arc, target: &str) -> Result merge_after_generations: Some(0), merge_max_generations: Some(8), merge_max_bytes: Some(config.merge_max_bytes), + key_index_type: state.config.key_index_type, merge_budget: Some(budget.clone()), session: Some(session.clone()), pending_generations_max: Some(0), diff --git a/crates/lance-context-master/src/catchup/kubernetes.rs b/crates/lance-context-master/src/catchup/kubernetes.rs index 003c5ca..fbb1042 100644 --- a/crates/lance-context-master/src/catchup/kubernetes.rs +++ b/crates/lance-context-master/src/catchup/kubernetes.rs @@ -235,6 +235,10 @@ pub(super) fn render_job(config: &MasterConfig, record: &Record, mut spec: Value } let overrides = [ ("DATA_DIR", config.data_dir.clone()), + ( + "ROLLOUT_KEY_INDEX_TYPE", + config.key_index_type.as_str().into(), + ), ("ETCD_ENDPOINTS", config.etcd.etcd_endpoints.join(",")), ("ETCD_PREFIX", config.etcd.etcd_prefix.clone()), ("CATCHUP_ENABLED", "false".into()), diff --git a/crates/lance-context-master/src/catchup/tests.rs b/crates/lance-context-master/src/catchup/tests.rs index 1b30f6c..8f35776 100644 --- a/crates/lance-context-master/src/catchup/tests.rs +++ b/crates/lance-context-master/src/catchup/tests.rs @@ -52,7 +52,8 @@ fn admission_requires_fresh_owned_pressure_even_for_manual_requests() { } #[test] fn job_is_native_and_bounded_and_overrides_unsafe_inherited_mode() { - let c = config(); + let mut c = config(); + c.key_index_type = lance_context_core::KeyIndexType::Zonemap; let r = Record { target: "hot".into(), job: "lc-catchup-test".into(), @@ -72,7 +73,10 @@ fn job_is_native_and_bounded_and_overrides_unsafe_inherited_mode() { progress: None, }; let mut pod = template(); - pod["containers"][0]["env"] = json!([{"name":"CATCHUP_ENABLED","value":"true"}]); + pod["containers"][0]["env"] = json!([ + {"name":"CATCHUP_ENABLED","value":"true"}, + {"name":"ROLLOUT_KEY_INDEX_TYPE","value":"btree"} + ]); pod["activeDeadlineSeconds"] = json!(1); let job = kubernetes::render_job(&c, &r, pod); assert!(job["spec"].get("activeDeadlineSeconds").is_none()); @@ -89,6 +93,18 @@ fn job_is_native_and_bounded_and_overrides_unsafe_inherited_mode() { let env = job["spec"]["template"]["spec"]["containers"][0]["env"] .as_array() .unwrap(); + assert_eq!( + env.iter() + .filter(|e| e["name"] == "ROLLOUT_KEY_INDEX_TYPE") + .count(), + 1 + ); + assert_eq!( + env.iter() + .find(|e| e["name"] == "ROLLOUT_KEY_INDEX_TYPE") + .unwrap()["value"], + "zonemap" + ); assert_eq!( env.iter() .filter(|e| e["name"] == "CATCHUP_ENABLED") diff --git a/crates/lance-context-master/src/config.rs b/crates/lance-context-master/src/config.rs index 7ef3f8b..bdb51f2 100644 --- a/crates/lance-context-master/src/config.rs +++ b/crates/lance-context-master/src/config.rs @@ -118,6 +118,16 @@ pub struct MasterConfig { #[arg(long, env = "INDEX_BEFORE_MERGE", default_value_t = true, action = clap::ArgAction::Set)] pub index_before_merge: bool, + /// Base key index policy. Configure every master and worker consistently. + /// ZoneMap uses bounded key-only predicate deletes during WAL merge. + #[arg( + long, + env = "ROLLOUT_KEY_INDEX_TYPE", + value_enum, + default_value = "btree" + )] + pub key_index_type: lance_context_core::KeyIndexType, + /// Maximum bytes per compacted output file. `0` uses Lance's default. #[arg( long, @@ -238,6 +248,18 @@ pub struct MasterConfig { #[cfg(test)] mod tests { + #[test] + fn key_index_policy_is_explicit_and_validated() { + use clap::Parser; + use lance_context_core::KeyIndexType; + let config = + super::MasterConfig::try_parse_from(["master", "--key-index-type", "zonemap"]).unwrap(); + assert_eq!(config.key_index_type, KeyIndexType::Zonemap); + assert!( + super::MasterConfig::try_parse_from(["master", "--key-index-type", "unknown",]) + .is_err() + ); + } use super::*; use clap::CommandFactory; diff --git a/crates/lance-context-master/src/routes.rs b/crates/lance-context-master/src/routes.rs index 6bef210..3177eab 100644 --- a/crates/lance-context-master/src/routes.rs +++ b/crates/lance-context-master/src/routes.rs @@ -798,6 +798,7 @@ mod tests { compaction_batch_size: 8, compaction_max_source_fragments: 32, index_after_compaction: false, + key_index_type: Default::default(), index_before_merge: false, compaction_max_bytes_per_file: 1024 * 1024 * 1024, merge_wal_interval_secs: 0, diff --git a/crates/lance-context-master/src/scheduler.rs b/crates/lance-context-master/src/scheduler.rs index 0650184..5c9a812 100644 --- a/crates/lance-context-master/src/scheduler.rs +++ b/crates/lance-context-master/src/scheduler.rs @@ -375,10 +375,13 @@ async fn index_id_inner(state: &Arc, name: &str) -> Result GenericStoreOptions { GenericStoreOptions { session: self.rollout_session.clone(), + key_index_type: self.config.key_index_type, ..Default::default() } } @@ -245,6 +246,7 @@ impl MasterState { pub(crate) fn rollout_store_options(&self) -> RolloutStoreOptions { RolloutStoreOptions { session: self.rollout_session.clone(), + key_index_type: self.config.key_index_type, // The master observes and merges shards *because* they are behind; // the worker-side read cap must never hide those from it. pending_generations_max: Some(0), @@ -313,6 +315,7 @@ mod tests { compaction_batch_size: 8, compaction_max_source_fragments: 32, index_after_compaction: false, + key_index_type: Default::default(), index_before_merge: false, compaction_max_bytes_per_file: 1024 * 1024 * 1024, merge_wal_interval_secs: 0, diff --git a/crates/lance-context-master/src/task_store.rs b/crates/lance-context-master/src/task_store.rs index 3fee22a..f42265f 100644 --- a/crates/lance-context-master/src/task_store.rs +++ b/crates/lance-context-master/src/task_store.rs @@ -1525,6 +1525,7 @@ mod tests { compaction_batch_size: 8, compaction_max_source_fragments: 32, index_after_compaction: false, + key_index_type: Default::default(), index_before_merge: false, compaction_max_bytes_per_file: 1024 * 1024 * 1024, merge_wal_interval_secs: 0, diff --git a/crates/lance-context-server/src/config.rs b/crates/lance-context-server/src/config.rs index d9d15f3..0fddc20 100644 --- a/crates/lance-context-server/src/config.rs +++ b/crates/lance-context-server/src/config.rs @@ -65,6 +65,16 @@ pub struct ServerConfig { #[arg(long, env = "ROLLOUT_MERGE_MAX_BYTES", default_value = "1073741824")] pub rollout_merge_max_bytes: usize, + /// Base key index policy. Configure every master and worker consistently. + /// ZoneMap uses bounded key-only predicate deletes during WAL merge. + #[arg( + long, + env = "ROLLOUT_KEY_INDEX_TYPE", + value_enum, + default_value = "btree" + )] + pub rollout_key_index_type: lance_context_core::KeyIndexType, + /// Warn when one MemWAL shard has at least this many flushed generations /// pending merge, sampled on every LSM read of any store kind. Every /// pending generation is a separate dataset a read must open, so this is @@ -244,6 +254,21 @@ impl ServerConfig { #[cfg(test)] mod tests { + #[test] + fn key_index_policy_is_explicit_and_validated() { + use clap::Parser; + use lance_context_core::KeyIndexType; + let config = + super::ServerConfig::try_parse_from(["server", "--rollout-key-index-type", "zonemap"]) + .unwrap(); + assert_eq!(config.rollout_key_index_type, KeyIndexType::Zonemap); + assert!(super::ServerConfig::try_parse_from([ + "server", + "--rollout-key-index-type", + "unknown", + ]) + .is_err()); + } use super::*; use clap::CommandFactory; diff --git a/crates/lance-context-server/src/routes/datagen.rs b/crates/lance-context-server/src/routes/datagen.rs index c714ebf..c1f5308 100644 --- a/crates/lance-context-server/src/routes/datagen.rs +++ b/crates/lance-context-server/src/routes/datagen.rs @@ -46,6 +46,7 @@ pub async fn create_datagen_store( merge_after_generations: None, merge_max_generations: Some(state.rollout_merge_max_generations), merge_max_bytes: Some(state.rollout_merge_max_bytes), + key_index_type: Default::default(), pending_generations_warn: Some(state.rollout_wal_pending_warn_generations), pending_generations_max: Some(state.rollout_wal_pending_max_generations), merge_budget: state.merge_budget.clone(), diff --git a/crates/lance-context-server/src/routes/generic.rs b/crates/lance-context-server/src/routes/generic.rs index ce4f6ec..d9bb7ba 100644 --- a/crates/lance-context-server/src/routes/generic.rs +++ b/crates/lance-context-server/src/routes/generic.rs @@ -57,6 +57,7 @@ pub async fn create_generic_store( merge_after_generations: None, merge_max_generations: Some(state.rollout_merge_max_generations), merge_max_bytes: Some(state.rollout_merge_max_bytes), + key_index_type: Default::default(), pending_generations_warn: Some(state.rollout_wal_pending_warn_generations), pending_generations_max: Some(state.rollout_wal_pending_max_generations), merge_budget: state.merge_budget.clone(), diff --git a/crates/lance-context-server/src/routes/rollouts.rs b/crates/lance-context-server/src/routes/rollouts.rs index d8777eb..2b39f7e 100644 --- a/crates/lance-context-server/src/routes/rollouts.rs +++ b/crates/lance-context-server/src/routes/rollouts.rs @@ -197,6 +197,7 @@ pub async fn create_rollout_store( .then_some(state.rollout_merge_after_generations), merge_max_generations: Some(state.rollout_merge_max_generations), merge_max_bytes: Some(state.rollout_merge_max_bytes), + key_index_type: Default::default(), pending_generations_warn: Some(state.rollout_wal_pending_warn_generations), pending_generations_max: Some(state.rollout_wal_pending_max_generations), merge_budget: state.merge_budget.clone(), diff --git a/crates/lance-context-server/src/state.rs b/crates/lance-context-server/src/state.rs index b17f502..00d5984 100644 --- a/crates/lance-context-server/src/state.rs +++ b/crates/lance-context-server/src/state.rs @@ -123,6 +123,7 @@ pub struct AppState { pub rollout_merge_max_generations: usize, /// Buffered Arrow array byte budget per merge pass; `0` disables this cap. pub rollout_merge_max_bytes: usize, + pub rollout_key_index_type: lance_context_core::KeyIndexType, /// Per-shard pending-generation count at which reads warn; `0` disables. pub rollout_wal_pending_warn_generations: usize, pub rollout_wal_pending_max_generations: usize, @@ -380,6 +381,7 @@ impl AppState { rollout_merge_after_generations: config.rollout_merge_after_generations, rollout_merge_max_generations: config.rollout_merge_max_generations, rollout_merge_max_bytes: config.rollout_merge_max_bytes, + rollout_key_index_type: config.rollout_key_index_type, rollout_wal_pending_warn_generations: config.rollout_wal_pending_warn_generations, rollout_wal_pending_max_generations: config.rollout_wal_pending_max_generations, merge_budget: (config.rollout_merge_memory_bytes > 0) @@ -465,6 +467,7 @@ impl AppState { rollout_merge_after_generations: 0, rollout_merge_max_generations: 8, rollout_merge_max_bytes: 1024 * 1024 * 1024, + rollout_key_index_type: Default::default(), rollout_wal_pending_warn_generations: 256, rollout_wal_pending_max_generations: 0, merge_budget: None, @@ -504,6 +507,7 @@ impl AppState { .then_some(self.rollout_merge_after_generations), merge_max_generations: Some(self.rollout_merge_max_generations), merge_max_bytes: Some(self.rollout_merge_max_bytes), + key_index_type: self.rollout_key_index_type, pending_generations_warn: Some(self.rollout_wal_pending_warn_generations), pending_generations_max: Some(self.rollout_wal_pending_max_generations), merge_budget: self.merge_budget.clone(), @@ -691,6 +695,7 @@ impl AppState { .then_some(self.rollout_merge_after_generations), merge_max_generations: Some(self.rollout_merge_max_generations), merge_max_bytes: Some(self.rollout_merge_max_bytes), + key_index_type: self.rollout_key_index_type, pending_generations_warn: Some(self.rollout_wal_pending_warn_generations), pending_generations_max: Some(self.rollout_wal_pending_max_generations), merge_budget: self.merge_budget.clone(), @@ -832,6 +837,7 @@ impl AppState { .then_some(self.rollout_merge_after_generations), merge_max_generations: Some(self.rollout_merge_max_generations), merge_max_bytes: Some(self.rollout_merge_max_bytes), + key_index_type: self.rollout_key_index_type, pending_generations_warn: Some(self.rollout_wal_pending_warn_generations), pending_generations_max: Some(self.rollout_wal_pending_max_generations), merge_budget: self.merge_budget.clone(), diff --git a/crates/lance-context/src/unified_datagen.rs b/crates/lance-context/src/unified_datagen.rs index 6fd6f31..4237962 100644 --- a/crates/lance-context/src/unified_datagen.rs +++ b/crates/lance-context/src/unified_datagen.rs @@ -38,6 +38,7 @@ impl DatagenStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, diff --git a/crates/lance-context/src/unified_generic.rs b/crates/lance-context/src/unified_generic.rs index 3af2b5a..8c653ec 100644 --- a/crates/lance-context/src/unified_generic.rs +++ b/crates/lance-context/src/unified_generic.rs @@ -41,6 +41,7 @@ impl GenericStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, @@ -65,6 +66,7 @@ impl GenericStore { merge_after_generations: None, merge_max_generations: None, merge_max_bytes: None, + key_index_type: Default::default(), pending_generations_warn: None, pending_generations_max: None, merge_budget: None, diff --git a/docs/benchmarks/merge-index-policy.md b/docs/benchmarks/merge-index-policy.md new file mode 100644 index 0000000..605c495 --- /dev/null +++ b/docs/benchmarks/merge-index-policy.md @@ -0,0 +1,106 @@ +# Configurable WAL merge key index + +`ROLLOUT_KEY_INDEX_TYPE=btree|zonemap` selects the base-table key index for +rollout, generic and datagen stores opened by workers. Masters use the same +setting for scheduled `IndexId` and propagate it to native catch-up Jobs. +The default remains `btree`. Core callers can set `key_index_type` in their +store options. This does not change ContextStore's separate ID-index setting. + +Use the same setting on every writer/maintenance process for a table. An +existing `id_idx` of a different type is replaced during merge/index +maintenance, not just by opening the table. Roll out consistent configuration +before admitting maintenance with the new policy; otherwise mixed policies +can repeatedly replace the index. Selecting ZoneMap does not remove other +user-created indexes. + +## Why index selection also changes the delete path + +With Lance 9.0.0, `merge_insert` requires an exact-answer index for its indexed +lookup path. ZoneMap does not qualify. A key-only source with +`WhenMatched::Delete` still produces a target scan projecting payload columns +on its non-indexed path. Both a Rust execution-plan probe and an Azure storage +benchmark reproduced this. Changing only `IndexType::BTree` to `ZoneMap` +therefore reintroduces payload scan amplification. + +The ZoneMap policy uses `DeleteBuilder::from_expr` with typed ID literals, +allowing the predicate scanner to prune zones and read only keys. Expressions +are bounded to 1,024 IDs per delete; every delete commits before the prepared +rows are appended. The original WAL remains until the append and manifest +drain succeed. A retry deletes the same keys and appends their newest values +again. Existing table ownership, commit fencing, WAL byte limits and the shared +merge reservation are unchanged. Many small keys can require multiple delete +commits; this tradeoff must be measured for the intended workload. + +BTree keeps the existing merge path and index-extension behavior. An `index` +phase metric now separately measures index preparation inside the encompassing +`append` phase; do not sum nested phase durations as independent work. + +## Reproduction + +Use a new isolated prefix for every invocation; the script writes new datasets +and must never point at a production table. Azure authentication uses the +normal Lance environment/workload identity. No production records are copied. + +``` +BENCH_ROOT=abfss://CONTAINER@ACCOUNT.dfs.core.windows.net/index-study \ +BENCH_RUN=run1 BENCH_CASE=large_payload_ordered \ +BENCH_KIND=zonemap_predicate BENCH_BATCH_ROWS=1024 \ +python docs/benchmarks/merge_index.py +``` + +Compare `BENCH_KIND=btree` (existing merge) with `zonemap_predicate` (new path). +`zonemap` and `none` reproduce the old non-indexed join as controls. Each +invocation runs in a fresh process, generates deterministic synthetic data, +measures Lance read bytes and I/O counts plus sampled process RSS, and tests +0%, 50% and 100% duplicate IDs. It then retries the last batch and verifies all +IDs/revisions plus one payload. It does not simulate a mid-commit process kill. +Use 128 rows for `many_ids_random` / `many_ids_ordered` (262,144 base rows, +256-byte payloads), and 1,024 for `large_payload_random` / +`large_payload_ordered` (4,096 base rows, 64-KiB payloads; 64-MiB merge batches). + +These measure index/delete/append costs on real object storage, not complete +WAL read/dedup/drain or production table throughput. Row distributions and +sizes are synthetic, not a sample of current production traffic. Small-table +results do not establish a fleet-wide default. Report storage failures and +exclude incomplete fixtures from successful-run aggregates. + +## Azure results, 2026-10-03 + +Lance 9.0.0; one isolated AMD64 Pod, 4 CPU limit / 8 GiB memory limit; +sequential fresh processes for each fixture. The first four rows below each +contain six batches (two fixtures, each with 0%, 50%, 100% overlap); the +fragmented row contains three batches. Numbers cover index maintenance + +delete + append, and exclude initial index creation and base-table loading. + +| Synthetic base / batch | BTree mean / median seconds | ZoneMap predicate mean / median seconds | BTree / ZoneMap mean read MiB | +|---|---:|---:|---:| +| 262k unordered IDs, 8 fragments / 128 rows | 1.213 / 1.309 | 0.687 / 0.681 | 4.930 / 2.292 | +| 262k ordered IDs, 8 fragments / 128 rows | 1.031 / 1.065 | 0.707 / 0.720 | 2.892 / 0.143 | +| 4k unordered IDs, 256 MiB payload, 8 fragments / 64 MiB | 1.128 / 1.116 | 1.011 / 1.051 | 0.303 / 0.460 | +| 4k ordered IDs, 256 MiB payload, 8 fragments / 64 MiB | 2.344 / 1.073 | 1.071 / 1.040 | 0.303 / 0.139 | +| 262k unordered IDs, 2,048 fragments / 128 rows | 1.102 / 1.266 | 1.354 / 1.681 | 5.208 / 57.068 | + +The large-payload ordered BTree mean includes a successful slow append; its +median shows why the mean alone must not be used to claim a 2x improvement. +In the fragmented case, predicate projection avoids materializing payloads, +but small object reads/prefetch and weak zone pruning still amplify physical +I/O. ZoneMap is not uniformly faster. Sampled peak process RSS was 278–625 MiB +for the new ZoneMap path and 343–619 MiB for BTree across these fixtures. + +A separate control using ZoneMap with the original `merge_insert` path read +263.27 MiB on average per delete against the 256 MiB payload fixture. The new +predicate path read 0.06 MiB per delete in the corresponding 128-row control. +Index selection without the delete-path change is therefore insufficient. + +Every completed fixture passed ID/revision conservation, same-batch retry, +and payload spot checks; the Pod recorded zero OOM events. One ordered +large-payload ZoneMap fixture failed on a 30-second Azure manifest PUT timeout +during retry append. It is excluded from successful aggregates and recorded +in `merge-index-results.json`; a new-prefix rerun completed. Raw logs are +retained in the task's `/tmp/lc-index-study` directory. + +These results support an opt-in ZoneMap policy for suitable layouts, while +retaining BTree as the default. A workload with thousands of overlapping +fragments needs further compaction/layout work or its own representative +benchmark before switching. To reproduce the fragmented case, use +`BENCH_CASE=fragmented BENCH_FRAGMENT_ROWS=128 BENCH_BATCH_ROWS=128`. diff --git a/docs/benchmarks/merge-index-results.json b/docs/benchmarks/merge-index-results.json new file mode 100644 index 0000000..2d61ce8 --- /dev/null +++ b/docs/benchmarks/merge-index-results.json @@ -0,0 +1,133 @@ +{ + "measurement": "index maintenance + key deletion + append; excludes WAL prepare/drain and initial index build", + "samples": [ + { + "case": "many_ids_random", + "kind": "btree", + "n": 6, + "mean_seconds": 1.212600218015723, + "median_seconds": 1.3087847646092996, + "min_seconds": 0.6173404389992356, + "max_seconds": 1.6881290629971772, + "mean_read_mib": 4.930051803588867, + "mean_iops": 347, + "max_rss_mib": 345.3046875 + }, + { + "case": "many_ids_random", + "kind": "zonemap_predicate", + "n": 6, + "mean_seconds": 0.6871739939088002, + "median_seconds": 0.6813553952379152, + "min_seconds": 0.4439579779282212, + "max_seconds": 0.9270498780533671, + "mean_read_mib": 2.291510581970215, + "mean_iops": 43.333333333333336, + "max_rss_mib": 277.54296875 + }, + { + "case": "many_ids_ordered", + "kind": "btree", + "n": 6, + "mean_seconds": 1.031405057486457, + "median_seconds": 1.0647503705695271, + "min_seconds": 0.35998480627313256, + "max_seconds": 1.5736293338704854, + "mean_read_mib": 2.891977310180664, + "mean_iops": 281, + "max_rss_mib": 342.578125 + }, + { + "case": "many_ids_ordered", + "kind": "zonemap_predicate", + "n": 6, + "mean_seconds": 0.7070462131717553, + "median_seconds": 0.7199272835860029, + "min_seconds": 0.4173455690033734, + "max_seconds": 1.0748638901859522, + "mean_read_mib": 0.14267381032307944, + "mean_iops": 29.333333333333332, + "max_rss_mib": 277.7578125 + }, + { + "case": "large_payload_random", + "kind": "btree", + "n": 6, + "mean_seconds": 1.1275531464489177, + "median_seconds": 1.1163200367009267, + "min_seconds": 0.9487479431554675, + "max_seconds": 1.4106423910707235, + "mean_read_mib": 0.3032808303833008, + "mean_iops": 28, + "max_rss_mib": 588.91015625 + }, + { + "case": "large_payload_random", + "kind": "zonemap_predicate", + "n": 6, + "mean_seconds": 1.0108064976520836, + "median_seconds": 1.0509298224933445, + "min_seconds": 0.6639390881173313, + "max_seconds": 1.3364728998858482, + "mean_read_mib": 0.4599571228027344, + "mean_iops": 42.333333333333336, + "max_rss_mib": 605.48828125 + }, + { + "case": "large_payload_ordered", + "kind": "btree", + "n": 6, + "mean_seconds": 2.3442257062221565, + "median_seconds": 1.0725305575178936, + "min_seconds": 0.6588206051383168, + "max_seconds": 7.812396941473708, + "mean_read_mib": 0.30266284942626953, + "mean_iops": 28, + "max_rss_mib": 619.015625 + }, + { + "case": "large_payload_ordered", + "kind": "zonemap_predicate", + "n": 6, + "mean_seconds": 1.0707172531401739, + "median_seconds": 1.0403590565547347, + "min_seconds": 0.8739815538283437, + "max_seconds": 1.2820485748816282, + "mean_read_mib": 0.13915506998697916, + "mean_iops": 30.333333333333332, + "max_rss_mib": 625.203125 + }, + { + "case": "fragmented", + "kind": "btree", + "n": 3, + "mean_seconds": 1.1019850313508261, + "median_seconds": 1.2661343852523714, + "min_seconds": 0.6936106679495424, + "max_seconds": 1.3462100408505648, + "mean_read_mib": 5.207733154296875, + "mean_iops": 347, + "max_rss_mib": 372.73046875 + }, + { + "case": "fragmented", + "kind": "zonemap_predicate", + "n": 3, + "mean_seconds": 1.354135996196419, + "median_seconds": 1.6811805807519704, + "min_seconds": 0.5980637588072568, + "max_seconds": 1.7831636490300298, + "mean_read_mib": 57.06763903299967, + "mean_iops": 4123.333333333333, + "max_rss_mib": 383.02734375 + } + ], + "failures": [ + { + "case": "large_payload_ordered", + "kind": "zonemap_predicate", + "stage": "retry_append", + "error": "Azure manifest PUT exceeded 30s; incomplete fixture excluded; rerun under fresh prefix succeeded" + } + ] +} diff --git a/docs/benchmarks/merge_index.py b/docs/benchmarks/merge_index.py new file mode 100644 index 0000000..3ce95ee --- /dev/null +++ b/docs/benchmarks/merge_index.py @@ -0,0 +1,95 @@ +"""Manual isolated object-store comparison; requires pylance==9.0.0. + +See merge-index-policy.md for scope, configuration and known limitations. +""" +import os, sys, time, json, threading, gc, uuid +os.environ.update(LANCE_CPU_THREADS='2', LANCE_MEM_POOL_SIZE=str(512 * 2 ** 20), TOKIO_WORKER_THREADS='2') +import lance, pyarrow as pa, numpy as np, psutil +ROOT = os.environ['BENCH_ROOT'].rstrip('/') + '/' + os.environ.get('BENCH_RUN', 'run1') +if 'index-study' not in ROOT: + raise ValueError('Use an isolated benchmark prefix containing index-study') +proc = psutil.Process() + +def emit(**v): + print(json.dumps(v), flush=True) + +def measured(label, fn, **meta): + peak = [proc.memory_info().rss] + stop = threading.Event() + + def watch(): + while not stop.wait(0.02): + peak[0] = max(peak[0], proc.memory_info().rss) + th = threading.Thread(target=watch) + th.start() + t = time.monotonic() + b = lance.bytes_read_counter() + i = lance.iops_counter() + try: + r = fn() + finally: + stop.set() + th.join() + emit(event='measurement', label=label, seconds=time.monotonic() - t, bytes_read=lance.bytes_read_counter() - b, iops=lance.iops_counter() - i, peak_rss=peak[0], **meta) + return r +schema = pa.schema([pa.field('id', pa.string()), pa.field('payload', pa.binary()), pa.field('revision', pa.int32())]) + +def batch(ids, blob_size, revision, seed): + rng = np.random.default_rng(seed) + data = rng.integers(0, 256, size=(len(ids), blob_size), dtype=np.uint8) + data[:, :4] = np.frombuffer(revision.to_bytes(4, 'little'), dtype=np.uint8) + return pa.Table.from_arrays([pa.array(ids), pa.array([row.tobytes() for row in data]), pa.array([revision] * len(ids), type=pa.int32())], schema=schema) + +def run(case, rows, blob_size, ordered, kind): + meta = dict(case=case, kind=kind, rows=rows, blob_size=blob_size, ordered=ordered) + uri = ROOT + '/' + case + '-' + kind + emit(event='start', uri=uri, **meta) + ids = [f'{i:032x}' for i in range(rows)] + if not ordered: + np.random.default_rng(42).shuffle(ids) + chunk = max(1, min(8192, 8 * 2 ** 20 // blob_size)) + + def batches(): + for off in range(0, rows, chunk): + yield batch(ids[off:off + chunk], blob_size, 0, off).to_batches()[0] + ds = measured('base_write', lambda: lance.write_dataset(pa.RecordBatchReader.from_batches(schema, batches()), uri, max_rows_per_file=int(os.environ.get('BENCH_FRAGMENT_ROWS', max(chunk, rows // 8)))), **meta) + if kind != 'none': + measured('index_build', lambda: ds.create_scalar_index('id', kind.split('_')[0].upper(), name='id_idx'), **meta) + emit(event='indices', indices=str(ds.list_indices()), **meta) + expected = {ident: 0 for ident in ids} + for round_no, overlap in enumerate([0, 0.5, 1]): + n = int(os.environ.get('BENCH_BATCH_ROWS', '128')) + old = int(n * overlap) + new_ids = [f'{rows + round_no * n + i:032x}' for i in range(n - old)] + update_ids = ids[round_no * old:round_no * old + old] + new_ids + source = batch(update_ids, blob_size, round_no + 1, 100000 + round_no) + m = dict(**meta, round=round_no, overlap=overlap, source_bytes=source.nbytes) + if kind != 'none': + measured('index_maintain', lambda: ds.optimize.optimize_indices(num_indices_to_merge=1, index_names=['id_idx']), **m) + builder = ds.merge_insert('id').when_matched_delete() + if round_no == 0: + try: + emit(event='plan', plan=builder.explain_plan(schema=source.select(['id']).schema), **m) + except Exception as e: + emit(event='plan_unavailable', error=str(e), **m) + measured('delete', lambda: ds.delete('id IN (' + ','.join(("'" + x + "'" for x in update_ids)) + ')') if kind.endswith('_predicate') else builder.execute(source.select(['id'])), **m) + ds = measured('append', lambda: lance.write_dataset(source, uri, mode='append'), **m) + expected.update({ident: round_no + 1 for ident in update_ids}) + if kind != 'none': + measured('retry_index', lambda: ds.optimize.optimize_indices(num_indices_to_merge=1, index_names=['id_idx']), **meta) + measured('retry_delete', lambda: ds.delete('id IN (' + ','.join(("'" + x + "'" for x in update_ids)) + ')') if kind.endswith('_predicate') else ds.merge_insert('id').when_matched_delete().execute(source.select(['id'])), **meta) + ds = measured('retry_append', lambda: lance.write_dataset(source, uri, mode='append'), **meta) + actual = ds.to_table(columns=['id', 'revision']).to_pydict() + assert len(actual['id']) == len(expected), (len(actual['id']), len(expected)) + assert dict(zip(actual['id'], actual['revision'])) == expected + ident = update_ids[0] + got = ds.to_table(columns=['payload'], filter=f"id = '{ident}'").column(0).to_pylist() + want = source.filter(pa.compute.equal(source['id'], ident))['payload'].to_pylist() + assert got == want + emit(event='verified', expected_rows=len(expected), fragments=len(ds.get_fragments()), **meta) + gc.collect() +emit(event='environment', lance=lance.__version__, python=sys.version, root=ROOT) +cases = {'fragmented': (262144, 256, False), 'many_ids_random': (262144, 256, False), 'many_ids_ordered': (262144, 256, True), 'large_payload_random': (4096, 65536, False), 'large_payload_ordered': (4096, 65536, True)} +case = os.environ['BENCH_CASE'] +run(case, *cases[case], os.environ['BENCH_KIND']) +emit(event='complete', memory_events=open('/sys/fs/cgroup/memory.events').read()) From 6ebe7b2b62c94099dd6db73698aae79c2b8d8f2e Mon Sep 17 00:00:00 2001 From: Beinan Wang <> Date: Sat, 3 Oct 2026 22:51:30 +0000 Subject: [PATCH 2/2] docs: distinguish predicate scan plans in the merge benchmark --- docs/benchmarks/merge_index.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/docs/benchmarks/merge_index.py b/docs/benchmarks/merge_index.py index 3ce95ee..7a679d0 100644 --- a/docs/benchmarks/merge_index.py +++ b/docs/benchmarks/merge_index.py @@ -69,7 +69,11 @@ def batches(): builder = ds.merge_insert('id').when_matched_delete() if round_no == 0: try: - emit(event='plan', plan=builder.explain_plan(schema=source.select(['id']).schema), **m) + if kind.endswith('_predicate'): + predicate = 'id IN (' + ','.join("'" + x + "'" for x in update_ids) + ')' + emit(event='predicate_scan_plan', plan=ds.scanner(columns=['id'], filter=predicate).explain_plan(), **m) + else: + emit(event='merge_plan', plan=builder.explain_plan(schema=source.select(['id']).schema), **m) except Exception as e: emit(event='plan_unavailable', error=str(e), **m) measured('delete', lambda: ds.delete('id IN (' + ','.join(("'" + x + "'" for x in update_ids)) + ')') if kind.endswith('_predicate') else builder.execute(source.select(['id'])), **m)