diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs index 712a5897..669948d0 100644 --- a/crates/lance-context-core/src/rollout_store.rs +++ b/crates/lance-context-core/src/rollout_store.rs @@ -4441,12 +4441,10 @@ mod tests { }); } - /// A merge into a base table that has rows but no id BTree builds the - /// BTree first: without it the delete-only `merge_insert` is a hash - /// join over the whole base table. The very first merge (empty base) - /// has nothing to join and builds nothing. + /// Merging does not publish table-wide indexes, even on a populated base. + /// Predicate deletion scans keys only when no index is available. #[test] - fn merge_builds_the_id_btree_when_the_base_has_rows_but_no_index() { + fn merge_does_not_build_an_id_index_on_the_write_path() { let dir = TempDir::new().unwrap(); let uri = dir.path().to_string_lossy().to_string(); let runtime = tokio::runtime::Runtime::new().unwrap(); @@ -4465,25 +4463,17 @@ mod tests { store.flush().await.unwrap(); store.cleanup_own_shard().await.unwrap(); assert!( - store.has_id_btree_index().await.unwrap(), - "second merge joins against a populated base and builds the BTree first" + !store.has_id_btree_index().await.unwrap(), + "merge must not race other shards to create a table-wide index" ); assert_eq!(store.list(None, None).await.unwrap().len(), 2); }); } - /// The WAL merge is a delete-only `merge_insert` on the key followed by - /// an append: an upsert would take the matched target rows with every - /// column (multi-megabyte inline blobs included) to rewrite them. - /// Observable shape: an overwritten key leaves the old fragment in - /// place with a deletion vector, and the new row lands in a fresh - /// fragment; last-write-wins still holds. - /// A BTree that exists but covers only some fragments leaves the rest to - /// a full scan in the delete-only merge_insert. Each merge extends it - /// over whatever the previous merges and compactions appended, so no - /// fragment is ever scanned twice. + /// Partial index coverage remains correct without extending the index in + /// each shard's write path. Explicit maintenance still extends coverage. #[test] - fn merge_extends_a_stale_id_btree_over_new_fragments() { + fn merge_uses_partial_btree_without_publishing_index_updates() { use lance::index::DatasetIndexInternalExt as _; let dir = TempDir::new().unwrap(); let uri = dir.path().to_string_lossy().to_string(); @@ -4493,8 +4483,14 @@ mod tests { store.add(&[assistant_record("a-0")]).await.unwrap(); store.flush().await.unwrap(); store.cleanup_own_shard().await.unwrap(); - // Three more merges, each appending a fragment the index (built by - // the first merge) does not cover. + store.create_id_btree_index().await.unwrap(); + let indices = store.base.dataset.load_indices().await.unwrap(); + let original = indices + .iter() + .find(|i| i.name == ROLLOUT_ID_INDEX_NAME) + .unwrap() + .uuid; + // Three more merges append fragments outside existing coverage. for id in ["a-1", "a-2", "a-3"] { store.add(&[assistant_record(id)]).await.unwrap(); store.flush().await.unwrap(); @@ -4507,12 +4503,24 @@ mod tests { .await .unwrap() .len(); - // The merge that appended the last fragment extended the index - // before its own write, so at most that one fragment is uncovered. - assert!( - unindexed <= 1, - "unindexed fragments after merges: {unindexed}" + assert_eq!(unindexed, 3); + let indices = store.base.dataset.load_indices().await.unwrap(); + assert_eq!( + indices + .iter() + .find(|i| i.name == ROLLOUT_ID_INDEX_NAME) + .unwrap() + .uuid, + original ); + store.extend_id_btree_index().await.unwrap(); + assert!(store + .base + .dataset + .unindexed_fragments(ROLLOUT_ID_INDEX_NAME) + .await + .unwrap() + .is_empty()); assert_eq!(store.list(None, None).await.unwrap().len(), 4); }); } @@ -4772,6 +4780,9 @@ mod tests { } 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)); + store.create_id_key_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() @@ -4790,11 +4801,119 @@ mod tests { .unwrap(); store.flush().await.unwrap(); store.cleanup_own_shard().await.unwrap(); - assert!(store.has_id_btree_index().await.unwrap()); + assert!(!store.has_id_btree_index().await.unwrap()); assert_eq!(store.list(None, None).await.unwrap().len(), ids.len()); + store.create_id_key_index().await.unwrap(); + assert!(store.has_id_btree_index().await.unwrap()); }); } + #[test] + fn parallel_shard_merges_preserve_partial_indexes_and_full_payloads() { + use crate::KeyIndexType; + use lance::index::DatasetIndexExt; + tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap() + .block_on(async { + for kind in [KeyIndexType::Btree, KeyIndexType::Zonemap] { + let dir = TempDir::new().unwrap(); + let uri = dir.path().to_string_lossy().to_string(); + let mut seed = RolloutStore::open_with_options( + &uri, + RolloutStoreOptions { + shard_id: Some("seed".into()), + key_index_type: kind, + ..Default::default() + }, + ) + .await + .unwrap(); + let records: Vec<_> = (0..8) + .map(|i| artifact_record(&format!("old-{i}"), &[1; 65536])) + .collect(); + seed.add(&records).await.unwrap(); + seed.flush().await.unwrap(); + seed.cleanup_own_shard().await.unwrap(); + seed.create_id_key_index().await.unwrap(); + let indices = seed.base.dataset.load_indices().await.unwrap(); + let index = indices + .iter() + .find(|i| i.name == ROLLOUT_ID_INDEX_NAME) + .unwrap() + .uuid; + // Half the existing rows are outside index coverage. The + // concurrent updates must delete those rows too, using key + // scans without extending the shared index. + let uncovered: Vec<_> = (8..16) + .map(|i| artifact_record(&format!("old-{i}"), &[1; 65536])) + .collect(); + seed.add(&uncovered).await.unwrap(); + seed.flush().await.unwrap(); + seed.cleanup_own_shard().await.unwrap(); + let mut shards = Vec::new(); + for shard in 0..4 { + let writer = RolloutStore::open_with_options( + &uri, + RolloutStoreOptions { + shard_id: Some(format!("shard-{shard}")), + key_index_type: kind, + ..Default::default() + }, + ) + .await + .unwrap(); + let records: Vec<_> = (shard * 4..shard * 4 + 4) + .flat_map(|i| { + [ + artifact_record(&format!("old-{i}"), &[2; 65536]), + artifact_record(&format!("new-{i}"), &[2; 65536]), + ] + }) + .collect(); + writer.add(&records).await.unwrap(); + writer.flush().await.unwrap(); + shards.push(writer); + } + let results = futures::future::join_all( + shards + .iter_mut() + .map(|writer| async move { writer.cleanup_own_shard().await }), + ) + .await; + for result in results { + assert_eq!(result.unwrap(), 1); + } + // Replay is empty and cannot duplicate data or mutate the index. + for writer in &mut shards { + assert_eq!(writer.cleanup_own_shard().await.unwrap(), 0); + } + let fresh = RolloutStore::open(&uri).await.unwrap(); + assert_eq!(fresh.base.dataset.count_rows(None).await.unwrap(), 32); + let rows = fresh.list(None, None).await.unwrap(); + assert_eq!(rows.len(), 32); + for prefix in ["old", "new"] { + for i in 0..16 { + let id = format!("{prefix}-{i}"); + assert!(rows.iter().any(|row| row.id == id)); + assert_eq!(fresh.get_blob(&id).await.unwrap().unwrap(), vec![2; 65536]); + } + } + let indices = fresh.base.dataset.load_indices().await.unwrap(); + assert_eq!( + indices + .iter() + .find(|i| i.name == ROLLOUT_ID_INDEX_NAME) + .unwrap() + .uuid, + index + ); + } + }); + } + #[test] fn merge_insert_probes_the_id_btree_instead_of_scanning() { use lance::dataset::MergeInsertBuilder; @@ -4881,9 +5000,7 @@ mod tests { store.create_id_btree_index().await.unwrap(); assert_eq!(store.extend_id_btree_index().await.unwrap(), 0); - // Two merges land two fragments. Each merge extends the index over - // what was uncovered *before* its own append, so only the last - // appended fragment is left for the explicit call here. + // Both merged fragments remain uncovered until explicit maintenance. for id in ["a-1", "a-2"] { store.add(&[assistant_record(id)]).await.unwrap(); store.flush().await.unwrap(); @@ -4899,9 +5016,9 @@ mod tests { .len() } }; - assert_eq!(unindexed(&store).await, 1); + assert_eq!(unindexed(&store).await, 2); - assert_eq!(store.extend_id_btree_index().await.unwrap(), 1); + assert_eq!(store.extend_id_btree_index().await.unwrap(), 2); assert_eq!(unindexed(&store).await, 0); assert!(store.has_id_btree_index().await.unwrap()); assert_eq!(store.extend_id_btree_index().await.unwrap(), 0); diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index 8bd2ffa6..0bb82edd 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -54,8 +54,7 @@ use lance::dataset::optimize::{ CompactionOptions, RewriteResult, }; use lance::dataset::{ - builder::DatasetBuilder, Dataset, MergeInsertBuilder, NewColumnTransform, WhenMatched, - WhenNotMatched, WriteMode, WriteParams, + builder::DatasetBuilder, Dataset, NewColumnTransform, WriteMode, WriteParams, }; use lance::index::DatasetIndexExt; use lance::io::{ObjectStoreParams, StorageOptionsAccessor}; @@ -1326,16 +1325,16 @@ impl StorageBase { /// /// `read_flushed_generations` has already reduced the source to one newest /// 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. + /// Bounded predicate deletes project only keys and row IDs, using any + /// existing BTree/ZoneMap coverage without publishing an index update. /// /// Not `WhenMatched::UpdateAll`, deliberately. An upsert takes the matched /// target rows *with every column* to join and rewrite them, and rollout /// rows carry multi-megabyte inline blobs: in production one merge of /// 17k matched rows read 7.6 GB (459 KB/row), one of 1.4k rows read /// 4.0 GB (3 MB/row), and two of those in flight held a worker at its - /// 32 GiB limit. The delete-only merge probes the id index and touches - /// only the key column and deletion vectors; the append streams the new + /// 32 GiB limit. Predicate deletion uses available index coverage and reads + /// only keys and row IDs for uncovered fragments; the append streams the new /// rows straight to fresh fragments. Neither ever holds a target blob. /// /// Last-write-wins and retry idempotence are preserved: a crash between @@ -1349,53 +1348,28 @@ impl StorageBase { merge_schema: Arc, watermark: Option, ) -> LanceResult<()> { - observe_phase!("index", self.ensure_merge_key_index().await)?; + // Predicate deletion projects only the key and row IDs, including for + // unindexed fragments. It can use existing BTree/ZoneMap coverage but + // does not need to publish an index before every shard merge. Doing + // that here made parallel shards compete on CreateIndex transactions + // and repeatedly rebuild the same table-wide index. 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>>()?; - 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?; - } + use datafusion::logical_expr::Expr; + let mut values = Vec::with_capacity(1024); + for batch in &batches { + for row in 0..batch.num_rows() { + values.push(Expr::Literal( + ScalarValue::try_from_array(batch.column(key_index), 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); + } + if !values.is_empty() { + self.delete_key_values(values).await?; } if let Some(watermark) = watermark { @@ -1445,48 +1419,6 @@ impl StorageBase { 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 { @@ -1809,32 +1741,16 @@ impl StorageBase { }) } - /// Build a BTree scalar index on the base table's key column, or report - /// whether one already exists. - /// - /// The key column is the table's (unenforced) primary key. The index does - /// two jobs: - /// - /// - point lookups and range scans on the already-merged base table; - /// - **the MemWAL merge itself**. `merge_insert` joins the WAL rows to the - /// base table on the key. With a scalar index that answers equality - /// exactly it probes only the fragments holding the source keys; without - /// one it reads the *whole* base table into a hash join. On a 2 TB / - /// 5.8M-row table that full join ran for every 64-generation merge and - /// took every worker with it (18/20 OOMKilled together, 2026-09-29). - /// - /// It must be a BTree: Lance's `merge_insert` only takes the indexed path - /// for an index whose plugin `provides_exact_answer()`, and ZoneMap (a - /// per-fragment min/max) does not, so a ZoneMap here changed nothing for - /// the merge. `replace(true)` makes rebuilding idempotent. + /// Build or replace a BTree scalar index on the base table's key column. /// - /// # MemWAL interaction + /// The index accelerates base-table point lookups and range scans. WAL + /// merges use available coverage but do not create or extend indexes; + /// their predicate deletes scan only keys and row IDs when uncovered. + /// Index publication belongs to explicit maintenance, avoiding conflicts + /// between independently merging shards. /// - /// Lance's MemWAL does not maintain this index across WAL flushes (it only - /// keeps the indices named in `maintained_indexes`). That does not affect - /// correctness: rows are de-duplicated by the key column at read time, so - /// the index only ever needs to describe the base table's already-merged - /// fragments, and compaction rebuilds it (see the master's `IndexId`). + /// MemWAL flushes do not extend base-table index coverage. Lance combines + /// indexed lookups with scans of uncovered fragments, preserving correctness. pub async fn create_key_btree_index(&mut self) -> LanceResult<()> { self.ensure_writable()?; info!(column = %self.key_column, "creating BTree index on key column"); @@ -1855,14 +1771,9 @@ impl StorageBase { /// Extend the key column's BTree index over every base-table fragment it /// does not yet cover, appending an index delta rather than rebuilding. /// - /// Lance's `merge_insert` probes the index for the fragments it covers and - /// **scans** every fragment it does not (its plan is a `Union` of the two). - /// Every WAL merge appends fragments the index has never seen, so between - /// rebuilds each merge reads those fragments whole: 6.5 GB for a 2,046-row - /// merge into a 22-fragment table was observed, and that read is what - /// spiked workers to 20-29 GiB after the BTree itself was in place. - /// Calling this before a merge keeps the unindexed set empty, so the scan - /// arm reads nothing. + /// This is explicit index maintenance. WAL merges leave new fragments + /// uncovered until maintenance runs, and query planning scans their key + /// columns rather than rebuilding the index on every merge. /// /// Returns how many fragments were unindexed beforehand (0 = no commit). /// A table without the BTree at all is left alone; the caller builds it diff --git a/docs/benchmarks/merge-index-policy.md b/docs/benchmarks/merge-index-policy.md index 605c4953..fb251e36 100644 --- a/docs/benchmarks/merge-index-policy.md +++ b/docs/benchmarks/merge-index-policy.md @@ -7,13 +7,13 @@ 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 +existing `id_idx` of a different type is replaced during explicit 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 +## Why merges use predicate deletion 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 @@ -22,7 +22,7 @@ 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, +Both BTree and ZoneMap policies use `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 @@ -31,9 +31,11 @@ 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. +WAL merges neither create nor extend the key index. Explicit `IndexId` +maintenance still manages coverage. This removes competing `CreateIndex` +transactions from parallel shard merges; uncovered fragments require key scans, +so measure scan costs and keep scheduled maintenance running. The historical +`index` phase metric is no longer emitted by the WAL merge path. ## Reproduction @@ -48,7 +50,10 @@ 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). +Compare `BENCH_KIND=btree BENCH_MAINTAIN_INDEX=1` (the previous eager-index +merge) with `BENCH_KIND=btree_predicate BENCH_MAINTAIN_INDEX=0` (the current +BTree path). `zonemap_predicate` with maintenance disabled exercises the current +ZoneMap path; set it to 1 to reproduce the earlier ZoneMap measurements below. `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 @@ -104,3 +109,32 @@ 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`. + + +## Deferred index maintenance, Azure results, 2026-10-05 + +One isolated AMD64 Pod with 4 CPU / 8 GiB limits, fresh processes and datasets, +Lance 9.0.0, synthetic unordered IDs. Each row sums three batches (0%, 50%, +100% overlap), excluding initial loading/index creation and the replay check. +All eight fixtures passed complete ID/revision comparison and payload spot +checks after replay; cgroup max/oom/oom_kill counters remained zero. + +| Base / batch | Index and delete policy | Three batches (s) | Read MiB | Peak process MiB | +|---|---|---:|---:|---:| +| 262k IDs / 128 rows | BTree eager index + merge_insert | 3.093 | 14.767 | 339.5 | +| 262k IDs / 128 rows | BTree deferred index + predicate | 2.656 | 6.288 | 308.7 | +| 262k IDs / 128 rows | ZoneMap eager index + predicate | 2.177 | 6.876 | 271.4 | +| 262k IDs / 128 rows | ZoneMap deferred index + predicate | 2.210 | 6.899 | 270.5 | +| 256 MiB payload / 64 MiB | BTree eager index + merge_insert | 3.416 | 0.909 | 585.7 | +| 256 MiB payload / 64 MiB | BTree deferred index + predicate | 2.747 | 0.474 | 653.3 | +| 256 MiB payload / 64 MiB | ZoneMap eager index + predicate | 2.962 | 1.380 | 586.8 | +| 256 MiB payload / 64 MiB | ZoneMap deferred index + predicate | 2.913 | 1.435 | 586.8 | + +BTree elapsed time fell 14% and 20% in this small run; ZoneMap differences +were small. BTree's large-payload peak RSS increased by about 68 MiB. +These serial component measurements do not quantify fleet throughput or +concurrent conflict reduction. The separate Rust regression runs four shard +merges concurrently for both index types, verifies all updated payload bytes, +and checks physical row counts, empty replay and unchanged index identity. +Long-running tables with large uncovered key ranges still need index +maintenance; this run does not bound their scan amplification. diff --git a/docs/benchmarks/merge_index.py b/docs/benchmarks/merge_index.py index 7a679d04..ae34c0e0 100644 --- a/docs/benchmarks/merge_index.py +++ b/docs/benchmarks/merge_index.py @@ -64,7 +64,7 @@ def batches(): 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': + if kind != 'none' and os.environ.get('BENCH_MAINTAIN_INDEX', '1') == '1': 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: @@ -79,7 +79,7 @@ def batches(): 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': + if kind != 'none' and os.environ.get('BENCH_MAINTAIN_INDEX', '1') == '1': 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)