From a3d3d9070fbe7cdf0739db5554b08b045c7beab5 Mon Sep 17 00:00:00 2001 From: liwuhen Date: Thu, 23 Jul 2026 20:13:02 +0800 Subject: [PATCH 1/5] fix(table): validate write batch schema at the core entry --- crates/paimon/src/table/table_write.rs | 490 ++++++++++++++++++++++++- 1 file changed, 488 insertions(+), 2 deletions(-) diff --git a/crates/paimon/src/table/table_write.rs b/crates/paimon/src/table/table_write.rs index 86d8085f..d6e9e876 100644 --- a/crates/paimon/src/table/table_write.rs +++ b/crates/paimon/src/table/table_write.rs @@ -88,6 +88,7 @@ impl FileWriter { /// Reference: [pypaimon BatchTableWrite](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/write/table_write.py) pub struct TableWrite { table: Table, + write_schema: Arc, partition_writers: HashMap, partition_computer: PartitionComputer, partition_keys: Vec, @@ -131,6 +132,7 @@ impl TableWrite { pub(crate) fn new(table: &Table, commit_user: String) -> crate::Result { let is_overwrite = false; let schema = table.schema(); + let write_schema = build_target_arrow_schema(schema.fields())?; let core_options = CoreOptions::new(schema.options()); let blob_descriptor_fields = core_options.blob_descriptor_fields(); let blob_view_fields = core_options.blob_view_fields(); @@ -344,6 +346,7 @@ impl TableWrite { Ok(Self { table: table.clone(), + write_schema, partition_writers: HashMap::new(), partition_computer, partition_keys, @@ -434,6 +437,8 @@ impl TableWrite { /// Write an Arrow RecordBatch. Rows are routed to the correct partition and bucket. pub async fn write_arrow_batch(&mut self, batch: &RecordBatch) -> Result<()> { + self.validate_write_batch_schema(batch)?; + if batch.num_rows() == 0 { return Ok(()); } @@ -451,6 +456,102 @@ impl TableWrite { Ok(()) } + fn validate_write_batch_schema(&self, batch: &RecordBatch) -> Result<()> { + let expected_schema = &self.write_schema; + let actual_schema = batch.schema(); + let table_field_count = expected_schema.fields().len(); + let actual_field_count = actual_schema.fields().len(); + let allows_value_kind = self.changelog_producer == ChangelogProducer::Input; + let includes_value_kind = allows_value_kind && actual_field_count == table_field_count + 1; + + if actual_field_count != table_field_count && !includes_value_kind { + let maximum_expected_count = table_field_count + usize::from(allows_value_kind); + let (mismatch_index, expected_field, actual_field) = + if actual_field_count < table_field_count { + let field = expected_schema.field(actual_field_count); + ( + actual_field_count, + format!("'{}': {:?}", field.name(), field.data_type()), + "".to_string(), + ) + } else { + let field = actual_schema.field(maximum_expected_count); + ( + maximum_expected_count, + "".to_string(), + format!("'{}': {:?}", field.name(), field.data_type()), + ) + }; + let expected_count = if allows_value_kind { + format!("{table_field_count} or {}", table_field_count + 1) + } else { + table_field_count.to_string() + }; + return Err(crate::Error::DataInvalid { + message: format!( + "write batch schema field count mismatch: expected {expected_count}, actual {actual_field_count}; first mismatch at index {mismatch_index}: expected {expected_field}, actual {actual_field}" + ), + source: None, + }); + } + + for (index, expected_field) in expected_schema.fields().iter().enumerate() { + let actual_field = actual_schema.field(index); + if actual_field.name() != expected_field.name() { + return Err(crate::Error::DataInvalid { + message: format!( + "write batch schema field name mismatch at index {index}: expected '{}', actual '{}'", + expected_field.name(), + actual_field.name() + ), + source: None, + }); + } + } + if includes_value_kind { + let actual_field = actual_schema.field(table_field_count); + if actual_field.name() != VALUE_KIND_FIELD_NAME { + return Err(crate::Error::DataInvalid { + message: format!( + "write batch schema field name mismatch at index {table_field_count}: expected '{VALUE_KIND_FIELD_NAME}', actual '{}'", + actual_field.name() + ), + source: None, + }); + } + } + + for (index, expected_field) in expected_schema.fields().iter().enumerate() { + let actual_field = actual_schema.field(index); + if actual_field.data_type() != expected_field.data_type() { + return Err(crate::Error::DataInvalid { + message: format!( + "write batch schema data type mismatch for field '{}' at index {index}: expected {:?}, actual {:?}", + expected_field.name(), + expected_field.data_type(), + actual_field.data_type() + ), + source: None, + }); + } + } + if includes_value_kind { + let actual_field = actual_schema.field(table_field_count); + if actual_field.data_type() != &arrow_schema::DataType::Int8 { + return Err(crate::Error::DataInvalid { + message: format!( + "write batch schema data type mismatch for field '{VALUE_KIND_FIELD_NAME}' at index {table_field_count}: expected {:?}, actual {:?}", + arrow_schema::DataType::Int8, + actual_field.data_type() + ), + source: None, + }); + } + } + + Ok(()) + } + /// Group rows by (partition_bytes, bucket) and return sub-batches. /// /// In cross-partition mode, also generates DELETE sub-batches for keys that @@ -758,7 +859,6 @@ impl TableWrite { || !self.blob_view_fields.is_empty() { let fields = self.table.schema().fields(); - let input_schema = build_target_arrow_schema(fields)?; Ok(FileWriter::AppendDedicated(Box::new( AppendDedicatedFormatFileWriter::new( self.table.file_io().clone(), @@ -774,7 +874,7 @@ impl TableWrite { self.file_format.clone(), self.vector_target_file_size, self.vector_file_format.as_deref(), - &input_schema, + &self.write_schema, fields, self.table.schema().options(), &self.blob_inline_fields, @@ -1191,6 +1291,300 @@ mod tests { .unwrap() } + fn make_id_only_batch(ids: Vec) -> RecordBatch { + RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ArrowField::new( + "id", + ArrowDataType::Int32, + false, + )])), + vec![Arc::new(Int32Array::from(ids))], + ) + .unwrap() + } + + fn assert_data_invalid_contains(error: crate::Error, expected_context: &[&str]) { + let message = match error { + crate::Error::DataInvalid { message, .. } => message, + error => panic!("expected DataInvalid, got {error:?}"), + }; + for context in expected_context { + assert!( + message.contains(context), + "expected error message to contain {context:?}, got {message:?}" + ); + } + } + + #[tokio::test] + async fn test_write_arrow_batch_rejects_missing_blob_field() { + let table = Table::new( + test_file_io(), + Identifier::new("default", "test_blob_schema_validation"), + "memory:/test_blob_schema_validation".to_string(), + test_blob_table_schema(), + None, + ); + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + + let error = table_write + .write_arrow_batch(&make_id_only_batch(vec![1])) + .await + .unwrap_err(); + + assert_data_invalid_contains( + error, + &[ + "field count", + "expected 2", + "actual 1", + "payload", + "", + ], + ); + } + + #[tokio::test] + async fn test_write_arrow_batch_rejects_missing_vector_field() { + let table = Table::new( + test_file_io(), + Identifier::new("default", "test_vector_schema_validation"), + "memory:/test_vector_schema_validation".to_string(), + test_vector_table_schema("parquet"), + None, + ); + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + + let error = table_write + .write_arrow_batch(&make_id_only_batch(vec![1])) + .await + .unwrap_err(); + + assert_data_invalid_contains( + error, + &[ + "field count", + "expected 2", + "actual 1", + "embedding", + "", + ], + ); + } + + #[tokio::test] + async fn test_write_arrow_batch_rejects_mismatched_table_fields() { + let table = test_table(&test_file_io(), "memory:/test_schema_field_validation"); + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + + let swapped = RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ + ArrowField::new("value", ArrowDataType::Int32, false), + ArrowField::new("id", ArrowDataType::Int32, false), + ])), + vec![ + Arc::new(Int32Array::from(vec![10])), + Arc::new(Int32Array::from(vec![1])), + ], + ) + .unwrap(); + let error = table_write.write_arrow_batch(&swapped).await.unwrap_err(); + assert_data_invalid_contains( + error, + &["field name", "index 0", "expected 'id'", "actual 'value'"], + ); + + let wrong_name = RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new("wrong", ArrowDataType::Int32, false), + ])), + vec![ + Arc::new(Int32Array::from(vec![1])), + Arc::new(Int32Array::from(vec![10])), + ], + ) + .unwrap(); + let error = table_write + .write_arrow_batch(&wrong_name) + .await + .unwrap_err(); + assert_data_invalid_contains( + error, + &[ + "field name", + "index 1", + "expected 'value'", + "actual 'wrong'", + ], + ); + + let wrong_type = RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int64, false), + ArrowField::new("value", ArrowDataType::Int32, false), + ])), + vec![ + Arc::new(Int64Array::from(vec![1_i64])), + Arc::new(Int32Array::from(vec![10])), + ], + ) + .unwrap(); + let error = table_write + .write_arrow_batch(&wrong_type) + .await + .unwrap_err(); + assert_data_invalid_contains( + error, + &["data type", "field 'id'", "expected Int32", "actual Int64"], + ); + + let unexpected_column = RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new("value", ArrowDataType::Int32, false), + ArrowField::new("extra", ArrowDataType::Int32, false), + ])), + vec![ + Arc::new(Int32Array::from(vec![1])), + Arc::new(Int32Array::from(vec![10])), + Arc::new(Int32Array::from(vec![20])), + ], + ) + .unwrap(); + let error = table_write + .write_arrow_batch(&unexpected_column) + .await + .unwrap_err(); + assert_data_invalid_contains( + error, + &[ + "field count", + "expected 2", + "actual 3", + "", + "extra", + ], + ); + } + + #[tokio::test] + async fn test_write_arrow_batch_rejects_invalid_empty_batch_schema() { + let table = test_table(&test_file_io(), "memory:/test_empty_schema_validation"); + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + let batch = RecordBatch::new_empty(Arc::new(ArrowSchema::new(vec![ArrowField::new( + "id", + ArrowDataType::Int32, + false, + )]))); + + let error = table_write.write_arrow_batch(&batch).await.unwrap_err(); + + assert_data_invalid_contains( + error, + &[ + "field count", + "expected 2", + "actual 1", + "value", + "", + ], + ); + } + + #[tokio::test] + async fn test_normal_write_rejects_appended_value_kind() { + let table = test_table(&test_file_io(), "memory:/test_unexpected_value_kind"); + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + + let error = table_write + .write_arrow_batch(&make_batch_with_value_kind(vec![1], vec![10], vec![0])) + .await + .unwrap_err(); + + assert_data_invalid_contains( + error, + &[ + "field count", + "expected 2", + "actual 3", + "", + VALUE_KIND_FIELD_NAME, + ], + ); + } + + #[tokio::test] + async fn test_rowkind_field_accepts_table_schema_and_rejects_caller_value_kind() { + let file_io = test_file_io(); + let table_path = "memory:/test_rowkind_schema_validation"; + 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("rowkind.field", "op") + .build() + .unwrap(); + let table = Table::new( + file_io, + Identifier::new("default", "test_rowkind_schema_validation"), + table_path.to_string(), + TableSchema::new(0, &schema), + None, + ); + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + + let caller_value_kind = 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), + ArrowField::new(VALUE_KIND_FIELD_NAME, ArrowDataType::Int8, false), + ])), + vec![ + Arc::new(Int32Array::from(vec![1])), + Arc::new(Int32Array::from(vec![10])), + Arc::new(StringArray::from(vec!["+I"])), + Arc::new(Int8Array::from(vec![0])), + ], + ) + .unwrap(); + let error = table_write + .write_arrow_batch(&caller_value_kind) + .await + .unwrap_err(); + assert_data_invalid_contains( + error, + &[ + "field count", + "expected 3", + "actual 4", + VALUE_KIND_FIELD_NAME, + ], + ); + + let valid = 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])), + Arc::new(Int32Array::from(vec![10])), + Arc::new(StringArray::from(vec!["+I"])), + ], + ) + .unwrap(); + table_write.write_arrow_batch(&valid).await.unwrap(); + let messages = table_write.prepare_commit().await.unwrap(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].new_files[0].row_count, 1); + } + #[tokio::test] async fn test_write_and_commit() { let file_io = test_file_io(); @@ -2677,6 +3071,69 @@ mod tests { } } + #[tokio::test] + async fn test_input_changelog_rejects_misplaced_or_wrong_type_value_kind() { + let table = Table::new( + test_file_io(), + Identifier::new("default", "test_input_changelog_schema_validation"), + "memory:/test_input_changelog_schema_validation".to_string(), + pk_changelog_schema(&[("changelog-producer", "input")]), + None, + ); + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + + let misplaced = RecordBatch::try_new( + Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new(VALUE_KIND_FIELD_NAME, ArrowDataType::Int8, false), + ArrowField::new("value", ArrowDataType::Int32, false), + ])), + vec![ + Arc::new(Int32Array::from(vec![1])), + Arc::new(Int8Array::from(vec![0])), + Arc::new(Int32Array::from(vec![10])), + ], + ) + .unwrap(); + let error = table_write.write_arrow_batch(&misplaced).await.unwrap_err(); + assert_data_invalid_contains( + error, + &[ + "field name", + "index 1", + "expected 'value'", + VALUE_KIND_FIELD_NAME, + ], + ); + + let wrong_type = 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::Int32, false), + ])), + vec![ + Arc::new(Int32Array::from(vec![1])), + Arc::new(Int32Array::from(vec![10])), + Arc::new(Int32Array::from(vec![0])), + ], + ) + .unwrap(); + let error = table_write + .write_arrow_batch(&wrong_type) + .await + .unwrap_err(); + assert_data_invalid_contains( + error, + &[ + "data type", + VALUE_KIND_FIELD_NAME, + "expected Int8", + "actual Int32", + ], + ); + } + #[tokio::test] async fn test_input_changelog_writes_raw_rows_separately_from_data_rows() { let file_io = test_file_io(); @@ -3473,6 +3930,35 @@ mod tests { )); } + #[tokio::test] + async fn test_cross_partition_write_rejects_caller_value_kind() { + let file_io = test_file_io(); + let table = + test_cross_partition_table(&file_io, "memory:/test_cross_partition_schema_validation"); + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + + let error = table_write + .write_arrow_batch(&make_partitioned_batch_with_value_kind( + vec!["a"], + vec![1], + vec![10], + vec![0], + )) + .await + .unwrap_err(); + + assert_data_invalid_contains( + error, + &[ + "field count", + "expected 3", + "actual 4", + "", + VALUE_KIND_FIELD_NAME, + ], + ); + } + #[test] fn test_rejects_cross_partition_partial_update() { let file_io = test_file_io(); From e72400fe0e4b883b07694adaa509d5aa7bfe4e34 Mon Sep 17 00:00:00 2001 From: liwuhen Date: Fri, 24 Jul 2026 08:58:23 +0800 Subject: [PATCH 2/5] fix(ci): align writes with schema validation --- .../datafusion/src/physical_plan/sink.rs | 75 ++++++++++++++++--- crates/paimon/tests/audit_log_table_test.rs | 2 +- 2 files changed, 65 insertions(+), 12 deletions(-) diff --git a/crates/integrations/datafusion/src/physical_plan/sink.rs b/crates/integrations/datafusion/src/physical_plan/sink.rs index 64575c02..dd07a9ab 100644 --- a/crates/integrations/datafusion/src/physical_plan/sink.rs +++ b/crates/integrations/datafusion/src/physical_plan/sink.rs @@ -32,25 +32,35 @@ use datafusion::execution::SendableRecordBatchStream; use datafusion::execution::TaskContext; use datafusion::physical_plan::DisplayAs; use futures::StreamExt; +use paimon::spec::{CoreOptions, ROW_ID_FIELD_NAME}; use paimon::table::Table; use crate::error::to_datafusion_error; -fn to_paimon_batch(batch: RecordBatch) -> DFResult { - if !batch - .schema() - .fields() - .iter() - .any(|field| field.data_type() == &ArrowDataType::Utf8View) +fn to_paimon_batch(batch: RecordBatch, strip_internal_row_id: bool) -> DFResult { + let input_schema = batch.schema(); + let row_id_index = strip_internal_row_id.then(|| { + input_schema + .fields() + .iter() + .rposition(|field| field.name() == ROW_ID_FIELD_NAME) + }); + let row_id_index = row_id_index.flatten(); + if row_id_index.is_none() + && !input_schema + .fields() + .iter() + .any(|field| field.data_type() == &ArrowDataType::Utf8View) { return Ok(batch); } - let fields = batch - .schema() + let fields = input_schema .fields() .iter() - .map(|field| { + .enumerate() + .filter(|(index, _)| Some(*index) != row_id_index) + .map(|(_, field)| { if field.data_type() == &ArrowDataType::Utf8View { Arc::new(field.as_ref().clone().with_data_type(ArrowDataType::Utf8)) } else { @@ -60,11 +70,14 @@ fn to_paimon_batch(batch: RecordBatch) -> DFResult { .collect::>(); let schema = Arc::new(Schema::new_with_metadata( fields, - batch.schema().metadata().clone(), + input_schema.metadata().clone(), )); let columns = batch .columns() .iter() + .enumerate() + .filter(|(index, _)| Some(*index) != row_id_index) + .map(|(_, column)| column) .zip(schema.fields()) .map(|(column, field)| { if column.data_type() == field.data_type() { @@ -88,14 +101,18 @@ fn to_paimon_batch(batch: RecordBatch) -> DFResult { pub struct PaimonDataSink { table: Table, schema: ArrowSchemaRef, + strip_internal_row_id: bool, overwrite: bool, } impl PaimonDataSink { pub fn new(table: Table, schema: ArrowSchemaRef, overwrite: bool) -> Self { + let strip_internal_row_id = + CoreOptions::new(table.schema().options()).data_evolution_enabled(); Self { table, schema, + strip_internal_row_id, overwrite, } } @@ -131,7 +148,7 @@ impl DataSink for PaimonDataSink { let mut row_count = 0u64; while let Some(batch) = data.next().await { - let batch = to_paimon_batch(batch?)?; + let batch = to_paimon_batch(batch?, self.strip_internal_row_id)?; row_count += batch.num_rows() as u64; tw.write_arrow_batch(&batch) .await @@ -153,3 +170,39 @@ impl DataSink for PaimonDataSink { Ok(row_count) } } + +#[cfg(test)] +mod tests { + use super::*; + use datafusion::arrow::array::{Int32Array, Int64Array, StringViewArray}; + use datafusion::arrow::datatypes::Field; + + #[test] + fn test_to_paimon_batch_strips_internal_row_id_and_casts_string_views() { + let schema = Arc::new(Schema::new(vec![ + Field::new("id", ArrowDataType::Int32, false), + Field::new("name", ArrowDataType::Utf8View, true), + Field::new(ROW_ID_FIELD_NAME, ArrowDataType::Int64, true), + ])); + let batch = RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from(vec![1, 2])), + Arc::new(StringViewArray::from(vec![Some("a"), None])), + Arc::new(Int64Array::from(vec![10, 11])), + ], + ) + .unwrap(); + + let converted = to_paimon_batch(batch, true).unwrap(); + + assert_eq!(converted.num_rows(), 2); + assert_eq!(converted.num_columns(), 2); + assert_eq!(converted.schema().field(0).name(), "id"); + assert_eq!(converted.schema().field(1).name(), "name"); + assert_eq!( + converted.schema().field(1).data_type(), + &ArrowDataType::Utf8 + ); + } +} diff --git a/crates/paimon/tests/audit_log_table_test.rs b/crates/paimon/tests/audit_log_table_test.rs index 8839673e..11fb9d48 100644 --- a/crates/paimon/tests/audit_log_table_test.rs +++ b/crates/paimon/tests/audit_log_table_test.rs @@ -197,7 +197,7 @@ async fn audit_log_delta_scan_preserves_pk_row_kinds() { let (file_io, table) = memory_table( table_path, pk_schema(&[ - ("changelog-producer", "none"), + ("changelog-producer", "input"), ("merge-engine", "deduplicate"), ("bucket", "1"), ]), From 4281dc87673c6532039daae6735a2858dc829bde Mon Sep 17 00:00:00 2001 From: liwuhen Date: Fri, 24 Jul 2026 09:26:55 +0800 Subject: [PATCH 3/5] fix(ci): use input changelog for value-kind test --- crates/integrations/datafusion/tests/pk_tables.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/integrations/datafusion/tests/pk_tables.rs b/crates/integrations/datafusion/tests/pk_tables.rs index 0bad6e3b..2b6fb269 100644 --- a/crates/integrations/datafusion/tests/pk_tables.rs +++ b/crates/integrations/datafusion/tests/pk_tables.rs @@ -186,6 +186,8 @@ async fn test_pk_partial_update_ignore_delete_alias_e2e() { .sql("CREATE SCHEMA paimon.test_db") .await .unwrap(); + // The batch below supplies explicit row kinds through `_VALUE_KIND`, which + // is part of the input changelog write contract rather than a normal write. sql_context .sql( "CREATE TABLE paimon.test_db.t_partial_update_ignore_delete ( @@ -194,6 +196,7 @@ async fn test_pk_partial_update_ignore_delete_alias_e2e() { ) WITH ( 'bucket' = '1', 'merge-engine' = 'partial-update', + 'changelog-producer' = 'input', 'partial-update.ignore-delete' = 'true' )", ) From 9e97ad5b9639094ec8616ad3ea02262f3e1a6a92 Mon Sep 17 00:00:00 2001 From: liwuhen Date: Fri, 24 Jul 2026 11:00:55 +0800 Subject: [PATCH 4/5] fix(table): reject value kind for cross-partition writes --- crates/paimon/src/table/table_write.rs | 27 ++++++++++++++++++++++---- 1 file changed, 23 insertions(+), 4 deletions(-) diff --git a/crates/paimon/src/table/table_write.rs b/crates/paimon/src/table/table_write.rs index d6e9e876..42fcd782 100644 --- a/crates/paimon/src/table/table_write.rs +++ b/crates/paimon/src/table/table_write.rs @@ -461,7 +461,12 @@ impl TableWrite { let actual_schema = batch.schema(); let table_field_count = expected_schema.fields().len(); let actual_field_count = actual_schema.fields().len(); - let allows_value_kind = self.changelog_producer == ChangelogProducer::Input; + // Cross-partition routing generates its own `_VALUE_KIND` column so that + // migrated rows can be written as deletes in their previous partition. + // Accepting a caller-provided column here would make the generated delete + // batch contain `_VALUE_KIND` twice. + let allows_value_kind = self.changelog_producer == ChangelogProducer::Input + && !matches!(self.bucket_assigner, BucketAssignerEnum::CrossPartition(_)); let includes_value_kind = allows_value_kind && actual_field_count == table_field_count + 1; if actual_field_count != table_field_count && !includes_value_kind { @@ -3931,10 +3936,24 @@ mod tests { } #[tokio::test] - async fn test_cross_partition_write_rejects_caller_value_kind() { + async fn test_input_changelog_cross_partition_write_rejects_caller_value_kind() { let file_io = test_file_io(); - let table = - test_cross_partition_table(&file_io, "memory:/test_cross_partition_schema_validation"); + let schema = Schema::builder() + .column("pt", DataType::VarChar(VarCharType::string_type())) + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .partition_keys(["pt"]) + .option("changelog-producer", "input") + .build() + .unwrap(); + let table = Table::new( + file_io, + Identifier::new("default", "test_cross_partition_schema_validation"), + "memory:/test_cross_partition_schema_validation".to_string(), + TableSchema::new(0, &schema), + None, + ); let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); let error = table_write From 5b3231a4253731318e420d7dd45566d308da42cf Mon Sep 17 00:00:00 2001 From: liwuhen Date: Sat, 25 Jul 2026 15:01:25 +0800 Subject: [PATCH 5/5] fix(table): accept value kind for primary-key writes --- crates/paimon/src/table/table_write.rs | 52 +++++++++++++++++++++++--- 1 file changed, 47 insertions(+), 5 deletions(-) diff --git a/crates/paimon/src/table/table_write.rs b/crates/paimon/src/table/table_write.rs index 42fcd782..490de635 100644 --- a/crates/paimon/src/table/table_write.rs +++ b/crates/paimon/src/table/table_write.rs @@ -461,11 +461,11 @@ impl TableWrite { let actual_schema = batch.schema(); let table_field_count = expected_schema.fields().len(); let actual_field_count = actual_schema.fields().len(); - // Cross-partition routing generates its own `_VALUE_KIND` column so that - // migrated rows can be written as deletes in their previous partition. - // Accepting a caller-provided column here would make the generated delete - // batch contain `_VALUE_KIND` twice. - let allows_value_kind = self.changelog_producer == ChangelogProducer::Input + // Primary-key data writers consume `_VALUE_KIND` independently of whether + // separate changelog files are enabled. Row-kind generation and + // cross-partition routing add the column themselves, so callers must not. + let allows_value_kind = !self.primary_key_indices.is_empty() + && self.row_kind_generator.is_none() && !matches!(self.bucket_assigner, BucketAssignerEnum::CrossPartition(_)); let includes_value_kind = allows_value_kind && actual_field_count == table_field_count + 1; @@ -3046,6 +3046,48 @@ mod tests { TableSchema::new(0, &schema) } + #[tokio::test] + async fn test_default_changelog_producer_accepts_value_kind() { + let file_io = test_file_io(); + let table_path = "memory:/test_default_changelog_value_kind"; + setup_dirs(&file_io, table_path).await; + let table = Table::new( + file_io.clone(), + Identifier::new("default", "test_default_changelog_value_kind"), + table_path.to_string(), + pk_changelog_schema(&[]), + None, + ); + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + + table_write + .write_arrow_batch(&make_batch_with_value_kind( + vec![1, 1, 2], + vec![10, 20, 30], + vec![1, 2, 3], + )) + .await + .unwrap(); + + let messages = table_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)); + assert!(messages[0].new_changelog_files.is_empty()); + + let data_file = &messages[0].new_files[0]; + let data_file_path = format!( + "{table_path}/{}/{}", + bucket_dir_name(messages[0].bucket), + data_file.file_name + ); + let data_batches = + read_physical_key_value_batches(&file_io, &data_file_path, data_file.file_size).await; + assert_eq!(collect_i8(&data_batches, 1), vec![2, 3]); + assert_eq!(collect_i32(&data_batches, 2), vec![1, 2]); + assert_eq!(collect_i32(&data_batches, 3), vec![20, 30]); + } + #[test] fn test_table_write_rejects_loaded_first_row_with_incompatible_changelog_producer() { let file_io = test_file_io();