This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new c6c5099b fix(table): validate partial-update delete config and mixed
batches (#957)
c6c5099b is described below
commit c6c5099bccb8f3ad47f78a81f8cbdcea85a2eae5
Author: QuakeWang <[email protected]>
AuthorDate: Thu Oct 8 21:07:45 2026 +0800
fix(table): validate partial-update delete config and mixed batches (#957)
---
crates/paimon/src/spec/partial_update.rs | 37 ++++++
crates/paimon/src/table/kv_file_reader.rs | 149 ++++++++++++++++++++++
crates/paimon/src/table/table_write.rs | 200 +++++++++++++++++++++++++++++-
3 files changed, 383 insertions(+), 3 deletions(-)
diff --git a/crates/paimon/src/spec/partial_update.rs
b/crates/paimon/src/spec/partial_update.rs
index c9d5ead0..93d2a31a 100644
--- a/crates/paimon/src/spec/partial_update.rs
+++ b/crates/paimon/src/spec/partial_update.rs
@@ -197,6 +197,19 @@ impl<'a> PartialUpdateConfig<'a> {
}
fn validate_delete_config(&self) -> crate::Result<()> {
+ if let Some(value) = self
+ .options
+ .get(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION)
+ {
+ if !value.eq_ignore_ascii_case("true") &&
!value.eq_ignore_ascii_case("false") {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Invalid value '{value}' for option \
+ '{PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION}':
expected true or false"
+ ),
+ });
+ }
+ }
let ignore_delete = self
.options
.get(IGNORE_DELETE_OPTION)
@@ -679,6 +692,30 @@ mod tests {
}
}
+ #[test]
+ fn test_remove_record_on_delete_rejects_invalid_boolean() {
+ for value in ["tru", "", "1"] {
+ let options =
+
partial_update_options(&[(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION,
value)]);
+ let config = PartialUpdateConfig::new(&options);
+ for error in [
+ config
+ .validate_create_fields(&[], &["id".to_string()])
+ .unwrap_err(),
+ config.validate_write_mode(true, "default.t").unwrap_err(),
+ config.validate_read_mode(true, "default.t").unwrap_err(),
+ ] {
+ assert!(
+ matches!(error, crate::Error::ConfigInvalid { ref message }
+ if
message.contains(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE_OPTION)
+ && message.contains(value)
+ && message.contains("true or false")),
+ "got {error:?}"
+ );
+ }
+ }
+ }
+
#[test]
fn test_validate_read_mode_rejects_unsupported_aggregation_modifiers() {
let key = "fields.price.ignore-delete";
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 3b603133..cbe1aa6c 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -3067,6 +3067,155 @@ mod tests {
assert_eq!(int_column(&batches, "b"), vec![7]);
}
+ #[tokio::test]
+ async fn kv_read_partial_update_remove_record_on_delete() {
+ let file_io = test_file_io();
+ let table_path = "memory:/kv_partial_update_remove_record_on_delete";
+ setup_dirs(&file_io, table_path).await;
+ let table = pk_table(
+ &file_io,
+ table_path,
+ &[
+ ("merge-engine", "partial-update"),
+ ("partial-update.remove-record-on-delete", "true"),
+ ],
+ );
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, true),
+ ArrowField::new(
+ crate::spec::VALUE_KIND_FIELD_NAME,
+ ArrowDataType::Int8,
+ false,
+ ),
+ ]));
+ let make = |value: Option<i32>, kind: i8| {
+ RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int32Array::from(vec![value])),
+ Arc::new(Int8Array::from(vec![kind])),
+ ],
+ )
+ .unwrap()
+ };
+
+ write_commit(&table, &make(Some(10), 0)).await;
+ assert_eq!(
+ int_column(&read_rows(&table, None, None).await, "value"),
+ vec![10]
+ );
+
+ write_commit(&table, &make(Some(99), 1)).await;
+ assert_eq!(
+ int_column(&read_rows(&table, None, None).await, "value"),
+ vec![10]
+ );
+
+ write_commit(&table, &make(Some(20), 2)).await;
+ assert_eq!(
+ int_column(&read_rows(&table, None, None).await, "value"),
+ vec![20]
+ );
+
+ write_commit(&table, &make(None, 3)).await;
+ let rows = read_rows(&table, None, None).await;
+ assert!(rows.iter().all(|batch| batch.num_rows() == 0), "{rows:?}");
+
+ write_commit(&table, &make(None, 0)).await;
+ let rows = read_rows(&table, None, None).await;
+ assert_eq!(int_column(&rows, "id"), vec![1]);
+ let value =
rows[0].column(rows[0].schema().index_of("value").unwrap());
+ assert!(value.is_null(0));
+
+ write_commit(&table, &make(Some(30), 3)).await;
+ write_commit(&table, &make(None, 0)).await;
+ assert_eq!(
+ int_column(&read_rows(&table, None, None).await, "value"),
+ vec![30]
+ );
+
+ let mut write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ write
+ .write_arrow_batch(&int_batch(vec![1], vec![Some(40)]))
+ .await
+ .unwrap();
+ write.write_arrow_batch(&make(None, 3)).await.unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].new_files.len(), 1);
+ assert_eq!(messages[0].new_files[0].delete_row_count, Some(1));
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+ let rows = read_rows(&table, None, None).await;
+ assert!(rows.iter().all(|batch| batch.num_rows() == 0), "{rows:?}");
+ }
+
+ #[tokio::test]
+ async fn kv_read_partial_update_mixed_value_kind_schemas() {
+ for (nullable, insert_kind) in [(true, None), (true, Some(0)), (false,
Some(0))] {
+ for kinds_first in [false, true] {
+ let file_io = test_file_io();
+ let table_path = format!(
+
"memory:/kv_partial_update_mixed_kinds_{nullable}_{insert_kind:?}_{kinds_first}"
+ );
+ setup_dirs(&file_io, &table_path).await;
+ let table = pk_table(
+ &file_io,
+ &table_path,
+ &[
+ ("merge-engine", "partial-update"),
+ ("partial-update.remove-record-on-delete", "true"),
+ ],
+ );
+ let ordinary = int_batch(vec![1, 3, 4], vec![Some(10),
Some(30), Some(40)]);
+ let explicit_kinds = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, true),
+ ArrowField::new(VALUE_KIND_FIELD_NAME,
ArrowDataType::Int8, nullable),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![2, 3, 4])),
+ Arc::new(Int32Array::from(vec![Some(20), None,
Some(42)])),
+ Arc::new(Int8Array::from(vec![insert_kind, Some(3),
Some(2)])),
+ ],
+ )
+ .unwrap();
+ let batches = if kinds_first {
+ [&explicit_kinds, &ordinary]
+ } else {
+ [&ordinary, &explicit_kinds]
+ };
+ let mut write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ for batch in batches {
+ write.write_arrow_batch(batch).await.unwrap();
+ }
+ let messages = write.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let batches = read_rows(&table, None, None).await;
+ let mut rows: Vec<_> = int_column(&batches, "id")
+ .into_iter()
+ .zip(int_column(&batches, "value"))
+ .collect();
+ rows.sort_unstable();
+ let expected = if kinds_first {
+ vec![(1, 10), (2, 20), (3, 30), (4, 40)]
+ } else {
+ vec![(1, 10), (2, 20), (4, 42)]
+ };
+ assert_eq!(rows, expected, "{table_path}");
+ }
+ }
+ }
+
#[tokio::test]
async fn kv_read_partial_update_sequence_groups_with_projection() {
let file_io = test_file_io();
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 59ed4a4c..a03eae62 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -26,8 +26,8 @@ use crate::resource::ResourceContext;
use crate::spec::PartitionComputer;
use crate::spec::{
first_row_supports_changelog_producer, BinaryRow, ChangelogProducer,
CoreOptions, DataField,
- DataType, MergeEngine, RowKind, RowKindFilter, EMPTY_SERIALIZED_ROW,
POSTPONE_BUCKET,
- VALUE_KIND_FIELD_NAME,
+ DataType, MergeEngine, PartialUpdateConfig, RowKind, RowKindFilter,
EMPTY_SERIALIZED_ROW,
+ POSTPONE_BUCKET, VALUE_KIND_FIELD_NAME,
};
use crate::table::bucket_assigner::{BatchAssignOutput, BucketAssignerEnum,
PartitionBucketKey};
use crate::table::bucket_assigner_constant::ConstantBucketAssigner;
@@ -48,7 +48,7 @@ use crate::table::row_kind_generator::RowKindGenerator;
use crate::table::write_batch_normalize::normalize_write_array;
use crate::table::{Snapshot, Table, TableScan};
use crate::Result;
-use arrow_array::{ArrayRef, RecordBatch};
+use arrow_array::{ArrayRef, Int8Array, RecordBatch};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
@@ -808,6 +808,49 @@ impl TableWrite {
self.table.schema().options(),
)?;
}
+ let config = PartialUpdateConfig::new(self.table.schema().options());
+ let batch = if !self.primary_key_indices.is_empty()
+ && config.is_enabled()
+ && config.remove_record_on_delete()
+ {
+ let batch = if batch
+ .schema()
+ .column_with_name(VALUE_KIND_FIELD_NAME)
+ .is_none()
+ {
+ Self::add_per_row_value_kind_column(
+ &batch,
+ vec![RowKind::Insert.to_value(); batch.num_rows()],
+ )?
+ } else {
+ batch
+ };
+ let schema = batch.schema();
+ let kind_index = schema.index_of(VALUE_KIND_FIELD_NAME).unwrap();
+ let mut columns = batch.columns().to_vec();
+ // Input changelogs require non-null kinds; NULL means Insert.
+ if columns[kind_index].null_count() > 0 {
+ let kinds = columns[kind_index]
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap();
+ columns[kind_index] = Arc::new(Int8Array::from_iter_values(
+ kinds
+ .iter()
+ .map(|kind|
kind.unwrap_or(RowKind::Insert.to_value())),
+ ));
+ }
+ let mut schema =
arrow_schema::SchemaBuilder::from(schema.as_ref());
+ Arc::make_mut(schema.field_mut(kind_index)).set_nullable(false);
+ RecordBatch::try_new(Arc::new(schema.finish()),
columns).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!("Failed to normalize _VALUE_KIND:
{error}"),
+ source: Some(Box::new(error)),
+ }
+ })?
+ } else {
+ batch
+ };
Ok(Some(batch))
}
@@ -2273,6 +2316,132 @@ pub(in crate::table) mod tests {
assert_eq!(messages[0].new_files[0].row_count, 1);
}
+ #[tokio::test]
+ async fn test_partial_update_remove_record_on_delete_with_rowkind_field() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_partial_update_delete_rowkind_field";
+ setup_dirs(&file_io, table_path).await;
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("merge-engine", "partial-update")
+ .option("partial-update.remove-record-on-delete", "true")
+ .option("rowkind.field", "op")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default",
"test_partial_update_delete_rowkind_field"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ArrowField::new("op", ArrowDataType::Utf8, false),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1])),
+ Arc::new(Int32Array::from(vec![10, 10])),
+ Arc::new(StringArray::from(vec!["+I", "-D"])),
+ ],
+ )
+ .unwrap();
+ let mut writer = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].new_files.len(), 1);
+ assert_eq!(messages[0].new_files[0].delete_row_count, Some(1));
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+ assert!(read_id_value_rows(&table).await.is_empty());
+ }
+
+ #[tokio::test]
+ async fn test_partial_update_input_changelog_normalizes_null_kinds() {
+ for kinds_first in [false, true] {
+ let file_io = test_file_io();
+ let table_path =
format!("memory:/test_partial_update_null_changelog_{kinds_first}");
+ setup_dirs(&file_io, &table_path).await;
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default",
"test_partial_update_null_changelog"),
+ table_path.clone(),
+ pk_changelog_schema(&[
+ ("merge-engine", "partial-update"),
+ ("partial-update.remove-record-on-delete", "true"),
+ ("changelog-producer", "input"),
+ ]),
+ None,
+ );
+ let ordinary = make_batch(vec![1, 3, 4], vec![10, 30, 40]);
+ let explicit_kinds = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ArrowField::new(VALUE_KIND_FIELD_NAME,
ArrowDataType::Int8, true),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![2, 3, 4])),
+ Arc::new(Int32Array::from(vec![20, 31, 42])),
+ Arc::new(Int8Array::from(vec![None, Some(3), Some(2)])),
+ ],
+ )
+ .unwrap();
+ let batches = if kinds_first {
+ [&explicit_kinds, &ordinary]
+ } else {
+ [&ordinary, &explicit_kinds]
+ };
+ let mut writer = TableWrite::new(&table,
"test-user".to_string()).unwrap();
+ for batch in batches {
+ writer.write_arrow_batch(batch).await.unwrap();
+ }
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].new_changelog_files.len(), 1);
+ let changelog = &messages[0].new_changelog_files[0];
+ let changelog_path = format!(
+ "{table_path}/{}/{}",
+ bucket_dir_name(messages[0].bucket),
+ changelog.file_name
+ );
+ let changelog_size = changelog.file_size;
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+
+ let expected = if kinds_first {
+ vec![(1, 10), (2, 20), (3, 30), (4, 40)]
+ } else {
+ vec![(1, 10), (2, 20), (4, 42)]
+ };
+ assert_eq!(read_id_value_rows(&table).await, expected,
"{table_path}");
+ let changelog =
+ read_physical_key_value_batches(&file_io, &changelog_path,
changelog_size).await;
+ assert!(changelog
+ .iter()
+ .all(|batch| batch.column(1).null_count() == 0));
+ assert_eq!(collect_i32(&changelog, 2), vec![1, 2, 3, 3, 4, 4]);
+ let (expected_kinds, expected_values) = if kinds_first {
+ (vec![0, 0, 3, 0, 2, 0], vec![10, 20, 31, 30, 42, 40])
+ } else {
+ (vec![0, 0, 0, 3, 0, 2], vec![10, 20, 30, 31, 40, 42])
+ };
+ assert_eq!(collect_i8(&changelog, 1), expected_kinds,
"{table_path}");
+ assert_eq!(collect_i32(&changelog, 3), expected_values,
"{table_path}");
+ }
+ }
+
#[tokio::test]
async fn test_write_and_commit() {
let file_io = test_file_io();
@@ -3004,6 +3173,31 @@ pub(in crate::table) mod tests {
assert_eq!(collect_i32(&batches, 1), vec![10, 20, 30]);
}
+ #[tokio::test]
+ async fn test_append_row_format_ignores_partial_update_options() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_append_row_partial_update_options";
+ setup_dirs(&file_io, table_path).await;
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("merge-engine", "partial-update")
+ .option("partial-update.remove-record-on-delete", "true")
+ .option("file.format", "row")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default",
"test_append_row_partial_update_options"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+
+ commit_one_batch(&table, vec![1], vec![10]).await;
+ assert_eq!(read_id_value_rows(&table).await, vec![(1, 10)]);
+ }
+
#[tokio::test]
async fn
test_primary_key_non_parquet_formats_roundtrip_with_typed_physical_fields() {
for format in ["row", "avro"] {
