diff --git a/crates/integrations/datafusion/tests/pk_tables.rs b/crates/integrations/datafusion/tests/pk_tables.rs index 4b0840c0..0bad6e3b 100644 --- a/crates/integrations/datafusion/tests/pk_tables.rs +++ b/crates/integrations/datafusion/tests/pk_tables.rs @@ -32,9 +32,13 @@ use common::{ collect_id_name, collect_id_value, collect_int_int_str, create_sql_context, create_test_env, row_count, setup_sql_context, string_value, }; -use datafusion::arrow::array::{Array, Int32Array, Int64Array}; +use datafusion::arrow::array::{Array, Int32Array, Int64Array, Int8Array, RecordBatch}; +use datafusion::arrow::datatypes::{ + DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema, +}; use paimon::catalog::Identifier; use paimon::Catalog; +use std::sync::Arc; // ======================= Basic PK Write + Read ======================= @@ -174,6 +178,71 @@ async fn test_pk_partial_update_fixed_bucket_e2e() { ); } +#[tokio::test] +async fn test_pk_partial_update_ignore_delete_alias_e2e() { + let (_tmp, catalog) = create_test_env(); + let sql_context = create_sql_context(catalog.clone()).await; + sql_context + .sql("CREATE SCHEMA paimon.test_db") + .await + .unwrap(); + sql_context + .sql( + "CREATE TABLE paimon.test_db.t_partial_update_ignore_delete ( + id INT NOT NULL, value INT, + PRIMARY KEY (id) + ) WITH ( + 'bucket' = '1', + 'merge-engine' = 'partial-update', + 'partial-update.ignore-delete' = 'true' + )", + ) + .await + .unwrap(); + + let table = catalog + .get_table(&Identifier::new( + "test_db", + "t_partial_update_ignore_delete", + )) + .await + .unwrap(); + let batch = RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new("value", ArrowDataType::Int32, true), + ArrowField::new("_VALUE_KIND", ArrowDataType::Int8, false), + ])), + vec![ + Arc::new(Int32Array::from(vec![1, 1, 2, 1])), + Arc::new(Int32Array::from(vec![ + Some(10), + Some(999), + Some(200), + Some(20), + ])), + Arc::new(Int8Array::from(vec![0, 3, 1, 2])), + ], + ) + .unwrap(); + let write_builder = table.new_write_builder(); + let mut write = write_builder.new_write().unwrap(); + write.write_arrow_batch(&batch).await.unwrap(); + let messages = write.prepare_commit().await.unwrap(); + write_builder.new_commit().commit(messages).await.unwrap(); + + assert_eq!( + collect_id_value( + &sql_context, + "SELECT id, value + FROM paimon.test_db.t_partial_update_ignore_delete + ORDER BY id", + ) + .await, + vec![(1, 20)] + ); +} + /// Partial updates of one key within a single INSERT are merged at flush /// (mirrors Java MergeTreeWriter#flushWriteBuffer): the flushed file holds /// one row per key, so SELECT and COUNT(*) agree. diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index e5a61b54..fd4809bc 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -1966,9 +1966,26 @@ mod tests { let opts = CoreOptions::new(&fallback); assert!(opts.ignore_delete()); + let partial_update = HashMap::from([( + "partial-update.ignore-delete".to_string(), + "true".to_string(), + )]); + let opts = CoreOptions::new(&partial_update); + assert!(opts.ignore_delete()); + let primary = HashMap::from([("ignore-delete".to_string(), "true".to_string())]); let opts = CoreOptions::new(&primary); assert!(opts.ignore_delete()); + + let primary_precedence = HashMap::from([ + ("ignore-delete".to_string(), "false".to_string()), + ( + "partial-update.ignore-delete".to_string(), + "true".to_string(), + ), + ]); + let opts = CoreOptions::new(&primary_precedence); + assert!(!opts.ignore_delete()); } #[test] diff --git a/crates/paimon/src/spec/partial_update.rs b/crates/paimon/src/spec/partial_update.rs index b7ae1b6d..07a57d73 100644 --- a/crates/paimon/src/spec/partial_update.rs +++ b/crates/paimon/src/spec/partial_update.rs @@ -21,6 +21,7 @@ const MERGE_ENGINE_OPTION: &str = "merge-engine"; const PARTIAL_UPDATE_ENGINE: &str = "partial-update"; const IGNORE_DELETE_OPTION: &str = "ignore-delete"; const IGNORE_DELETE_SUFFIX: &str = ".ignore-delete"; +const PARTIAL_UPDATE_IGNORE_DELETE_OPTION: &str = "partial-update.ignore-delete"; const PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION: &str = "partial-update.remove-record-on-delete"; const PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP_OPTION: &str = @@ -116,8 +117,9 @@ impl<'a> PartialUpdateConfig<'a> { } fn is_unsupported_partial_update_option(key: &str) -> bool { - key == IGNORE_DELETE_OPTION - || key.ends_with(IGNORE_DELETE_SUFFIX) + (key.ends_with(IGNORE_DELETE_SUFFIX) + && key != IGNORE_DELETE_OPTION + && key != PARTIAL_UPDATE_IGNORE_DELETE_OPTION) || key == PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION || key == PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP_OPTION || key == FIELDS_DEFAULT_AGG_FUNCTION_OPTION @@ -157,6 +159,32 @@ mod tests { ); } + #[test] + fn test_validate_create_mode_accepts_partial_update_ignore_delete() { + for value in ["true", "false"] { + let options = partial_update_options(&[(PARTIAL_UPDATE_IGNORE_DELETE_OPTION, value)]); + let config = PartialUpdateConfig::new(&options); + + assert_eq!( + config.validate_create_mode(true).unwrap(), + Some(PartialUpdateMode::Basic) + ); + } + } + + #[test] + fn test_validate_create_mode_accepts_ignore_delete() { + for value in ["true", "false"] { + let options = partial_update_options(&[(IGNORE_DELETE_OPTION, value)]); + let config = PartialUpdateConfig::new(&options); + + assert_eq!( + config.validate_create_mode(true).unwrap(), + Some(PartialUpdateMode::Basic) + ); + } + } + #[test] fn test_validate_create_mode_ignores_non_pk_tables() { let options = partial_update_options(&[(IGNORE_DELETE_OPTION, "true")]); @@ -168,10 +196,10 @@ mod tests { #[test] fn test_validate_create_mode_rejects_unsupported_partial_update_options() { for key in [ - IGNORE_DELETE_OPTION, - "partial-update.ignore-delete", PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION, PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP_OPTION, + "deduplicate.ignore-delete", + "fields.price.ignore-delete", "fields.price.sequence-group", "fields.price.aggregate-function", FIELDS_DEFAULT_AGG_FUNCTION_OPTION, diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 413c0d99..0245cf41 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -453,6 +453,15 @@ impl TableSchema { new_schema.highest_field_id = highest_field_id.max(Self::current_highest_field_id(&new_schema.fields)); + if PartialUpdateConfig::new(&new_schema.options).is_enabled() + && self.core_options().ignore_delete() + && !CoreOptions::new(&new_schema.options).ignore_delete() + { + return Err(crate::Error::Unsupported { + message: "Cannot change ignore-delete from true to false.".to_string(), + }); + } + // Re-run create-time validations on the final schema, mirroring Java // `SchemaValidation.validateTableSchema` after applying changes. Schema::validate_key_field_types( @@ -2099,7 +2108,7 @@ mod tests { #[test] fn test_partial_update_schema_validation_rejects_unsupported_options() { for (key, value) in [ - ("ignore-delete", "true"), + ("fields.value.ignore-delete", "true"), ("fields.value.sequence-group", "g1"), ("fields.default-aggregate-function", "last_non_null"), ] { @@ -2119,6 +2128,27 @@ mod tests { } } + #[test] + fn test_partial_update_schema_validation_accepts_ignore_delete_options() { + for (key, value) in [ + ("ignore-delete", "true"), + ("ignore-delete", "false"), + ("partial-update.ignore-delete", "true"), + ("partial-update.ignore-delete", "false"), + ] { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("merge-engine", "partial-update") + .option(key, value) + .build() + .unwrap(); + + assert_eq!(schema.options().get(key).map(String::as_str), Some(value)); + } + } + #[test] fn test_aggregation_schema_validation_accepts_basic_options() { let schema = Schema::builder() @@ -2709,7 +2739,7 @@ mod tests { } #[test] - fn test_partial_update_apply_changes_rejects_unsupported_option() { + fn test_partial_update_apply_changes_accepts_ignore_delete_option() { let table_schema = TableSchema::new( 0, &Schema::builder() @@ -2721,21 +2751,70 @@ mod tests { .unwrap(), ); - let err = table_schema + let new_schema = table_schema .apply_changes(vec![crate::spec::SchemaChange::set_option( "ignore-delete".to_string(), "true".to_string(), )]) - .unwrap_err(); + .unwrap(); - assert!( - matches!(err, crate::Error::ConfigInvalid { ref message } - if message.contains("merge-engine=partial-update") - && message.contains("ignore-delete")), - "partial-update alter should reject unsupported option, got {err:?}" + assert_eq!( + new_schema + .options() + .get("ignore-delete") + .map(String::as_str), + Some("true") ); } + #[test] + fn test_partial_update_apply_changes_rejects_disabling_ignore_delete() { + for (existing_options, change) in [ + ( + vec![("ignore-delete", "true")], + crate::spec::SchemaChange::set_option( + "ignore-delete".to_string(), + "false".to_string(), + ), + ), + ( + vec![("ignore-delete", "true")], + crate::spec::SchemaChange::remove_option("ignore-delete".to_string()), + ), + ( + vec![("partial-update.ignore-delete", "true")], + crate::spec::SchemaChange::set_option( + "ignore-delete".to_string(), + "false".to_string(), + ), + ), + ( + vec![("partial-update.ignore-delete", "true")], + crate::spec::SchemaChange::remove_option( + "partial-update.ignore-delete".to_string(), + ), + ), + ] { + let mut builder = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("merge-engine", "partial-update"); + for (key, value) in existing_options { + builder = builder.option(key, value); + } + let table_schema = TableSchema::new(0, &builder.build().unwrap()); + + let err = table_schema.apply_changes(vec![change]).unwrap_err(); + + assert!( + matches!(err, crate::Error::Unsupported { ref message } + if message.contains("Cannot change ignore-delete from true to false")), + "got {err:?}" + ); + } + } + #[test] fn test_aggregation_apply_changes_accepts_valid_option() { let table_schema = TableSchema::new( diff --git a/crates/paimon/src/table/kv_file_writer.rs b/crates/paimon/src/table/kv_file_writer.rs index 54ffa463..86873e8a 100644 --- a/crates/paimon/src/table/kv_file_writer.rs +++ b/crates/paimon/src/table/kv_file_writer.rs @@ -30,13 +30,13 @@ use crate::arrow::format::create_format_writer; use crate::io::FileIO; use crate::spec::stats::{compute_column_stats, BinaryTableStats}; use crate::spec::{ - extract_datum_from_arrow, AggregationConfig, BinaryRowBuilder, DataFileMeta, DataType, - MergeEngine, PartialUpdateConfig, RowKind, EMPTY_SERIALIZED_ROW, SEQUENCE_NUMBER_FIELD_NAME, - VALUE_KIND_FIELD_NAME, + extract_datum_from_arrow, AggregationConfig, BinaryRowBuilder, CoreOptions, DataFileMeta, + DataType, MergeEngine, PartialUpdateConfig, RowKind, EMPTY_SERIALIZED_ROW, + SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_NAME, }; use crate::table::prepared_files::PreparedFiles; use crate::Result; -use arrow_array::{Array, Int64Array, Int8Array, RecordBatch, UInt32Array}; +use arrow_array::{Array, BooleanArray, Int64Array, Int8Array, RecordBatch, UInt32Array}; use arrow_ord::sort::{lexsort_to_indices, SortColumn, SortOptions}; use arrow_row::{RowConverter, SortField}; use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema}; @@ -49,6 +49,7 @@ use std::sync::Arc; pub(crate) struct KeyValueFileWriter { file_io: FileIO, config: KeyValueWriteConfig, + ignore_delete: bool, /// Next sequence number to assign (bucket-local, always auto-incremented). next_sequence_number: i64, /// Buffered batches (user schema). @@ -105,6 +106,8 @@ impl KeyValueFileWriter { config: KeyValueWriteConfig, next_sequence_number: i64, ) -> Result { + let ignore_delete = config.merge_engine == MergeEngine::PartialUpdate + && CoreOptions::new(&config.table_options).ignore_delete(); if config.merge_engine == MergeEngine::PartialUpdate { PartialUpdateConfig::new(&config.table_options) .validate_runtime_mode(true, &config.table_name)?; @@ -136,6 +139,7 @@ impl KeyValueFileWriter { Ok(Self { file_io, config, + ignore_delete, next_sequence_number, buffer: Vec::new(), buffer_bytes: 0, @@ -147,6 +151,13 @@ impl KeyValueFileWriter { /// Buffer a RecordBatch. Flushes when buffer exceeds write_buffer_size. /// Sequence numbers are assigned per-bucket on flush, matching Java Paimon behavior. pub(crate) async fn write(&mut self, batch: &RecordBatch) -> Result<()> { + // Filter before byte accounting and buffering so ignored retracts do + // not consume write-buffer memory or automatic sequence numbers. + let batch = if self.ignore_delete { + Self::filter_retract_rows(batch)? + } else { + batch.clone() + }; if batch.num_rows() == 0 { return Ok(()); } @@ -155,7 +166,7 @@ impl KeyValueFileWriter { .iter() .map(|c| c.get_buffer_memory_size()) .sum(); - self.buffer.push(batch.clone()); + self.buffer.push(batch); self.buffer_bytes += batch_bytes; if self.buffer_bytes as i64 >= self.config.write_buffer_size { @@ -310,6 +321,44 @@ impl KeyValueFileWriter { Ok(()) } + fn filter_retract_rows(batch: &RecordBatch) -> Result { + let Some(vk_idx) = batch + .schema() + .fields() + .iter() + .position(|field| field.name() == VALUE_KIND_FIELD_NAME) + else { + return Ok(batch.clone()); + }; + let value_kinds = batch + .column(vk_idx) + .as_any() + .downcast_ref::() + .ok_or_else(|| crate::Error::DataInvalid { + message: "_VALUE_KIND column must be Int8".to_string(), + source: None, + })?; + let keep = BooleanArray::from( + (0..batch.num_rows()) + .map(|row| { + let value = if value_kinds.is_null(row) { + RowKind::Insert.to_value() + } else { + value_kinds.value(row) + }; + RowKind::from_value(value).map(|kind| kind.is_add()) + }) + .collect::>>()?, + ); + + arrow_select::filter::filter_record_batch(batch, &keep).map_err(|e| { + crate::Error::DataInvalid { + message: format!("Failed to filter ignored retract rows: {e}"), + source: None, + } + }) + } + async fn write_indexed_file( &self, batch: &RecordBatch, @@ -551,7 +600,8 @@ impl KeyValueFileWriter { /// the read-side `PartialUpdateMergeFunction`: rows are visited in /// ascending (sequence fields, auto-seq) order and every column keeps its /// latest non-null value; a column that is null in every row stays null. - /// DELETE / UPDATE_BEFORE rows are rejected, matching the read side. + /// DELETE / UPDATE_BEFORE rows are rejected defensively. When + /// `ignore-delete=true`, `write` filters them before buffering. /// /// Returns the merged batch (user schema, in primary-key order) and its /// `_SEQUENCE_NUMBER` column; each merged row keeps the highest sequence @@ -866,6 +916,138 @@ mod tests { .unwrap() } + #[tokio::test] + async fn test_flush_partial_update_ignore_delete_skips_retract_only_batch() { + let schema = Arc::new(ArrowSchema::new(vec![ + Arc::new(ArrowField::new("id", ArrowDataType::Int32, false)), + Arc::new(ArrowField::new("seq", ArrowDataType::Int64, false)), + Arc::new(ArrowField::new( + VALUE_KIND_FIELD_NAME, + ArrowDataType::Int8, + false, + )), + ])); + let batch = RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from(vec![1, 2])) as Arc, + Arc::new(Int64Array::from(vec![10, 20])) as Arc, + Arc::new(Int8Array::from(vec![1, 3])) as Arc, + ], + ) + .unwrap(); + let mut config = test_write_config(MergeEngine::PartialUpdate); + config + .table_options + .insert("ignore-delete".to_string(), "true".to_string()); + let mut writer = + KeyValueFileWriter::new(FileIOBuilder::new("memory").build().unwrap(), config, 7) + .unwrap(); + + writer.write(&batch).await.unwrap(); + assert!(writer.buffer.is_empty()); + assert_eq!(writer.buffer_bytes, 0); + assert_eq!(writer.next_sequence_number, 7); + + let prepared = writer.prepare_commit().await.unwrap(); + + assert!(prepared.data_files.is_empty()); + assert!(prepared.changelog_files.is_empty()); + assert_eq!(writer.next_sequence_number, 7); + } + + #[tokio::test] + async fn test_flush_partial_update_ignore_delete_filters_data_and_changelog() { + let schema = Arc::new(ArrowSchema::new(vec![ + Arc::new(ArrowField::new("id", ArrowDataType::Int32, false)), + Arc::new(ArrowField::new("seq", ArrowDataType::Int64, false)), + Arc::new(ArrowField::new( + VALUE_KIND_FIELD_NAME, + ArrowDataType::Int8, + false, + )), + Arc::new(ArrowField::new("value", ArrowDataType::Int32, true)), + ])); + let batch = RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from(vec![1, 1, 2])) as Arc, + Arc::new(Int64Array::from(vec![10, 20, 30])) as Arc, + Arc::new(Int8Array::from(vec![0, 3, 1])) as Arc, + Arc::new(Int32Array::from(vec![Some(100), None, None])) + as Arc, + ], + ) + .unwrap(); + let mut config = test_write_config(MergeEngine::PartialUpdate); + config.input_changelog = true; + config.table_options.insert( + "partial-update.ignore-delete".to_string(), + "true".to_string(), + ); + let mut writer = + KeyValueFileWriter::new(FileIOBuilder::new("memory").build().unwrap(), config, 7) + .unwrap(); + + writer.write(&batch).await.unwrap(); + let prepared = writer.prepare_commit().await.unwrap(); + + assert_eq!(prepared.data_files.len(), 1); + assert_eq!(prepared.changelog_files.len(), 1); + for file in prepared + .data_files + .iter() + .chain(prepared.changelog_files.iter()) + { + assert_eq!(file.row_count, 1); + assert_eq!(file.min_sequence_number, 7); + assert_eq!(file.max_sequence_number, 7); + assert_eq!(file.delete_row_count, Some(0)); + } + assert_eq!(writer.next_sequence_number, 8); + } + + #[tokio::test] + async fn test_flush_partial_update_explicit_false_rejects_retract() { + let schema = Arc::new(ArrowSchema::new(vec![ + Arc::new(ArrowField::new("id", ArrowDataType::Int32, false)), + Arc::new(ArrowField::new("seq", ArrowDataType::Int64, false)), + Arc::new(ArrowField::new( + VALUE_KIND_FIELD_NAME, + ArrowDataType::Int8, + false, + )), + ])); + let batch = RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from(vec![1])) as Arc, + Arc::new(Int64Array::from(vec![10])) as Arc, + Arc::new(Int8Array::from(vec![3])) as Arc, + ], + ) + .unwrap(); + let mut config = test_write_config(MergeEngine::PartialUpdate); + config + .table_options + .insert("ignore-delete".to_string(), "false".to_string()); + let mut writer = + KeyValueFileWriter::new(FileIOBuilder::new("memory").build().unwrap(), config, 0) + .unwrap(); + + writer.write(&batch).await.unwrap(); + let err = match writer.prepare_commit().await { + Ok(_) => panic!("explicit ignore-delete=false must reject retract rows"), + Err(err) => err, + }; + + assert!(matches!( + err, + crate::Error::Unsupported { message } + if message.contains("does not support DELETE or UPDATE_BEFORE") + )); + } + /// Partial-update merges each key group down to one row at flush: every /// column keeps its latest non-null value (different columns may come /// from different source rows) and the merged row carries the group's diff --git a/crates/paimon/src/table/sort_merge.rs b/crates/paimon/src/table/sort_merge.rs index 523f9552..abbb6513 100644 --- a/crates/paimon/src/table/sort_merge.rs +++ b/crates/paimon/src/table/sort_merge.rs @@ -26,7 +26,7 @@ //! - DataFusion: `SortPreservingMergeStream` (LoserTree layout) //! - Arrow-row: `RowConverter` for efficient key comparison -use crate::spec::{AggregationConfig, DataField, PartialUpdateConfig, RowKind}; +use crate::spec::{AggregationConfig, CoreOptions, DataField, PartialUpdateConfig, RowKind}; use crate::table::aggregator::{new_aggregator, FieldAggregator}; use crate::table::ArrowRecordBatchStream; use crate::Error; @@ -184,9 +184,12 @@ impl MergeFunction for DeduplicateMergeFunction { /// Basic partial-update merge: for each non-key column, keep the latest /// non-null value ordered by user sequence (if configured) then system sequence. /// -/// DELETE / UPDATE_BEFORE rows are treated as unsupported in this mode. +/// DELETE / UPDATE_BEFORE rows are ignored when `ignore-delete=true` and +/// treated as unsupported otherwise. #[derive(Debug, Clone, Copy)] -pub(crate) struct PartialUpdateMergeFunction(()); +pub(crate) struct PartialUpdateMergeFunction { + ignore_delete: bool, +} impl PartialUpdateMergeFunction { pub(crate) fn new( @@ -194,7 +197,9 @@ impl PartialUpdateMergeFunction { table_name: &str, ) -> crate::Result { PartialUpdateConfig::new(table_options).validate_runtime_mode(true, table_name)?; - Ok(Self(())) + Ok(Self { + ignore_delete: CoreOptions::new(table_options).ignore_delete(), + }) } } @@ -221,14 +226,19 @@ impl MergeFunction for PartialUpdateMergeFunction { let mut latest_non_null_by_col: Vec> = vec![None; output_schema.fields().len()]; + let mut saw_add = false; for row_idx in ordered_row_indices { let row = &rows[row_idx]; if !RowKind::from_value(row.value_kind)?.is_add() { + if self.ignore_delete { + continue; + } return Err(crate::Error::Unsupported { message: "merge-engine=partial-update basic mode does not support DELETE or UPDATE_BEFORE rows".to_string(), }); } + saw_add = true; for (output_col_idx, latest_non_null) in latest_non_null_by_col.iter_mut().enumerate() { let source_array = batch_buffer[row.batch_idx] @@ -239,6 +249,10 @@ impl MergeFunction for PartialUpdateMergeFunction { } } + if !saw_add { + return Ok(MergeResult::Omit); + } + let output_columns: Vec = output_schema .fields() .iter() @@ -1911,6 +1925,158 @@ mod tests { )); } + #[tokio::test] + async fn test_partial_update_merge_ignores_delete_when_configured() { + let schema = make_schema(); + let output_schema = make_output_schema(); + let s0 = stream_from_batches(vec![make_batch_with_kind( + &schema, + vec![1], + vec![1], + vec![0], + vec![Some("old")], + )]); + let s1 = stream_from_batches(vec![make_batch_with_kind( + &schema, + vec![1], + vec![2], + vec![3], + vec![Some("delete")], + )]); + let options = HashMap::from([ + ("merge-engine".to_string(), "partial-update".to_string()), + ("ignore-delete".to_string(), "true".to_string()), + ]); + + let result = SortMergeReaderBuilder::new( + vec![s0, s1], + schema, + vec![0], + 1, + 2, + vec![], + vec![3], + output_schema, + Box::new(PartialUpdateMergeFunction::new(&options, "test_table").unwrap()), + ) + .build() + .unwrap() + .try_collect::>() + .await + .unwrap(); + + assert_eq!(result.len(), 1); + assert_eq!(result[0].num_rows(), 1); + assert_eq!( + result[0] + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + "old" + ); + } + + #[tokio::test] + async fn test_partial_update_merge_alias_ignores_update_before() { + let schema = make_schema(); + let output_schema = make_output_schema(); + let s0 = stream_from_batches(vec![make_batch_with_kind( + &schema, + vec![1], + vec![1], + vec![0], + vec![Some("old")], + )]); + let s1 = stream_from_batches(vec![make_batch_with_kind( + &schema, + vec![1], + vec![2], + vec![1], + vec![Some("before")], + )]); + let s2 = stream_from_batches(vec![make_batch_with_kind( + &schema, + vec![1], + vec![3], + vec![2], + vec![Some("new")], + )]); + let options = HashMap::from([ + ("merge-engine".to_string(), "partial-update".to_string()), + ( + "partial-update.ignore-delete".to_string(), + "true".to_string(), + ), + ]); + + let result = SortMergeReaderBuilder::new( + vec![s0, s1, s2], + schema, + vec![0], + 1, + 2, + vec![], + vec![3], + output_schema, + Box::new(PartialUpdateMergeFunction::new(&options, "test_table").unwrap()), + ) + .build() + .unwrap() + .try_collect::>() + .await + .unwrap(); + + assert_eq!(result.len(), 1); + assert_eq!(result[0].num_rows(), 1); + assert_eq!( + result[0] + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + "new" + ); + } + + #[tokio::test] + async fn test_partial_update_merge_omits_retract_only_key_when_configured() { + let schema = make_schema(); + let output_schema = make_output_schema(); + let stream = stream_from_batches(vec![make_batch_with_kind( + &schema, + vec![1], + vec![1], + vec![3], + vec![Some("delete")], + )]); + let options = HashMap::from([ + ("merge-engine".to_string(), "partial-update".to_string()), + ("ignore-delete".to_string(), "true".to_string()), + ]); + + let result = SortMergeReaderBuilder::new( + vec![stream], + schema, + vec![0], + 1, + 2, + vec![], + vec![3], + output_schema, + Box::new(PartialUpdateMergeFunction::new(&options, "test_table").unwrap()), + ) + .build() + .unwrap() + .try_collect::>() + .await + .unwrap(); + + assert_eq!(result.iter().map(RecordBatch::num_rows).sum::(), 0); + } + #[test] fn test_partial_update_merge_function_new_rejects_unsupported_options() { let options = HashMap::from([ diff --git a/crates/paimon/tests/incremental_batch_scan_test.rs b/crates/paimon/tests/incremental_batch_scan_test.rs index c1a0a8c2..bc007851 100644 --- a/crates/paimon/tests/incremental_batch_scan_test.rs +++ b/crates/paimon/tests/incremental_batch_scan_test.rs @@ -69,6 +69,19 @@ async fn read_incremental_pairs( collect_pairs(&batches) } +async fn read_current_pairs(table: &paimon::table::Table) -> Vec<(i32, i32)> { + let builder = table.new_read_builder(); + let plan = builder.new_scan().plan().await.unwrap(); + let read = table.new_read_builder().new_read().unwrap(); + let batches: Vec = read + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + collect_pairs(&batches) +} + async fn plan_incremental( table: &paimon::table::Table, mode: IncrementalScanMode, @@ -269,6 +282,34 @@ async fn changelog_between_snapshots_reads_changelog_manifest_files() { assert_eq!(rows, vec![(1, 10), (1, 20)]); } +#[tokio::test] +async fn partial_update_ignore_delete_filters_data_and_input_changelog() { + let table_path = "memory:/incremental_batch/partial_update_ignore_delete"; + let (file_io, table) = memory_table( + table_path, + pk_schema(&[ + ("changelog-producer", "input"), + ("merge-engine", "partial-update"), + ("partial-update.ignore-delete", "true"), + ("bucket", "1"), + ]), + ); + setup_dirs(&file_io, table_path).await; + persist_table_schema(&file_io, table_path, table.schema()).await; + + write_batch( + &table, + &make_batch_with_kinds(vec![1, 1, 2, 1], vec![10, 999, 200, 20], vec![0, 3, 1, 2]), + ) + .await; + + assert_eq!(read_current_pairs(&table).await, vec![(1, 20)]); + assert_eq!( + read_incremental_pairs(&table, IncrementalScanMode::Changelog, 0, 1).await, + vec![(1, 10), (1, 20)] + ); +} + /// Multi-snapshot changelog range is left-open / right-closed and ordered by snapshot id. #[tokio::test] async fn changelog_multi_snapshot_range_is_ordered_and_left_open() { diff --git a/docs/src/sql.md b/docs/src/sql.md index 0cd5e255..ac3d0ee9 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -1880,6 +1880,15 @@ Set via `WITH ('key' = 'value')` at table creation time, or dynamically via `SET | `'merge-engine' = 'partial-update'` | Basic partial-update engine for PK tables | | `'merge-engine' = 'aggregation'` | Basic aggregation engine for PK tables | +Rust supports the basic partial-update engine with latest-non-null semantics. +Set either `'ignore-delete' = 'true'` or +`'partial-update.ignore-delete' = 'true'` to ignore `DELETE` and +`UPDATE_BEFORE` rows during writes and when reading existing files. The default +and an explicit `false` continue to reject these retract rows. Once enabled on +an existing partial-update table, `ignore-delete` cannot be changed back to +`false`. Advanced partial-update features such as sequence groups, partial +aggregation, and remove-record-on-delete are not supported. + Rust currently supports `merge-engine=aggregation` in basic mode only. It works with fixed buckets and ordinary dynamic buckets (`'bucket' = '-1'`) when the primary key includes all partition columns. It supports per-field aggregate