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 4aa617ee fix(table): validate write batch schema at the core entry
(#602)
4aa617ee is described below
commit 4aa617eecbb27235499a6ef3599ceceab4ede202
Author: Wuhen- Li <[email protected]>
AuthorDate: Sun Jul 26 19:41:41 2026 +0800
fix(table): validate write batch schema at the core entry (#602)
---
.../datafusion/src/physical_plan/sink.rs | 75 ++-
crates/integrations/datafusion/tests/pk_tables.rs | 3 +
crates/paimon/src/table/table_write.rs | 551 ++++++++++++++++++++-
crates/paimon/tests/audit_log_table_test.rs | 2 +-
4 files changed, 617 insertions(+), 14 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<RecordBatch> {
- if !batch
- .schema()
- .fields()
- .iter()
- .any(|field| field.data_type() == &ArrowDataType::Utf8View)
+fn to_paimon_batch(batch: RecordBatch, strip_internal_row_id: bool) ->
DFResult<RecordBatch> {
+ 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<RecordBatch> {
.collect::<Vec<_>>();
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<RecordBatch> {
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/integrations/datafusion/tests/pk_tables.rs
b/crates/integrations/datafusion/tests/pk_tables.rs
index d51b0052..39e25972 100644
--- a/crates/integrations/datafusion/tests/pk_tables.rs
+++ b/crates/integrations/datafusion/tests/pk_tables.rs
@@ -290,6 +290,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 (
@@ -298,6 +300,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'
)",
)
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 86d8085f..490de635 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<arrow_schema::Schema>,
partition_writers: HashMap<PartitionBucketKey, FileWriter>,
partition_computer: PartitionComputer,
partition_keys: Vec<String>,
@@ -131,6 +132,7 @@ impl TableWrite {
pub(crate) fn new(table: &Table, commit_user: String) ->
crate::Result<Self> {
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,107 @@ 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();
+ // 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;
+
+ 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()),
+ "<missing>".to_string(),
+ )
+ } else {
+ let field = actual_schema.field(maximum_expected_count);
+ (
+ maximum_expected_count,
+ "<no field>".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 +864,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 +879,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 +1296,300 @@ mod tests {
.unwrap()
}
+ fn make_id_only_batch(ids: Vec<i32>) -> 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",
+ "<missing>",
+ ],
+ );
+ }
+
+ #[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",
+ "<missing>",
+ ],
+ );
+ }
+
+ #[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",
+ "<no field>",
+ "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",
+ "<missing>",
+ ],
+ );
+ }
+
+ #[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",
+ "<no field>",
+ 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();
@@ -2647,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();
@@ -2677,6 +3118,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 +3977,49 @@ mod tests {
));
}
+ #[tokio::test]
+ async fn
test_input_changelog_cross_partition_write_rejects_caller_value_kind() {
+ let file_io = test_file_io();
+ 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
+ .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",
+ "<no field>",
+ VALUE_KIND_FIELD_NAME,
+ ],
+ );
+ }
+
#[test]
fn test_rejects_cross_partition_partial_update() {
let file_io = test_file_io();
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"),
]),
