diff --git a/rust/lance/src/io/commit.rs b/rust/lance/src/io/commit.rs index 26f9567609c..321d3700414 100644 --- a/rust/lance/src/io/commit.rs +++ b/rust/lance/src/io/commit.rs @@ -1371,7 +1371,6 @@ pub(crate) async fn commit_transaction( // has not necessarily. So for anything involving writing, use `object_store`. let read_version = transaction.read_version; let mut target_version = read_version + 1; - let original_dataset = dataset.clone(); // read_version sometimes defaults to zero for overwrite. // If num_retries is zero, we are in "strict overwrite" mode. @@ -1388,6 +1387,9 @@ pub(crate) async fn commit_transaction( dataset.clone() }; + // `dataset` advances past the read version inside the retry loop. + let read_version_dataset = dataset.clone(); + let mut transaction = transaction.clone(); let num_attempts = std::cmp::max(commit_config.num_retries, 1); @@ -1418,7 +1420,8 @@ pub(crate) async fn commit_transaction( // started at exact same time better. let mut rebase = - TransactionRebase::try_new(&original_dataset, transaction, affected_rows).await?; + TransactionRebase::try_new(&read_version_dataset, transaction, affected_rows) + .await?; for (other_version, other_transaction) in other_transactions.iter() { rebase.check_txn(other_transaction, *other_version)?; diff --git a/rust/lance/src/io/commit/conflict_resolver.rs b/rust/lance/src/io/commit/conflict_resolver.rs index 0c4e3d28889..1e665ec613e 100644 --- a/rust/lance/src/io/commit/conflict_resolver.rs +++ b/rust/lance/src/io/commit/conflict_resolver.rs @@ -7,15 +7,16 @@ use crate::index::mem_wal::{load_mem_wal_index_details, new_mem_wal_index_meta}; use crate::io::deletion::read_dataset_deletion_file; use crate::{ Dataset, - dataset::transaction::{DataOverlayGroup, Operation, Transaction, UpdateMode}, + dataset::transaction::{DataOverlayGroup, Operation, Transaction, UpdateMap, UpdateMode}, }; use futures::{StreamExt, TryStreamExt}; +use lance_core::datatypes::{Field, Schema}; use lance_core::{Error, Result, utils::deletion::DeletionVector}; use lance_index::frag_reuse::FRAG_REUSE_INDEX_NAME; use lance_index::mem_wal::{CompactedSsTable, MEM_WAL_INDEX_NAME}; use lance_select::{RowAddrTreeMap, RowSetOps}; -use lance_table::format::IndexMetadata; use lance_table::format::overlay::OverlayCoverage; +use lance_table::format::{DataFile, IndexMetadata}; use lance_table::{format::Fragment, io::deletion::write_deletion_file}; use roaring::RoaringBitmap; use std::{ @@ -37,6 +38,89 @@ pub struct TransactionRebase<'a> { /// Compacted SSTables from conflicting UpdateMemWalState transactions. /// Used when rebasing CreateIndex of MemWalIndex. conflicting_mem_wal_compacted_sstables: Vec, + /// Field tree at `transaction.read_version`. + read_version_schema: Schema, +} + +/// `id` plus its ancestors and descendants. +/// +/// A packed struct or a blob records the parent's id in the data file where a +/// normal struct records its leaves (`collect_columns` in dataset.rs), so a +/// write and a metadata change can name either end. Containment runs both ways. +fn field_lineage(schema: &Schema, id: i32) -> HashSet { + let mut lineage = HashSet::from([id]); + + // On popping a field at `depth`, `path` truncated to `depth` is its + // ancestry. `field_by_id` finds the field but not the way down to it. + let mut path: Vec = Vec::new(); + let mut stack: Vec<(&Field, usize)> = schema.fields.iter().map(|f| (f, 0)).collect(); + let mut found = None; + while let Some((field, depth)) = stack.pop() { + path.truncate(depth); + if field.id == id { + found = Some(field); + break; + } + path.push(field.id); + stack.extend(field.children.iter().map(|c| (c, depth + 1))); + } + + // Absent field: `path` is the branch the walk gave up on, not an ancestry. + let Some(field) = found else { + return lineage; + }; + + lineage.extend(path); + let mut subtree = vec![field]; + while let Some(field) = subtree.pop() { + lineage.insert(field.id); + subtree.extend(field.children.iter()); + } + lineage +} + +/// Fields these data files carry values for. A negative id is a tombstone. +fn live_field_ids<'a>(files: impl Iterator) -> Vec { + files + .flat_map(|f| f.fields.iter().copied()) + .filter(|id| *id >= 0) + .collect() +} + +/// The fields whose values `operation` produced. +/// +/// A row rewrite leaves `fields_modified` empty, naming the columns it +/// recomputed only in the set index maintenance consults -- which is +/// meaningful only once it wrote a fragment. +fn value_producing_fields(operation: &Operation) -> Vec { + match operation { + Operation::Update { + fields_modified, + new_fragments, + fields_for_preserving_frag_bitmap, + .. + } => { + let rewritten = if new_fragments.is_empty() { + &[][..] + } else { + fields_for_preserving_frag_bitmap + }; + fields_modified + .iter() + .chain(rewritten) + .map(|id| *id as i32) + .collect() + } + Operation::DataReplacement { replacements } => { + live_field_ids(replacements.iter().map(|r| &r.1)) + } + Operation::DataOverlay { groups } => live_field_ids( + groups + .iter() + .flat_map(|g| g.overlays.iter().map(|o| &o.data_file)), + ), + _ => Vec::new(), + } } /// Whether `operation` may make a nullability-affecting schema change: a @@ -72,12 +156,30 @@ fn supplies_values(operation: &Operation) -> bool { ) } +/// The dataset at `transaction.read_version`; callers may submit through a +/// handle at any version. A no-op when the caller already resolved it. +async fn dataset_at_read_version<'d>( + dataset: &'d Dataset, + transaction: &Transaction, +) -> Result> { + if transaction.read_version == 0 || dataset.manifest.version == transaction.read_version { + return Ok(Cow::Borrowed(dataset)); + } + Ok(Cow::Owned( + dataset.checkout_version(transaction.read_version).await?, + )) +} + impl<'a> TransactionRebase<'a> { pub async fn try_new( dataset: &Dataset, transaction: Transaction, affected_rows: Option<&'a RowAddrTreeMap>, ) -> Result { + let dataset = dataset_at_read_version(dataset, &transaction).await?; + let dataset = dataset.as_ref(); + let read_version_schema = dataset.schema().clone(); + match &transaction.operation { // These operations add new fragments or don't modify any. Operation::Append { .. } @@ -96,6 +198,7 @@ impl<'a> TransactionRebase<'a> { modified_fragment_ids: HashSet::new(), conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema, }), Operation::Delete { updated_fragments, @@ -124,12 +227,12 @@ impl<'a> TransactionRebase<'a> { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema, }); } let initial_fragments = - initial_fragments_for_rebase(dataset, &transaction, &modified_fragment_ids) - .await; + initial_fragments_for_rebase(dataset, &modified_fragment_ids); Ok(Self { transaction, affected_rows, @@ -137,6 +240,7 @@ impl<'a> TransactionRebase<'a> { modified_fragment_ids, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema, }) } Operation::Rewrite { groups, .. } => { @@ -146,8 +250,7 @@ impl<'a> TransactionRebase<'a> { .collect::>(); let initial_fragments = - initial_fragments_for_rebase(dataset, &transaction, &modified_fragment_ids) - .await; + initial_fragments_for_rebase(dataset, &modified_fragment_ids); Ok(Self { transaction, affected_rows, @@ -155,14 +258,14 @@ impl<'a> TransactionRebase<'a> { modified_fragment_ids, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema, }) } Operation::DataReplacement { replacements } => { let modified_fragment_ids = replacements.iter().map(|r| r.0).collect::>(); let initial_fragments = - initial_fragments_for_rebase(dataset, &transaction, &modified_fragment_ids) - .await; + initial_fragments_for_rebase(dataset, &modified_fragment_ids); Ok(Self { transaction, affected_rows, @@ -170,14 +273,14 @@ impl<'a> TransactionRebase<'a> { modified_fragment_ids, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema, }) } Operation::DataOverlay { groups } => { let modified_fragment_ids = groups.iter().map(|g| g.fragment_id).collect::>(); let initial_fragments = - initial_fragments_for_rebase(dataset, &transaction, &modified_fragment_ids) - .await; + initial_fragments_for_rebase(dataset, &modified_fragment_ids); Ok(Self { transaction, affected_rows, @@ -185,13 +288,13 @@ impl<'a> TransactionRebase<'a> { modified_fragment_ids, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema, }) } Operation::Merge { fragments, .. } => { let modified_fragment_ids = fragments.iter().map(|f| f.id).collect::>(); let initial_fragments = - initial_fragments_for_rebase(dataset, &transaction, &modified_fragment_ids) - .await; + initial_fragments_for_rebase(dataset, &modified_fragment_ids); Ok(Self { transaction, affected_rows, @@ -199,11 +302,33 @@ impl<'a> TransactionRebase<'a> { modified_fragment_ids, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema, }) } } } + /// A field's metadata is part of its definition, so a write and a metadata + /// change to a field it produced values for are not independent. Called + /// from both sides: commit order decides only which one rebases. + fn check_field_metadata( + &self, + write: &Operation, + field_metadata_updates: &HashMap, + other_transaction: &Transaction, + other_version: u64, + ) -> Result<()> { + let written = value_producing_fields(write); + let redefined = field_metadata_updates.keys().any(|id| { + let lineage = field_lineage(&self.read_version_schema, *id); + written.iter().any(|field| lineage.contains(field)) + }); + if redefined { + return Err(self.retryable_conflict_err(other_transaction, other_version)); + } + Ok(()) + } + #[track_caller] fn retryable_conflict_err(&self, other_transaction: &Transaction, other_version: u64) -> Error { Error::retryable_commit_conflict_source( @@ -465,8 +590,16 @@ impl<'a> TransactionRebase<'a> { | Operation::ReserveFragments { .. } | Operation::Project { .. } | Operation::Clone { .. } - | Operation::UpdateConfig { .. } | Operation::UpdateBases { .. } => Ok(()), + Operation::UpdateConfig { + field_metadata_updates, + .. + } => self.check_field_metadata( + &self.transaction.operation, + field_metadata_updates, + other_transaction, + other_version, + ), Operation::DataOverlay { groups } => { // Our update recomputed rows from the pre-overlay base, so if // it commits over an overlay it would silently undo the @@ -1082,9 +1215,17 @@ impl<'a> TransactionRebase<'a> { ) -> Result<()> { if let Operation::DataReplacement { replacements } = &self.transaction.operation { match &other_transaction.operation { + Operation::UpdateConfig { + field_metadata_updates, + .. + } => self.check_field_metadata( + &self.transaction.operation, + field_metadata_updates, + other_transaction, + other_version, + ), Operation::Append { .. } | Operation::Clone { .. } - | Operation::UpdateConfig { .. } | Operation::ReserveFragments { .. } | Operation::Project { .. } // Both a column replacement and an overlay preserve physical row @@ -1247,11 +1388,19 @@ impl<'a> TransactionRebase<'a> { | Operation::CreateIndex { .. } | Operation::ReserveFragments { .. } | Operation::Project { .. } - | Operation::UpdateConfig { .. } | Operation::UpdateBases { .. } | Operation::Clone { .. } | Operation::DataReplacement { .. } | Operation::DataOverlay { .. } => Ok(()), + Operation::UpdateConfig { + field_metadata_updates, + .. + } => self.check_field_metadata( + &self.transaction.operation, + field_metadata_updates, + other_transaction, + other_version, + ), // A concurrent Delete only tombstones rows via a deletion vector, // which preserves physical offsets; the overlay value for a deleted // offset is simply inert. Conflict only if the whole overlaid @@ -1490,17 +1639,22 @@ impl<'a> TransactionRebase<'a> { Ok(()) } } + write @ (Operation::Update { .. } + | Operation::DataReplacement { .. } + | Operation::DataOverlay { .. }) => self.check_field_metadata( + write, + field_metadata_updates, + other_transaction, + other_version, + ), Operation::Append { .. } | Operation::Clone { .. } | Operation::Delete { .. } | Operation::CreateIndex { .. } | Operation::Rewrite { .. } - | Operation::DataReplacement { .. } - | Operation::DataOverlay { .. } | Operation::Merge { .. } | Operation::Restore { .. } | Operation::ReserveFragments { .. } - | Operation::Update { .. } | Operation::Project { .. } | Operation::UpdateMemWalState { .. } | Operation::UpdateBases { .. } => Ok(()), @@ -2117,26 +2271,14 @@ impl<'a> TransactionRebase<'a> { } } -async fn initial_fragments_for_rebase( +fn initial_fragments_for_rebase( dataset: &Dataset, - transaction: &Transaction, modified_fragment_ids: &HashSet, ) -> HashMap { if modified_fragment_ids.is_empty() { return HashMap::new(); } - let dataset = if dataset.manifest.version != transaction.read_version { - Cow::Owned( - dataset - .checkout_version(transaction.read_version) - .await - .unwrap(), - ) - } else { - Cow::Borrowed(dataset) - }; - dataset .fragments() .iter() @@ -2191,6 +2333,7 @@ mod tests { use std::{num::NonZero, sync::Arc}; use crate::dataset::transaction::UpdateMode::{RewriteColumns, RewriteRows}; + use crate::dataset::transaction::{UpdateMap, UpdateMapEntry}; use arrow_array::{Int32Array, RecordBatch}; use arrow_schema::{DataType, Field, Schema}; use lance_core::Error; @@ -2210,6 +2353,168 @@ mod tests { io, }; use lance_table::format::DataFile; + use lance_table::format::overlay::DataOverlayFile; + + /// Names the three `Operation::Update` fields these cases turn on and + /// defaults the rest. Ids are `i32` to match the schema. + #[derive(Default)] + struct UpdateForTest { + new_fragments: Vec, + fields_modified: Vec, + fields_for_preserving_frag_bitmap: Vec, + update_mode: Option, + } + + impl From for Operation { + fn from(update: UpdateForTest) -> Self { + let ids = |fields: Vec| fields.into_iter().map(|id| id as u32).collect(); + Self::Update { + removed_fragment_ids: vec![], + updated_fragments: vec![Fragment::new(1)], + new_fragments: update.new_fragments, + fields_modified: ids(update.fields_modified), + compacted_sstables: Vec::new(), + fields_for_preserving_frag_bitmap: ids(update.fields_for_preserving_frag_bitmap), + update_mode: update.update_mode, + inserted_rows_filter: None, + updated_fragment_offsets: None, + } + } + } + + /// Two structs, so a case can name one and rewrite the other. + fn struct_schema() -> lance_core::datatypes::Schema { + let st = |name: &str, children: [&str; 2]| { + Field::new( + name, + DataType::Struct( + children + .iter() + .map(|c| Field::new(*c, DataType::Int32, true)) + .collect(), + ), + true, + ) + }; + let arrow = Schema::new(vec![ + Field::new("id", DataType::Int32, false), + st("a_st", ["x", "y"]), + st("b_st", ["p", "q"]), + ]); + lance_core::datatypes::Schema::try_from(&arrow).unwrap() + } + + /// A metadata change naming field `id`. + fn metadata_on(id: i32) -> Operation { + Operation::UpdateConfig { + config_updates: None, + table_metadata_updates: None, + schema_metadata_updates: None, + field_metadata_updates: HashMap::from([( + id, + UpdateMap { + update_entries: vec![UpdateMapEntry { + key: "definition".to_string(), + value: Some("changed".to_string()), + }], + replace: false, + }, + )]), + } + } + + fn field_id(schema: &lance_core::datatypes::Schema, name: &str) -> i32 { + schema.field(name).unwrap().id + } + + /// The leaf field ids beneath `name`. + fn leaves_of(schema: &lance_core::datatypes::Schema, name: &str) -> Vec { + schema + .field(name) + .unwrap() + .children + .iter() + .map(|f| f.id) + .collect() + } + + /// An in-place column rewrite, which names the columns it recomputed. + fn update_rewriting(fields_modified: &[i32]) -> Operation { + UpdateForTest { + fields_modified: fields_modified.to_vec(), + update_mode: Some(RewriteColumns), + ..Default::default() + } + .into() + } + + /// A row rewrite, as `Dataset::update` and merge insert build it: rows move + /// to a new fragment, so the recomputed columns are named only for indices. + fn update_moving_rows(value_fields: &[i32]) -> Operation { + UpdateForTest { + new_fragments: vec![Fragment::new(2)], + fields_for_preserving_frag_bitmap: value_fields.to_vec(), + update_mode: Some(RewriteRows), + ..Default::default() + } + .into() + } + + /// A delete-only merge, which names the whole schema for index maintenance + /// but writes no fragment. + fn update_deleting(schema: &lance_core::datatypes::Schema) -> Operation { + UpdateForTest { + fields_for_preserving_frag_bitmap: schema.fields_pre_order().map(|f| f.id).collect(), + update_mode: Some(RewriteRows), + ..Default::default() + } + .into() + } + + /// A replacement of `fields`; a negative id is a tombstone. + fn replacement(fields: &[i32]) -> Operation { + Operation::DataReplacement { + replacements: vec![DataReplacementGroup( + 1, + DataFile::new_legacy_from_fields("f.lance", fields.to_vec(), None), + )], + } + } + + /// An overlay carrying new cell values for `fields`. + fn overlay(fields: &[i32]) -> Operation { + Operation::DataOverlay { + groups: vec![DataOverlayGroup { + fragment_id: 1, + overlays: vec![DataOverlayFile { + data_file: DataFile::new_legacy_from_fields("o.lance", fields.to_vec(), None), + coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])), + committed_version: 0, + }], + }], + } + } + + /// Whether `ours` retryably conflicts with `other` under `schema`. + fn conflicts( + ours: &Operation, + other: &Operation, + schema: &lance_core::datatypes::Schema, + ) -> bool { + let mut rebase = TransactionRebase { + transaction: Transaction::new(0, ours.clone(), None), + initial_fragments: HashMap::new(), + modified_fragment_ids: modified_fragment_ids(ours).collect::>(), + affected_rows: None, + conflicting_frag_reuse_indices: Vec::new(), + conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: schema.clone(), + }; + matches!( + rebase.check_txn(&Transaction::new(0, other.clone(), None), 1), + Err(Error::RetryableCommitConflict { .. }) + ) + } async fn test_dataset(num_rows: usize, num_fragments: usize) -> Dataset { let write_params = WriteParams { @@ -2969,7 +3274,9 @@ mod tests { Retryable, // rewrite Compatible, // reserve Retryable, // update - Compatible, // update config + // This update rewrote field 0, whose metadata that + // transaction changes. + Retryable, // update config ], ), ( @@ -3087,7 +3394,7 @@ mod tests { ), ( // Changing field metadata conflicts with another update changing same field - // metadata or overwrite + // metadata or overwrite, and with a write that rewrote that field's data create_update_config_for_test( None, None, @@ -3108,7 +3415,9 @@ mod tests { NotCompatible, // overwrite Compatible, // rewrite Compatible, // reserve - Compatible, // update + // That update rewrote field 0, which this transaction + // redefines; the next case redefines a field it did not. + Retryable, // update NotCompatible, // update config ], ), @@ -3149,6 +3458,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; for (other, expected_conflict) in other_transactions.iter().zip(expected_conflicts) { @@ -3353,6 +3663,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let other_txn = Transaction::new(0, other.clone(), None); let result = rebase.check_txn(&other_txn, 1); @@ -3412,6 +3723,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let other_txn = Transaction::new(0, other.clone(), None); let result = rebase.check_txn(&other_txn, 1); @@ -3524,6 +3836,7 @@ mod tests { affected_rows: affected_rows.as_ref(), conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let other_txn = Transaction::new(0, other.clone(), None); let result = rebase.check_txn(&other_txn, 1); @@ -3566,6 +3879,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = append_rebase.check_txn(&Transaction::new(0, merge.clone(), None), 1); assert_eq!( @@ -3581,6 +3895,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = merge_rebase.check_txn(&Transaction::new(0, append, None), 1); assert!( @@ -3657,6 +3972,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = rebase.check_txn(&Transaction::new(0, theirs, None), 1); assert_eq!( @@ -3790,6 +4106,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let update = Transaction::new( 1, @@ -3853,6 +4170,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let same_name = Transaction::new( @@ -3907,6 +4225,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let different_name_result = rebase.check_txn(&different_name, 1); assert!( @@ -3973,6 +4292,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = rebase.check_txn(&rewrite, 2); if expect_conflict { @@ -4369,6 +4689,154 @@ mod tests { } } + /// Asserts the pair conflicts in both commit orders. + fn assert_conflicts_both_ways( + write: &Operation, + metadata: &Operation, + schema: &lance_core::datatypes::Schema, + ) { + assert!(conflicts(write, metadata, schema), "{write}, {metadata}"); + assert!(conflicts(metadata, write, schema), "{metadata}, {write}"); + } + + /// Asserts the pair is independent in both commit orders. + fn assert_rebases_past( + write: &Operation, + metadata: &Operation, + schema: &lance_core::datatypes::Schema, + ) { + assert!(!conflicts(write, metadata, schema), "{write}, {metadata}"); + assert!(!conflicts(metadata, write, schema), "{metadata}, {write}"); + } + + /// Every shape of write that produces values for `fields`. + fn writes_of(fields: &[i32]) -> [Operation; 4] { + [ + update_rewriting(fields), + update_moving_rows(fields), + replacement(fields), + overlay(fields), + ] + } + + /// A write and a metadata change to a field it wrote are not independent, + /// whichever commits first. + #[test] + fn test_a_write_conflicts_with_a_metadata_change_to_a_field_it_wrote() { + let schema = struct_schema(); + let scalar = field_id(&schema, "id"); + let a_st = field_id(&schema, "a_st"); + let a_leaves = leaves_of(&schema, "a_st"); + + for write in writes_of(&[scalar]) { + assert_conflicts_both_ways(&write, &metadata_on(scalar), &schema); + } + + // Written as leaves, named by the parent: walks down. + for write in writes_of(&a_leaves) { + assert_conflicts_both_ways(&write, &metadata_on(a_st), &schema); + } + + // Written under the parent, metadata names a child: walks up. + for write in writes_of(&[a_st]) { + assert_conflicts_both_ways(&write, &metadata_on(a_leaves[0]), &schema); + } + } + + /// `commit_transaction` resolves the transaction's read version before + /// rebasing, so submitting through an older handle keeps the lineage. + #[tokio::test] + async fn test_lineage_holds_through_an_older_dataset_handle() { + use crate::dataset::NewColumnTransform; + + let mut dataset = test_dataset(5, 1).await; + let stale = dataset.clone(); + dataset + .add_columns( + NewColumnTransform::AllNulls(Arc::new(Schema::new(vec![Field::new_struct( + "a_st", + vec![Field::new("x", DataType::Int32, true)], + true, + )]))), + None, + None, + ) + .await + .unwrap(); + let read_version = dataset.manifest.version; + assert!(stale.manifest.version < read_version); + + // The write names the parent and the metadata names the child, and + // `stale` carries neither field. + let parent = dataset.schema().field("a_st").unwrap().clone(); + CommitBuilder::new(Arc::new(dataset)) + .execute(Transaction::new( + read_version, + metadata_on(parent.children[0].id), + None, + )) + .await + .unwrap(); + + let write: Operation = UpdateForTest { + fields_modified: vec![parent.id], + update_mode: Some(RewriteColumns), + ..Default::default() + } + .into(); + let result = CommitBuilder::new(Arc::new(stale)) + .execute(Transaction::new(read_version, write, None)) + .await; + assert!( + matches!(result, Err(Error::RetryableCommitConflict { .. })), + "{result:?}" + ); + } + + /// Everything the rule deliberately leaves alone. + #[test] + fn test_a_write_rebases_past_metadata_for_a_field_it_did_not_write() { + let schema = struct_schema(); + let a_st = field_id(&schema, "a_st"); + let b_st = field_id(&schema, "b_st"); + let a_leaves = leaves_of(&schema, "a_st"); + + // A sibling subtree, neither above nor below anything written. + for write in writes_of(&a_leaves) { + assert_rebases_past(&write, &metadata_on(b_st), &schema); + } + + // A write that names no field, and a delete-only merge. + assert_rebases_past(&update_rewriting(&[]), &metadata_on(a_st), &schema); + assert_rebases_past(&update_deleting(&schema), &metadata_on(a_st), &schema); + + // Tombstoned ids name no live field. + assert_rebases_past(&replacement(&[-2, -2]), &metadata_on(a_st), &schema); + assert_rebases_past(&overlay(&[-2, -2]), &metadata_on(a_st), &schema); + + // An absent field must not pick up the branch the search gave up on -- + // the last one walked, which ends on the field this write names. + let absent = schema.fields_pre_order().map(|f| f.id).max().unwrap() + 1; + let names_last_walked = update_rewriting(&[field_id(&schema, "id")]); + assert_rebases_past(&names_last_walked, &metadata_on(absent), &schema); + + // Metadata that describes no column at all. + let entry = UpdateMap { + update_entries: vec![UpdateMapEntry { + key: "k".to_string(), + value: Some("v".to_string()), + }], + replace: false, + }; + let no_field = Operation::UpdateConfig { + config_updates: Some(entry.clone()), + table_metadata_updates: Some(entry.clone()), + schema_metadata_updates: Some(entry), + field_metadata_updates: HashMap::new(), + }; + assert_rebases_past(&update_rewriting(&a_leaves), &no_field, &schema); + } + #[tokio::test] async fn test_conflicts_data_replacement() { use io::commit::conflict_resolver::tests::{ConflictResult::*, modified_fragment_ids}; @@ -4619,6 +5087,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = rebase.check_txn(&txn2, 1); @@ -4682,6 +5151,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = rebase.check_txn(&committed_txn, 1); @@ -4720,6 +5190,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = rebase.check_txn(&committed_txn, 1); @@ -4759,6 +5230,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = rebase.check_txn(&committed_txn, 1); @@ -4798,6 +5270,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = rebase.check_txn(&committed_txn, 1); @@ -4848,6 +5321,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result = rebase.check_txn(&committed_txn, 1); @@ -4873,6 +5347,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; let result_higher = rebase_higher.check_txn(&committed_txn, 1); @@ -4919,6 +5394,7 @@ mod tests { affected_rows: None, conflicting_frag_reuse_indices: Vec::new(), conflicting_mem_wal_compacted_sstables: Vec::new(), + read_version_schema: lance_core::datatypes::Schema::default(), }; // CreateIndex of MemWalIndex should be compatible with UpdateMemWalState