diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 35e152a1a..bf257eb75 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -23,7 +23,7 @@ use crate::spec::core_options::{ QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION, TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION, TABLE_TYPE_OPTION, }; -use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType, VarCharType}; +use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType}; use crate::spec::{ remove_field_scoped_options, rename_field_scoped_options, validate_no_aggregation_on_sequence_field, AggregationConfig, BlobType, ColumnMove, @@ -1416,7 +1416,7 @@ impl Schema { Self::validate_first_row_changelog_producer(options)?; Self::validate_changelog_producer_requires_primary_keys(options, primary_keys)?; Self::validate_read_sequence_number_requires_primary_keys(options, primary_keys)?; - Self::validate_rowkind_field(options, primary_keys, fields)?; + Self::validate_rowkind_field(options, fields)?; Self::validate_deletion_vectors(options)?; Self::validate_bucket_keys(options, fields, partition_keys, primary_keys)?; Self::validate_sequence_field(options, fields, partition_keys, primary_keys)?; @@ -2120,28 +2120,14 @@ impl Schema { fn validate_rowkind_field( options: &HashMap, - primary_keys: &[String], fields: &[DataField], ) -> crate::Result<()> { let core = CoreOptions::new(options); let Some(field_name) = core.rowkind_field() else { return Ok(()); }; - - if primary_keys.is_empty() { - return Err(crate::Error::ConfigInvalid { - message: "rowkind.field requires a primary-key table".to_string(), - }); - } - - let producer = core - .try_changelog_producer() + core.try_changelog_producer() .map_err(Self::options_error_to_config_invalid)?; - if producer == ChangelogProducer::Input { - return Err(crate::Error::ConfigInvalid { - message: "rowkind.field cannot be used with changelog-producer=input".to_string(), - }); - } let Some(field) = fields.iter().find(|f| f.name() == field_name) else { return Err(crate::Error::ConfigInvalid { @@ -2149,13 +2135,10 @@ impl Schema { }); }; - if !matches!( - field.data_type(), - DataType::VarChar(v) if v.length() == VarCharType::MAX_LENGTH - ) { + if !matches!(field.data_type(), DataType::Char(_) | DataType::VarChar(_)) { return Err(crate::Error::ConfigInvalid { message: format!( - "rowkind.field '{field_name}' must be STRING (unbounded VARCHAR) type" + "rowkind.field '{field_name}' must be CHARACTER_STRING (CHAR or VARCHAR) type" ), }); } @@ -6591,17 +6574,13 @@ mod tests { } #[test] - fn rowkind_field_requires_primary_key_table() { - let err = Schema::builder() + fn rowkind_field_accepts_append_tables() { + Schema::builder() .column("id", DataType::Int(IntType::new())) .column("op", DataType::VarChar(VarCharType::string_type())) .option("rowkind.field", "op") .build() - .unwrap_err(); - assert!( - matches!(err, crate::Error::ConfigInvalid { ref message } if message.contains("rowkind.field")), - "got {err:?}" - ); + .unwrap(); } #[test] @@ -6690,20 +6669,15 @@ mod tests { } #[test] - fn rowkind_field_rejects_changelog_producer_input() { - let err = Schema::builder() + fn rowkind_field_accepts_changelog_producer_input() { + Schema::builder() .column("id", DataType::Int(IntType::new())) .column("op", DataType::VarChar(VarCharType::string_type())) .primary_key(["id"]) .option("changelog-producer", "input") .option("rowkind.field", "op") .build() - .unwrap_err(); - assert!( - matches!(err, crate::Error::ConfigInvalid { ref message } - if message.contains("changelog-producer=input")), - "got {err:?}" - ); + .unwrap(); } #[test] diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index f11c34267..80c7e3964 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -302,7 +302,8 @@ pub struct Table { query_auth_session: Option, rest_env: Option, /// True when this table copy was switched to a historical schema by - /// [`Table::copy_with_time_travel`]. Such a copy is read-only. + /// [`Table::copy_with_time_travel`]. Ordinary writes and publication reject + /// this copy; row-level updates may use it as their read snapshot. time_traveled: bool, /// Captured snapshot or empty view, so scans don't resolve latest again. /// Cleared when [`Table::copy_with_options`] changes the selector. @@ -801,7 +802,7 @@ impl Table { }) } - /// Create a read-only copy pinned to an already resolved snapshot. + /// Create a read view pinned to an already resolved snapshot. /// /// Replaces any selector that originally resolved the snapshot with an /// explicit `scan.snapshot-id`, so every subsequent scan stage observes the @@ -817,7 +818,7 @@ impl Table { Ok(table) } - /// Create a read-only copy pinned to a snapshot from this table and branch, + /// Create a read view pinned to a snapshot from this table and branch, /// preserving the current read schema and options other than scan selectors. /// /// Unlike time travel, pinning must not change the fields used by an already @@ -1133,8 +1134,9 @@ impl Table { }) } - /// Whether this copy is read-only because its schema was time-traveled or - /// its read view was explicitly pinned (including an empty view). + /// Whether this copy has a time-traveled schema or an explicitly pinned + /// read view (including an empty view). Ordinary writes and commits reject + /// it; row-level updates retain its snapshot as their conflict baseline. pub fn is_time_traveled(&self) -> bool { self.time_traveled } diff --git a/crates/paimon/src/table/row_kind_generator.rs b/crates/paimon/src/table/row_kind_generator.rs index f037dabe5..9631182a1 100644 --- a/crates/paimon/src/table/row_kind_generator.rs +++ b/crates/paimon/src/table/row_kind_generator.rs @@ -19,17 +19,17 @@ //! //! Reference: Java `org.apache.paimon.table.sink.RowKindGenerator`. -use arrow_array::{Array, LargeStringArray, RecordBatch, StringArray}; +use arrow_array::{LargeStringArray, RecordBatch, StringArray}; use crate::spec::{RowKind, TableSchema}; pub struct RowKindGenerator { - index: usize, + field_name: String, } impl RowKindGenerator { pub fn create(schema: &TableSchema, field_name: &str) -> crate::Result { - let index = schema + schema .fields() .iter() .position(|f| f.name() == field_name) @@ -40,28 +40,42 @@ impl RowKindGenerator { ), source: None, })?; - Ok(Self { index }) + Ok(Self { + field_name: field_name.to_owned(), + }) } - pub fn generate(&self, batch: &RecordBatch, row: usize) -> crate::Result { - let col = batch.column(self.index); - if col.is_null(row) { - return Err(crate::Error::DataInvalid { - message: "Row kind cannot be null.".to_string(), - source: None, - }); - } - let value = if let Some(strings) = col.as_any().downcast_ref::() { - strings.value(row) + pub fn generate_batch(&self, batch: &RecordBatch) -> crate::Result> { + // with_write_type may reorder the physical columns. Resolve once for + // the batch, then parse values without a per-row field-name search. + let col = + batch + .column_by_name(&self.field_name) + .ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "Cannot find rowkind field '{}' in write input", + self.field_name + ), + source: None, + })?; + let parse = |value: Option<&str>| { + value + .ok_or_else(|| crate::Error::DataInvalid { + message: "Row kind cannot be null.".into(), + source: None, + }) + .and_then(RowKind::from_short_string) + }; + if let Some(strings) = col.as_any().downcast_ref::() { + strings.iter().map(parse).collect() } else if let Some(strings) = col.as_any().downcast_ref::() { - strings.value(row) + strings.iter().map(parse).collect() } else { - return Err(crate::Error::DataInvalid { - message: "rowkind.field column must be Utf8/String".to_string(), + Err(crate::Error::DataInvalid { + message: "rowkind.field column must be Utf8/String".into(), source: None, - }); - }; - RowKind::from_short_string(value) + }) + } } } @@ -104,6 +118,6 @@ mod tests { ], ) .unwrap(); - assert_eq!(gen.generate(&batch, 0).unwrap(), RowKind::Delete); + assert_eq!(gen.generate_batch(&batch).unwrap(), vec![RowKind::Delete]); } } diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index df9ec40a8..20846ea6f 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -1796,11 +1796,10 @@ impl TableCommit { self.check_global_index_sources(latest_snapshot, entries, new_index_entries) .await?; let mut index_entries = new_index_entries.clone(); - index_entries.extend(self.global_index_update_entries( - &previous, - entries, - new_index_entries, - )?); + index_entries.extend( + self.global_index_update_entries(&previous, entries, new_index_entries) + .await?, + ); let all = Self::merge_index_entries(&previous, &index_entries, false)?; self.check_deletion_vector_references( latest_snapshot, @@ -2019,34 +2018,69 @@ impl TableCommit { Ok(all) } - fn global_index_update_entries( + async fn global_index_update_entries( &self, previous_entries: &[IndexManifestEntry], commit_entries: &[ManifestEntry], new_index_entries: &[IndexManifestEntry], ) -> Result> { + if !previous_entries + .iter() + .any(|entry| entry.index_file.global_index_meta.is_some()) + { + return Ok(vec![]); + } + let action = + CoreOptions::new(self.table.schema().options()).global_index_column_update_action()?; + if action == GlobalIndexColumnUpdateAction::Ignore { + return Ok(vec![]); + } let mut updated_cols = HashSet::new(); + let mut updated_field_ids = HashSet::new(); let mut written_partitions: Vec> = Vec::new(); for entry in commit_entries .iter() .filter(|entry| *entry.kind() == FileKind::Add) { - let Some(write_cols) = entry.file().write_cols.as_ref() else { - continue; - }; + let file = entry.file(); // Dedicated normal/vector/blob INSERT files also carry write_cols. // Only files targeting existing row IDs can invalidate an existing index. - if entry.file().first_row_id.is_none() - && !write_cols - .iter() - .any(|col| col == crate::spec::ROW_ID_FIELD_NAME) + if file.first_row_id.is_none() + && !file.write_cols.as_ref().is_some_and(|columns| { + columns + .iter() + .any(|col| col == crate::spec::ROW_ID_FIELD_NAME) + }) { continue; } - for col in write_cols { - if !is_system_field(col) { - updated_cols.insert(col.clone()); - } + // Partial files retain their writer's schema, including historical + // names. Index metadata identifies fields by ID; comparing against + // today's names would miss renames and confuse dropped/readded names. + let historical_schema = if file.schema_id == self.table.schema().id() { + None + } else { + Some(self.table.schema_manager().schema(file.schema_id).await?) + }; + let schema = historical_schema + .as_deref() + .unwrap_or_else(|| self.table.schema()); + let fields = match &file.write_cols { + Some(columns) => super::data_evolution_fields::project_by_paths( + schema.fields(), + &columns + .iter() + .filter(|name| !is_system_field(name)) + .cloned() + .collect::>(), + )?, + // A complete-row update can omit write_cols, as can a compact + // normal file which excludes dedicated Blob/vector columns. + None => schema.data_file_fields(None), + }; + for field in fields { + updated_field_ids.insert(field.id()); + updated_cols.insert(field.name().to_string()); } if !written_partitions .iter() @@ -2089,8 +2123,8 @@ impl TableCommit { } let matched = indexed_field_ids .iter() + .filter(|field_id| updated_field_ids.contains(field_id)) .filter_map(|field_id| field_by_id.get(field_id)) - .filter(|field_name| updated_cols.contains(*field_name)) .cloned() .collect::>(); if !matched.is_empty() { @@ -2102,7 +2136,7 @@ impl TableCommit { return Ok(vec![]); } - match CoreOptions::new(self.table.schema().options()).global_index_column_update_action()? { + match action { GlobalIndexColumnUpdateAction::Ignore => Ok(vec![]), GlobalIndexColumnUpdateAction::DropPartitionIndex => Ok(affected .into_iter() @@ -5206,6 +5240,94 @@ mod tests { assert_eq!(index_entries[0].index_file.file_name, "lumina-0.index"); } + #[tokio::test] + async fn test_index_update_policy_uses_file_schema_identity() { + for mode in ["rename", "drop_readd", "complete_row"] { + for action in ["THROW_ERROR", "DROP_PARTITION_INDEX", "IGNORE"] { + for extra_field in [false, true] { + let file_io = test_file_io(); + let path = format!("memory:/index_identity_{mode}_{action}_{extra_field}"); + setup_dirs(&file_io, &path).await; + let old = test_schema(); + let mut encoded = serde_json::to_value(&old).unwrap(); + encoded["id"] = serde_json::json!(1); + if mode == "rename" { + encoded["fields"][0]["name"] = serde_json::json!("renamed_id"); + } else if mode == "drop_readd" { + encoded["fields"][0]["id"] = serde_json::json!(2); + encoded["highestFieldId"] = serde_json::json!(2); + } + let current: TableSchema = serde_json::from_value(encoded).unwrap(); + let current = current.copy_with_options(HashMap::from([( + "global-index.column-update-action".into(), + action.into(), + )])); + let index_id = current.fields()[0].id(); + let table = Table::new( + file_io.clone(), + Identifier::new("default", "t"), + path.clone(), + current, + None, + ); + file_io.mkdirs(&format!("{path}/schema/")).await.unwrap(); + file_io + .new_output(&format!("{path}/schema/schema-0")) + .unwrap() + .write(serde_json::to_vec(&old).unwrap().into()) + .await + .unwrap(); + let commit = TableCommit::new(table, "test-user".into()); + let index_file = if extra_field { + test_global_index_file_with_extra_fields( + "id.index", + 1, + vec![index_id], + 0, + 9, + ) + } else { + test_global_index_file("id.index", index_id, 0, 9) + }; + let previous = vec![IndexManifestEntry { + kind: FileKind::Add, + partition: vec![], + bucket: 0, + index_file, + version: 1, + }]; + let mut file = test_data_file("update.parquet", 10); + file.first_row_id = Some(0); + if mode == "complete_row" { + file.schema_id = 1; + file.write_cols = None; + } else { + file.write_cols = Some(vec!["id".into()]); + } + let entries = + commit.messages_to_entries(&[CommitMessage::new(vec![], 0, vec![file])]); + let result = commit + .global_index_update_entries(&previous, &entries, &[]) + .await; + if mode == "drop_readd" || action == "IGNORE" { + assert!(result.unwrap().is_empty(), "{mode}/{action}/{extra_field}"); + } else if action == "THROW_ERROR" { + let error = result.unwrap_err().to_string(); + assert!(error.contains("globally indexed columns"), "{error}"); + if mode == "rename" { + assert!(error.contains("renamed_id"), "{error}"); + } + } else { + let deleted = result.unwrap(); + assert_eq!(deleted.len(), 1, "{mode}/{action}/{extra_field}"); + assert_eq!(deleted[0].kind, FileKind::Delete); + assert_eq!(deleted[0].index_file, previous[0].index_file); + } + } + } + } + } + #[tokio::test] async fn test_partial_update_indexed_column_rejects_by_default() { let file_io = test_file_io(); diff --git a/crates/paimon/src/table/table_merge_input.rs b/crates/paimon/src/table/table_merge_input.rs index 4cb83ed30..8c95ee071 100644 --- a/crates/paimon/src/table/table_merge_input.rs +++ b/crates/paimon/src/table/table_merge_input.rs @@ -84,7 +84,8 @@ impl MergeInput { }); } crate::spec::CoreOptions::new(table.schema().options()).validate_scan_options()?; - let schema = crate::arrow::build_target_arrow_schema(table.schema().fields())?; + let schema = + crate::arrow::build_target_arrow_schema(&table.schema().fields_with_read_metadata())?; for column in columns { schema .field_with_name(column) diff --git a/crates/paimon/src/table/table_update.rs b/crates/paimon/src/table/table_update.rs index 490010808..de07c4d3a 100644 --- a/crates/paimon/src/table/table_update.rs +++ b/crates/paimon/src/table/table_update.rs @@ -46,6 +46,10 @@ fn invalid(message: impl Into) -> crate::Error { /// Row-ID updates infer columns from the input unless configured with /// [`with_update_type`](Self::with_update_type). Key-based upserts accept a /// column subset including the keys and match keys within each partition. +/// Row updates, deletes and shard updates may select a historical read view; +/// their messages retain its snapshot as the conflict baseline. Publish those +/// messages using a current table. Upsert and MERGE with INSERT clauses +/// require a current table as ordinary writes do. #[derive(Clone)] pub struct TableUpdate { table: Table, @@ -256,6 +260,8 @@ impl TableUpdate { batches: Vec, upsert_keys: Vec, ) -> crate::Result> { + // Unlike partial row updates, an upsert may append complete rows. + super::write_builder::ensure_table_write_allowed(&self.table)?; let mut writer = super::table_upsert::TableUpsert::new( &self.table, self.commit_user.clone(), @@ -281,6 +287,9 @@ impl TableUpdate { when_matched: Vec, when_not_matched: Vec, ) -> crate::Result> { + if !when_not_matched.is_empty() { + super::write_builder::ensure_table_write_allowed(&self.table)?; + } super::table_merge_into::merge_into( &self.table, &self.commit_user, @@ -314,11 +323,9 @@ impl TableUpdate { if row_ids.is_empty() { return Ok(Vec::new()); } - let mut writer = self - .table - .new_write_builder() - .with_commit_user(self.commit_user.clone())? - .new_delete()?; + // The high-level updater selects a read snapshot; the low-level + // ordinary write builder continues rejecting historical handles. + let mut writer = super::DataEvolutionDeleteWriter::new(&self.table)?; writer.add_row_ids(row_ids)?; writer.prepare_commit().await } diff --git a/crates/paimon/src/table/table_update_delete.rs b/crates/paimon/src/table/table_update_delete.rs index bd2d99e35..d7029350b 100644 --- a/crates/paimon/src/table/table_update_delete.rs +++ b/crates/paimon/src/table/table_update_delete.rs @@ -30,7 +30,7 @@ pub(super) async fn delete( table: &Table, predicate: Option, ) -> crate::Result> { - super::write_builder::ensure_table_write_allowed(table)?; + super::write_builder::ensure_paimon_table(table)?; // Partition-only deletes use manifest metadata directly, but still need // the same unrestricted whole-table grant as row-level deletes. Authorize // before the empty-target and partition-optimization branches. diff --git a/crates/paimon/src/table/table_write.rs b/crates/paimon/src/table/table_write.rs index 80edf85a6..39842497a 100644 --- a/crates/paimon/src/table/table_write.rs +++ b/crates/paimon/src/table/table_write.rs @@ -424,9 +424,7 @@ impl TableWrite { } let row_kind_generator = match core_options.rowkind_field() { - Some(field_name) if has_primary_keys => { - Some(RowKindGenerator::create(schema, field_name)?) - } + Some(field_name) => Some(RowKindGenerator::create(schema, field_name)?), _ => None, }; let row_kind_filter = RowKindFilter::of(core_options); @@ -784,8 +782,9 @@ impl TableWrite { return writer.write(batch).await; } self.ensure_active()?; - let input = batch; - let Some(batch) = self.normalize_write_batch(input)? else { + let filtered = self.filter_write_input(batch)?; + let input = filtered.as_ref().unwrap_or(batch); + let Some(batch) = self.normalize_filtered_write_batch(input)? else { return Ok(()); }; let routing = self.normalize_automatic_routing_input(input)?; @@ -807,7 +806,11 @@ impl TableWrite { pub(super) fn normalize_write_batch(&self, batch: &RecordBatch) -> Result> { let filtered = self.filter_write_input(batch)?; - let batch = self.validate_write_batch_schema(filtered.as_ref().unwrap_or(batch))?; + self.normalize_filtered_write_batch(filtered.as_ref().unwrap_or(batch)) + } + + fn normalize_filtered_write_batch(&self, batch: &RecordBatch) -> Result> { + let batch = self.validate_write_batch_schema(batch)?; // Filtering already precedes value normalization. Generate the system // column now so every surviving PK batch has the same buffer schema. let batch = self.generate_rowkind_batch(&batch)?; @@ -870,16 +873,16 @@ impl TableWrite { /// its optional RowKindFilter, independently of physical KV nullability. /// Ignored rows never reach value extraction or nested writers. fn filter_write_input(&self, batch: &RecordBatch) -> Result> { - if self.primary_key_indices.is_empty() { + // Ordinary Arrow append input has no event header. It represents + // INSERTs and needs no kind allocation or pre-routing filter. + if self.primary_key_indices.is_empty() && self.row_kind_generator.is_none() { return Ok(None); } // Reject malformed schemas even when every event would be ignored. // Empty arrays check equivalent layouts without examining row values. self.validate_write_batch_schema(&RecordBatch::new_empty(batch.schema()))?; let kinds = if let Some(generator) = &self.row_kind_generator { - (0..batch.num_rows()) - .map(|row| generator.generate(batch, row)) - .collect::>>()? + generator.generate_batch(batch)? } else if let Some(column) = batch.column_by_name(VALUE_KIND_FIELD_NAME) { column .as_any() @@ -896,15 +899,18 @@ impl TableWrite { // still identify the event; other engines and partition migrations // require every declared top-level NOT NULL field. let delete_required_fields = self.delete_not_null_fields(); + let projected = self.project_full_write_input(batch)?; + let physical = projected.as_ref().unwrap_or(batch); for (index, field) in self.write_schema.fields().iter().enumerate() { - if field.is_nullable() || batch.column(index).null_count() == 0 { + if field.is_nullable() || physical.column(index).null_count() == 0 { continue; } let delete_required = delete_required_fields .as_ref() .is_none_or(|fields| fields.contains(field.name())); for (row, kind) in kinds.iter().enumerate() { - if batch.column(index).is_null(row) && (*kind != RowKind::Delete || delete_required) + if physical.column(index).is_null(row) + && (*kind != RowKind::Delete || delete_required) { return Err(crate::Error::DataInvalid { message: format!("Cannot write null to non-null column({})", field.name()), @@ -913,14 +919,25 @@ impl TableWrite { } } } - let Some(filter) = &self.row_kind_filter else { - return Ok(None); - }; - let rows = kinds - .into_iter() - .enumerate() - .filter_map(|(row, kind)| filter.test(kind).then_some(row)) - .collect::>(); + let mut rows = Vec::with_capacity(batch.num_rows()); + for (row, kind) in kinds.into_iter().enumerate() { + if self + .row_kind_filter + .as_ref() + .is_some_and(|filter| !filter.test(kind)) + { + continue; + } + if self.primary_key_indices.is_empty() + && !matches!(kind, RowKind::Insert | RowKind::UpdateAfter) + { + return Err(crate::Error::DataInvalid { + message: format!("Append only writer cannot accept row with RowKind {kind:?}"), + source: None, + }); + } + rows.push(row); + } if rows.len() == batch.num_rows() { Ok(None) } else { @@ -1265,6 +1282,11 @@ impl TableWrite { } fn generate_rowkind_batch(&self, batch: &RecordBatch) -> Result { + // Append rows have already been filtered and checked. Their events do + // not become a physical KV system column. + if self.primary_key_indices.is_empty() { + return Ok(batch.clone()); + } let Some(generator) = &self.row_kind_generator else { // Every PK batch sent to a buffered writer uses the same schema. // Mixing default INSERT batches with explicit kinds must neither @@ -1291,9 +1313,11 @@ impl TableWrite { }); } - let kinds = (0..batch.num_rows()) - .map(|row| generator.generate(batch, row).map(|kind| kind.to_value())) - .collect::>>()?; + let kinds = generator + .generate_batch(batch)? + .into_iter() + .map(|kind| kind.to_value()) + .collect(); Self::add_per_row_value_kind_column(batch, kinds) } diff --git a/crates/paimon/src/table/table_write_precomputed.rs b/crates/paimon/src/table/table_write_precomputed.rs index 35f6ad301..3730d3d0f 100644 --- a/crates/paimon/src/table/table_write_precomputed.rs +++ b/crates/paimon/src/table/table_write_precomputed.rs @@ -63,8 +63,9 @@ impl TableWrite { upper_bound - 1, ))); } - let input = batch; - let Some(batch) = self.normalize_write_batch(input)? else { + let filtered = self.filter_write_input(batch)?; + let input = filtered.as_ref().unwrap_or(batch); + let Some(batch) = self.normalize_filtered_write_batch(input)? else { return Ok(()); }; let partitions = self.precomputed_partition_bytes(input, &batch)?; diff --git a/crates/paimon/src/table/write_builder.rs b/crates/paimon/src/table/write_builder.rs index a073c7ecf..2c3744ed3 100644 --- a/crates/paimon/src/table/write_builder.rs +++ b/crates/paimon/src/table/write_builder.rs @@ -262,7 +262,11 @@ impl<'a> PaimonWriteBuilder<'a> { /// Create a configured table update object. pub fn new_update(&self) -> crate::Result { - ensure_table_write_allowed(self.table)?; + // Row-level updates can select an older read snapshot. Their partial + // files retain that schema and their messages carry the snapshot used + // for conflict detection. Appending and publishing still require a + // current table, including append-capable operations on TableUpdate. + ensure_paimon_table(self.table)?; if self.overwrite { return Err(crate::Error::Unsupported { message: "TableUpdate does not support overwrite".to_string(), @@ -304,16 +308,7 @@ pub(super) fn ensure_table_write_authorized(table: &Table) -> crate::Result<()> } pub(super) fn ensure_table_write_allowed(table: &Table) -> crate::Result<()> { - let options = crate::spec::CoreOptions::new(table.schema().options()); - let declared = options.table_type()?; - if declared.requires_table_engine() { - return Err(crate::Error::Unsupported { - message: format!( - "table '{}' is declared '{declared}' and cannot be written as a Paimon table", - table.identifier().full_name() - ), - }); - } + ensure_paimon_table(table)?; // A time-travel table may carry a historical schema. let selector = crate::spec::CoreOptions::new(table.schema().options()).try_time_travel_selector(); @@ -327,6 +322,20 @@ pub(super) fn ensure_table_write_allowed(table: &Table) -> crate::Result<()> { Ok(()) } +pub(super) fn ensure_paimon_table(table: &Table) -> crate::Result<()> { + let options = crate::spec::CoreOptions::new(table.schema().options()); + let declared = options.table_type()?; + if declared.requires_table_engine() { + return Err(crate::Error::Unsupported { + message: format!( + "table '{}' is declared '{declared}' and cannot be written as a Paimon table", + table.identifier().full_name() + ), + }); + } + Ok(()) +} + pub(super) fn validate_commit_user(commit_user: &str) -> crate::Result<()> { let is_invalid = commit_user.is_empty() || commit_user == "." diff --git a/crates/paimon/tests/append_row_kind_test.rs b/crates/paimon/tests/append_row_kind_test.rs new file mode 100644 index 000000000..313e5d400 --- /dev/null +++ b/crates/paimon/tests/append_row_kind_test.rs @@ -0,0 +1,310 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod common; + +use arrow_array::{ArrayRef, Int32Array, Int8Array, RecordBatch, StringArray}; +use common::incremental_helpers::{memory_table, persist_table_schema, setup_dirs}; +use futures::TryStreamExt; +use paimon::spec::{CharType, DataType, IntType, Schema, TableSchema, VarCharType}; +use paimon::table::Table; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use std::sync::Arc; + +async fn make_table(evolution: bool, options: &[(&str, &str)]) -> Table { + let mut schema = Schema::builder() + .column("id", DataType::Int(IntType::with_nullable(false))) + .column("op", DataType::VarChar(VarCharType::string_type())) + .column("pt", DataType::VarChar(VarCharType::string_type())) + .partition_keys(["pt"]) + .option("rowkind.field", "op") + .option("data-evolution.enabled", evolution.to_string()) + .option("row-tracking.enabled", evolution.to_string()); + for (key, value) in options { + schema = schema.option(*key, *value); + } + let path = "memory:/append-row-kind"; + let (io, table) = memory_table(path, TableSchema::new(0, &schema.build().unwrap())); + setup_dirs(&io, path).await; + persist_table_schema(&io, path, table.schema()).await; + table +} + +fn batch(rows: &[(Option, &str, &str)]) -> RecordBatch { + RecordBatch::try_from_iter(vec![ + ( + "id", + Arc::new(Int32Array::from_iter(rows.iter().map(|x| x.0))) as ArrayRef, + ), + ( + "op", + Arc::new(StringArray::from_iter_values(rows.iter().map(|x| x.1))) as ArrayRef, + ), + ( + "pt", + Arc::new(StringArray::from_iter_values(rows.iter().map(|x| x.2))) as ArrayRef, + ), + ]) + .unwrap() +} + +async fn ids(table: &Table) -> Vec { + let read = table.new_read_builder(); + let plan = read.new_scan().plan().await.unwrap(); + let batches: Vec<_> = read + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let mut ids: Vec<_> = batches + .iter() + .flat_map(|batch| { + batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect(); + ids.sort(); + ids +} + +#[tokio::test] +async fn append_rejects_retracts_and_filters_only_configured_events() { + for evolution in [false, true] { + for kind in ["-D", "-U"] { + let table = make_table(evolution, &[]).await; + let mut writer = table.new_write_builder().new_write().unwrap(); + let error = writer + .write_arrow_batch(&batch(&[(Some(1), kind, "bad")])) + .await + .unwrap_err(); + assert!(error.to_string().contains("Append"), "{error}"); + assert!(writer.prepare_commit().await.unwrap().is_empty()); + writer.close().await; + } + let table = make_table(evolution, &[("ignore-delete", "true")]).await; + let builder = table.new_write_builder(); + let mut writer = builder.new_write().unwrap(); + writer + .write_arrow_batch(&batch(&[ + (Some(1), "-D", "bad"), + (Some(2), "+U", "keep"), + (Some(3), "-U", "bad"), + (Some(4), "+I", "keep"), + ])) + .await + .unwrap(); + let messages = writer.prepare_commit().await.unwrap(); + assert_eq!( + messages + .iter() + .flat_map(|m| &m.new_files) + .map(|f| f.row_count) + .sum::(), + 2 + ); + builder.new_commit().commit(messages).await.unwrap(); + writer.close().await; + assert_eq!(ids(&table).await, vec![2, 4]); + } +} + +#[tokio::test] +async fn append_checks_required_values_before_filter_and_keeps_ignore_before_distinct() { + let table = make_table(false, &[("ignore-delete", "true")]).await; + let mut writer = table.new_write_builder().new_write().unwrap(); + let error = writer + .write_arrow_batch(&batch(&[(None, "-D", "bad")])) + .await + .unwrap_err(); + assert!(error.to_string().contains("non-null column(id)"), "{error}"); + assert!(writer.prepare_commit().await.unwrap().is_empty()); + writer.close().await; + + let table = make_table(false, &[("ignore-update-before", "true")]).await; + let mut writer = table.new_write_builder().new_write().unwrap(); + writer + .write_arrow_batch(&batch(&[(Some(1), "-U", "bad")])) + .await + .unwrap(); + let error = writer + .write_arrow_batch(&batch(&[(Some(2), "-D", "bad")])) + .await + .unwrap_err(); + assert!(error.to_string().contains("Append"), "{error}"); + assert!(writer.prepare_commit().await.unwrap().is_empty()); + writer.close().await; +} + +#[tokio::test] +async fn append_selected_write_type_filters_the_same_rows_as_routing() { + for precomputed in [false, true] { + for reordered in [false, true] { + let table = make_table( + false, + &[ + ("ignore-delete", "true"), + ("bucket", "2"), + ("bucket-key", "id"), + ], + ) + .await; + let builder = table.new_write_builder(); + let mut writer = builder.new_write().unwrap(); + // Full input retains omitted event/routing fields, whereas a + // reordered input resolves the generator at its current position. + let names = if reordered { + vec!["op", "id", "pt"] + } else { + vec!["id"] + }; + writer + .with_write_type(names.into_iter().map(str::to_owned).collect()) + .unwrap(); + let input = batch(&[(Some(1), "-D", "bad"), (Some(2), "+I", "keep")]); + let input = if reordered { + input.project(&[1, 0, 2]).unwrap() + } else { + input + }; + writer + .write_arrow(&[input], precomputed.then_some(0)) + .await + .unwrap(); + let messages = writer.prepare_commit().await.unwrap(); + assert_eq!( + messages + .iter() + .flat_map(|m| &m.new_files) + .map(|f| f.row_count) + .sum::(), + 1 + ); + builder.new_commit().commit(messages).await.unwrap(); + writer.close().await; + assert_eq!(ids(&table).await, vec![2]); + } + } +} + +#[tokio::test] +async fn generated_primary_key_kinds_preserve_input_changelog_events() { + for op_type in [ + DataType::VarChar(VarCharType::string_type()), + DataType::VarChar(VarCharType::new(2).unwrap()), + DataType::Char(CharType::new(2).unwrap()), + ] { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .column("op", op_type) + .primary_key(["id"]) + .option("bucket", "1") + .option("rowkind.field", "op") + .option("changelog-producer", "input") + .build() + .unwrap(); + let path = "memory:/generated-input-changelog"; + let (io, table) = memory_table(path, TableSchema::new(0, &schema)); + setup_dirs(&io, path).await; + persist_table_schema(&io, path, table.schema()).await; + let builder = table.new_write_builder(); + let mut writer = builder.new_write().unwrap(); + let input = RecordBatch::try_from_iter(vec![ + ( + "id", + Arc::new(Int32Array::from(vec![1, 1, 2, 2])) as ArrayRef, + ), + ( + "value", + Arc::new(Int32Array::from(vec![10, 20, 25, 30])) as ArrayRef, + ), + ( + "op", + Arc::new(StringArray::from(vec!["+I", "-D", "-U", "+U"])) as ArrayRef, + ), + ]) + .unwrap(); + writer.write_arrow_batch(&input).await.unwrap(); + let messages = writer.prepare_commit().await.unwrap(); + let logs: Vec<_> = messages + .iter() + .flat_map(|m| &m.new_changelog_files) + .collect(); + assert_eq!(logs.len(), 1); + assert_eq!(logs[0].row_count, 4); + assert_eq!(logs[0].delete_row_count, Some(2)); + let bytes = io + .new_input(&format!("{path}/bucket-0/{}", logs[0].file_name)) + .unwrap() + .read() + .await + .unwrap(); + let batches = ParquetRecordBatchReaderBuilder::try_new(bytes) + .unwrap() + .build() + .unwrap() + .collect::, _>>() + .unwrap(); + let values: Vec<_> = batches + .iter() + .flat_map(|batch| { + batch + .column_by_name("value") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect(); + let kinds: Vec<_> = batches + .iter() + .flat_map(|batch| { + batch + .column_by_name("_VALUE_KIND") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect(); + assert_eq!(values, vec![10, 20, 25, 30]); + assert_eq!(kinds, vec![0, 3, 1, 2]); + builder.new_commit().commit(messages).await.unwrap(); + assert_eq!(ids(&table).await, vec![2]); + let snapshot = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + assert_eq!(snapshot.changelog_record_count(), Some(4)); + writer.close().await; + } +} diff --git a/crates/paimon/tests/historical_table_update_test.rs b/crates/paimon/tests/historical_table_update_test.rs new file mode 100644 index 000000000..4a2a92eb8 --- /dev/null +++ b/crates/paimon/tests/historical_table_update_test.rs @@ -0,0 +1,581 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod common; + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow_array::{Array, ArrayRef, Int32Array, Int64Array, RecordBatch}; +use common::incremental_helpers::{memory_table, persist_table_schema, setup_dirs, write_batch}; +use futures::TryStreamExt; +use paimon::spec::{DataType, Datum, IntType, PredicateBuilder, Schema, TableSchema}; +use paimon::table::{ + CommitMessage, MergeAssignment, MergeSource, Table, UpdateAssignment, WhenMatched, + WhenNotMatched, +}; + +fn batch(names: &[&str], values: &[Vec]) -> RecordBatch { + RecordBatch::try_from_iter(names.iter().zip(values).map(|(name, values)| { + ( + *name, + Arc::new(Int32Array::from(values.clone())) as ArrayRef, + ) + })) + .unwrap() +} + +async fn evolved_table(path: &str) -> (Table, Table) { + let schema = |extra| { + let mut builder = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .column("keep", DataType::Int(IntType::new())) + .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") + .option("deletion-vectors.enabled", "true"); + if extra { + builder = builder.column("extra", DataType::Int(IntType::new())); + } + TableSchema::new(i64::from(extra), &builder.build().unwrap()) + }; + let (io, original) = memory_table(path, schema(false)); + setup_dirs(&io, path).await; + persist_table_schema(&io, path, original.schema()).await; + write_batch( + &original, + &batch( + &["id", "value", "keep"], + &[vec![1, 2], vec![10, 20], vec![100, 200]], + ), + ) + .await; + let latest_schema = schema(true); + persist_table_schema(&io, path, &latest_schema).await; + let latest = original + .copy_with_resolved_schema(latest_schema, "main") + .unwrap(); + write_batch( + &latest, + &batch( + &["id", "value", "keep", "extra"], + &[vec![3], vec![30], vec![300], vec![999]], + ), + ) + .await; + let historical = latest + .copy_with_time_travel(HashMap::from([("scan.snapshot-id".into(), "1".into())])) + .await + .unwrap(); + assert_eq!(historical.schema().id(), 0); + assert!(historical.is_time_traveled()); + (latest, historical) +} + +async fn commit(table: &Table, messages: Vec) { + table + .new_write_builder() + .new_commit() + .commit(messages) + .await + .unwrap(); +} + +async fn rows(table: &Table) -> Vec>> { + let builder = table.new_read_builder(); + let plan = builder.new_scan().plan().await.unwrap(); + let batches: Vec<_> = builder + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let mut rows: Vec>> = Vec::new(); + for batch in batches { + for row in 0..batch.num_rows() { + rows.push( + batch + .columns() + .iter() + .map(|array| { + array.is_valid(row).then(|| { + if let Some(array) = array.as_any().downcast_ref::() { + i64::from(array.value(row)) + } else { + array + .as_any() + .downcast_ref::() + .unwrap() + .value(row) + } + }) + }) + .collect(), + ); + } + } + rows.sort_by_key(|row| row[0]); + rows +} + +fn changed() -> RecordBatch { + RecordBatch::try_from_iter([ + ("_ROW_ID", Arc::new(Int64Array::from(vec![0])) as ArrayRef), + ("value", Arc::new(Int32Array::from(vec![11])) as ArrayRef), + ]) + .unwrap() +} + +#[tokio::test] +async fn historical_updates_keep_schema_and_conflict_baseline() { + for operation in [ + "row_id", + "incremental", + "groups", + "predicate", + "shard", + "merge", + ] { + let (latest, historical) = + evolved_table(&format!("memory:/historical_update_{operation}")).await; + let mut update = historical.new_write_builder().new_update().unwrap(); + update.with_update_type(vec!["value".into()]).unwrap(); + let messages = match operation { + "row_id" => update + .update_by_arrow_with_row_id(vec![changed()]) + .await + .unwrap(), + "groups" => update + .update_by_arrow_batches_with_row_id([Ok(vec![changed()])]) + .await + .unwrap(), + "incremental" => update + .new_update_by_row_id() + .await + .unwrap() + .update_columns(vec![changed()], vec!["value".into()]) + .await + .unwrap(), + "predicate" => update + .update_by_predicate( + Some( + PredicateBuilder::new(historical.schema().fields()) + .equal("id", Datum::Int(1)) + .unwrap(), + ), + vec![( + "value".into(), + UpdateAssignment::Array(vec![Arc::new(Int32Array::from(vec![11]))]), + )], + Vec::new(), + ) + .await + .unwrap(), + "merge" => update + .merge_into( + MergeSource::Batches(vec![batch(&["id", "value"], &[vec![1], vec![11]])]), + vec![("id".into(), "id".into())], + vec![WhenMatched { + condition: None, + delete: false, + assignments: vec![( + "value".into(), + MergeAssignment::SourceColumn("value".into()), + )], + }], + Vec::new(), + ) + .await + .unwrap(), + "shard" => { + let mut shard = update.new_shard_updater(0, 1, None).await.unwrap(); + let inputs: Vec<_> = shard.arrow_reader().unwrap().try_collect().await.unwrap(); + assert_eq!(inputs.iter().map(RecordBatch::num_rows).sum::(), 2); + for input in inputs { + let ids = input + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + shard + .update_by_arrow_batch(batch( + &["value"], + &[ids + .values() + .iter() + .map(|id| if *id == 1 { 11 } else { 20 }) + .collect()], + )) + .await + .unwrap(); + } + shard.prepare_commit().await.unwrap() + } + _ => unreachable!(), + }; + assert!(!messages.is_empty(), "{operation}"); + for message in &messages { + assert_eq!(message.check_from_snapshot, Some(1), "{operation}"); + for file in &message.new_files { + assert_eq!(file.schema_id, 0); + assert_eq!(file.write_cols, Some(vec!["value".into()])); + } + } + commit(&latest, messages).await; + let expected = vec![ + vec![Some(1), Some(11), Some(100), None], + vec![Some(2), Some(20), Some(200), None], + vec![Some(3), Some(30), Some(300), Some(999)], + ]; + assert_eq!(rows(&latest).await, expected, "{operation}"); + // Selection and check_from_snapshot must both keep snapshot 1. A + // second update of the same old file must not silently overwrite it. + let stale = update + .update_by_arrow_with_row_id(vec![changed()]) + .await + .unwrap(); + assert!(stale + .iter() + .all(|message| message.check_from_snapshot == Some(1))); + let error = latest + .new_write_builder() + .new_commit() + .commit(stale) + .await + .unwrap_err(); + assert!( + error.to_string().contains("conflict"), + "{operation}: {error}" + ); + assert_eq!(rows(&latest).await, expected); + } +} + +#[tokio::test] +async fn historical_deletes_preserve_newer_rows() { + for operation in ["row_id", "predicate"] { + let (latest, historical) = + evolved_table(&format!("memory:/historical_delete_{operation}")).await; + let update = historical.new_write_builder().new_update().unwrap(); + let messages = if operation == "row_id" { + update.delete_by_row_id(vec![0, 0]).await.unwrap() + } else { + update + .delete_by_predicate(Some( + PredicateBuilder::new(historical.schema().fields()) + .equal("id", Datum::Int(1)) + .unwrap(), + )) + .await + .unwrap() + }; + assert!(!messages.is_empty()); + assert!(messages + .iter() + .all(|message| message.check_from_snapshot == Some(1))); + commit(&latest, messages).await; + assert_eq!( + rows(&latest).await, + vec![ + vec![Some(2), Some(20), Some(200), None], + vec![Some(3), Some(30), Some(300), Some(999)], + ] + ); + } +} + +#[tokio::test] +async fn historical_updater_cannot_append_or_publish() { + let (latest, historical) = evolved_table("memory:/historical_write_guards").await; + assert!(historical.new_write_builder().new_write().is_err()); + assert!(historical.new_write_builder().try_new_commit().is_err()); + assert!(historical.new_write_builder().new_delete().is_err()); + assert!(historical + .new_write_builder() + .new_data_evolution_writer(vec!["value".into()]) + .is_err()); + let update = historical.new_write_builder().new_update().unwrap(); + let upsert = update + .upsert_by_arrow_with_key( + vec![batch( + &["id", "value", "keep"], + &[vec![1], vec![11], vec![100]], + )], + vec!["id".into()], + ) + .await + .unwrap_err(); + assert!(upsert.to_string().contains("time-travel option")); + let merge = update + .merge_into( + MergeSource::Batches(Vec::new()), + Vec::new(), + Vec::new(), + vec![WhenNotMatched { + condition: None, + assignments: Vec::new(), + }], + ) + .await + .unwrap_err(); + assert!(merge.to_string().contains("time-travel option")); + assert_eq!( + rows(&latest).await, + vec![ + vec![Some(1), Some(10), Some(100), None], + vec![Some(2), Some(20), Some(200), None], + vec![Some(3), Some(30), Some(300), Some(999)], + ] + ); +} + +#[tokio::test] +async fn historical_partial_files_follow_field_ids_after_schema_changes() { + for change in ["rename", "drop_readd", "widen"] { + let (table, historical) = + evolved_table(&format!("memory:/historical_fields_{change}")).await; + let mut schema = serde_json::to_value(table.schema()).unwrap(); + schema["id"] = 2.into(); + match change { + "rename" => schema["fields"][1]["name"] = "renamed".into(), + "drop_readd" => { + schema["fields"][1]["id"] = 4.into(); + schema["highestFieldId"] = 4.into(); + } + "widen" => schema["fields"][1]["type"] = "BIGINT".into(), + _ => unreachable!(), + } + let schema: TableSchema = serde_json::from_value(schema).unwrap(); + persist_table_schema(table.file_io(), table.location(), &schema).await; + let latest = table.copy_with_resolved_schema(schema, "main").unwrap(); + let value = if change == "widen" { 1i64 << 40 } else { 40 }; + let values: ArrayRef = if change == "widen" { + Arc::new(Int64Array::from(vec![value])) + } else { + Arc::new(Int32Array::from(vec![40])) + }; + let input = RecordBatch::try_new( + paimon::arrow::build_target_arrow_schema(latest.schema().fields()).unwrap(), + vec![ + Arc::new(Int32Array::from(vec![4])), + values, + Arc::new(Int32Array::from(vec![400])), + Arc::new(Int32Array::from(vec![1000])), + ], + ) + .unwrap(); + write_batch(&latest, &input).await; + let messages = historical + .new_write_builder() + .new_update() + .unwrap() + .update_by_arrow_with_row_id(vec![changed()]) + .await + .unwrap(); + assert!(messages + .iter() + .all(|message| message.check_from_snapshot == Some(1))); + assert!(messages + .iter() + .flat_map(|message| &message.new_files) + .all(|file| file.schema_id == 0 && file.write_cols == Some(vec!["value".into()]))); + commit(&latest, messages).await; + let expected = if change == "drop_readd" { + [None, None, None] + } else { + [Some(11), Some(20), Some(30)] + }; + assert_eq!( + rows(&latest).await, + vec![ + vec![Some(1), expected[0], Some(100), None], + vec![Some(2), expected[1], Some(200), None], + vec![Some(3), expected[2], Some(300), Some(999)], + vec![Some(4), Some(value), Some(400), Some(1000)], + ], + "{change}" + ); + } +} + +#[tokio::test] +async fn historical_deletion_vectors_cannot_restore_a_concurrent_deletion() { + let (latest, historical) = evolved_table("memory:/historical_delete_conflict").await; + let newer = latest + .new_write_builder() + .new_update() + .unwrap() + .delete_by_row_id(vec![1]) + .await + .unwrap(); + commit(&latest, newer).await; + let stale = historical + .new_write_builder() + .new_update() + .unwrap() + .delete_by_row_id(vec![0]) + .await + .unwrap(); + assert!(stale + .iter() + .all(|message| message.check_from_snapshot == Some(1))); + let error = latest + .new_write_builder() + .new_commit() + .commit(stale) + .await + .unwrap_err(); + assert!( + error.to_string().to_ascii_lowercase().contains("conflict"), + "{error}" + ); + assert_eq!( + rows(&latest).await, + vec![ + vec![Some(1), Some(10), Some(100), None], + vec![Some(3), Some(30), Some(300), Some(999)], + ] + ); +} + +#[tokio::test] +async fn captured_empty_update_views_do_not_observe_later_rows() { + let (latest, _) = evolved_table("memory:/captured_empty_update").await; + let empty = latest.copy_with_pinned_snapshot(None); + let update = empty.new_write_builder().new_update().unwrap(); + let messages = update + .update_by_predicate( + None, + vec![( + "value".into(), + UpdateAssignment::Function(Arc::new(|_| panic!("empty view evaluated a callback"))), + )], + vec!["value".into()], + ) + .await + .unwrap(); + assert!(messages.is_empty()); + assert!(update.delete_by_predicate(None).await.unwrap().is_empty()); + let mut shard = update.new_shard_updater(0, 1, None).await.unwrap(); + let batches: Vec<_> = shard.arrow_reader().unwrap().try_collect().await.unwrap(); + assert!(batches.is_empty()); + assert!(shard.prepare_commit().await.unwrap().is_empty()); + let messages = update + .merge_into( + MergeSource::Batches(vec![batch(&["id", "value"], &[vec![1], vec![11]])]), + vec![("id".into(), "id".into())], + vec![WhenMatched { + condition: None, + delete: false, + assignments: vec![( + "value".into(), + MergeAssignment::SourceColumn("value".into()), + )], + }], + Vec::new(), + ) + .await + .unwrap(); + assert!(messages.is_empty()); + let error = update + .update_by_arrow_with_row_id(vec![changed()]) + .await + .unwrap_err(); + assert!( + error.to_string().contains("No files with row tracking"), + "{error}" + ); + assert_eq!(rows(&latest).await.len(), 3); +} + +#[tokio::test] +async fn row_id_merge_reads_the_independently_selected_source_view() { + for source_view in ["current", "snapshot", "empty"] { + let (latest, historical) = + evolved_table(&format!("memory:/merge_source_view_{source_view}")).await; + let changed = RecordBatch::try_from_iter([ + ("_ROW_ID", Arc::new(Int64Array::from(vec![0])) as ArrayRef), + ("value", Arc::new(Int32Array::from(vec![99])) as ArrayRef), + ]) + .unwrap(); + commit( + &latest, + latest + .new_write_builder() + .new_update() + .unwrap() + .update_by_arrow_with_row_id(vec![changed]) + .await + .unwrap(), + ) + .await; + let source = match source_view { + "snapshot" => latest + .copy_with_time_travel(HashMap::from([("scan.snapshot-id".into(), "3".into())])) + .await + .unwrap(), + "empty" => latest.copy_with_pinned_snapshot(None), + _ => latest.clone(), + }; + let assignment = if source_view == "empty" { + MergeAssignment::Value(UpdateAssignment::Function(Arc::new(|_| { + panic!("an empty MERGE source evaluated an assignment") + }))) + } else { + MergeAssignment::SourceColumn("value".into()) + }; + let messages = historical + .new_write_builder() + .new_update() + .unwrap() + .merge_into( + MergeSource::Table(Arc::new(source)), + vec![("_ROW_ID".into(), "_ROW_ID".into())], + vec![WhenMatched { + condition: None, + delete: false, + assignments: vec![("keep".into(), assignment)], + }], + Vec::new(), + ) + .await + .unwrap(); + let kept = if source_view == "empty" { + assert!(messages.is_empty()); + [100, 200] + } else { + assert!(!messages.is_empty()); + assert!(messages + .iter() + .all(|message| message.check_from_snapshot == Some(1))); + commit(&latest, messages).await; + [99, 20] + }; + assert_eq!( + rows(&latest).await, + vec![ + vec![Some(1), Some(99), Some(kept[0]), None], + vec![Some(2), Some(20), Some(kept[1]), None], + vec![Some(3), Some(30), Some(300), Some(999)], + ], + "{source_view}" + ); + } +} diff --git a/crates/paimon/tests/table_predicate_delete_test.rs b/crates/paimon/tests/table_predicate_delete_test.rs index cc30f3447..1dc8e4414 100644 --- a/crates/paimon/tests/table_predicate_delete_test.rs +++ b/crates/paimon/tests/table_predicate_delete_test.rs @@ -495,7 +495,7 @@ async fn row_delete_rejects_concurrent_dv_replacement() { } #[tokio::test] -async fn predicate_delete_rejects_time_travel_auth_and_inconsistent_field_indices() { +async fn historical_predicate_delete_keeps_auth_and_field_index_validation() { let table = table(&[]).await; seed(&table).await; let pb = PredicateBuilder::new(table.schema().fields()); @@ -516,7 +516,18 @@ async fn predicate_delete_rejects_time_travel_auth_and_inconsistent_field_indice .contains("does not match the table schema")); let historical = table.copy_with_options(HashMap::from([("scan.snapshot-id".into(), "1".into())])); - assert!(historical.new_write_builder().new_update().is_err()); + let messages = historical + .new_write_builder() + .new_update() + .unwrap() + .delete_by_predicate(Some(predicate.clone())) + .await + .unwrap(); + assert!(!messages.is_empty()); + assert!(messages + .iter() + .all(|message| message.check_from_snapshot == Some(1))); + assert!(historical.new_write_builder().new_write().is_err()); let auth = table.copy_with_options(HashMap::from([( "query-auth.enabled".into(), "true".into(), @@ -530,6 +541,9 @@ async fn predicate_delete_rejects_time_travel_auth_and_inconsistent_field_indice .unwrap_err(); assert!(error.to_string().contains("query-auth.enabled")); assert_eq!(ids(&table).await, vec![0, 1, 2, 3]); + write_batch(&table, &batch(vec![4], vec![Some("a")], vec![50], vec![1])).await; + commit(&table, messages).await; + assert_eq!(ids(&table).await, vec![2, 3, 4]); } #[tokio::test] diff --git a/crates/paimon/tests/table_update_test.rs b/crates/paimon/tests/table_update_test.rs index 8689dd5e8..cd39b34d3 100644 --- a/crates/paimon/tests/table_update_test.rs +++ b/crates/paimon/tests/table_update_test.rs @@ -1373,16 +1373,34 @@ async fn empty_update_type_means_all_columns_for_upsert_only() { } #[tokio::test] -async fn update_builder_rejects_overwrite_and_time_travel() { +async fn update_builder_rejects_overwrite_but_allows_historical_partial_updates() { let table = evolution_table().await; assert!(table .new_write_builder() .with_overwrite() .new_update() .is_err()); + seed(&table).await; let historical = table.copy_with_options(HashMap::from([("scan.snapshot-id".into(), "1".into())])); - assert!(historical.new_write_builder().new_update().is_err()); + let messages = historical + .new_write_builder() + .new_update() + .unwrap() + .update_by_arrow_with_row_id(vec![matched(vec![0], &[("value", vec![11])])]) + .await + .unwrap(); + assert!(!messages.is_empty()); + assert!(messages + .iter() + .all(|message| message.check_from_snapshot == Some(1))); + assert!(historical.new_write_builder().new_write().is_err()); + assert!(historical.new_write_builder().try_new_commit().is_err()); + commit(&table, messages).await; + assert_eq!( + read_rows(&table).await, + vec![vec![1, 11, 100], vec![2, 20, 200], vec![3, 30, 300]] + ); } #[tokio::test]