From 3b564534a2478a566a30d0f085819e268793d54a Mon Sep 17 00:00:00 2001 From: aaltshuler Date: Thu, 2 Jul 2026 02:00:15 +0300 Subject: [PATCH] test(engine): red regression for merge-update row-id overlap breaking filtered reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit iss-merge-rowid-overlap-corrupts-filtered-reads / lance#7444: an update-style merge over a merge-written fragment reuses the updated rows' stable row ids (spec-legal overlap); a later delete leaves the overlapping range sparsely tiled and unpatched lance-table 7.0.0's RowIdIndex::new fails every filtered read that builds the id→address map. Two boundaries: - writes.rs::filtered_read_after_merge_update_and_delete_keeps_row_ids_consistent (engine walk: merge-load → same-key merge-load → delete → keyed point lookups), plus the green isolation control filtered_read_after_append_and_delete_is_consistent. - lance_surface_guards.rs::filtered_scan_tolerates_merge_update_row_id_overlap (faithful transcription of lance#7444's minimal repro at the raw Lance API; pins the surface so a future bump that regresses lance#7480 turns red). Both fail on unpatched Lance 7.0.0 with the predicted symptom (rowids/index.rs:50 "Wrong range" via the take path); green arrives with the vendored lance-table patch in the next commit. --- .../omnigraph/tests/lance_surface_guards.rs | 106 ++++++++++++++++ crates/omnigraph/tests/writes.rs | 113 ++++++++++++++++++ 2 files changed, 219 insertions(+) diff --git a/crates/omnigraph/tests/lance_surface_guards.rs b/crates/omnigraph/tests/lance_surface_guards.rs index 4933166e..e56fe222 100644 --- a/crates/omnigraph/tests/lance_surface_guards.rs +++ b/crates/omnigraph/tests/lance_surface_guards.rs @@ -1136,3 +1136,109 @@ async fn camelcase_index_equality_routes_to_scalar_index() { for camelCase index routing. got plan:\n{err:?}" ); } + +// --- Guard: filtered scans tolerate merge_insert's overlapping row-id ranges +// (lance#7444, fixed by lance#7480; consumed via the vendored lance-table +// patch) ------------------------------------------------------------- +// +// An update-style merge_insert over a fragment that was itself merge-written +// reuses the updated rows' stable row ids in its rewritten fragments (row-id +// lineage spec: updates preserve `_rowid`) while the superseded fragment keeps +// its full id sequence plus a deletion vector — legal, overlapping +// cross-fragment id ranges. A later delete leaves the overlap sparsely tiled; +// unpatched lance-table 7.0.0's `RowIdIndex::new` asserted dense tiling and +// failed any filtered read that builds the id→address map: "Wrong range" +// debug assert, "all columns in a record batch must have the same length" (or +// a silently-wrong batch) in release. Faithful transcription of lance#7444's +// minimal repro: merge-seed → merge-update → delete → filter + with_row_id. +// This guard turns red if a Lance bump regresses the fix, or if the vendored +// patch is dropped before the pinned lance-table ships lance#7480. +#[tokio::test] +async fn filtered_scan_tolerates_merge_update_row_id_overlap() { + use futures::TryStreamExt; + + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + + let schema = Arc::new(Schema::new(vec![ + Field::new("slug", DataType::Utf8, false), + Field::new("title", DataType::Utf8, false), + ])); + let mk_batch = |slugs: Vec, titles: Vec| { + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(slugs)) as arrow_array::ArrayRef, + Arc::new(StringArray::from(titles)) as arrow_array::ArrayRef, + ], + ) + .unwrap() + }; + + // Empty dataset WITH stable row ids; both data writes are merge_inserts + // (merge-on-merge is lance#7444's trigger qualifier — a plain + // Dataset::write seed does not reproduce). + let empty = mk_batch(Vec::new(), Vec::new()); + let reader = RecordBatchIterator::new(vec![Ok(empty)], schema.clone()); + let params = WriteParams { + mode: WriteMode::Create, + enable_stable_row_ids: true, + data_storage_version: Some(LanceFileVersion::V2_2), + ..Default::default() + }; + let ds = Dataset::write(reader, uri, Some(params)).await.unwrap(); + + let merge = |ds: Dataset, batch: RecordBatch, schema: Arc| async move { + let job = MergeInsertBuilder::try_new(Arc::new(ds), vec!["slug".to_string()]) + .unwrap() + .when_matched(WhenMatched::UpdateAll) + .when_not_matched(WhenNotMatched::InsertAll) + .try_build() + .unwrap(); + let source = RecordBatchIterator::new(vec![Ok(batch)], schema); + let (ds, _stats) = job.execute_reader(source).await.unwrap(); + (*ds).clone() + }; + + // Merge #1 seeds 40 rows; merge #2 rewrites 15 of them (keeping their + // stable ids — the overlap with the merge-written seed fragment). + let seed = mk_batch( + (1..=40).map(|i| format!("t{i}")).collect(), + (1..=40).map(|i| format!("r{i}")).collect(), + ); + let ds = merge(ds, seed, schema.clone()).await; + let updates = mk_batch( + (1..=15).map(|i| format!("t{i}")).collect(), + (1..=15).map(|i| format!("e{i}")).collect(), + ); + let ds = Arc::new(merge(ds, updates, schema.clone()).await); + + // The delete's deletion vector makes the overlapping id region sparse. + let staged = lance::dataset::DeleteBuilder::new(ds.clone(), "slug = 't20'") + .execute_uncommitted() + .await + .unwrap(); + assert_eq!(staged.num_deleted_rows, 1, "expected exactly t20 deleted"); + let ds = CommitBuilder::new(ds) + .execute(staged.transaction) + .await + .unwrap(); + + // filter + with_row_id forces the RowIdIndex build (a full scan does + // not). On the broken index this errors/panics; on the fixed one every + // live id resolves. + for (slug, expected) in [("t3", 1usize), ("t20", 0usize)] { + let mut scan = ds.scan(); + scan.with_row_id(); + scan.filter(&format!("slug = '{slug}'")).unwrap(); + let batches: Vec = scan + .try_into_stream() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + let rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!(rows, expected, "filtered read for {slug}"); + } +} diff --git a/crates/omnigraph/tests/writes.rs b/crates/omnigraph/tests/writes.rs index 72821fef..08dedf7c 100644 --- a/crates/omnigraph/tests/writes.rs +++ b/crates/omnigraph/tests/writes.rs @@ -1952,3 +1952,116 @@ async fn post_publish_fold_matches_fresh_reopen() { "post-publish fold diverged from a fresh re-scan (folded {folded} vs scanned {scanned})" ); } + +const FIND_PERSON_QUERY: &str = r#" +query find_person($name: String) { + match { $p: Person { name: $name } } + return { $p.name } +} +"#; + +/// Regression: iss-merge-rowid-overlap-corrupts-filtered-reads / lance#7444. +/// +/// An update-style merge (same-key merge load) reuses the updated rows' +/// stable row ids in the rewritten fragments while the superseded fragment +/// keeps its full row-id sequence plus a deletion vector — overlapping +/// cross-fragment id ranges, legal per the Lance row-id-lineage spec. A +/// later delete punches a hole in that overlapping range; on unpatched +/// Lance 7.0.0 `RowIdIndex::new` then fails any filtered scan that needs +/// the id→address map ("all columns in a record batch must have the same +/// length" in release, a "Wrong range" debug assert). Fixed upstream by +/// lance#7480; consumed here via the vendored `lance-table` patch. +#[tokio::test] +async fn filtered_read_after_merge_update_and_delete_keeps_row_ids_consistent() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let mut db = Omnigraph::init(uri, TEST_SCHEMA).await.unwrap(); + + let seed: String = (1..=40) + .map(|i| format!("{{\"type\":\"Person\",\"data\":{{\"name\":\"p{i}\",\"age\":{i}}}}}\n")) + .collect(); + load_jsonl(&mut db, &seed, LoadMode::Merge).await.unwrap(); + + // Same-key updates: Lance Operation::Update rewrites these 15 rows into + // new fragments that keep their original stable row ids (the overlap). + let updates: String = (1..=15) + .map(|i| { + format!( + "{{\"type\":\"Person\",\"data\":{{\"name\":\"p{i}\",\"age\":{}}}}}\n", + 100 + i + ) + }) + .collect(); + load_jsonl(&mut db, &updates, LoadMode::Merge).await.unwrap(); + + // The delete adds a deletion vector, so the overlapping region no longer + // densely tiles its id range — the shape lance#7444 choked on. + mutate_main( + &mut db, + MUTATION_QUERIES, + "remove_person", + &mixed_params(&[("$name", "p20")], &[]), + ) + .await + .unwrap(); + + // Filtered point lookups must still resolve: an updated row, an + // untouched row, and the deleted row (absent), each via the key filter. + for (name, expected) in [("p3", vec!["p3"]), ("p30", vec!["p30"]), ("p20", vec![])] { + let result = query_main( + &mut db, + FIND_PERSON_QUERY, + "find_person", + &mixed_params(&[("$name", name)], &[]), + ) + .await + .unwrap_or_else(|e| panic!("filtered read for {name} failed: {e}")); + let got = first_column_sorted(&result); + assert_eq!(got, expected, "filtered read for {name}"); + } +} + +/// Isolation control for the regression above: the same load/delete/filtered +/// read walk WITHOUT same-key updates (append-only merges, disjoint keys) +/// never produces overlapping row-id ranges and passes on unpatched Lance. +/// If this one fails alongside the merge-update case, the defect is not the +/// lance#7444 overlap shape. +#[tokio::test] +async fn filtered_read_after_append_and_delete_is_consistent() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let mut db = Omnigraph::init(uri, TEST_SCHEMA).await.unwrap(); + + let seed: String = (1..=40) + .map(|i| format!("{{\"type\":\"Person\",\"data\":{{\"name\":\"p{i}\",\"age\":{i}}}}}\n")) + .collect(); + load_jsonl(&mut db, &seed, LoadMode::Merge).await.unwrap(); + + // Disjoint keys: plain inserts, no fragment rewrite, no id reuse. + let more: String = (41..=55) + .map(|i| format!("{{\"type\":\"Person\",\"data\":{{\"name\":\"p{i}\",\"age\":{i}}}}}\n")) + .collect(); + load_jsonl(&mut db, &more, LoadMode::Merge).await.unwrap(); + + mutate_main( + &mut db, + MUTATION_QUERIES, + "remove_person", + &mixed_params(&[("$name", "p20")], &[]), + ) + .await + .unwrap(); + + for (name, expected) in [("p3", vec!["p3"]), ("p50", vec!["p50"]), ("p20", vec![])] { + let result = query_main( + &mut db, + FIND_PERSON_QUERY, + "find_person", + &mixed_params(&[("$name", name)], &[]), + ) + .await + .unwrap_or_else(|e| panic!("filtered read for {name} failed: {e}")); + let got = first_column_sorted(&result); + assert_eq!(got, expected, "filtered read for {name}"); + } +}