From defe7b93f9560a3e5496855a20f9a2b5182d258a Mon Sep 17 00:00:00 2001 From: aaltshuler Date: Sat, 5 Sep 2026 13:41:22 +0300 Subject: [PATCH 1/3] fix(merge): preserve state when source table versions lag target --- crates/omnigraph/src/exec/merge.rs | 34 +- crates/omnigraph/tests/branching.rs | 433 +++++++++++++++++++ crates/omnigraph/tests/failpoints.rs | 266 ++++++++---- crates/omnigraph/tests/merge_fast_forward.rs | 321 ++++++++------ docs/dev/merge.md | 13 +- docs/releases/v0.11.0.md | 8 + docs/user/branching/merge.md | 5 + 7 files changed, 857 insertions(+), 223 deletions(-) diff --git a/crates/omnigraph/src/exec/merge.rs b/crates/omnigraph/src/exec/merge.rs index 45b1fca2b..2f6877e6a 100644 --- a/crates/omnigraph/src/exec/merge.rs +++ b/crates/omnigraph/src/exec/merge.rs @@ -3499,11 +3499,27 @@ fn row_id_at(batch: &RecordBatch, row: usize) -> Result { Ok(ids.value(row).to_string()) } +/// The manifest projects the greatest numeric version for each table identity. +/// Native refs have independent version histories, so adopting an equal or +/// lower source version cannot replace the target's current registration. +/// This selects a publication route; row comparison still determines the delta. +fn adopt_requires_target_lineage( + source_entry: &crate::db::DatasetEntry, + target_entry: Option<&crate::db::DatasetEntry>, +) -> bool { + target_entry.is_some_and(|target| { + source_entry.published_dataset_version <= target.published_dataset_version + }) +} + fn adopt_advances_head( target_active: Option<&str>, source_entry: &crate::db::DatasetEntry, target_entry: Option<&crate::db::DatasetEntry>, ) -> bool { + if adopt_requires_target_lineage(source_entry, target_entry) { + return true; + } match (target_active, source_entry.native_dataset_branch.as_deref()) { // Source on a branch, target on main — delta applied onto main's lineage. (None, Some(_)) => true, @@ -3512,7 +3528,7 @@ fn adopt_advances_head( target_entry.and_then(|entry| entry.native_dataset_branch.as_deref()) == Some(target_branch) } - // Source on main (pointer switch) or target doesn't own (fork): no advance. + // A newer source on main (pointer switch) or an unowned target (fork). _ => false, } } @@ -3525,10 +3541,9 @@ fn adopt_advances_head( /// forks become [`CandidateTableState::AdoptSourceState`] and do not advance /// data HEAD. /// -/// The HEAD-advancing subcases mirror [`publish_adopted_source_state`]: source -/// on a branch with the target either on main or owning the table. Computing the -/// delta here (rather than inside the publish) is what closes the recovery gap — -/// the classifier knows whether the publish will move Lance HEAD. +/// The HEAD-advancing subcases also include adoption that cannot replace the +/// target's greatest registered version. Computing the delta here (rather than +/// inside the publish) lets recovery own every required target-lineage write. async fn classify_adopt( target_db: &Omnigraph, catalog: &Catalog, @@ -3740,10 +3755,11 @@ fn keep_publishing_candidate( CandidateTableState::AdoptSourceState { validation_delta: None } - ) && matches!( - plan_adopted_source_state(target_active, source_entry, target_entry, table_key), - AdoptPublish::Nothing - ); + ) && (adopt_requires_target_lineage(source_entry, target_entry) + || matches!( + plan_adopted_source_state(target_active, source_entry, target_entry, table_key), + AdoptPublish::Nothing + )); if publishes_nothing { return None; } diff --git a/crates/omnigraph/tests/branching.rs b/crates/omnigraph/tests/branching.rs index ac2fcccfd..c0580000a 100644 --- a/crates/omnigraph/tests/branching.rs +++ b/crates/omnigraph/tests/branching.rs @@ -1828,9 +1828,35 @@ async fn branch_merge_applies_node_insert_to_main() { let dir = tempfile::tempdir().unwrap(); let uri = dir.path().to_str().unwrap(); let main = init_and_load(&dir).await; + // Warm this handle before another writer adds an edge. Creating a fresh + // branch must inherit the new edge even though this handle predates it. + assert_eq!(count_rows(&main, "edge:Knows").await, 3); + let writer = Omnigraph::open(uri).await.unwrap(); + writer + .mutate( + "main", + MUTATION_QUERIES, + "add_friend", + ¶ms(&[("$from", "Alice"), ("$to", "Diana")]), + ) + .await + .unwrap(); + let expected_edge = snapshot_main(&writer) + .await + .unwrap() + .dataset("edge:Knows") + .unwrap() + .clone(); + let mut expected_ids = collect_column_strings(&read_table(&writer, "edge:Knows").await, "id"); + expected_ids.sort(); + assert_eq!(expected_ids.len(), 4); main.branch_create("feature").await.unwrap(); let mut feature = Omnigraph::open(uri).await.unwrap(); + assert_eq!( + count_rows_branch(&feature, "feature", "edge:Knows").await, + 4 + ); mutate_branch( &mut feature, "feature", @@ -1841,6 +1867,32 @@ async fn branch_merge_applies_node_insert_to_main() { .await .unwrap(); + let branch_edge = snapshot_branch(&feature, "feature") + .await + .unwrap() + .dataset("edge:Knows") + .unwrap() + .clone(); + assert_eq!(branch_edge.dataset_path, expected_edge.dataset_path); + assert_eq!( + branch_edge.native_dataset_branch, + expected_edge.native_dataset_branch + ); + assert_eq!( + branch_edge.published_dataset_version, + expected_edge.published_dataset_version + ); + assert_eq!(branch_edge.entity_count, expected_edge.entity_count); + let mut branch_ids = collect_column_strings( + &read_table_branch(&feature, "feature", "edge:Knows").await, + "id", + ); + branch_ids.sort(); + assert_eq!( + branch_ids, expected_ids, + "an unrelated node write must preserve every inherited edge" + ); + let outcome = feature.branch_merge("feature", "main").await.unwrap(); assert_eq!(outcome, MergeOutcome::FastForward); @@ -1854,6 +1906,387 @@ async fn branch_merge_applies_node_insert_to_main() { .await .unwrap(); assert_eq!(qr.num_rows(), 1); + for handle in [&main, &feature, &reopened] { + let mut ids = collect_column_strings(&read_table(handle, "edge:Knows").await, "id"); + ids.sort(); + assert_eq!( + ids, expected_ids, + "fast-forward must preserve untouched edge IDs on warm and fresh handles" + ); + } +} + +// The .gqt runner exposes query, mutate, and restart, but no branch create or +// merge steps. Rust also lets these regressions inspect native table versions +// and arrange an external commit between a cached read and merge planning. +#[tokio::test] +async fn branch_merge_preserves_untouched_edges_after_external_commit_and_read() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let main = init_and_load(&dir).await; + main.branch_create("feature").await.unwrap(); + assert_eq!(count_rows(&main, "edge:Knows").await, 3); + + let writer = Omnigraph::open(uri).await.unwrap(); + writer + .mutate( + "main", + MUTATION_QUERIES, + "add_friend", + ¶ms(&[("$from", "Alice"), ("$to", "Diana")]), + ) + .await + .unwrap(); + writer + .mutate( + "feature", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Eve")], &[("$age", 22)]), + ) + .await + .unwrap(); + // Refresh main's table view through the old handle before asking that same + // handle to classify the now-divergent branches. + assert_eq!(count_rows(&main, "edge:Knows").await, 4); + let outcome = main.branch_merge("feature", "main").await.unwrap(); + assert_eq!(outcome, MergeOutcome::Merged); + for handle in [&main, &Omnigraph::open(uri).await.unwrap()] { + assert_eq!(count_rows(handle, "edge:Knows").await, 4); + let friends = handle + .query( + ReadTarget::branch("main"), + TEST_QUERIES, + "friends_of", + ¶ms(&[("$name", "Alice")]), + ) + .await + .unwrap(); + assert_eq!(first_column_sorted(&friends), ["Bob", "Charlie", "Diana"]); + assert_eq!(count_rows(handle, "node:Person").await, 5); + } +} + +#[tokio::test] +async fn branch_merge_preserves_untouched_edges_after_round_trip() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let db = init_and_load(&dir).await; + db.branch_create("feature").await.unwrap(); + // Build a longer native history on the branch than replaying its net + // changes will produce on main. Every pair returns to the same three rows. + for _ in 0..8 { + db.mutate( + "feature", + MUTATION_QUERIES, + "add_friend", + ¶ms(&[("$from", "Diana"), ("$to", "Alice")]), + ) + .await + .unwrap(); + db.mutate( + "feature", + MUTATION_QUERIES, + "remove_friendship", + ¶ms(&[("$from", "Diana")]), + ) + .await + .unwrap(); + } + db.mutate( + "feature", + MUTATION_QUERIES, + "remove_friendship", + ¶ms(&[("$from", "Bob")]), + ) + .await + .unwrap(); + assert_eq!( + db.branch_merge("feature", "main").await.unwrap(), + MergeOutcome::FastForward + ); + db.mutate( + "main", + MUTATION_QUERIES, + "add_friend", + ¶ms(&[("$from", "Alice"), ("$to", "Diana")]), + ) + .await + .unwrap(); + let expected_rows = read_table(&db, "edge:Knows").await; + let mut expected_ids = collect_column_strings(&expected_rows, "id"); + expected_ids.sort(); + assert_eq!(expected_ids.len(), 3); + let main_edge = snapshot_main(&db) + .await + .unwrap() + .dataset("edge:Knows") + .unwrap() + .clone(); + let feature_edge = snapshot_branch(&db, "feature") + .await + .unwrap() + .dataset("edge:Knows") + .unwrap() + .clone(); + assert!(feature_edge.published_dataset_version > main_edge.published_dataset_version); + assert_eq!( + db.branch_merge("main", "feature").await.unwrap(), + MergeOutcome::FastForward + ); + let inherited_count = count_rows_branch(&db, "feature", "edge:Knows").await; + // Only the Person table changes before the final merge. The edge added on + // main must survive the round trip, including a fresh handle's raw scan. + db.mutate( + "feature", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Eve")], &[("$age", 22)]), + ) + .await + .unwrap(); + assert_eq!( + db.branch_merge("feature", "main").await.unwrap(), + MergeOutcome::FastForward + ); + let reopened = Omnigraph::open(uri).await.unwrap(); + let mut actual_ids = collect_column_strings(&read_table(&reopened, "edge:Knows").await, "id"); + actual_ids.sort(); + assert_eq!( + actual_ids, expected_ids, + "an unrelated node edit and merge must not remove the edge inserted on main" + ); + assert_eq!( + inherited_count, 3, + "main's added edge must already be visible on feature before its unrelated edit" + ); +} + +#[tokio::test] +async fn branch_merge_preserves_state_when_native_versions_differ() { + // The short histories make native versions equal; the long histories put + // the source below the target. A lazy child additionally requires + // recovery-owned first-touch forking without changing the indexed schema. + for branch_updates in [8, 2] { + for lazy_target in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let main = init_and_load(&dir).await; + main.branch_create("feature").await.unwrap(); + let history_branch = if lazy_target { "main" } else { "feature" }; + let history_updates = branch_updates + i64::from(lazy_target); + for age in 40..40 + history_updates { + main.mutate( + history_branch, + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "Alice")], &[("$age", age)]), + ) + .await + .unwrap(); + } + let (source, target) = if lazy_target { + // Keep the changes disjoint: a long main history and one Bob + // edit on feature merge into one additional feature commit. + main.mutate( + "feature", + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "Bob")], &[("$age", 26)]), + ) + .await + .unwrap(); + assert_eq!( + main.branch_merge("main", "feature").await.unwrap(), + MergeOutcome::Merged + ); + main.mutate( + "feature", + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "Alice")], &[("$age", 50)]), + ) + .await + .unwrap(); + // Inherit main's root-owned indexed table. Cloning feature's + // already-cloned indexes would hit the separate Lance #7840 + // bug pinned by second_generation_branch_index_reads_fail_upstream + // in lance_surface_guards.rs. This first-generation fork still + // exercises the merge's lazy-target recovery route. + main.branch_create_from(ReadTarget::branch("main"), "child") + .await + .unwrap(); + ("feature", "child") + } else { + assert_eq!( + main.branch_merge("feature", "main").await.unwrap(), + MergeOutcome::FastForward + ); + main.mutate( + "main", + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "Alice")], &[("$age", 50)]), + ) + .await + .unwrap(); + ("main", "feature") + }; + let target_native = graph_native_ref(uri, target).await; + let source_entry = snapshot_branch(&main, source) + .await + .unwrap() + .dataset("node:Person") + .unwrap() + .clone(); + let target_entry = snapshot_branch(&main, target) + .await + .unwrap() + .dataset("node:Person") + .unwrap() + .clone(); + if branch_updates == 2 { + assert_eq!( + source_entry.published_dataset_version, target_entry.published_dataset_version, + "fixture must exercise equal numeric versions on different refs" + ); + } else { + assert!( + source_entry.published_dataset_version < target_entry.published_dataset_version, + "fixture must exercise a lower source version" + ); + } + assert_ne!( + source_entry.native_dataset_branch, + target_entry.native_dataset_branch + ); + assert_eq!( + target_entry.native_dataset_branch.as_deref() == Some(target_native.as_str()), + !lazy_target + ); + assert_eq!( + main.branch_merge(source, target).await.unwrap(), + MergeOutcome::FastForward + ); + let merged_entry = snapshot_branch(&main, target) + .await + .unwrap() + .dataset("node:Person") + .unwrap() + .clone(); + assert!( + merged_entry.published_dataset_version > target_entry.published_dataset_version, + "{target}, {branch_updates} updates: changed rows must advance target's own version" + ); + assert_eq!( + merged_entry.native_dataset_branch.as_deref(), + Some(target_native.as_str()) + ); + let reopened = Omnigraph::open(uri).await.unwrap(); + for handle in [&main, &reopened] { + let result = handle + .query( + ReadTarget::branch(target), + TEST_QUERIES, + "get_person", + ¶ms(&[("$name", "Alice")]), + ) + .await + .unwrap(); + let batch = result.concat_batches().unwrap(); + assert_eq!( + batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 50, + "{target}, {branch_updates} updates: source value must survive adoption" + ); + } + main.mutate( + target, + MUTATION_QUERIES, + "add_friend", + ¶ms(&[("$from", "Alice"), ("$to", "Diana")]), + ) + .await + .unwrap(); + assert_eq!( + main.branch_merge(target, "main").await.unwrap(), + MergeOutcome::FastForward + ); + let result = main + .query( + ReadTarget::branch("main"), + TEST_QUERIES, + "get_person", + ¶ms(&[("$name", "Alice")]), + ) + .await + .unwrap(); + let batch = result.concat_batches().unwrap(); + assert_eq!( + batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 50, + "{target}, {branch_updates} updates: an unrelated edit must not roll back main" + ); + + // Main now has the same Person rows at a lower or equal version. + // Bringing it back is an empty delta: retain the target's complete + // public registration and do not create a physical table commit. + let before_empty = snapshot_branch(&main, target) + .await + .unwrap() + .dataset("node:Person") + .unwrap() + .clone(); + let table_uri = format!("{uri}/{}", before_empty.dataset_path); + let head_before = + open_dataset_head(&table_uri, before_empty.native_dataset_branch.as_deref()) + .await + .version() + .version; + assert_eq!( + main.branch_merge("main", target).await.unwrap(), + MergeOutcome::FastForward + ); + let reopened = Omnigraph::open(uri).await.unwrap(); + for handle in [&main, &reopened] { + let after_empty = snapshot_branch(handle, target) + .await + .unwrap() + .dataset("node:Person") + .unwrap() + .clone(); + assert_eq!(after_empty.type_key, before_empty.type_key); + assert_eq!(after_empty.dataset_path, before_empty.dataset_path); + assert_eq!( + after_empty.native_dataset_branch, + before_empty.native_dataset_branch + ); + assert_eq!( + after_empty.published_dataset_version, + before_empty.published_dataset_version + ); + assert_eq!(after_empty.entity_count, before_empty.entity_count); + } + assert_eq!( + open_dataset_head(&table_uri, before_empty.native_dataset_branch.as_deref()) + .await + .version() + .version, + head_before, + "empty adoption must not advance the physical target HEAD" + ); + } + } } #[tokio::test] diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs index b7c0ea5ab..91aacd649 100644 --- a/crates/omnigraph/tests/failpoints.rs +++ b/crates/omnigraph/tests/failpoints.rs @@ -10603,7 +10603,8 @@ async fn pre_upgrade_v1_branch_merge_sidecar_rolls_forward_not_back() { /// (D2) — without `CommitGraph::open_at_branch`, the recovery sweep /// would record the global head as the merge parent on a non-main /// target, and future merges between the same pair would lose -/// already-up-to-date detection. +/// already-up-to-date detection. Fast-forwards from an equal or lower native +/// source version must also recover the delta written onto the target ref. #[tokio::test] #[serial] #[serial(branch_merge_phase_b)] @@ -10611,16 +10612,13 @@ async fn branch_merge_phase_b_failure_recovered_on_non_main_target() { use omnigraph::loader::{LoadMode, load_jsonl}; let _scenario = FailScenario::setup(); - let dir = tempfile::tempdir().unwrap(); - let uri = dir.path().to_str().unwrap().to_string(); - let operation_id; - - // Setup: - // main: alice - // target_branch (off main): + bob (target moved past base) - // source_branch (off main): + carol (source moved past base) - // Merge: source_branch → target_branch - { + for (case, target_updates, source_branch) in [ + ("three-way", 0, "source_branch"), + ("equal native versions", 2, "main"), + ("lower source native version", 8, "main"), + ] { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); let db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap(); load_jsonl( &db, @@ -10631,85 +10629,201 @@ async fn branch_merge_phase_b_failure_recovered_on_non_main_target() { .await .unwrap(); db.branch_create("target_branch").await.unwrap(); - db.mutate( - "target_branch", - MUTATION_QUERIES, - "insert_person", - &mixed_params(&[("$name", "Bob")], &[("$age", 40)]), - ) - .await - .unwrap(); - db.branch_create("source_branch").await.unwrap(); - db.mutate( - "source_branch", - MUTATION_QUERIES, - "insert_person", - &mixed_params(&[("$name", "Carol")], &[("$age", 50)]), - ) - .await - .unwrap(); - } + if target_updates == 0 { + // Preserve the divergent source/target merge coverage. + db.mutate( + "target_branch", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Bob")], &[("$age", 40)]), + ) + .await + .unwrap(); + db.branch_create(source_branch).await.unwrap(); + db.mutate( + source_branch, + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Carol")], &[("$age", 50)]), + ) + .await + .unwrap(); + } else { + // Replaying these updates onto main takes one native commit. + // Main then advances graph lineage while its table version remains + // equal to or lower than the target's independent native history. + for age in 40..40 + target_updates { + db.mutate( + "target_branch", + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "alice")], &[("$age", age)]), + ) + .await + .unwrap(); + } + assert_eq!( + db.branch_merge("target_branch", "main").await.unwrap(), + omnigraph::db::MergeOutcome::FastForward, + "{case}" + ); + db.mutate( + "main", + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "alice")], &[("$age", 50)]), + ) + .await + .unwrap(); + } - let main_person_pin = { - let db = Omnigraph::open(&uri).await.unwrap(); - db.snapshot_of(omnigraph::db::ReadTarget::branch("main")) + let main_person_pin = db + .snapshot_of(ReadTarget::branch("main")) .await .unwrap() .dataset("node:Person") .expect("main must have Person") - .published_dataset_version - }; - let target_parent_commit_id = branch_head_commit_id(dir.path(), "target_branch") + .published_dataset_version; + let source_person = db + .snapshot_of(ReadTarget::branch(source_branch)) + .await + .unwrap() + .dataset("node:Person") + .unwrap() + .clone(); + if target_updates > 0 { + let target_version = db + .snapshot_of(ReadTarget::branch("target_branch")) + .await + .unwrap() + .dataset("node:Person") + .unwrap() + .published_dataset_version; + assert_eq!( + source_person.published_dataset_version.cmp(&target_version), + if target_updates == 2 { + std::cmp::Ordering::Equal + } else { + std::cmp::Ordering::Less + }, + "fixture must exercise {case}" + ); + } + let source_head = branch_head_commit_id(dir.path(), source_branch) + .await + .unwrap(); + let target_parent_commit_id = branch_head_commit_id(dir.path(), "target_branch") + .await + .unwrap(); + drop(db); + + // The fixed effect set is confirmed, but graph publication has not + // happened. Recovery must publish it onto this non-main target. + let operation_id = { + let db = Omnigraph::open(&uri).await.unwrap(); + let _failpoint = ScopedFailPoint::new( + names::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT, + "return", + ); + let err = db + .branch_merge(source_branch, "target_branch") + .await + .unwrap_err(); + assert!( + err.to_string().contains( + "injected failpoint triggered: branch_merge.post_phase_b_pre_manifest_commit" + ), + "{case}: unexpected error: {err}" + ); + single_sidecar_operation_id(dir.path()) + }; + + let db = Omnigraph::open(&uri).await.unwrap(); + drop(db); + assert_post_recovery_invariants( + dir.path(), + &operation_id, + RecoveryExpectation::RolledForwardOriginalLineage { + tables: vec![ + TableExpectation::branch("node:Person", "target_branch") + .expected_main_manifest_pin(main_person_pin) + .expected_recovery_parent_commit_id(target_parent_commit_id), + ], + }, + ) .await .unwrap(); - // Setup: failpoint fires after the per-table publish loop completes - // but before commit_manifest_updates. Sidecar persists with - // branch=Some("target_branch"). - { let db = Omnigraph::open(&uri).await.unwrap(); - let _failpoint = ScopedFailPoint::new( - names::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT, - "return", - ); - let err = db - .branch_merge("source_branch", "target_branch") + let recovered_source = db + .snapshot_of(ReadTarget::branch(source_branch)) .await - .unwrap_err(); - assert!( - err.to_string().contains( - "injected failpoint triggered: branch_merge.post_phase_b_pre_manifest_commit" - ), - "unexpected error: {err}" + .unwrap(); + let recovered_source_person = recovered_source.dataset("node:Person").unwrap(); + assert_eq!( + recovered_source_person.published_dataset_version, + source_person.published_dataset_version, + "{case}: recovery must not advance the source table" ); - let recovery_dir = dir.path().join("__recovery"); - let sidecar_count = std::fs::read_dir(&recovery_dir).unwrap().count(); assert_eq!( - sidecar_count, 1, - "exactly one sidecar must persist after non-main branch_merge failure" + recovered_source_person.native_dataset_branch, source_person.native_dataset_branch, + "{case}: recovery must preserve the source ref" ); - operation_id = single_sidecar_operation_id(dir.path()); + assert_eq!( + branch_head_commit_id(dir.path(), source_branch) + .await + .unwrap(), + source_head, + "{case}: recovery must not change source lineage" + ); + let recovered_commit = + omnigraph::db::commit_graph::CommitGraph::open_at_branch(&uri, "target_branch") + .await + .unwrap() + .head_commit() + .await + .unwrap() + .unwrap(); + assert_eq!( + recovered_commit.merged_parent_commit_id.as_deref(), + Some(source_head.as_str()), + "{case}: recovery must retain the captured source parent" + ); + let expected = if target_updates == 0 { + vec![("Bob", 40), ("Carol", 50), ("alice", 30)] + } else { + vec![("alice", 50)] + }; + let rows = helpers::read_table_branch(&db, "target_branch", "node:Person").await; + assert_eq!( + rows.iter().map(RecordBatch::num_rows).sum::(), + expected.len(), + "{case}: recovery must retain every merged row" + ); + for (name, expected_age) in expected { + let result = db + .query( + ReadTarget::branch("target_branch"), + TEST_QUERIES, + "get_person", + ¶ms(&[("$name", name)]), + ) + .await + .unwrap(); + let batch = result.concat_batches().unwrap(); + assert_eq!(batch.num_rows(), 1, "{case}: missing {name}"); + assert_eq!( + batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + expected_age, + "{case}: recovery must publish {name}'s captured source value" + ); + } } - - // Recovery: reopen runs full sweep. The BranchMerge sidecar's branch - // = Some("target_branch"); D2 fix opens a per-branch CommitGraph - // for the audit append so the merge-parent linkage is correct. - let db = Omnigraph::open(&uri).await.unwrap(); - drop(db); - - assert_post_recovery_invariants( - dir.path(), - &operation_id, - RecoveryExpectation::RolledForwardOriginalLineage { - tables: vec![ - TableExpectation::branch("node:Person", "target_branch") - .expected_main_manifest_pin(main_person_pin) - .expected_recovery_parent_commit_id(target_parent_commit_id), - ], - }, - ) - .await - .unwrap(); } /// Contract: the BranchMerge sidecar's per-table `table_branch` MUST be diff --git a/crates/omnigraph/tests/merge_fast_forward.rs b/crates/omnigraph/tests/merge_fast_forward.rs index cb1c38115..56b2e7ef2 100644 --- a/crates/omnigraph/tests/merge_fast_forward.rs +++ b/crates/omnigraph/tests/merge_fast_forward.rs @@ -1392,149 +1392,196 @@ async fn fast_forward_merge_streams_blob_columns() { /// so publication must use the update-only keyed stage introduced by #481. /// Overwrite retains the admitted external descriptor on the source branch; /// merge owns the copied bytes while leaving unchanged valid-empty and null -/// siblings distinct. +/// siblings distinct. Cover both feature-to-main adoption and main-to-feature +/// adoption when main's native version is lower than the owned target's. #[tokio::test] async fn blob_changed_only_adopt_uses_known_present_update() { - let dir = tempfile::tempdir().unwrap(); - let uri = dir.path().to_str().unwrap(); - let external_dir = tempfile::tempdir().unwrap(); - let external_path = external_dir.path().join("changed.txt"); - std::fs::write(&external_path, b"Changed externally").unwrap(); - let external_uri = url::Url::from_file_path(std::fs::canonicalize(&external_path).unwrap()) - .expect("external Blob path is absolute") - .to_string(); - let empty_path = external_dir.path().join("valid-empty.txt"); - std::fs::write(&empty_path, b"").unwrap(); - let empty_uri = url::Url::from_file_path(std::fs::canonicalize(&empty_path).unwrap()) - .expect("empty external Blob path is absolute") - .to_string(); - let external_base = url::Url::from_directory_path(external_dir.path()) - .expect("external Blob base is absolute") - .to_string(); - let policy = ExternalBlobPolicy::allow(vec![ - ExternalBlobBase::new(external_base, ExternalBlobExecutionScope::EmbeddedOnly).unwrap(), - ]) - .unwrap(); - - let main = Omnigraph::init(uri, BLOB_SCHEMA) - .await - .unwrap() - .with_external_blob_policy(policy.clone()) - .unwrap(); - let base_data = [ - serde_json::json!({ - "type": "Document", - "data": {"title": "changed", "content": "base64:QmFzZQ==", "note": "base"}, - }), - serde_json::json!({ - "type": "Document", - "data": {"title": "valid-empty", "content": empty_uri.clone(), "note": "empty"}, - }), - serde_json::json!({ - "type": "Document", - "data": {"title": "null", "content": null, "note": "null"}, - }), - ] - .into_iter() - .map(|row| row.to_string()) - .collect::>() - .join("\n"); - main.load("main", &base_data, LoadMode::Overwrite) - .await - .unwrap(); - main.branch_create("feature").await.unwrap(); + const SET_NOTE: &str = r#" +query set_note($title: String, $note: String) { + update Document set { note: $note } where title = $title +} +"#; - let feature = Omnigraph::open(uri) - .await - .unwrap() - .with_external_blob_policy(policy.clone()) - .unwrap(); - let source_data = [ - serde_json::json!({ - "type": "Document", - "data": {"title": "changed", "content": external_uri, "note": "source"}, - }), - serde_json::json!({ - "type": "Document", - "data": {"title": "valid-empty", "content": empty_uri.clone(), "note": "empty"}, - }), - serde_json::json!({ - "type": "Document", - "data": {"title": "null", "content": null, "note": "null"}, - }), - ] - .into_iter() - .map(|row| row.to_string()) - .collect::>() - .join("\n"); - feature - .load("feature", &source_data, LoadMode::Overwrite) - .await + for source_main in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let external_dir = tempfile::tempdir().unwrap(); + let external_path = external_dir.path().join("changed.txt"); + std::fs::write(&external_path, b"Changed externally").unwrap(); + let external_uri = url::Url::from_file_path(std::fs::canonicalize(&external_path).unwrap()) + .expect("external Blob path is absolute") + .to_string(); + let empty_path = external_dir.path().join("valid-empty.txt"); + std::fs::write(&empty_path, b"").unwrap(); + let empty_uri = url::Url::from_file_path(std::fs::canonicalize(&empty_path).unwrap()) + .expect("empty external Blob path is absolute") + .to_string(); + let external_base = url::Url::from_directory_path(external_dir.path()) + .expect("external Blob base is absolute") + .to_string(); + let policy = ExternalBlobPolicy::allow(vec![ + ExternalBlobBase::new(external_base, ExternalBlobExecutionScope::EmbeddedOnly).unwrap(), + ]) .unwrap(); - let merger = Omnigraph::open(uri) - .await - .unwrap() - .with_external_blob_policy(policy) - .unwrap(); - let probes = MergeWriteProbes::default(); - let outcome = with_merge_write_probes(probes.clone(), merger.branch_merge("feature", "main")) - .await - .unwrap(); - assert_eq!(outcome, MergeOutcome::FastForward); - assert_eq!(probes.stage_known_present_update_calls(), 1); - assert_eq!(probes.stage_known_present_update_rows(), 1); - assert_eq!(probes.stage_merge_insert_calls(), 0); - assert_eq!(probes.stage_fenced_insert_calls(), 0); - assert_eq!(probes.strict_insert_preflight_calls(), 0); - assert_eq!( - probes.stage_vector_index_calls(), - 0, - "general Blob adoption must also defer derived index work" - ); - assert_eq!( - probes.external_blob_probe_inputs(), - 1, - "only the changed external descriptor belongs to the adopt delta" - ); - assert_eq!(probes.external_blob_probe_calls(), 1); - assert_eq!(probes.external_blob_payload_read_calls(), 1); + let main = Omnigraph::init(uri, BLOB_SCHEMA) + .await + .unwrap() + .with_external_blob_policy(policy.clone()) + .unwrap(); + let base_data = [ + serde_json::json!({ + "type": "Document", + "data": {"title": "changed", "content": "base64:QmFzZQ==", "note": "base"}, + }), + serde_json::json!({ + "type": "Document", + "data": {"title": "valid-empty", "content": empty_uri.clone(), "note": "empty"}, + }), + serde_json::json!({ + "type": "Document", + "data": {"title": "null", "content": null, "note": "null"}, + }), + ] + .into_iter() + .map(|row| row.to_string()) + .collect::>() + .join("\n"); + main.load("main", &base_data, LoadMode::Overwrite) + .await + .unwrap(); + main.branch_create("feature").await.unwrap(); - assert_eq!(count_rows(&merger, "node:Document").await, 3); - let changed = read_managed_blob_bytes( - &merger, - ReadTarget::branch("main"), - node_blob_cell("Document", "changed", "content"), - ) - .await; - assert_eq!(&changed[..], b"Changed externally"); + let feature = Omnigraph::open(uri) + .await + .unwrap() + .with_external_blob_policy(policy.clone()) + .unwrap(); + let source_data = [ + serde_json::json!({ + "type": "Document", + "data": {"title": "changed", "content": external_uri, "note": "source"}, + }), + serde_json::json!({ + "type": "Document", + "data": {"title": "valid-empty", "content": empty_uri.clone(), "note": "empty"}, + }), + serde_json::json!({ + "type": "Document", + "data": {"title": "null", "content": null, "note": "null"}, + }), + ] + .into_iter() + .map(|row| row.to_string()) + .collect::>() + .join("\n"); + let (source, target) = if source_main { + // Churn only one note, then merge its net change into main. The owned + // feature table now has a higher native version than main while both + // have the same logical rows. Main's next overwrite remains below it. + for step in 0..8 { + feature + .mutate( + "feature", + SET_NOTE, + "set_note", + ¶ms(&[("$title", "changed"), ("$note", &format!("step-{step}"))]), + ) + .await + .unwrap(); + } + assert_eq!( + main.branch_merge("feature", "main").await.unwrap(), + MergeOutcome::FastForward + ); + ("main", "feature") + } else { + ("feature", "main") + }; + feature + .load(source, &source_data, LoadMode::Overwrite) + .await + .unwrap(); + if source_main { + let source_snapshot = snapshot_branch(&main, source).await.unwrap(); + let target_snapshot = snapshot_branch(&main, target).await.unwrap(); + assert!( + source_snapshot + .dataset("node:Document") + .unwrap() + .published_dataset_version + < target_snapshot + .dataset("node:Document") + .unwrap() + .published_dataset_version, + "source-main fixture must require target-lineage adoption" + ); + } - let empty = merger - .read_blob_at( - ReadTarget::branch("main"), - node_blob_cell("Document", "valid-empty", "content"), - ) - .await - .unwrap(); - let BlobContent::External(empty) = empty.content else { - panic!("an unchanged retained descriptor must stay pointer-only"); - }; - assert_eq!(empty.uri, empty_uri); - assert_eq!(empty.offset, 0); - assert_eq!(empty.length, None); - - let null = merger - .read_blob_at( - ReadTarget::branch("main"), - node_blob_cell("Document", "null", "content"), + let merger = Omnigraph::open(uri) + .await + .unwrap() + .with_external_blob_policy(policy) + .unwrap(); + let probes = MergeWriteProbes::default(); + let outcome = with_merge_write_probes(probes.clone(), merger.branch_merge(source, target)) + .await + .unwrap(); + assert_eq!(outcome, MergeOutcome::FastForward); + assert_eq!(probes.stage_known_present_update_calls(), 1); + assert_eq!(probes.stage_known_present_update_rows(), 1); + assert_eq!(probes.stage_merge_insert_calls(), 0); + assert_eq!(probes.stage_fenced_insert_calls(), 0); + assert_eq!(probes.strict_insert_preflight_calls(), 0); + assert_eq!( + probes.stage_vector_index_calls(), + 0, + "general Blob adoption must also defer derived index work" + ); + assert_eq!( + probes.external_blob_probe_inputs(), + 1, + "only the changed external descriptor belongs to the adopt delta" + ); + assert_eq!(probes.external_blob_probe_calls(), 1); + assert_eq!(probes.external_blob_payload_read_calls(), 1); + + assert_eq!(count_rows_branch(&merger, target, "node:Document").await, 3); + let changed = read_managed_blob_bytes( + &merger, + ReadTarget::branch(target), + node_blob_cell("Document", "changed", "content"), ) - .await - .unwrap_err(); - assert!( - matches!( - null, - OmniError::Manifest(ref error) if error.kind == ManifestErrorKind::NotFound - ), - "an unchanged null Blob must remain null rather than becoming valid-empty: {null:?}" - ); + .await; + assert_eq!(&changed[..], b"Changed externally"); + + let empty = merger + .read_blob_at( + ReadTarget::branch(target), + node_blob_cell("Document", "valid-empty", "content"), + ) + .await + .unwrap(); + let BlobContent::External(empty) = empty.content else { + panic!("an unchanged retained descriptor must stay pointer-only"); + }; + assert_eq!(empty.uri, empty_uri); + assert_eq!(empty.offset, 0); + assert_eq!(empty.length, None); + + let null = merger + .read_blob_at( + ReadTarget::branch(target), + node_blob_cell("Document", "null", "content"), + ) + .await + .unwrap_err(); + assert!( + matches!( + null, + OmniError::Manifest(ref error) if error.kind == ManifestErrorKind::NotFound + ), + "an unchanged null Blob must remain null rather than becoming valid-empty: {null:?}" + ); + } } diff --git a/docs/dev/merge.md b/docs/dev/merge.md index 63c4c6748..bb49aca22 100644 --- a/docs/dev/merge.md +++ b/docs/dev/merge.md @@ -26,7 +26,8 @@ The table classifier chooses one of four routes: 1. **No change:** source contributes nothing. 2. **Pointer adoption:** target still equals the base and the exact source table state can become the target's visible pointer without copying rows. A - first-touch lazy target stays on this ref-only route. + first-touch lazy target can use this ref-only route when the version check + below permits it. 3. **Proven insertion replay:** target still permits data replay and the complete retained source interval proves a contiguous sequence of exact-ID, insertion-only transactions. @@ -37,6 +38,16 @@ An optimization miss is not a merge failure. Missing transaction history, unknown certificate fields, incomplete ancestry, or an unfamiliar Lance shape falls back to the general route. +Native Lance version numbers are local to each native branch. The manifest +projection nevertheless selects the highest registered version for a table +lifetime, so adopting a source version less than or equal to the target's +version cannot safely replace its pointer. A nonempty source delta uses the +existing target-lineage writer, creating a native target ref lazily when +needed. A proved empty delta retains the target's complete table entry, +including its native ref and version metadata. The version comparison selects +the publication route; it never proves row equality. This rule also governs +Blob preflight so validation and publication agree on which rows are copied. + ## Proven insertion route The internal `omnigraph.insert_absence = "v1"` transaction property says that diff --git a/docs/releases/v0.11.0.md b/docs/releases/v0.11.0.md index c2740736f..45f69cbc9 100644 --- a/docs/releases/v0.11.0.md +++ b/docs/releases/v0.11.0.md @@ -100,6 +100,14 @@ Notes accumulate here until the release is cut. `Float64` and `count` still returns 0. A JSON row omits the null cell's key, so the type shows only in the Arrow IPC result schema. +- **Repeated branch merges preserve previously merged rows and values.** + After a branch accumulated more table versions than `main`, merging newer + data from `main` into that branch could report success while retaining stale + data. Merging the branch back after an unrelated edit could then delete an + edge or roll back a value on `main`. Merge now applies these changes through + the target's table history and preserves its existing entry when the rows + are unchanged. No graph-storage or recovery-format change is required. + - **Two projections producing one result column name are refused at compile time (`T25`).** `return { $a.num1 as number, $a.num2 as number }` used to run and emit two columns both named `number`; every reader that diff --git a/docs/user/branching/merge.md b/docs/user/branching/merge.md index 86b0dfcf4..91a28a7d7 100644 --- a/docs/user/branching/merge.md +++ b/docs/user/branching/merge.md @@ -26,6 +26,11 @@ composition: follow it with `branch delete `. A merge statement takes no commit precondition -- `POST /mutate/if-graph-commit` and `--if-commit` refuse one beside it -- and no front offers a conditional merge today. +A merge preserves changes already integrated into a branch when you later +merge that branch back. For example, after merging a new edge from `main` +into `review`, editing only a node on `review` and merging it into `main` +preserves that edge. + ## Outcomes - **Already up to date**: the target already contains the source changes. From 47cb4c02f5e9befd34c220e6f375b127681815bd Mon Sep 17 00:00:00 2001 From: aaltshuler Date: Mon, 7 Sep 2026 15:42:38 +0300 Subject: [PATCH 2/3] test(merge): cover stale values and untouched edges in GQT --- ...anch_round_trip_preserves_newer_values.gqt | 124 ++++++++++++++++ ...h_round_trip_preserves_untouched_edges.gqt | 135 ++++++++++++++++++ 2 files changed, 259 insertions(+) create mode 100644 crates/omnigraph-gqt/cases/branch_round_trip_preserves_newer_values.gqt create mode 100644 crates/omnigraph-gqt/cases/branch_round_trip_preserves_untouched_edges.gqt diff --git a/crates/omnigraph-gqt/cases/branch_round_trip_preserves_newer_values.gqt b/crates/omnigraph-gqt/cases/branch_round_trip_preserves_newer_values.gqt new file mode 100644 index 000000000..ace0d4b0b --- /dev/null +++ b/crates/omnigraph-gqt/cases/branch_round_trip_preserves_newer_values.gqt @@ -0,0 +1,124 @@ +# issue: none +# red_on: 2026-09-07, main 5f94a741: after merging main into feature, Alice was age 47 instead of 50. +# notes: A feature accumulates more native Person versions than its net merge +# notes: creates on main. Bringing main's newer value back must replace the +# notes: feature's old value, including after a restart and an unrelated write. +# notes: PR 630; branching.rs::branch_merge_preserves_state_when_native_versions_differ +# notes: retains the equal/lower-version, lazy-target, and physical HEAD proofs. + +--- schema +node Person { + name: String @key + age: I32 +} +node Company { + name: String @key +} + +--- seed +{"type":"Person","data":{"name":"alice","age":30}} +{"type":"Company","data":{"name":"acme"}} + +--- mutate +branch create feature + +--- expect ok + +--- loop $age 40 48 + +--- mutate branch: feature +query change_age($age: I32) { + update Person set { age: $age } where name = "alice" +} + +--- params +{"age":${age}} + +--- expect affected: nodes=1 edges=0 + +--- endloop + +--- mutate +branch merge feature into main + +--- expect outcome: fast_forward + +--- mutate +query update_main() { + update Person set { age: 50 } where name = "alice" +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +branch merge main into feature + +--- expect outcome: fast_forward + +--- query branch: feature +query adopted_value() { + match { $p: Person } + return { $p.name, $p.age } +} + +--- expect unordered +{"p.name":"alice","p.age":50} + +--- expect shape +p.name: String +p.age: I32 + +--- restart + +--- query branch: feature +query reopened_value() { + match { $p: Person } + return { $p.name, $p.age } +} + +--- expect unordered +{"p.name":"alice","p.age":50} + +--- expect shape +p.name: String +p.age: I32 + +--- mutate branch: feature +query unrelated_company() { + insert Company { name: "globex" } +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +branch merge feature into main + +--- expect outcome: fast_forward + +--- restart + +--- query +query main_kept_newer_value() { + match { $p: Person } + return { $p.name, $p.age } +} + +--- expect unordered +{"p.name":"alice","p.age":50} + +--- expect shape +p.name: String +p.age: I32 + +--- query +query main_received_company() { + match { $c: Company } + return { $c.name } +} + +--- expect unordered +{"c.name":"acme"} +{"c.name":"globex"} + +--- expect shape +c.name: String diff --git a/crates/omnigraph-gqt/cases/branch_round_trip_preserves_untouched_edges.gqt b/crates/omnigraph-gqt/cases/branch_round_trip_preserves_untouched_edges.gqt new file mode 100644 index 000000000..6446d3962 --- /dev/null +++ b/crates/omnigraph-gqt/cases/branch_round_trip_preserves_untouched_edges.gqt @@ -0,0 +1,135 @@ +# issue: none +# red_on: 2026-09-07, main 5f94a741: after merging main into feature, only two of the expected three bound edge rows remained. +# notes: Churn a feature's edge history, merge it, add an edge on main, merge +# notes: main back, then edit only a node. The final merge must retain all three +# notes: edges after reopening. A bound edge makes duplicate or missing rows visible. +# notes: PR 630; branching.rs::branch_merge_preserves_untouched_edges_after_round_trip +# notes: retains native-version and exact stored-edge ID assertions. + +--- schema +node Person { + name: String @key +} +edge Knows: Person -> Person + +--- seed +{"type":"Person","data":{"name":"alice"}} +{"type":"Person","data":{"name":"bob"}} +{"type":"Person","data":{"name":"charlie"}} +{"type":"Person","data":{"name":"diana"}} +{"edge":"Knows","from":"alice","to":"bob"} +{"edge":"Knows","from":"alice","to":"charlie"} +{"edge":"Knows","from":"bob","to":"diana"} + +--- mutate +branch create feature + +--- expect ok + +--- loop $i 0 8 + +--- mutate branch: feature +query add_temporary_edge() { + insert Knows { from: "diana", to: "alice" } +} + +--- expect affected: nodes=0 edges=1 + +--- mutate branch: feature +query remove_temporary_edge() { + delete Knows where from = "diana" +} + +--- expect affected: nodes=0 edges=1 + +--- endloop + +--- mutate branch: feature +query remove_bob_edge() { + delete Knows where from = "bob" +} + +--- expect affected: nodes=0 edges=1 + +--- mutate +branch merge feature into main + +--- expect outcome: fast_forward + +--- mutate +query add_main_edge() { + insert Knows { from: "alice", to: "diana" } +} + +--- expect affected: nodes=0 edges=1 + +--- mutate +branch merge main into feature + +--- expect outcome: fast_forward + +--- query branch: feature +query feature_received_every_edge() { + match { + $a: Person + $a $e:knows $b + } + return { $a.name as source, $b.name as target } +} + +--- expect unordered +{"source":"alice","target":"bob"} +{"source":"alice","target":"charlie"} +{"source":"alice","target":"diana"} + +--- expect shape +source: String +target: String + +--- mutate branch: feature +query unrelated_node() { + insert Person { name: "eve" } +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +branch merge feature into main + +--- expect outcome: fast_forward + +--- restart + +--- query +query main_kept_every_edge() { + match { + $a: Person + $a $e:knows $b + } + return { $a.name as source, $b.name as target } +} + +--- expect unordered +{"source":"alice","target":"bob"} +{"source":"alice","target":"charlie"} +{"source":"alice","target":"diana"} + +--- expect shape +source: String +target: String + +--- query +query main_received_unrelated_node() { + match { $p: Person } + return { $p.name } +} + +--- expect unordered +{"p.name":"alice"} +{"p.name":"bob"} +{"p.name":"charlie"} +{"p.name":"diana"} +{"p.name":"eve"} + +--- expect shape +p.name: String From ab3232ab956d5693e4b3fca28a43eba3a258453f Mon Sep 17 00:00:00 2001 From: azim afroozeh <13484327+azimafroozeh@users.noreply.github.com> Date: Mon, 7 Sep 2026 15:14:04 +0200 Subject: [PATCH 3/3] fix(merge): fence retained tables, refuse a losing pointer, pin lagging adopts in GQT --- ...anch_round_trip_preserves_newer_values.gqt | 124 ---------- ...h_round_trip_preserves_untouched_edges.gqt | 135 ----------- ...ge_adopt_equal_source_version_collides.gqt | 74 ++++++ ...pt_lazy_target_below_inherited_version.gqt | 92 +++++++ ...adopt_lower_source_version_keeps_edges.gqt | 113 +++++++++ crates/omnigraph/src/db/manifest/state.rs | 13 + crates/omnigraph/src/exec/merge.rs | 78 +++--- crates/omnigraph/tests/branching.rs | 226 +----------------- crates/omnigraph/tests/failpoints.rs | 65 +++-- crates/omnigraph/tests/merge_fast_forward.rs | 6 +- docs/dev/merge.md | 23 +- docs/releases/v0.11.0.md | 20 +- docs/user/branching/merge.md | 4 +- 13 files changed, 431 insertions(+), 542 deletions(-) delete mode 100644 crates/omnigraph-gqt/cases/branch_round_trip_preserves_newer_values.gqt delete mode 100644 crates/omnigraph-gqt/cases/branch_round_trip_preserves_untouched_edges.gqt create mode 100644 crates/omnigraph-gqt/cases/merge_adopt_equal_source_version_collides.gqt create mode 100644 crates/omnigraph-gqt/cases/merge_adopt_lazy_target_below_inherited_version.gqt create mode 100644 crates/omnigraph-gqt/cases/merge_adopt_lower_source_version_keeps_edges.gqt diff --git a/crates/omnigraph-gqt/cases/branch_round_trip_preserves_newer_values.gqt b/crates/omnigraph-gqt/cases/branch_round_trip_preserves_newer_values.gqt deleted file mode 100644 index ace0d4b0b..000000000 --- a/crates/omnigraph-gqt/cases/branch_round_trip_preserves_newer_values.gqt +++ /dev/null @@ -1,124 +0,0 @@ -# issue: none -# red_on: 2026-09-07, main 5f94a741: after merging main into feature, Alice was age 47 instead of 50. -# notes: A feature accumulates more native Person versions than its net merge -# notes: creates on main. Bringing main's newer value back must replace the -# notes: feature's old value, including after a restart and an unrelated write. -# notes: PR 630; branching.rs::branch_merge_preserves_state_when_native_versions_differ -# notes: retains the equal/lower-version, lazy-target, and physical HEAD proofs. - ---- schema -node Person { - name: String @key - age: I32 -} -node Company { - name: String @key -} - ---- seed -{"type":"Person","data":{"name":"alice","age":30}} -{"type":"Company","data":{"name":"acme"}} - ---- mutate -branch create feature - ---- expect ok - ---- loop $age 40 48 - ---- mutate branch: feature -query change_age($age: I32) { - update Person set { age: $age } where name = "alice" -} - ---- params -{"age":${age}} - ---- expect affected: nodes=1 edges=0 - ---- endloop - ---- mutate -branch merge feature into main - ---- expect outcome: fast_forward - ---- mutate -query update_main() { - update Person set { age: 50 } where name = "alice" -} - ---- expect affected: nodes=1 edges=0 - ---- mutate -branch merge main into feature - ---- expect outcome: fast_forward - ---- query branch: feature -query adopted_value() { - match { $p: Person } - return { $p.name, $p.age } -} - ---- expect unordered -{"p.name":"alice","p.age":50} - ---- expect shape -p.name: String -p.age: I32 - ---- restart - ---- query branch: feature -query reopened_value() { - match { $p: Person } - return { $p.name, $p.age } -} - ---- expect unordered -{"p.name":"alice","p.age":50} - ---- expect shape -p.name: String -p.age: I32 - ---- mutate branch: feature -query unrelated_company() { - insert Company { name: "globex" } -} - ---- expect affected: nodes=1 edges=0 - ---- mutate -branch merge feature into main - ---- expect outcome: fast_forward - ---- restart - ---- query -query main_kept_newer_value() { - match { $p: Person } - return { $p.name, $p.age } -} - ---- expect unordered -{"p.name":"alice","p.age":50} - ---- expect shape -p.name: String -p.age: I32 - ---- query -query main_received_company() { - match { $c: Company } - return { $c.name } -} - ---- expect unordered -{"c.name":"acme"} -{"c.name":"globex"} - ---- expect shape -c.name: String diff --git a/crates/omnigraph-gqt/cases/branch_round_trip_preserves_untouched_edges.gqt b/crates/omnigraph-gqt/cases/branch_round_trip_preserves_untouched_edges.gqt deleted file mode 100644 index 6446d3962..000000000 --- a/crates/omnigraph-gqt/cases/branch_round_trip_preserves_untouched_edges.gqt +++ /dev/null @@ -1,135 +0,0 @@ -# issue: none -# red_on: 2026-09-07, main 5f94a741: after merging main into feature, only two of the expected three bound edge rows remained. -# notes: Churn a feature's edge history, merge it, add an edge on main, merge -# notes: main back, then edit only a node. The final merge must retain all three -# notes: edges after reopening. A bound edge makes duplicate or missing rows visible. -# notes: PR 630; branching.rs::branch_merge_preserves_untouched_edges_after_round_trip -# notes: retains native-version and exact stored-edge ID assertions. - ---- schema -node Person { - name: String @key -} -edge Knows: Person -> Person - ---- seed -{"type":"Person","data":{"name":"alice"}} -{"type":"Person","data":{"name":"bob"}} -{"type":"Person","data":{"name":"charlie"}} -{"type":"Person","data":{"name":"diana"}} -{"edge":"Knows","from":"alice","to":"bob"} -{"edge":"Knows","from":"alice","to":"charlie"} -{"edge":"Knows","from":"bob","to":"diana"} - ---- mutate -branch create feature - ---- expect ok - ---- loop $i 0 8 - ---- mutate branch: feature -query add_temporary_edge() { - insert Knows { from: "diana", to: "alice" } -} - ---- expect affected: nodes=0 edges=1 - ---- mutate branch: feature -query remove_temporary_edge() { - delete Knows where from = "diana" -} - ---- expect affected: nodes=0 edges=1 - ---- endloop - ---- mutate branch: feature -query remove_bob_edge() { - delete Knows where from = "bob" -} - ---- expect affected: nodes=0 edges=1 - ---- mutate -branch merge feature into main - ---- expect outcome: fast_forward - ---- mutate -query add_main_edge() { - insert Knows { from: "alice", to: "diana" } -} - ---- expect affected: nodes=0 edges=1 - ---- mutate -branch merge main into feature - ---- expect outcome: fast_forward - ---- query branch: feature -query feature_received_every_edge() { - match { - $a: Person - $a $e:knows $b - } - return { $a.name as source, $b.name as target } -} - ---- expect unordered -{"source":"alice","target":"bob"} -{"source":"alice","target":"charlie"} -{"source":"alice","target":"diana"} - ---- expect shape -source: String -target: String - ---- mutate branch: feature -query unrelated_node() { - insert Person { name: "eve" } -} - ---- expect affected: nodes=1 edges=0 - ---- mutate -branch merge feature into main - ---- expect outcome: fast_forward - ---- restart - ---- query -query main_kept_every_edge() { - match { - $a: Person - $a $e:knows $b - } - return { $a.name as source, $b.name as target } -} - ---- expect unordered -{"source":"alice","target":"bob"} -{"source":"alice","target":"charlie"} -{"source":"alice","target":"diana"} - ---- expect shape -source: String -target: String - ---- query -query main_received_unrelated_node() { - match { $p: Person } - return { $p.name } -} - ---- expect unordered -{"p.name":"alice"} -{"p.name":"bob"} -{"p.name":"charlie"} -{"p.name":"diana"} -{"p.name":"eve"} - ---- expect shape -p.name: String diff --git a/crates/omnigraph-gqt/cases/merge_adopt_equal_source_version_collides.gqt b/crates/omnigraph-gqt/cases/merge_adopt_equal_source_version_collides.gqt new file mode 100644 index 000000000..f2c08ae5c --- /dev/null +++ b/crates/omnigraph-gqt/cases/merge_adopt_equal_source_version_collides.gqt @@ -0,0 +1,74 @@ +# issue: none +# red_on: 2026-09-07, main @ 5f94a741: `branch merge main into feature` failed with `storage: Concurrent modification: table version 4 already exists for identity … (edge:Knows) with different state` +# notes: sibling of merge_adopt_lower_source_version_keeps_edges: two edge writes +# notes: on `feature` instead of three make the two edge tables reach the same +# notes: version number on different native branches. Adopting `main` into +# notes: `feature` must fast-forward and show all four edges. + +--- schema +node Person { + name: String @key +} + +edge Knows: Person -> Person + +--- seed +{"type":"Person","data":{"name":"a"}} +{"type":"Person","data":{"name":"b"}} +{"type":"Person","data":{"name":"c"}} +{"edge":"Knows","from":"a","to":"b"} + +--- mutate +branch create feature + +--- expect ok + +--- mutate branch: feature +query add_b_c() { + insert Knows { from: "b", to: "c" } +} + +--- expect affected: nodes=0 edges=1 + +--- mutate branch: feature +query add_c_a() { + insert Knows { from: "c", to: "a" } +} + +--- expect affected: nodes=0 edges=1 + +--- mutate +branch merge feature + +--- expect outcome: fast_forward + +--- mutate +query add_a_c_on_main() { + insert Knows { from: "a", to: "c" } +} + +--- expect affected: nodes=0 edges=1 + +--- mutate +branch merge main into feature + +--- expect outcome: fast_forward + +--- query branch: feature +query edges_on_feature() { + match { + $x: Person + $x knows $y + } + return { $x.name, $y.name } +} + +--- expect unordered +{"x.name": "a", "y.name": "b"} +{"x.name": "b", "y.name": "c"} +{"x.name": "c", "y.name": "a"} +{"x.name": "a", "y.name": "c"} + +--- expect shape +x.name: String +y.name: String diff --git a/crates/omnigraph-gqt/cases/merge_adopt_lazy_target_below_inherited_version.gqt b/crates/omnigraph-gqt/cases/merge_adopt_lazy_target_below_inherited_version.gqt new file mode 100644 index 000000000..e54f87756 --- /dev/null +++ b/crates/omnigraph-gqt/cases/merge_adopt_lazy_target_below_inherited_version.gqt @@ -0,0 +1,92 @@ +# issue: none +# notes: from PR 630, the lazy-target arm. `main` writes its node table four +# notes: times; `feature` writes once, takes `main` by a three-way merge, then +# notes: sets a's age; `child` forks from `main` and never writes, so its node +# notes: table is `main`'s at a version above `feature`'s. Merging `feature` +# notes: into `child` must show feature's value on child. + +--- schema +node Person { + name: String @key + age: I32 +} + +--- seed +{"type":"Person","data":{"name":"a","age":1}} +{"type":"Person","data":{"name":"b","age":1}} + +--- mutate +branch create feature + +--- expect ok + +--- mutate +query a_2_on_main() { + update Person set { age: 2 } where name = "a" +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +query a_3_on_main() { + update Person set { age: 3 } where name = "a" +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +query a_4_on_main() { + update Person set { age: 4 } where name = "a" +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +query a_5_on_main() { + update Person set { age: 5 } where name = "a" +} + +--- expect affected: nodes=1 edges=0 + +--- mutate branch: feature +query b_9_on_feature() { + update Person set { age: 9 } where name = "b" +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +branch merge main into feature + +--- expect outcome: merged + +--- mutate branch: feature +query a_50_on_feature() { + update Person set { age: 50 } where name = "a" +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +branch create child + +--- expect ok + +--- mutate +branch merge feature into child + +--- expect outcome: fast_forward + +--- query branch: child +query ages_on_child() { + match { $p: Person } + return { $p.name, $p.age } +} + +--- expect unordered +{"p.name": "a", "p.age": 50} +{"p.name": "b", "p.age": 9} + +--- expect shape +p.name: String +p.age: I32 diff --git a/crates/omnigraph-gqt/cases/merge_adopt_lower_source_version_keeps_edges.gqt b/crates/omnigraph-gqt/cases/merge_adopt_lower_source_version_keeps_edges.gqt new file mode 100644 index 000000000..1caff7a50 --- /dev/null +++ b/crates/omnigraph-gqt/cases/merge_adopt_lower_source_version_keeps_edges.gqt @@ -0,0 +1,113 @@ +# issue: none +# red_on: 2026-09-07, main @ 5f94a741: after `branch merge main into feature` reported fast_forward, feature's edges lacked a->c; merging feature back then dropped a->c from main +# notes: from PR 630. `feature` writes its edge table three times, so its table +# notes: version passes `main`'s. Merge `feature` into `main`, add an edge on +# notes: `main`, merge `main` into `feature`: the adopted edge must be visible +# notes: on `feature`. Then edit only a node on `feature` and merge back: `main` +# notes: must keep the edge. + +--- schema +node Person { + name: String @key +} + +edge Knows: Person -> Person + +--- seed +{"type":"Person","data":{"name":"a"}} +{"type":"Person","data":{"name":"b"}} +{"type":"Person","data":{"name":"c"}} +{"edge":"Knows","from":"a","to":"b"} + +--- mutate +branch create feature + +--- expect ok + +--- mutate branch: feature +query add_b_c() { + insert Knows { from: "b", to: "c" } +} + +--- expect affected: nodes=0 edges=1 + +--- mutate branch: feature +query add_c_a() { + insert Knows { from: "c", to: "a" } +} + +--- expect affected: nodes=0 edges=1 + +--- mutate branch: feature +query drop_b_c() { + delete Knows where from = "b" +} + +--- expect affected: nodes=0 edges=1 + +--- mutate +branch merge feature + +--- expect outcome: fast_forward + +--- mutate +query add_a_c_on_main() { + insert Knows { from: "a", to: "c" } +} + +--- expect affected: nodes=0 edges=1 + +--- mutate +branch merge main into feature + +--- expect outcome: fast_forward + +--- query branch: feature +query edges_on_feature() { + match { + $x: Person + $x knows $y + } + return { $x.name, $y.name } +} + +--- expect unordered +{"x.name": "a", "y.name": "b"} +{"x.name": "a", "y.name": "c"} +{"x.name": "c", "y.name": "a"} + +--- expect shape +x.name: String +y.name: String + +--- mutate branch: feature +query add_d_on_feature() { + insert Person { name: "d" } +} + +--- expect affected: nodes=1 edges=0 + +--- mutate +branch merge feature + +--- expect outcome: fast_forward + +--- restart + +--- query +query edges_on_main() { + match { + $x: Person + $x knows $y + } + return { $x.name, $y.name } +} + +--- expect unordered +{"x.name": "a", "y.name": "b"} +{"x.name": "a", "y.name": "c"} +{"x.name": "c", "y.name": "a"} + +--- expect shape +x.name: String +y.name: String diff --git a/crates/omnigraph/src/db/manifest/state.rs b/crates/omnigraph/src/db/manifest/state.rs index 18b5c3600..25449c191 100644 --- a/crates/omnigraph/src/db/manifest/state.rs +++ b/crates/omnigraph/src/db/manifest/state.rs @@ -26,6 +26,19 @@ pub struct DatasetEntry { pub(crate) version_metadata: TableVersionMetadata, } +impl DatasetEntry { + /// Field-for-field equal registration, the Lance manifest metadata included. + pub fn same_registration(&self, other: &DatasetEntry) -> bool { + self.identity == other.identity + && self.type_key == other.type_key + && self.dataset_path == other.dataset_path + && self.published_dataset_version == other.published_dataset_version + && self.native_dataset_branch == other.native_dataset_branch + && self.entity_count == other.entity_count + && self.version_metadata == other.version_metadata + } +} + #[derive(Debug, Clone)] pub(super) struct ManifestState { pub(super) version: u64, diff --git a/crates/omnigraph/src/exec/merge.rs b/crates/omnigraph/src/exec/merge.rs index 2f6877e6a..3cfe7e0ed 100644 --- a/crates/omnigraph/src/exec/merge.rs +++ b/crates/omnigraph/src/exec/merge.rs @@ -3499,10 +3499,8 @@ fn row_id_at(batch: &RecordBatch, row: usize) -> Result { Ok(ids.value(row).to_string()) } -/// The manifest projects the greatest numeric version for each table identity. -/// Native refs have independent version histories, so adopting an equal or -/// lower source version cannot replace the target's current registration. -/// This selects a publication route; row comparison still determines the delta. +/// A source version at or below the target's must be written onto the target's +/// lineage (`docs/dev/merge.md` §Table classification); rows still decide the delta. fn adopt_requires_target_lineage( source_entry: &crate::db::DatasetEntry, target_entry: Option<&crate::db::DatasetEntry>, @@ -3528,22 +3526,13 @@ fn adopt_advances_head( target_entry.and_then(|entry| entry.native_dataset_branch.as_deref()) == Some(target_branch) } - // A newer source on main (pointer switch) or an unowned target (fork). _ => false, } } -/// Classify a table whose target state equals base (the adopt / fast-forward -/// case). A proven insertion-only descendant becomes -/// [`CandidateTableState::AdoptPureInserts`]; every other non-empty delta that -/// advances target HEAD becomes [`CandidateTableState::AdoptWithDelta`] with -/// its write payload pre-computed for recovery planning. Pointer switches and -/// forks become [`CandidateTableState::AdoptSourceState`] and do not advance -/// data HEAD. -/// -/// The HEAD-advancing subcases also include adoption that cannot replace the -/// target's greatest registered version. Computing the delta here (rather than -/// inside the publish) lets recovery own every required target-lineage write. +/// Classify a table whose target equals base: a proven insertion-only descendant +/// is `AdoptPureInserts`, any other HEAD-advancing delta (a source at or below the +/// target's version included) is `AdoptWithDelta`, a newer pointer or fork is `AdoptSourceState`. async fn classify_adopt( target_db: &Omnigraph, catalog: &Catalog, @@ -3637,12 +3626,9 @@ async fn classify_general_adopt( } } -/// What publishing a table's adopted source state does to `__manifest`. -/// -/// An empty delta does not imply an empty publish: source and target can hold -/// the same content at different Lance versions. Planning purely lets -/// classification drop a table whose registration is already stored, which the -/// registry guard would otherwise reject (#473). +/// What publishing a table's adopted source state does to `__manifest`; planning +/// lets classification drop a table whose registration is already stored (#473) +/// or whose source version is at or below the target's. #[must_use = "the adopt plan decides whether this table is a merge candidate"] enum AdoptPublish { /// The planned registration is field-for-field the stored entry. @@ -3657,11 +3643,8 @@ enum AdoptPublish { }, } -/// Plan what adopting the source's table state publishes, without an effect. -/// -/// Reaching a branch-bearing arm means the delta was empty: the HEAD-advancing -/// case is classified [`CandidateTableState::AdoptWithDelta`] and published by -/// [`publish_adopted_delta`]. +/// Plan what adopting the source's table state publishes, without an effect; +/// only a source newer than the target reaches a branch-bearing arm. fn plan_adopted_source_state( target_active: Option<&str>, source_entry: &crate::db::DatasetEntry, @@ -3800,6 +3783,21 @@ mod adopt_plan_tests { } } + /// `adopt_requires_target_lineage` mirrors the projection fold's `>=` + /// (`db/manifest/state.rs`): a source at or below the target's version + /// advances HEAD; a newer source on main does not; no target entry, no rule. + #[test] + fn source_at_or_below_target_version_advances_head() { + let target = entry(4, Some("feature"), 3, "manifest-v4"); + let below = entry(3, None, 3, "manifest-v3"); + let equal = entry(4, None, 3, "manifest-v4-main"); + let above = entry(5, None, 3, "manifest-v5"); + assert!(adopt_advances_head(Some("feature"), &below, Some(&target))); + assert!(adopt_advances_head(Some("feature"), &equal, Some(&target))); + assert!(!adopt_advances_head(Some("feature"), &above, Some(&target))); + assert!(!adopt_advances_head(Some("feature"), &below, None)); + } + /// The #473 shape: the source advanced two Lance versions on a branch and /// came back to the target's content. The plan must be `Nothing`. /// @@ -3930,6 +3928,16 @@ async fn publish_adopted_source_state( )?; match plan_adopted_source_state(target_active, source_entry, target_entry, table_key) { + AdoptPublish::Pointer(update) + if target_entry.is_some_and(|current| { + update.published_dataset_version <= current.published_dataset_version + }) => + { + Err(OmniError::manifest_internal(format!( + "branch merge table '{table_key}' plans a pointer at version {} at or below the target's registration; classification must route that onto the target lineage", + update.published_dataset_version + ))) + } AdoptPublish::Pointer(update) => Ok(update), AdoptPublish::Fork { source_branch, @@ -5074,6 +5082,7 @@ impl Omnigraph { }), }; let mut candidates: HashMap = HashMap::new(); + let mut fenced_but_unpublished_table_keys: Vec = Vec::new(); let empty_external_preflight = crate::table_store::ExternalBlobPreflight::default(); let mut blob_table_keys = HashSet::new(); let mut blob_selection = crate::table_store::PersistedBlobSelection::default(); @@ -5104,7 +5113,7 @@ impl Omnigraph { let has_blob = schema_has_blob(&schema_for_table_key(catalog, table_key)?)?; if !has_blob { if same_manifest_state(base_entry, target_entry) { - if let Some(candidate) = classify_adopt( + match classify_adopt( self, catalog, base_snapshot, @@ -5116,7 +5125,10 @@ impl Omnigraph { ) .await? { - candidates.insert(table_key.clone(), candidate); + Some(candidate) => { + candidates.insert(table_key.clone(), candidate); + } + None => fenced_but_unpublished_table_keys.push(table_key.clone()), } } else { let table_walk_timing = crate::instrumentation::start_merge_timing( @@ -5317,8 +5329,11 @@ impl Omnigraph { ) .await? }; - if let Some(candidate) = candidate { - candidates.insert(table_key.clone(), candidate); + match candidate { + Some(candidate) => { + candidates.insert(table_key.clone(), candidate); + } + None => fenced_but_unpublished_table_keys.push(table_key.clone()), } continue; } @@ -5491,6 +5506,7 @@ impl Omnigraph { let expected_versions = candidates .keys() + .chain(fenced_but_unpublished_table_keys.iter()) .filter_map(|table_key| { let identity = target_snapshot .dataset(table_key) diff --git a/crates/omnigraph/tests/branching.rs b/crates/omnigraph/tests/branching.rs index c0580000a..86aeffcf6 100644 --- a/crates/omnigraph/tests/branching.rs +++ b/crates/omnigraph/tests/branching.rs @@ -1828,35 +1828,9 @@ async fn branch_merge_applies_node_insert_to_main() { let dir = tempfile::tempdir().unwrap(); let uri = dir.path().to_str().unwrap(); let main = init_and_load(&dir).await; - // Warm this handle before another writer adds an edge. Creating a fresh - // branch must inherit the new edge even though this handle predates it. - assert_eq!(count_rows(&main, "edge:Knows").await, 3); - let writer = Omnigraph::open(uri).await.unwrap(); - writer - .mutate( - "main", - MUTATION_QUERIES, - "add_friend", - ¶ms(&[("$from", "Alice"), ("$to", "Diana")]), - ) - .await - .unwrap(); - let expected_edge = snapshot_main(&writer) - .await - .unwrap() - .dataset("edge:Knows") - .unwrap() - .clone(); - let mut expected_ids = collect_column_strings(&read_table(&writer, "edge:Knows").await, "id"); - expected_ids.sort(); - assert_eq!(expected_ids.len(), 4); main.branch_create("feature").await.unwrap(); let mut feature = Omnigraph::open(uri).await.unwrap(); - assert_eq!( - count_rows_branch(&feature, "feature", "edge:Knows").await, - 4 - ); mutate_branch( &mut feature, "feature", @@ -1867,32 +1841,6 @@ async fn branch_merge_applies_node_insert_to_main() { .await .unwrap(); - let branch_edge = snapshot_branch(&feature, "feature") - .await - .unwrap() - .dataset("edge:Knows") - .unwrap() - .clone(); - assert_eq!(branch_edge.dataset_path, expected_edge.dataset_path); - assert_eq!( - branch_edge.native_dataset_branch, - expected_edge.native_dataset_branch - ); - assert_eq!( - branch_edge.published_dataset_version, - expected_edge.published_dataset_version - ); - assert_eq!(branch_edge.entity_count, expected_edge.entity_count); - let mut branch_ids = collect_column_strings( - &read_table_branch(&feature, "feature", "edge:Knows").await, - "id", - ); - branch_ids.sort(); - assert_eq!( - branch_ids, expected_ids, - "an unrelated node write must preserve every inherited edge" - ); - let outcome = feature.branch_merge("feature", "main").await.unwrap(); assert_eq!(outcome, MergeOutcome::FastForward); @@ -1906,167 +1854,13 @@ async fn branch_merge_applies_node_insert_to_main() { .await .unwrap(); assert_eq!(qr.num_rows(), 1); - for handle in [&main, &feature, &reopened] { - let mut ids = collect_column_strings(&read_table(handle, "edge:Knows").await, "id"); - ids.sort(); - assert_eq!( - ids, expected_ids, - "fast-forward must preserve untouched edge IDs on warm and fresh handles" - ); - } -} - -// The .gqt runner exposes query, mutate, and restart, but no branch create or -// merge steps. Rust also lets these regressions inspect native table versions -// and arrange an external commit between a cached read and merge planning. -#[tokio::test] -async fn branch_merge_preserves_untouched_edges_after_external_commit_and_read() { - let dir = tempfile::tempdir().unwrap(); - let uri = dir.path().to_str().unwrap(); - let main = init_and_load(&dir).await; - main.branch_create("feature").await.unwrap(); - assert_eq!(count_rows(&main, "edge:Knows").await, 3); - - let writer = Omnigraph::open(uri).await.unwrap(); - writer - .mutate( - "main", - MUTATION_QUERIES, - "add_friend", - ¶ms(&[("$from", "Alice"), ("$to", "Diana")]), - ) - .await - .unwrap(); - writer - .mutate( - "feature", - MUTATION_QUERIES, - "insert_person", - &mixed_params(&[("$name", "Eve")], &[("$age", 22)]), - ) - .await - .unwrap(); - // Refresh main's table view through the old handle before asking that same - // handle to classify the now-divergent branches. - assert_eq!(count_rows(&main, "edge:Knows").await, 4); - let outcome = main.branch_merge("feature", "main").await.unwrap(); - assert_eq!(outcome, MergeOutcome::Merged); - for handle in [&main, &Omnigraph::open(uri).await.unwrap()] { - assert_eq!(count_rows(handle, "edge:Knows").await, 4); - let friends = handle - .query( - ReadTarget::branch("main"), - TEST_QUERIES, - "friends_of", - ¶ms(&[("$name", "Alice")]), - ) - .await - .unwrap(); - assert_eq!(first_column_sorted(&friends), ["Bob", "Charlie", "Diana"]); - assert_eq!(count_rows(handle, "node:Person").await, 5); - } -} - -#[tokio::test] -async fn branch_merge_preserves_untouched_edges_after_round_trip() { - let dir = tempfile::tempdir().unwrap(); - let uri = dir.path().to_str().unwrap(); - let db = init_and_load(&dir).await; - db.branch_create("feature").await.unwrap(); - // Build a longer native history on the branch than replaying its net - // changes will produce on main. Every pair returns to the same three rows. - for _ in 0..8 { - db.mutate( - "feature", - MUTATION_QUERIES, - "add_friend", - ¶ms(&[("$from", "Diana"), ("$to", "Alice")]), - ) - .await - .unwrap(); - db.mutate( - "feature", - MUTATION_QUERIES, - "remove_friendship", - ¶ms(&[("$from", "Diana")]), - ) - .await - .unwrap(); - } - db.mutate( - "feature", - MUTATION_QUERIES, - "remove_friendship", - ¶ms(&[("$from", "Bob")]), - ) - .await - .unwrap(); - assert_eq!( - db.branch_merge("feature", "main").await.unwrap(), - MergeOutcome::FastForward - ); - db.mutate( - "main", - MUTATION_QUERIES, - "add_friend", - ¶ms(&[("$from", "Alice"), ("$to", "Diana")]), - ) - .await - .unwrap(); - let expected_rows = read_table(&db, "edge:Knows").await; - let mut expected_ids = collect_column_strings(&expected_rows, "id"); - expected_ids.sort(); - assert_eq!(expected_ids.len(), 3); - let main_edge = snapshot_main(&db) - .await - .unwrap() - .dataset("edge:Knows") - .unwrap() - .clone(); - let feature_edge = snapshot_branch(&db, "feature") - .await - .unwrap() - .dataset("edge:Knows") - .unwrap() - .clone(); - assert!(feature_edge.published_dataset_version > main_edge.published_dataset_version); - assert_eq!( - db.branch_merge("main", "feature").await.unwrap(), - MergeOutcome::FastForward - ); - let inherited_count = count_rows_branch(&db, "feature", "edge:Knows").await; - // Only the Person table changes before the final merge. The edge added on - // main must survive the round trip, including a fresh handle's raw scan. - db.mutate( - "feature", - MUTATION_QUERIES, - "insert_person", - &mixed_params(&[("$name", "Eve")], &[("$age", 22)]), - ) - .await - .unwrap(); - assert_eq!( - db.branch_merge("feature", "main").await.unwrap(), - MergeOutcome::FastForward - ); - let reopened = Omnigraph::open(uri).await.unwrap(); - let mut actual_ids = collect_column_strings(&read_table(&reopened, "edge:Knows").await, "id"); - actual_ids.sort(); - assert_eq!( - actual_ids, expected_ids, - "an unrelated node edit and merge must not remove the edge inserted on main" - ); - assert_eq!( - inherited_count, 3, - "main's added edge must already be visible on feature before its unrelated edit" - ); } +/// Rust because the pins are native table versions, the target ref's physical +/// HEAD, and the entry retained on an empty delta; the row-visible half is +/// `merge_adopt_*.gqt`. #[tokio::test] async fn branch_merge_preserves_state_when_native_versions_differ() { - // The short histories make native versions equal; the long histories put - // the source below the target. A lazy child additionally requires - // recovery-owned first-touch forking without changing the indexed schema. for branch_updates in [8, 2] { for lazy_target in [false, true] { let dir = tempfile::tempdir().unwrap(); @@ -2086,8 +1880,6 @@ async fn branch_merge_preserves_state_when_native_versions_differ() { .unwrap(); } let (source, target) = if lazy_target { - // Keep the changes disjoint: a long main history and one Bob - // edit on feature merge into one additional feature commit. main.mutate( "feature", MUTATION_QUERIES, @@ -2108,11 +1900,6 @@ async fn branch_merge_preserves_state_when_native_versions_differ() { ) .await .unwrap(); - // Inherit main's root-owned indexed table. Cloning feature's - // already-cloned indexes would hit the separate Lance #7840 - // bug pinned by second_generation_branch_index_reads_fail_upstream - // in lance_surface_guards.rs. This first-generation fork still - // exercises the merge's lazy-target recovery route. main.branch_create_from(ReadTarget::branch("main"), "child") .await .unwrap(); @@ -2238,9 +2025,6 @@ async fn branch_merge_preserves_state_when_native_versions_differ() { "{target}, {branch_updates} updates: an unrelated edit must not roll back main" ); - // Main now has the same Person rows at a lower or equal version. - // Bringing it back is an empty delta: retain the target's complete - // public registration and do not create a physical table commit. let before_empty = snapshot_branch(&main, target) .await .unwrap() @@ -2276,6 +2060,10 @@ async fn branch_merge_preserves_state_when_native_versions_differ() { before_empty.published_dataset_version ); assert_eq!(after_empty.entity_count, before_empty.entity_count); + assert!( + after_empty.same_registration(&before_empty), + "empty adoption must retain the target's Lance manifest metadata" + ); } assert_eq!( open_dataset_head(&table_uri, before_empty.native_dataset_branch.as_deref()) diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs index 91aacd649..ee1c7ca4c 100644 --- a/crates/omnigraph/tests/failpoints.rs +++ b/crates/omnigraph/tests/failpoints.rs @@ -10603,12 +10603,15 @@ async fn pre_upgrade_v1_branch_merge_sidecar_rolls_forward_not_back() { /// (D2) — without `CommitGraph::open_at_branch`, the recovery sweep /// would record the global head as the merge parent on a non-main /// target, and future merges between the same pair would lose -/// already-up-to-date detection. Fast-forwards from an equal or lower native -/// source version must also recover the delta written onto the target ref. -#[tokio::test] +/// already-up-to-date detection. +#[test] #[serial] #[serial(branch_merge_phase_b)] -async fn branch_merge_phase_b_failure_recovered_on_non_main_target() { +fn branch_merge_phase_b_failure_recovered_on_non_main_target() { + on_big_stack(branch_merge_phase_b_failure_recovered_on_non_main_target_inner); +} + +async fn branch_merge_phase_b_failure_recovered_on_non_main_target_inner() { use omnigraph::loader::{LoadMode, load_jsonl}; let _scenario = FailScenario::setup(); @@ -10616,7 +10619,9 @@ async fn branch_merge_phase_b_failure_recovered_on_non_main_target() { ("three-way", 0, "source_branch"), ("equal native versions", 2, "main"), ("lower source native version", 8, "main"), + ("lower source native version, lazy target", 8, "feature"), ] { + let lazy_target = source_branch == "feature"; let dir = tempfile::tempdir().unwrap(); let uri = dir.path().to_str().unwrap().to_string(); let db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap(); @@ -10628,9 +10633,44 @@ async fn branch_merge_phase_b_failure_recovered_on_non_main_target() { ) .await .unwrap(); - db.branch_create("target_branch").await.unwrap(); - if target_updates == 0 { - // Preserve the divergent source/target merge coverage. + if lazy_target { + db.branch_create(source_branch).await.unwrap(); + for age in 40..40 + target_updates + 1 { + db.mutate( + "main", + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "alice")], &[("$age", age)]), + ) + .await + .unwrap(); + } + db.mutate( + source_branch, + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Bob")], &[("$age", 26)]), + ) + .await + .unwrap(); + assert_eq!( + db.branch_merge("main", source_branch).await.unwrap(), + omnigraph::db::MergeOutcome::Merged, + "{case}" + ); + db.mutate( + source_branch, + MUTATION_QUERIES, + "set_age", + &mixed_params(&[("$name", "alice")], &[("$age", 50)]), + ) + .await + .unwrap(); + db.branch_create("target_branch").await.unwrap(); + } else { + db.branch_create("target_branch").await.unwrap(); + } + if !lazy_target && target_updates == 0 { db.mutate( "target_branch", MUTATION_QUERIES, @@ -10648,10 +10688,7 @@ async fn branch_merge_phase_b_failure_recovered_on_non_main_target() { ) .await .unwrap(); - } else { - // Replaying these updates onto main takes one native commit. - // Main then advances graph lineage while its table version remains - // equal to or lower than the target's independent native history. + } else if !lazy_target { for age in 40..40 + target_updates { db.mutate( "target_branch", @@ -10717,8 +10754,6 @@ async fn branch_merge_phase_b_failure_recovered_on_non_main_target() { .unwrap(); drop(db); - // The fixed effect set is confirmed, but graph publication has not - // happened. Recovery must publish it onto this non-main target. let operation_id = { let db = Omnigraph::open(&uri).await.unwrap(); let _failpoint = ScopedFailPoint::new( @@ -10789,7 +10824,9 @@ async fn branch_merge_phase_b_failure_recovered_on_non_main_target() { Some(source_head.as_str()), "{case}: recovery must retain the captured source parent" ); - let expected = if target_updates == 0 { + let expected = if lazy_target { + vec![("Bob", 26), ("alice", 50)] + } else if target_updates == 0 { vec![("Bob", 40), ("Carol", 50), ("alice", 30)] } else { vec![("alice", 50)] diff --git a/crates/omnigraph/tests/merge_fast_forward.rs b/crates/omnigraph/tests/merge_fast_forward.rs index 56b2e7ef2..f34934916 100644 --- a/crates/omnigraph/tests/merge_fast_forward.rs +++ b/crates/omnigraph/tests/merge_fast_forward.rs @@ -1392,8 +1392,7 @@ async fn fast_forward_merge_streams_blob_columns() { /// so publication must use the update-only keyed stage introduced by #481. /// Overwrite retains the admitted external descriptor on the source branch; /// merge owns the copied bytes while leaving unchanged valid-empty and null -/// siblings distinct. Cover both feature-to-main adoption and main-to-feature -/// adoption when main's native version is lower than the owned target's. +/// siblings distinct. #[tokio::test] async fn blob_changed_only_adopt_uses_known_present_update() { const SET_NOTE: &str = r#" @@ -1476,9 +1475,6 @@ query set_note($title: String, $note: String) { .collect::>() .join("\n"); let (source, target) = if source_main { - // Churn only one note, then merge its net change into main. The owned - // feature table now has a higher native version than main while both - // have the same logical rows. Main's next overwrite remains below it. for step in 0..8 { feature .mutate( diff --git a/docs/dev/merge.md b/docs/dev/merge.md index bb49aca22..03d85d19a 100644 --- a/docs/dev/merge.md +++ b/docs/dev/merge.md @@ -42,11 +42,24 @@ Native Lance version numbers are local to each native branch. The manifest projection nevertheless selects the highest registered version for a table lifetime, so adopting a source version less than or equal to the target's version cannot safely replace its pointer. A nonempty source delta uses the -existing target-lineage writer, creating a native target ref lazily when -needed. A proved empty delta retains the target's complete table entry, -including its native ref and version metadata. The version comparison selects -the publication route; it never proves row equality. This rule also governs -Blob preflight so validation and publication agree on which rows are copied. +target-lineage writer, creating a native target ref lazily when needed. A +proved empty delta retains the target's complete table entry, including its +native ref and version metadata. The version comparison selects the +publication route; it never proves row equality. This rule also governs Blob +preflight so validation and publication agree on which rows are copied, which +means a `main`-to-branch adoption that carries Blob payloads is subject to the +keyed-write byte cap like any other delta. The proven insertion replay never +applies to this shape: the proof needs a source version above the base, and +the base equals the target here, so the delta is computed by the ordered walk +even when it turns out empty. + +This comparison is a fence, not the class fix: the projection and the +registry guard order registrations by the native version number, a +per-branch counter, and the same comparison produced #473 and the projection +miss this rule works around. Ordering registrations by the `__manifest` +version that published them (unique and increasing within every branch's +lineage, which is the only scope the projection ever compares) would retire +the rule. ## Proven insertion route diff --git a/docs/releases/v0.11.0.md b/docs/releases/v0.11.0.md index 45f69cbc9..94440d2e0 100644 --- a/docs/releases/v0.11.0.md +++ b/docs/releases/v0.11.0.md @@ -4,6 +4,18 @@ Notes accumulate here until the release is cut. ## Highlights +- **Repeated branch merges preserve previously merged rows and values.** + After a branch accumulated more table versions than `main`, merging newer + data from `main` into that branch could report success while retaining stale + data. Merging the branch back after an unrelated edit could then delete an + edge or roll back a value on `main`. When the two counts were equal, the + merge failed instead with `Concurrent modification: table version N already + exists for identity … with different state`. Merge now applies these changes + through the target's table history and preserves its existing entry when the + rows are unchanged; the equal-count merge fast-forwards. A `main`-to-branch + merge that carries Blob payloads now checks the same keyed-write byte cap as + a branch-to-`main` merge. + - **Managed cluster creation, config upload and retirement.** `cluster create` requests an empty cluster, `cluster push` conditionally prepares referenced config files, and `cluster delete` / `cluster undo-delete` target exact @@ -100,14 +112,6 @@ Notes accumulate here until the release is cut. `Float64` and `count` still returns 0. A JSON row omits the null cell's key, so the type shows only in the Arrow IPC result schema. -- **Repeated branch merges preserve previously merged rows and values.** - After a branch accumulated more table versions than `main`, merging newer - data from `main` into that branch could report success while retaining stale - data. Merging the branch back after an unrelated edit could then delete an - edge or roll back a value on `main`. Merge now applies these changes through - the target's table history and preserves its existing entry when the rows - are unchanged. No graph-storage or recovery-format change is required. - - **Two projections producing one result column name are refused at compile time (`T25`).** `return { $a.num1 as number, $a.num2 as number }` used to run and emit two columns both named `number`; every reader that diff --git a/docs/user/branching/merge.md b/docs/user/branching/merge.md index 91a28a7d7..105718ef0 100644 --- a/docs/user/branching/merge.md +++ b/docs/user/branching/merge.md @@ -29,7 +29,9 @@ refuse one beside it -- and no front offers a conditional merge today. A merge preserves changes already integrated into a branch when you later merge that branch back. For example, after merging a new edge from `main` into `review`, editing only a node on `review` and merging it into `main` -preserves that edge. +preserves that edge. A merge whose source and target tables reached the same +version count on different branches fast-forwards; earlier releases refused +it with a `table version … already exists` error. ## Outcomes