From 0ea0f47019f82c971017726a2f602389dccfe510 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 17:58:38 +0800 Subject: [PATCH 1/5] [core] Normalize upsert payloads after matching --- crates/paimon/src/table/table_update.rs | 3 + crates/paimon/src/table/table_upsert.rs | 353 +++++++++++++++++++++++- 2 files changed, 341 insertions(+), 15 deletions(-) diff --git a/crates/paimon/src/table/table_update.rs b/crates/paimon/src/table/table_update.rs index 546dc8a22..35b0114b2 100644 --- a/crates/paimon/src/table/table_update.rs +++ b/crates/paimon/src/table/table_update.rs @@ -243,6 +243,9 @@ impl TableUpdate { /// columns are omitted from `upsert_keys`. Existing keys update every /// matching row ID; new keys append all input columns. Omitted columns on /// matched rows retain their values; on new rows they are absent from the file. + /// After last-write-wins, matched values follow row-ID update conversion + /// rules; new rows must satisfy the ordinary Arrow write type. + /// Binary values, including nested ones, retain ordinary write layout rules. /// Input batches may have different non-key columns. After last-write-wins /// deduplication, matched rows must supply their update columns and appended /// rows in each partition must share a field set. diff --git a/crates/paimon/src/table/table_upsert.rs b/crates/paimon/src/table/table_upsert.rs index 8869e527e..c5336ed39 100644 --- a/crates/paimon/src/table/table_upsert.rs +++ b/crates/paimon/src/table/table_upsert.rs @@ -46,6 +46,32 @@ fn invalid(message: impl Into) -> crate::Error { } } +// BYTES and BLOB keep their declared Arrow representation. Reusing row-ID +// conversions for matched values must not introduce Binary/LargeBinary casts. +fn contains_binary(data_type: &DataType) -> bool { + match data_type { + DataType::Binary + | DataType::LargeBinary + | DataType::FixedSizeBinary(_) + | DataType::BinaryView => true, + DataType::Dictionary(_, value) => contains_binary(value), + DataType::Struct(fields) => fields + .iter() + .any(|field| contains_binary(field.data_type())), + DataType::List(field) + | DataType::LargeList(field) + | DataType::ListView(field) + | DataType::LargeListView(field) + | DataType::FixedSizeList(field, _) + | DataType::Map(field, _) + | DataType::RunEndEncoded(_, field) => contains_binary(field.data_type()), + DataType::Union(fields, _) => fields + .iter() + .any(|(_, field)| contains_binary(field.data_type())), + _ => false, + } +} + fn selected_rows(batch: &RecordBatch, rows: &[usize]) -> crate::Result { let indices = rows .iter() @@ -150,9 +176,9 @@ impl TableUpsert { } /// Add rows containing the keys and columns to write. Column order may - /// differ from the table schema; names and Arrow layouts follow the same - /// normalization as ordinary writes. Batches form one logical upsert input, - /// but may carry different non-key columns, preserving absent fields. + /// differ from the table schema. Normalize match/routing keys here; keep + /// payloads until last-write-wins and matching have selected their action. + /// Batches may carry different non-key columns, preserving absent fields. pub(super) fn add_batch(&mut self, batch: RecordBatch) -> crate::Result<()> { let target = crate::arrow::build_target_arrow_schema(self.table.schema().fields())?; let input_schema = batch.schema(); @@ -170,29 +196,92 @@ impl TableUpsert { } let mut columns = Vec::with_capacity(target.fields().len()); let mut fields = Vec::new(); + for field in target.fields() { + let Some(column) = batch.column_by_name(field.name()) else { + continue; + }; + if self.keys.contains(field.name()) { + columns.push(normalize_write_array(column, field.data_type()).map_err( + |error| invalid(format!("Invalid upsert key '{}': {error}", field.name())), + )?); + fields.push(field.clone()); + } else { + columns.push(column.clone()); + fields.push( + input_schema + .field_with_name(field.name()) + .unwrap() + .clone() + .into(), + ); + } + } + let ordered = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns) + .map_err(|error| invalid(format!("cannot order upsert columns: {error}")))?; + self.source.push(ordered); + Ok(()) + } + + /// Appends use the ordinary write contract; matched values use the row-ID + /// updater's conversion rules. Neither action touches discarded source rows. + fn normalize_selected_batch( + &self, + batch: RecordBatch, + append: bool, + ) -> crate::Result { + let target = crate::arrow::build_target_arrow_schema(self.table.schema().fields())?; + let update_fields = super::data_evolution_fields::project_by_paths( + self.table.schema().fields(), + &self.update_columns, + )?; let inline = CoreOptions::new(self.table.schema().options()).blob_inline_fields(); + let mut fields = Vec::new(); + let mut columns = Vec::new(); for (index, field) in target.fields().iter().enumerate() { let Some(column) = batch.column_by_name(field.name()) else { + // The row-ID updater diagnoses missing update columns. Absent + // append columns stay absent in the file's write_cols metadata. continue; }; let logical = &self.table.schema().fields()[index]; - let normalized = - if logical.data_type().is_blob_file_field() && !inline.contains(logical.name()) { - normalize_blob_write_field(column, field) - } else { - normalize_write_array(column, field.data_type()) - .map(|column| (field.as_ref().clone(), column)) - }; + let update_field = update_fields + .iter() + .find(|updated| updated.name() == field.name()); + let normalized = if !append && update_field.is_none() { + Ok(( + batch + .schema() + .field_with_name(field.name()) + .unwrap() + .clone(), + column.clone(), + )) + } else if logical.data_type().is_blob_file_field() && !inline.contains(logical.name()) { + normalize_blob_write_field(column, field) + } else if append + || contains_binary(column.data_type()) + || contains_binary(field.data_type()) + { + normalize_write_array(column, field.data_type()) + .map(|column| (field.as_ref().clone(), column)) + } else { + Ok(( + batch + .schema() + .field_with_name(field.name()) + .unwrap() + .clone(), + column.clone(), + )) + }; let (field, column) = normalized.map_err(|error| { invalid(format!("Invalid upsert column '{}': {error}", field.name())) })?; + fields.push(field); columns.push(column); - fields.push(Arc::new(field)); } - let ordered = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns) - .map_err(|error| invalid(format!("cannot order upsert columns: {error}")))?; - self.source.push(ordered); - Ok(()) + RecordBatch::try_new(Arc::new(Schema::new(fields)), columns) + .map_err(|error| invalid(format!("cannot normalize selected upsert rows: {error}"))) } /// Concatenate only match keys. Padding payload columns with NULL would @@ -277,6 +366,7 @@ impl TableUpsert { fn append_groups(&self, indices: &[usize]) -> crate::Result>> { let mut groups: IndexMap, Vec> = IndexMap::new(); for (batch, _) in self.selected_batches(indices)? { + let batch = self.normalize_selected_batch(batch, true)?; let mut by_partition: IndexMap, Vec> = IndexMap::new(); for (row, partition) in self.serialized_partitions(&batch)?.into_iter().enumerate() { by_partition.entry(partition).or_default().push(row); @@ -357,6 +447,7 @@ impl TableUpsert { update.pin_read_snapshot(snapshot_id); } for (selected, range) in matched_batches { + let selected = self.normalize_selected_batch(selected, false)?; let mut fields = selected.schema().fields().to_vec(); fields.push(Arc::new(Field::new(ROW_ID, DataType::Int64, false))); let mut columns: Vec = selected.columns().to_vec(); @@ -404,7 +495,239 @@ impl TableUpsert { #[cfg(test)] mod tests { use super::*; + use crate::catalog::Identifier; + use crate::io::FileIOBuilder; use crate::spec::{DataType as PaimonType, DecimalType, TimestampType, VarCharType}; + use crate::spec::{IntType, Schema as PaimonSchema, TableSchema}; + use crate::table::TableCommit; + use arrow_array::{Array, Int32Array, StringArray}; + + async fn numeric_table(path: &str) -> Table { + let schema = PaimonSchema::builder() + .column("key", PaimonType::Int(IntType::new())) + .column("value", PaimonType::Int(IntType::new())) + .column("keep", PaimonType::Int(IntType::new())) + .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") + .build() + .unwrap(); + let table = Table::new( + FileIOBuilder::new("memory").build().unwrap(), + Identifier::new("db", "upsert"), + path.into(), + TableSchema::new(0, &schema), + None, + ); + let initial = RecordBatch::try_from_iter([ + ("key", Arc::new(Int32Array::from(vec![1, 1, 2])) as ArrayRef), + ("value", Arc::new(Int32Array::from(vec![0; 3])) as ArrayRef), + ( + "keep", + Arc::new(Int32Array::from(vec![10, 11, 12])) as ArrayRef, + ), + ]) + .unwrap(); + let mut write = table.new_write_builder().new_write().unwrap(); + write.write_arrow_batch(&initial).await.unwrap(); + TableCommit::new(table.clone(), "seed".into()) + .commit(write.prepare_commit().await.unwrap()) + .await + .unwrap(); + write.close().await; + table + } + + fn numeric_source(keys: Vec, value: ArrayRef) -> RecordBatch { + RecordBatch::try_from_iter([ + ("key", Arc::new(Int32Array::from(keys)) as ArrayRef), + ("value", value), + ]) + .unwrap() + } + + async fn numeric_rows(table: &Table) -> Vec<(i64, i32, Option, Option)> { + let mut builder = table.new_read_builder(); + builder + .with_projection(&["_ROW_ID", "key", "value", "keep"]) + .unwrap(); + let plan = builder.new_scan().plan().await.unwrap(); + let batches: Vec<_> = builder + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let mut rows = Vec::new(); + for batch in batches { + let ids = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let keys = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(2) + .as_any() + .downcast_ref::() + .unwrap(); + let keep = batch + .column(3) + .as_any() + .downcast_ref::() + .unwrap(); + rows.extend((0..batch.num_rows()).map(|row| { + ( + ids.value(row), + keys.value(row), + (!values.is_null(row)).then(|| values.value(row)), + (!keep.is_null(row)).then(|| keep.value(row)), + ) + })); + } + rows.sort(); + rows + } + + #[tokio::test] + async fn upsert_casts_surviving_matched_values_and_ignores_other_payloads() { + let table = numeric_table("memory:/upsert_selected").await; + let first = numeric_source(vec![1], Arc::new(StringArray::from(vec!["not-an-int"]))); + let mut last = numeric_source(vec![1], Arc::new(StringArray::from(vec!["42"]))); + last = RecordBatch::try_from_iter([ + ("key", last.column(0).clone()), + ("value", last.column(1).clone()), + ( + "keep", + Arc::new(StringArray::from(vec!["unwritten"])) as ArrayRef, + ), + ]) + .unwrap(); + let mut update = table.new_write_builder().new_update().unwrap(); + update.with_update_type(vec!["value".into()]).unwrap(); + let messages = update + .upsert_by_arrow_with_key(vec![first, last], vec!["key".into()]) + .await + .unwrap(); + assert!(messages + .iter() + .all(|message| message.check_from_snapshot == Some(1))); + TableCommit::new(table.clone(), "update".into()) + .commit(messages) + .await + .unwrap(); + assert_eq!( + numeric_rows(&table).await, + vec![ + (0, 1, Some(42), Some(10)), + (1, 1, Some(42), Some(11)), + (2, 2, Some(0), Some(12)), + ] + ); + } + + #[tokio::test] + async fn upsert_keeps_update_casts_separate_from_append_validation() { + let table = numeric_table("memory:/upsert_mixed_types").await; + let matched = numeric_source(vec![1], Arc::new(StringArray::from(vec!["42"]))); + let append = numeric_source(vec![3], Arc::new(Int32Array::from(vec![7]))); + let mut update = table.new_write_builder().new_update().unwrap(); + update.with_update_type(vec!["value".into()]).unwrap(); + let messages = update + .upsert_by_arrow_with_key(vec![matched, append], vec!["key".into()]) + .await + .unwrap(); + TableCommit::new(table.clone(), "update".into()) + .commit(messages) + .await + .unwrap(); + assert_eq!( + numeric_rows(&table).await, + vec![ + (0, 1, Some(42), Some(10)), + (1, 1, Some(42), Some(11)), + (2, 2, Some(0), Some(12)), + (3, 3, Some(7), None), + ] + ); + } + + #[tokio::test] + async fn upsert_rejects_invalid_appends_before_staging_matched_updates() { + let table = numeric_table("memory:/upsert_invalid_append").await; + let batch = numeric_source(vec![1, 3], Arc::new(StringArray::from(vec!["42", "7"]))); + let mut update = table.new_write_builder().new_update().unwrap(); + update.with_update_type(vec!["value".into()]).unwrap(); + let error = update + .upsert_by_arrow_with_key(vec![batch], vec!["key".into()]) + .await + .unwrap_err(); + assert!(error.to_string().contains("Invalid upsert column 'value'")); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + Some(1) + ); + assert_eq!( + numeric_rows(&table).await, + vec![ + (0, 1, Some(0), Some(10)), + (1, 1, Some(0), Some(11)), + (2, 2, Some(0), Some(12)), + ] + ); + } + + #[tokio::test] + async fn upsert_update_casts_keep_binary_write_layouts() { + use arrow_array::{BinaryArray, LargeBinaryArray}; + let schema = PaimonSchema::builder() + .column("key", PaimonType::Int(IntType::new())) + .column( + "value", + PaimonType::VarBinary(crate::spec::VarBinaryType::default()), + ) + .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") + .build() + .unwrap(); + let table = Table::new( + FileIOBuilder::new("memory").build().unwrap(), + Identifier::new("db", "upsert"), + "memory:/upsert_binary".into(), + TableSchema::new(0, &schema), + None, + ); + let initial = numeric_source( + vec![1], + Arc::new(BinaryArray::from(vec![b"old".as_slice()])), + ); + let mut writer = table.new_write_builder().new_write().unwrap(); + writer.write_arrow_batch(&initial).await.unwrap(); + TableCommit::new(table.clone(), "seed".into()) + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); + writer.close().await; + let update = table.new_write_builder().new_update().unwrap(); + let input = numeric_source( + vec![1], + Arc::new(LargeBinaryArray::from(vec![b"new".as_slice()])), + ); + let error = update + .upsert_by_arrow_with_key(vec![input], vec!["key".into()]) + .await + .unwrap_err(); + assert!(error.to_string().contains("Invalid upsert column 'value'")); + } #[tokio::test] async fn upsert_normalizes_string_offsets_before_matching_and_writing() { From b0ff77f2740cd5655582a3e9ce1839ee3349d245 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 18:39:36 +0800 Subject: [PATCH 2/5] [core] Normalize logical keys for upsert and MERGE --- crates/paimon/src/table/key_normalization.rs | 180 ++++++++++++++++++ crates/paimon/src/table/mod.rs | 1 + crates/paimon/src/table/table_merge_into.rs | 22 ++- .../src/table/table_merge_into_tests.rs | 145 ++++++++++++++ crates/paimon/src/table/table_upsert.rs | 92 ++++++--- crates/paimon/src/table/update_input.rs | 2 +- 6 files changed, 411 insertions(+), 31 deletions(-) create mode 100644 crates/paimon/src/table/key_normalization.rs diff --git a/crates/paimon/src/table/key_normalization.rs b/crates/paimon/src/table/key_normalization.rs new file mode 100644 index 000000000..c3d2c28b5 --- /dev/null +++ b/crates/paimon/src/table/key_normalization.rs @@ -0,0 +1,180 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Logical keys used by row-level matching, independent of Arrow encoding. + +use arrow_array::{new_null_array, ArrayRef}; +use arrow_schema::DataType; + +use super::update_input::{cast_update_value, decode_dictionary, CastMode}; +use super::write_batch_normalize::normalize_write_array; + +/// Dictionary indices are local to each batch, never row identity. NULL keys +/// retain their validity bitmap; the operation decides whether NULL matches. +pub(super) fn normalize_match_key( + array: &ArrayRef, + expected: &DataType, +) -> crate::Result { + let decoded = decode_dictionary(array)?; + if decoded.data_type() == &DataType::Null { + return Ok(new_null_array(expected, decoded.len())); + } + normalize_write_array(&decoded, expected) +} + +/// Upsert matches logical values before deciding update versus append. Integer +/// widths, decimal scales and units within a temporal family can differ, but +/// conversion must be exact so distinct source keys never collapse. Payloads, +/// including the original key columns, keep their own action's write contract. +pub(super) fn normalize_upsert_key( + array: &ArrayRef, + expected: &DataType, +) -> crate::Result { + let decoded = decode_dictionary(array)?; + let source = decoded.data_type(); + let exact_family = source.is_integer() && expected.is_integer() + || source.is_decimal() && expected.is_decimal() + || matches!( + (source, expected), + ( + DataType::Date32 | DataType::Date64, + DataType::Date32 | DataType::Date64 + ) | ( + DataType::Time32(_) | DataType::Time64(_), + DataType::Time32(_) | DataType::Time64(_) + ) + ) + || matches!((source, expected), (DataType::Timestamp(_, from), DataType::Timestamp(_, to)) + if from.is_some() == to.is_some()); + if exact_family { + cast_update_value(&decoded, expected, CastMode::Assignment) + } else { + normalize_match_key(&decoded, expected) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow_array::types::Int8Type; + use arrow_array::{ + Array, Decimal128Array, DictionaryArray, Int32Array, Int64Array, Int8Array, StringArray, + TimestampMicrosecondArray, TimestampMillisecondArray, + }; + use arrow_schema::TimeUnit; + use std::sync::Arc; + + #[test] + fn dictionary_keys_use_value_validity_as_well_as_index_validity() { + let input: ArrayRef = Arc::new( + DictionaryArray::::try_new( + Int8Array::from(vec![Some(0), None, Some(1), Some(0)]), + Arc::new(StringArray::from(vec![None, Some("key")])), + ) + .unwrap(), + ); + let output = normalize_match_key(&input, &DataType::Utf8).unwrap(); + assert_eq!( + output.as_any().downcast_ref::().unwrap(), + &StringArray::from(vec![None, None, Some("key"), None]) + ); + } + + #[test] + fn integer_keys_are_exact_and_keep_nulls() { + let input: ArrayRef = Arc::new(Int64Array::from(vec![ + Some(-1), + None, + Some(i32::MAX as i64), + ])); + let output = normalize_upsert_key(&input, &DataType::Int32).unwrap(); + assert_eq!( + output.as_any().downcast_ref::().unwrap(), + &Int32Array::from(vec![Some(-1), None, Some(i32::MAX)]) + ); + let overflow: ArrayRef = Arc::new(Int64Array::from(vec![i32::MAX as i64 + 1])); + assert!(normalize_upsert_key(&overflow, &DataType::Int32).is_err()); + let text: ArrayRef = Arc::new(StringArray::from(vec!["42"])); + assert!(normalize_upsert_key(&text, &DataType::Int32).is_err()); + } + + #[test] + fn decimal_keys_cannot_round_into_existing_keys() { + let array = |value| { + Arc::new( + Decimal128Array::from(vec![value]) + .with_precision_and_scale(15, 4) + .unwrap(), + ) as ArrayRef + }; + let expected = DataType::Decimal128(10, 2); + let output = normalize_upsert_key(&array(12500), &expected).unwrap(); + assert_eq!( + output + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 125 + ); + assert!(normalize_upsert_key(&array(12501), &expected).is_err()); + } + + #[test] + fn temporal_keys_require_lossless_units_and_preserve_instant() { + let input: ArrayRef = Arc::new(TimestampMicrosecondArray::from(vec![ + Some(-1000), + None, + Some(2000), + ])); + let output = + normalize_upsert_key(&input, &DataType::Timestamp(TimeUnit::Millisecond, None)) + .unwrap(); + assert_eq!( + output + .as_any() + .downcast_ref::() + .unwrap(), + &TimestampMillisecondArray::from(vec![Some(-1), None, Some(2)]) + ); + let fractional: ArrayRef = Arc::new(TimestampMicrosecondArray::from(vec![-1001])); + assert!(normalize_upsert_key( + &fractional, + &DataType::Timestamp(TimeUnit::Millisecond, None) + ) + .is_err()); + let local: ArrayRef = + Arc::new(TimestampMicrosecondArray::from(vec![0]).with_timezone("America/New_York")); + let output = normalize_upsert_key( + &local, + &DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())), + ) + .unwrap(); + assert_eq!( + output + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 0 + ); + assert!( + normalize_upsert_key(&local, &DataType::Timestamp(TimeUnit::Microsecond, None)) + .is_err() + ); + } +} diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 33382b119..4b0b155df 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -133,6 +133,7 @@ mod query_auth; mod system_table; pub use system_table::SystemRead; +mod key_normalization; mod read_builder; mod read_limit; pub mod referenced_files; diff --git a/crates/paimon/src/table/table_merge_into.rs b/crates/paimon/src/table/table_merge_into.rs index 0004e3b7a..00652277c 100644 --- a/crates/paimon/src/table/table_merge_into.rs +++ b/crates/paimon/src/table/table_merge_into.rs @@ -427,14 +427,18 @@ pub(super) async fn merge_into( .map_err(|error| invalid(error.to_string()))? .data_type() }; - if source_type != target_type + let mut logical_source_type = source_type; + while let DataType::Dictionary(_, value) = logical_source_type { + logical_source_type = value; + } + if logical_source_type != target_type || !super::upsert_key_matcher::supported_key_type(target_type) { return Err(invalid( "MERGE ON key types must match and support row encoding", )); } - Ok(SortField::new(source_type.clone())) + Ok(SortField::new(target_type.clone())) }) .collect::>()?; let key_converter = @@ -442,8 +446,18 @@ pub(super) async fn merge_into( for (chunk, batch) in source.batches.iter().enumerate() { let keys = on .iter() - .map(|(_, name)| batch.column_by_name(name).unwrap().clone()) - .collect::>(); + .map(|(target, name)| { + let expected = if target == ROW_ID { + &DataType::Int64 + } else { + schema.field_with_name(target).unwrap().data_type() + }; + super::key_normalization::normalize_match_key( + batch.column_by_name(name).unwrap(), + expected, + ) + }) + .collect::>>()?; let rows = key_converter .convert_columns(&keys) .map_err(|error| invalid(error.to_string()))?; diff --git a/crates/paimon/src/table/table_merge_into_tests.rs b/crates/paimon/src/table/table_merge_into_tests.rs index b20c47530..2c520d217 100644 --- a/crates/paimon/src/table/table_merge_into_tests.rs +++ b/crates/paimon/src/table/table_merge_into_tests.rs @@ -107,6 +107,151 @@ fn update(condition: Option) -> WhenMatched { } } +fn dictionary_batch(ids: Vec>, values: Vec>) -> RecordBatch { + let keys = arrow_cast::cast( + &Int32Array::from(ids), + &arrow_schema::DataType::Dictionary( + Box::new(arrow_schema::DataType::Int8), + Box::new(arrow_schema::DataType::Int32), + ), + ) + .unwrap(); + let schema = arrow_schema::Schema::new(vec![ + arrow_schema::Field::new("id", keys.data_type().clone(), true), + arrow_schema::Field::new("value", arrow_schema::DataType::Int32, true), + ]); + RecordBatch::try_new( + Arc::new(schema), + vec![keys, Arc::new(Int32Array::from(values))], + ) + .unwrap() +} + +#[tokio::test] +async fn dictionary_merge_keys_use_values_and_sql_null_semantics() { + let target = table_at("dictionary_merge"); + seed( + &target, + &batch( + vec![Some(1), Some(1), Some(2), None], + vec![Some(10), Some(11), Some(20), Some(90)], + ), + ) + .await; + // Every non-null source key has dictionary index zero in its own chunk. + let source = vec![ + dictionary_batch(vec![Some(1)], vec![Some(42)]), + dictionary_batch(vec![Some(3)], vec![Some(30)]), + dictionary_batch(vec![None], vec![Some(91)]), + ]; + let messages = target + .new_write_builder() + .new_update() + .unwrap() + .merge_into( + MergeSource::Batches(source), + vec![("id".into(), "id".into())], + vec![update(None)], + vec![WhenNotMatched { + condition: None, + assignments: vec![ + ("id".into(), MergeAssignment::SourceColumn("id".into())), + ( + "value".into(), + MergeAssignment::SourceColumn("value".into()), + ), + ], + }], + ) + .await + .unwrap(); + TableCommit::new(target.clone(), "merge".into()) + .commit(messages) + .await + .unwrap(); + assert_eq!( + rows(&target).await, + vec![ + (None, Some(90)), + (None, Some(91)), + (Some(1), Some(42)), + (Some(1), Some(42)), + (Some(2), Some(20)), + (Some(3), Some(30)) + ] + ); +} + +#[tokio::test] +async fn dictionary_merge_cardinality_is_checked_before_conditions() { + let target = table_at("dictionary_cardinality"); + seed(&target, &batch(vec![Some(1)], vec![Some(10)])).await; + let source = vec![ + dictionary_batch(vec![Some(1)], vec![Some(11)]), + dictionary_batch(vec![Some(3), Some(1)], vec![Some(30), Some(12)]), + ]; + let calls = Arc::new(AtomicUsize::new(0)); + let observed = calls.clone(); + let condition = MergeCondition { + target_columns: Vec::new(), + source_columns: Vec::new(), + evaluate: Arc::new(move |batch| { + observed.fetch_add(1, Ordering::Relaxed); + Box::pin(async move { Ok(BooleanArray::from(vec![false; batch.num_rows()])) }) + }), + }; + let error = target + .new_write_builder() + .new_update() + .unwrap() + .merge_into( + MergeSource::Batches(source), + vec![("id".into(), "id".into())], + vec![update(Some(condition))], + Vec::new(), + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("multiple source rows")); + assert_eq!(calls.load(Ordering::Relaxed), 0); + assert_eq!(rows(&target).await, vec![(Some(1), Some(10))]); +} + +#[tokio::test] +async fn dictionary_merge_duplicate_unconditional_delete_is_allowed() { + let target = table_at("dictionary_delete"); + seed( + &target, + &batch(vec![Some(1), Some(2)], vec![Some(10), Some(20)]), + ) + .await; + let source = vec![ + dictionary_batch(vec![Some(1)], vec![Some(11)]), + dictionary_batch(vec![Some(3), Some(1)], vec![Some(30), Some(12)]), + ]; + let messages = target + .new_write_builder() + .new_update() + .unwrap() + .merge_into( + MergeSource::Batches(source), + vec![("id".into(), "id".into())], + vec![WhenMatched { + condition: None, + delete: true, + assignments: Vec::new(), + }], + Vec::new(), + ) + .await + .unwrap(); + TableCommit::new(target.clone(), "merge".into()) + .commit(messages) + .await + .unwrap(); + assert_eq!(rows(&target).await, vec![(Some(2), Some(20))]); +} + #[tokio::test] async fn core_merge_preserves_row_ids_and_handles_sql_null_keys() { let table = table(); diff --git a/crates/paimon/src/table/table_upsert.rs b/crates/paimon/src/table/table_upsert.rs index c5336ed39..b0c8e6fc2 100644 --- a/crates/paimon/src/table/table_upsert.rs +++ b/crates/paimon/src/table/table_upsert.rs @@ -176,8 +176,8 @@ impl TableUpsert { } /// Add rows containing the keys and columns to write. Column order may - /// differ from the table schema. Normalize match/routing keys here; keep - /// payloads until last-write-wins and matching have selected their action. + /// differ from the table schema. Keep original payloads, including keys, + /// until last-write-wins and matching have selected their action. /// Batches may carry different non-key columns, preserving absent fields. pub(super) fn add_batch(&mut self, batch: RecordBatch) -> crate::Result<()> { let target = crate::arrow::build_target_arrow_schema(self.table.schema().fields())?; @@ -200,21 +200,8 @@ impl TableUpsert { let Some(column) = batch.column_by_name(field.name()) else { continue; }; - if self.keys.contains(field.name()) { - columns.push(normalize_write_array(column, field.data_type()).map_err( - |error| invalid(format!("Invalid upsert key '{}': {error}", field.name())), - )?); - fields.push(field.clone()); - } else { - columns.push(column.clone()); - fields.push( - input_schema - .field_with_name(field.name()) - .unwrap() - .clone() - .into(), - ); - } + columns.push(column.clone()); + fields.push(input_schema.field_with_name(field.name()).unwrap().clone()); } let ordered = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns) .map_err(|error| invalid(format!("cannot order upsert columns: {error}")))?; @@ -262,7 +249,12 @@ impl TableUpsert { || contains_binary(column.data_type()) || contains_binary(field.data_type()) { - normalize_write_array(column, field.data_type()) + let value = if append { + column.clone() + } else { + super::update_input::decode_dictionary(column)? + }; + normalize_write_array(&value, field.data_type()) .map(|column| (field.as_ref().clone(), column)) } else { Ok(( @@ -287,20 +279,32 @@ impl TableUpsert { /// Concatenate only match keys. Padding payload columns with NULL would /// lose the distinction between an absent field and an explicit NULL. fn key_rows(&self) -> crate::Result { + let target = crate::arrow::build_target_arrow_schema(self.table.schema().fields())?; + let fields = target + .fields() + .iter() + .filter(|field| self.keys.contains(field.name())) + .cloned() + .collect::>(); + let key_schema = Arc::new(Schema::new(fields)); let batches = self .source .iter() .map(|batch| { - let columns = batch - .schema() + let columns = key_schema .fields() .iter() - .enumerate() - .filter(|(_, field)| self.keys.contains(field.name())) - .map(|(index, _)| index) - .collect::>(); - batch - .project(&columns) + .map(|field| { + super::key_normalization::normalize_upsert_key( + batch.column_by_name(field.name()).unwrap(), + field.data_type(), + ) + .map_err(|error| { + invalid(format!("Invalid upsert key '{}': {error}", field.name())) + }) + }) + .collect::>>()?; + RecordBatch::try_new(key_schema.clone(), columns) .map_err(|error| invalid(error.to_string())) }) .collect::>>()?; @@ -729,6 +733,42 @@ mod tests { assert!(error.to_string().contains("Invalid upsert column 'value'")); } + #[tokio::test] + async fn upsert_matches_integer_key_widths_but_keeps_append_input_types() { + let table = numeric_table("memory:/upsert_integer_key_widths").await; + let source = |key, value| { + RecordBatch::try_from_iter([ + ("key", Arc::new(Int64Array::from(vec![key])) as ArrayRef), + ("value", Arc::new(Int32Array::from(vec![value])) as ArrayRef), + ]) + .unwrap() + }; + let mut update = table.new_write_builder().new_update().unwrap(); + update.with_update_type(vec!["value".into()]).unwrap(); + let messages = update + .upsert_by_arrow_with_key(vec![source(1, 42)], vec!["key".into()]) + .await + .unwrap(); + TableCommit::new(table.clone(), "update".into()) + .commit(messages) + .await + .unwrap(); + assert_eq!( + numeric_rows(&table).await, + vec![ + (0, 1, Some(42), Some(10)), + (1, 1, Some(42), Some(11)), + (2, 2, Some(0), Some(12)) + ] + ); + let error = update + .upsert_by_arrow_with_key(vec![source(3, 7)], vec!["key".into()]) + .await + .unwrap_err(); + assert!(error.to_string().contains("Invalid upsert column 'key'")); + assert_eq!(numeric_rows(&table).await.len(), 3); + } + #[tokio::test] async fn upsert_normalizes_string_offsets_before_matching_and_writing() { use crate::catalog::Identifier; diff --git a/crates/paimon/src/table/update_input.rs b/crates/paimon/src/table/update_input.rs index d68de1185..bc8a781da 100644 --- a/crates/paimon/src/table/update_input.rs +++ b/crates/paimon/src/table/update_input.rs @@ -98,7 +98,7 @@ pub(super) fn normalize_row_ids(batch: RecordBatch) -> crate::Result crate::Result { +pub(super) fn decode_dictionary(array: &ArrayRef) -> crate::Result { let mut array = array.clone(); while let DataType::Dictionary(_, values) = array.data_type() { array = arrow_cast::cast_with_options( From cca4d9c703a1aca8f95b94251e3a4b68cc8c5ccb Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 19:59:35 +0800 Subject: [PATCH 3/5] [core] Preserve upsert casts and Java Blob map key semantics --- .../paimon/src/table/data_evolution_blob.rs | 7 +- .../paimon/src/table/data_evolution_writer.rs | 31 ++- crates/paimon/src/table/key_normalization.rs | 54 +++++ crates/paimon/src/table/table_update.rs | 3 +- crates/paimon/src/table/table_upsert.rs | 52 ++--- crates/paimon/src/table/update_input.rs | 192 ++++++++++++++---- .../paimon/src/table/write_batch_normalize.rs | 169 ++++++++++++++- 7 files changed, 422 insertions(+), 86 deletions(-) diff --git a/crates/paimon/src/table/data_evolution_blob.rs b/crates/paimon/src/table/data_evolution_blob.rs index 0ddacd59c..e171aba97 100644 --- a/crates/paimon/src/table/data_evolution_blob.rs +++ b/crates/paimon/src/table/data_evolution_blob.rs @@ -48,6 +48,7 @@ pub(super) async fn file_updates( batches: &[RecordBatch], matches: &[(usize, usize, usize)], file: &FileRowRange, + preserve_binary_layout: bool, ) -> Result> { let inline = CoreOptions::new(table.schema().options()).blob_inline_fields(); let fields = fields @@ -67,6 +68,7 @@ pub(super) async fn file_updates( matches, file.row_count as usize, baseline.contains(&field.id()), + preserve_binary_layout, )?; Ok(BlobUpdateBatch { field: field.clone(), @@ -140,6 +142,7 @@ pub(super) fn delta_batch( matches: &[(usize, usize, usize)], row_count: usize, has_baseline: bool, + preserve_binary_layout: bool, ) -> Result<(RecordBatch, Option>)> { let length = if has_baseline { matches.last().map_or(0, |(offset, _, _)| offset + 1) @@ -149,10 +152,10 @@ pub(super) fn delta_batch( let target = update_arrow_type(field, batches)?; let mut arrays = vec![arrow_array::new_null_array(&target, 1)]; for batch in batches { - arrays.push(super::super::update_input::cast_update_value( + arrays.push(super::super::update_input::cast_row_update_value( &matched_column(batch, field.name())?, &target, - super::super::update_input::CastMode::RowUpdate, + preserve_binary_layout, )?); } let mut indices = vec![(0, 0); length]; diff --git a/crates/paimon/src/table/data_evolution_writer.rs b/crates/paimon/src/table/data_evolution_writer.rs index 446f7f94e..96713d168 100644 --- a/crates/paimon/src/table/data_evolution_writer.rs +++ b/crates/paimon/src/table/data_evolution_writer.rs @@ -85,6 +85,7 @@ pub struct DataEvolutionWriter { matched_batch_groups: Vec, next_group_id: usize, blob_uri_reader_factory: Option>, + preserve_binary_layout: bool, } impl DataEvolutionWriter { @@ -176,6 +177,7 @@ impl DataEvolutionWriter { matched_batch_groups: Vec::new(), next_group_id: 1, blob_uri_reader_factory: None, + preserve_binary_layout: false, }) } @@ -186,6 +188,10 @@ impl DataEvolutionWriter { self.blob_uri_reader_factory = factory; } + pub(super) fn preserve_binary_layout(&mut self) { + self.preserve_binary_layout = true; + } + fn validate_partition_values( &self, files: &[FileRowRange], @@ -200,10 +206,10 @@ impl DataEvolutionWriter { .matched_batches .iter() .map(|batch| { - super::update_input::cast_update_value( + super::update_input::cast_row_update_value( &matched_column(batch, name)?, &target_type, - super::update_input::CastMode::RowUpdate, + self.preserve_binary_layout, ) }) .collect::>>()?; @@ -386,6 +392,7 @@ impl DataEvolutionWriter { &self.matched_batches, &sorted_matches, file_range, + self.preserve_binary_layout, ) .await?; if normal_fields.is_empty() { @@ -468,6 +475,7 @@ impl DataEvolutionWriter { &self.update_columns, &self.matched_batches, &sorted_matches, + self.preserve_binary_layout, ) }) .collect::>>()?; @@ -1119,9 +1127,16 @@ fn apply_field_updates( update_columns: &[String], matched_batches: &[RecordBatch], matches: &[(usize, usize, usize)], + preserve_binary_layout: bool, ) -> Result { if update_columns.iter().any(|column| column == path) { - let updated = interleave_updated_leaf(path, original, matched_batches, matches)?; + let updated = interleave_updated_leaf( + path, + original, + matched_batches, + matches, + preserve_binary_layout, + )?; // Readers allow runtime NULLs and relax nested field nullability. // Restore the declared write layout, validating only visible children, // before constructing a partial-column file with the table schema. @@ -1160,6 +1175,7 @@ fn apply_field_updates( update_columns, matched_batches, matches, + preserve_binary_layout, ) }) .collect::>>()?; @@ -1190,6 +1206,7 @@ fn interleave_updated_leaf( original: &ArrayRef, matched_batches: &[RecordBatch], matches: &[(usize, usize, usize)], + preserve_binary_layout: bool, ) -> Result { let mut batch_arrays = Vec::::new(); let mut batch_id_map = HashMap::::new(); @@ -1199,10 +1216,10 @@ fn interleave_updated_leaf( *index } else { let value = matched_column(&matched_batches[batch_idx], path)?; - let value = super::update_input::cast_update_value( + let value = super::update_input::cast_row_update_value( &value, original.data_type(), - super::update_input::CastMode::RowUpdate, + preserve_binary_layout, )?; let index = batch_arrays.len(); batch_arrays.push(value); @@ -2016,6 +2033,7 @@ mod tests { &["value".into()], &[updates], &[(0, 0, 0), (1, 0, 1)], + false, ) .unwrap(); let expected = crate::arrow::paimon_type_to_arrow(write_field.data_type()).unwrap(); @@ -2044,7 +2062,8 @@ mod tests { &original, &["value".into()], &[invalid], - &[(0, 0, 0)] + &[(0, 0, 0)], + false, ) .is_err()); } diff --git a/crates/paimon/src/table/key_normalization.rs b/crates/paimon/src/table/key_normalization.rs index c3d2c28b5..863007c88 100644 --- a/crates/paimon/src/table/key_normalization.rs +++ b/crates/paimon/src/table/key_normalization.rs @@ -46,6 +46,34 @@ pub(super) fn normalize_upsert_key( ) -> crate::Result { let decoded = decode_dictionary(array)?; let source = decoded.data_type(); + if source.is_decimal() && matches!(expected, DataType::Decimal128(..)) { + let source_scale = match source { + DataType::Decimal32(_, scale) + | DataType::Decimal64(_, scale) + | DataType::Decimal128(_, scale) + | DataType::Decimal256(_, scale) => *scale, + _ => unreachable!(), + }; + let DataType::Decimal128(_, target_scale) = expected else { + unreachable!() + }; + let difference = i16::from(*target_scale) - i16::from(source_scale); + let text_fallback = || { + // Arrow's decimal rescaler can reject a large scale difference + // before inspecting values, even for zero and NULL. Reuse the + // exact text parser as a fallback; it still rejects nonzero values + // that exceed precision or lose fractional digits. + let text = cast_update_value(&decoded, &DataType::Utf8, CastMode::Assignment)?; + cast_update_value(&text, expected, CastMode::Assignment) + }; + // Arrow computes scale differences in i8 and uses a Decimal128 power + // table. Skip that kernel before either can overflow. + return if !(-38..=38).contains(&difference) { + text_fallback() + } else { + cast_update_value(&decoded, expected, CastMode::Assignment).or_else(|_| text_fallback()) + }; + } let exact_family = source.is_integer() && expected.is_integer() || source.is_decimal() && expected.is_decimal() || matches!( @@ -134,6 +162,32 @@ mod tests { assert!(normalize_upsert_key(&array(12501), &expected).is_err()); } + #[test] + fn decimal_zero_and_null_keys_rescale_across_large_scale_differences() { + for (from, to) in [(-2, 38), (38, -2), (-90, 38), (-128, 38), (38, -128)] { + let input: ArrayRef = Arc::new( + Decimal128Array::from(vec![Some(0), None]) + .with_precision_and_scale(38, from) + .unwrap(), + ); + let expected = DataType::Decimal128(38, to); + let output = normalize_upsert_key(&input, &expected).unwrap(); + assert_eq!(output.data_type(), &expected); + assert_eq!( + output.as_any().downcast_ref::().unwrap(), + &Decimal128Array::from(vec![Some(0), None]) + .with_precision_and_scale(38, to) + .unwrap() + ); + let nonzero: ArrayRef = Arc::new( + Decimal128Array::from(vec![1]) + .with_precision_and_scale(38, from) + .unwrap(), + ); + assert!(normalize_upsert_key(&nonzero, &expected).is_err()); + } + } + #[test] fn temporal_keys_require_lossless_units_and_preserve_instant() { let input: ArrayRef = Arc::new(TimestampMicrosecondArray::from(vec![ diff --git a/crates/paimon/src/table/table_update.rs b/crates/paimon/src/table/table_update.rs index 35b0114b2..c20aa50ca 100644 --- a/crates/paimon/src/table/table_update.rs +++ b/crates/paimon/src/table/table_update.rs @@ -245,7 +245,8 @@ impl TableUpdate { /// matched rows retain their values; on new rows they are absent from the file. /// After last-write-wins, matched values follow row-ID update conversion /// rules; new rows must satisfy the ordinary Arrow write type. - /// Binary values, including nested ones, retain ordinary write layout rules. + /// Binary and LargeBinary are not implicitly converted to each other, + /// including inside nested values. /// Input batches may have different non-key columns. After last-write-wins /// deduplication, matched rows must supply their update columns and appended /// rows in each partition must share a field set. diff --git a/crates/paimon/src/table/table_upsert.rs b/crates/paimon/src/table/table_upsert.rs index b0c8e6fc2..7017375e7 100644 --- a/crates/paimon/src/table/table_upsert.rs +++ b/crates/paimon/src/table/table_upsert.rs @@ -46,32 +46,6 @@ fn invalid(message: impl Into) -> crate::Error { } } -// BYTES and BLOB keep their declared Arrow representation. Reusing row-ID -// conversions for matched values must not introduce Binary/LargeBinary casts. -fn contains_binary(data_type: &DataType) -> bool { - match data_type { - DataType::Binary - | DataType::LargeBinary - | DataType::FixedSizeBinary(_) - | DataType::BinaryView => true, - DataType::Dictionary(_, value) => contains_binary(value), - DataType::Struct(fields) => fields - .iter() - .any(|field| contains_binary(field.data_type())), - DataType::List(field) - | DataType::LargeList(field) - | DataType::ListView(field) - | DataType::LargeListView(field) - | DataType::FixedSizeList(field, _) - | DataType::Map(field, _) - | DataType::RunEndEncoded(_, field) => contains_binary(field.data_type()), - DataType::Union(fields, _) => fields - .iter() - .any(|(_, field)| contains_binary(field.data_type())), - _ => false, - } -} - fn selected_rows(batch: &RecordBatch, rows: &[usize]) -> crate::Result { let indices = rows .iter() @@ -243,18 +217,19 @@ impl TableUpsert { .clone(), column.clone(), )) - } else if logical.data_type().is_blob_file_field() && !inline.contains(logical.name()) { - normalize_blob_write_field(column, field) - } else if append - || contains_binary(column.data_type()) - || contains_binary(field.data_type()) + } else if logical.data_type().is_blob_file_field() + && !inline.contains(logical.name()) + && (append || matches!(logical.data_type(), crate::spec::DataType::Map(_))) { - let value = if append { - column.clone() + // Named-row Blob maps carry nullable keys and decimal text in + // their dedicated physical layout. Keep that transport intact. + if append { + normalize_blob_write_field(column, field) } else { - super::update_input::decode_dictionary(column)? - }; - normalize_write_array(&value, field.data_type()) + super::write_batch_normalize::normalize_blob_update_field(column, field) + } + } else if append { + normalize_write_array(column, field.data_type()) .map(|column| (field.as_ref().clone(), column)) } else { Ok(( @@ -446,6 +421,7 @@ impl TableUpsert { &self.table, self.update_columns.clone(), )?; + update.preserve_binary_layout(); update.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone()); if let Some(snapshot_id) = plan.snapshot_id() { update.pin_read_snapshot(snapshot_id); @@ -730,7 +706,9 @@ mod tests { .upsert_by_arrow_with_key(vec![input], vec!["key".into()]) .await .unwrap_err(); - assert!(error.to_string().contains("Invalid upsert column 'value'")); + assert!(error + .to_string() + .contains("incompatible with table type Binary")); } #[tokio::test] diff --git a/crates/paimon/src/table/update_input.rs b/crates/paimon/src/table/update_input.rs index bc8a781da..2249c15c7 100644 --- a/crates/paimon/src/table/update_input.rs +++ b/crates/paimon/src/table/update_input.rs @@ -131,32 +131,66 @@ pub(super) fn cast_update_value( mode: CastMode, ) -> crate::Result { if mode == CastMode::RowUpdate { - cast_values(array, target, CastMode::Assignment).or_else(|_| { - validate_map_constructor_input(array, target)?; - let decoded = decode_dictionary(array)?; - // Python converts lists of pairs to tuples only for a top-level MAP. - if let DataType::Map(field, sorted) = target { - if matches!( - decoded.data_type(), - DataType::List(_) - | DataType::LargeList(_) - | DataType::ListView(_) - | DataType::LargeListView(_) - | DataType::FixedSizeList(..) - ) { - return constructor_map(&decoded, field, *sorted); - } - } - cast_values(array, target, CastMode::Constructor) - }) + cast_row_update_value(array, target, false) } else { - cast_values(array, target, mode) + cast_values(array, target, mode, false) } } +/// Upserts preserve binary write layouts without disabling other row-ID casts. +/// Keep this policy through the whole-column constructor fallback and all +/// recursive conversions, including MAP/ROW constructor paths. +pub(super) fn cast_row_update_value( + array: &ArrayRef, + target: &DataType, + preserve_binary_layout: bool, +) -> crate::Result { + cast_values(array, target, CastMode::Assignment, preserve_binary_layout).or_else(|_| { + validate_map_constructor_input(array, target)?; + let decoded = decode_dictionary(array)?; + // Python converts lists of pairs to tuples only for a top-level MAP. + if let DataType::Map(field, sorted) = target { + if matches!( + decoded.data_type(), + DataType::List(_) + | DataType::LargeList(_) + | DataType::ListView(_) + | DataType::LargeListView(_) + | DataType::FixedSizeList(..) + ) { + return constructor_map(&decoded, field, *sorted, preserve_binary_layout); + } + } + cast_values(array, target, CastMode::Constructor, preserve_binary_layout) + }) +} + /// Cast logical values recursively, so dictionary encoding and nesting cannot /// bypass primitive validation. The final outer cast only converts the layout. -fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Result { +fn cast_values( + array: &ArrayRef, + target: &DataType, + mode: CastMode, + preserve_binary_layout: bool, +) -> crate::Result { + if preserve_binary_layout && matches!(array.data_type(), DataType::Dictionary(..)) { + return cast_values( + &decode_dictionary(array)?, + target, + mode, + preserve_binary_layout, + ); + } + if preserve_binary_layout + && matches!( + (array.data_type(), target), + (DataType::Binary, DataType::LargeBinary) | (DataType::LargeBinary, DataType::Binary) + ) + { + // Check before the constructor's all-NULL shortcut. A failed safe + // cast must not retry a prohibited width conversion via NULL values. + return super::write_batch_normalize::normalize_write_array(array, target); + } if mode == CastMode::Constructor && array.null_count() == array.len() { return Ok(new_null_array(target, array.len())); } @@ -164,7 +198,12 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re return Ok(array.clone()); } if matches!(array.data_type(), DataType::Dictionary(..)) { - return cast_values(&decode_dictionary(array)?, target, mode); + return cast_values( + &decode_dictionary(array)?, + target, + mode, + preserve_binary_layout, + ); } if let DataType::ListView(field) | DataType::LargeListView(field) = array.data_type() { // Convert only the offsets/layout first. Casting children here would @@ -179,7 +218,7 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re }, ) .map_err(|error| invalid(error.to_string()))?; - return cast_values(&materialized, target, mode); + return cast_values(&materialized, target, mode, preserve_binary_layout); } if matches!(array.data_type(), DataType::RunEndEncoded(..)) { return Err(invalid( @@ -189,12 +228,13 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re if mode == CastMode::Constructor { match (array.data_type(), target) { (DataType::Struct(_), DataType::Map(field, sorted)) => { - return constructor_map(array, field, *sorted); + return constructor_map(array, field, *sorted, preserve_binary_layout); } (DataType::Map(..), DataType::Struct(fields)) => { return constructor_struct_from_map( array.as_any().downcast_ref::().unwrap(), fields, + preserve_binary_layout, ); } _ => {} @@ -211,7 +251,7 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re .map(|field| match input.column_by_name(field.name()) { Some(column) => { let column = constructor_child(column, input.nulls(), 1, mode)?; - cast_values(&column, field.data_type(), mode) + cast_values(&column, field.data_type(), mode, preserve_binary_layout) } None => Ok(new_null_array(field.data_type(), input.len())), }) @@ -234,17 +274,19 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re array.as_any().downcast_ref::().unwrap(), field, mode, + preserve_binary_layout, )?), DataType::LargeList(_) => Some(cast_list( array.as_any().downcast_ref::().unwrap(), field, mode, + preserve_binary_layout, )?), DataType::FixedSizeList(_, size) => { let input = array.as_any().downcast_ref::().unwrap(); let values = constructor_child(input.values(), input.nulls(), *size as usize, mode)?; - let values = cast_values(&values, field.data_type(), mode)?; + let values = cast_values(&values, field.data_type(), mode, preserve_binary_layout)?; Some(Arc::new( FixedSizeListArray::try_new_with_length( field.clone(), @@ -279,7 +321,7 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Re .map_err(|error| invalid(error.to_string()))?, None => values.slice(start as usize, length), }; - cast_values(&values, field.data_type(), mode) + cast_values(&values, field.data_type(), mode, preserve_binary_layout) }) .collect::>>()?; let entries = StructArray::try_new_with_length(fields.clone(), columns, None, length) @@ -368,6 +410,7 @@ fn validate_struct_cast_fields( fn constructor_struct_from_map( input: &MapArray, fields: &arrow_schema::Fields, + preserve_binary_layout: bool, ) -> crate::Result { let keys = decode_dictionary(input.keys())?; let columns = fields @@ -412,7 +455,12 @@ fn constructor_struct_from_map( None, ) .map_err(|error| invalid(error.to_string()))?; - cast_values(&values, field.data_type(), CastMode::Constructor) + cast_values( + &values, + field.data_type(), + CastMode::Constructor, + preserve_binary_layout, + ) }) .collect::>>()?; Ok(Arc::new( @@ -430,6 +478,7 @@ fn constructor_map( input: &ArrayRef, entries_field: &arrow_schema::FieldRef, sorted: bool, + preserve_binary_layout: bool, ) -> crate::Result { let DataType::Struct(fields) = entries_field.data_type() else { return Err(invalid("Map entries must have a struct type")); @@ -452,7 +501,12 @@ fn constructor_map( .iter() .map(|column| { let visible = constructor_child(column, input.nulls(), 1, CastMode::Constructor)?; - cast_values(&visible, fields[1].data_type(), CastMode::Constructor) + cast_values( + &visible, + fields[1].data_type(), + CastMode::Constructor, + preserve_binary_layout, + ) }) .collect::>>()?; let values = if columns.is_empty() { @@ -489,11 +543,16 @@ fn constructor_map( offsets.push(i32::try_from(keys.len()).map_err(|_| invalid("Map offset overflow"))?); } ( - concat_constructor_values(&keys, fields[0].data_type())?, - concat_constructor_values(&values, fields[1].data_type())?, + concat_constructor_values(&keys, fields[0].data_type(), preserve_binary_layout)?, + concat_constructor_values(&values, fields[1].data_type(), preserve_binary_layout)?, ) }; - let keys = cast_values(&keys, fields[0].data_type(), CastMode::Constructor)?; + let keys = cast_values( + &keys, + fields[0].data_type(), + CastMode::Constructor, + preserve_binary_layout, + )?; let entries = StructArray::try_new(fields.clone(), vec![keys, values], None) .map_err(|error| invalid(error.to_string()))?; Ok(Arc::new( @@ -539,7 +598,11 @@ fn constructor_list_value(input: &ArrayRef, index: usize) -> crate::Result crate::Result { +fn concat_constructor_values( + values: &[ArrayRef], + target: &DataType, + preserve_binary_layout: bool, +) -> crate::Result { if values.is_empty() { return Ok(arrow_array::new_empty_array(target)); } @@ -549,13 +612,19 @@ fn concat_constructor_values(values: &[ArrayRef], target: &DataType) -> crate::R .collect::>(); let combined = arrow_select::concat::concat(&values).map_err(|error| invalid(error.to_string()))?; - cast_values(&combined, target, CastMode::Constructor) + cast_values( + &combined, + target, + CastMode::Constructor, + preserve_binary_layout, + ) } fn cast_list( input: &GenericListArray, field: &arrow_schema::FieldRef, mode: CastMode, + preserve_binary_layout: bool, ) -> crate::Result { let offsets = input.value_offsets(); let start = offsets[0]; @@ -570,7 +639,7 @@ fn cast_list( .values() .slice(start.as_usize(), (end - start).as_usize()), }; - let values = cast_values(&values, field.data_type(), mode)?; + let values = cast_values(&values, field.data_type(), mode, preserve_binary_layout)?; let offsets = OffsetBuffer::new(offsets.into()); Ok(Arc::new( GenericListArray::::try_new(field.clone(), offsets, values, input.nulls().cloned()) @@ -2014,6 +2083,61 @@ mod tests { assert!(cast_assignment(&input, &target).is_err()); } + #[test] + fn preserved_binary_layouts_allow_text_numeric_and_null_update_casts() { + let text: ArrayRef = Arc::new(StringArray::from(vec![Some("42"), None])); + let binary = cast_row_update_value(&text, &DataType::Binary, true).unwrap(); + let integers = cast_row_update_value(&binary, &DataType::Int32, true).unwrap(); + assert_eq!( + integers.to_data(), + Int32Array::from(vec![Some(42), None]).to_data() + ); + assert_eq!( + cast_row_update_value(&new_null_array(&DataType::Null, 2), &DataType::Binary, true) + .unwrap() + .null_count(), + 2 + ); + } + + #[test] + fn preserved_binary_layouts_reject_width_changes_before_null_fallbacks() { + use arrow_array::{types::Int8Type, DictionaryArray, Int8Array}; + for (source, target) in [ + (DataType::Binary, DataType::LargeBinary), + (DataType::LargeBinary, DataType::Binary), + ] { + let array = new_null_array(&source, 1); + assert!(cast_row_update_value(&array, &target, true).is_err()); + // Other row-ID/MERGE callers keep their established cast contract. + assert!(cast_update_value(&array, &target, CastMode::RowUpdate).is_ok()); + let dictionary: ArrayRef = Arc::new( + DictionaryArray::::try_new(Int8Array::from(vec![Some(0)]), array.clone()) + .unwrap(), + ); + assert!(cast_row_update_value(&dictionary, &target, true).is_err()); + let field = Arc::new(arrow_schema::Field::new("data", source, true)); + let nested: ArrayRef = Arc::new( + StructArray::try_new( + vec![field].into(), + vec![array], + Some(arrow_buffer::NullBuffer::from(vec![false])), + ) + .unwrap(), + ); + let expected = + DataType::Struct(vec![arrow_schema::Field::new("data", target, true)].into()); + // A NULL container reconstructs no child payload and follows the + // existing whole-value constructor behavior. + assert_eq!( + cast_row_update_value(&nested, &expected, true) + .unwrap() + .null_count(), + 1 + ); + } + } + #[test] fn binary_assignments_parse_text_and_never_reinterpret_numeric_bytes() { let binary: ArrayRef = Arc::new(arrow_array::BinaryArray::from(vec![ diff --git a/crates/paimon/src/table/write_batch_normalize.rs b/crates/paimon/src/table/write_batch_normalize.rs index e7de7ec7e..a3b0cc8c2 100644 --- a/crates/paimon/src/table/write_batch_normalize.rs +++ b/crates/paimon/src/table/write_batch_normalize.rs @@ -65,6 +65,33 @@ pub(super) fn blob_map_row_type(expected: &DataType) -> Option { pub(super) fn normalize_blob_write_field( array: &ArrayRef, expected: &Field, +) -> Result<(Field, ArrayRef)> { + normalize_blob_field(array, expected, true) +} + +/// Normalize Blob map keys and their physical transport, retaining the value +/// input type for the row-ID updater's conversion at the existing read layout. +pub(super) fn normalize_blob_update_field( + array: &ArrayRef, + expected: &Field, +) -> Result<(Field, ArrayRef)> { + let array = super::update_input::decode_dictionary(array)?; + if matches!(array.data_type(), DataType::Map(..) | DataType::List(_)) { + normalize_blob_field(&array, expected, false) + } else { + // NULLs and other row-ID constructor inputs retain their conversion + // at the existing DE writer position, rather than the declared schema. + Ok(( + expected.clone().with_data_type(array.data_type().clone()), + array, + )) + } +} + +fn normalize_blob_field( + array: &ArrayRef, + expected: &Field, + normalize_values: bool, ) -> Result<(Field, ArrayRef)> { let Some(target) = blob_map_row_type(expected.data_type()) else { return Ok(( @@ -121,17 +148,39 @@ pub(super) fn normalize_blob_write_field( .enumerate() .map(|(index, (column, field))| { if index == 0 { - normalize_blob_map_key(column, field.data_type()) - } else { + if normalize_values { + normalize_blob_map_key(column, field.data_type()) + } else { + normalize_blob_update_key(column, field.data_type()) + } + } else if normalize_values { normalize_write_array(column, field.data_type()) + } else { + Ok(column.clone()) } }) .collect::>>()?; - let entries = - StructArray::try_new(fields.clone(), columns, None).map_err(normalization_error)?; - let list = ListArray::try_new(child.clone(), offsets, Arc::new(entries), nulls.cloned()) + let mut fields = fields.to_vec(); + fields[1] = Arc::new( + fields[1] + .as_ref() + .clone() + .with_data_type(columns[1].data_type().clone()), + ); + let fields = fields.into(); + let entries = StructArray::try_new(fields, columns, None).map_err(normalization_error)?; + let child = Arc::new( + child + .as_ref() + .clone() + .with_data_type(entries.data_type().clone()), + ); + let list = ListArray::try_new(child, offsets, Arc::new(entries), nulls.cloned()) .map_err(normalization_error)?; - Ok((expected.clone().with_data_type(target), Arc::new(list))) + Ok(( + expected.clone().with_data_type(list.data_type().clone()), + Arc::new(list), + )) } fn normalize_blob_map_key(array: &ArrayRef, expected: &DataType) -> Result { @@ -165,6 +214,36 @@ fn normalize_blob_map_key(array: &ArrayRef, expected: &DataType) -> Result Result { + let decoded = super::update_input::decode_dictionary(array)?; + if matches!(expected, DataType::Decimal128(..)) && decoded.data_type() != &DataType::Null { + // Blob map keys follow Java Decimal.fromBigDecimal (HALF_UP), rather + // than the exact decimal conversion used to match table upsert keys. + let text = if decoded.data_type().is_decimal() { + super::update_input::cast_update_value( + &decoded, + &DataType::Utf8, + super::update_input::CastMode::Assignment, + )? + } else { + decoded + }; + return normalize_blob_map_key(&text, expected); + } + if matches!( + (decoded.data_type(), expected), + ( + DataType::Time32(_) | DataType::Time64(_), + DataType::Time32(_) | DataType::Time64(_) + ) + ) { + // Java/Python serialize TIME Blob keys in milliseconds. This is a + // value normalization step, unlike lossless table-key matching. + return super::update_input::cast_row_update_value(&decoded, expected, true); + } + super::key_normalization::normalize_upsert_key(&decoded, expected) +} + /// Decimal.fromBigDecimal: round magnitude HALF_UP, then check precision. /// Discarded fractional digits are never accumulated into i128, so source /// decimals can have more digits than the target's maximum precision of 38. @@ -972,6 +1051,84 @@ mod tests { assert!(normalize_blob_write_field(&visible_overflow, &expected).is_err()); } + #[test] + fn blob_map_updates_normalize_keys_and_transport_before_value_casts() { + let entries = StructArray::new( + vec![ + Field::new("k", DataType::Utf8, false), + Field::new("v", DataType::Utf8, true), + ] + .into(), + vec![ + Arc::new(StringArray::from(vec!["1.235"])), + Arc::new(StringArray::from(vec!["new"])), + ], + None, + ); + let entry_field = Arc::new(Field::new("input", entries.data_type().clone(), false)); + let offsets = OffsetBuffer::new(vec![0_i32, 1].into()); + let map: ArrayRef = Arc::new(MapArray::new( + entry_field.clone(), + offsets.clone(), + entries.clone(), + None, + false, + )); + let list: ArrayRef = Arc::new(ListArray::new( + entry_field, + offsets, + Arc::new(entries), + None, + )); + let expected = Field::new( + "payload", + DataType::Map( + Arc::new(Field::new( + "entries", + DataType::Struct( + vec![ + Field::new("key", DataType::Decimal128(10, 2), false), + Field::new("value", DataType::LargeBinary, true), + ] + .into(), + ), + false, + )), + false, + ), + true, + ); + let normalized = [map, list] + .iter() + .map(|array| normalize_blob_update_field(array, &expected).unwrap()) + .collect::>(); + assert_eq!(normalized[0].0, normalized[1].0); + assert_eq!(normalized[0].1.to_data(), normalized[1].1.to_data()); + let list = normalized[0] + .1 + .as_any() + .downcast_ref::() + .unwrap(); + let entries = list + .values() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + entries + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 124 + ); + // Preserve the original input value type until the row-ID updater + // casts against the actual read layout, including its nullability. + assert_eq!(entries.column(1).data_type(), &DataType::Utf8); + assert!(normalize_blob_write_field(&normalized[0].1, &expected).is_err()); + } + #[test] fn sliced_blob_map_inputs_drop_unreferenced_entries() { let input = blob_map_rows( From 581e698230d9d4c1791cbc6299736524cd02c11a Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 20:56:59 +0800 Subject: [PATCH 4/5] Reuse row-ID binary conversions for matched upserts --- .../paimon/src/table/data_evolution_blob.rs | 7 +- .../paimon/src/table/data_evolution_writer.rs | 31 +-- crates/paimon/src/table/table_update.rs | 2 - crates/paimon/src/table/table_upsert.rs | 34 +++- crates/paimon/src/table/update_input.rs | 191 ++++-------------- .../paimon/src/table/write_batch_normalize.rs | 6 +- 6 files changed, 83 insertions(+), 188 deletions(-) diff --git a/crates/paimon/src/table/data_evolution_blob.rs b/crates/paimon/src/table/data_evolution_blob.rs index e171aba97..0ddacd59c 100644 --- a/crates/paimon/src/table/data_evolution_blob.rs +++ b/crates/paimon/src/table/data_evolution_blob.rs @@ -48,7 +48,6 @@ pub(super) async fn file_updates( batches: &[RecordBatch], matches: &[(usize, usize, usize)], file: &FileRowRange, - preserve_binary_layout: bool, ) -> Result> { let inline = CoreOptions::new(table.schema().options()).blob_inline_fields(); let fields = fields @@ -68,7 +67,6 @@ pub(super) async fn file_updates( matches, file.row_count as usize, baseline.contains(&field.id()), - preserve_binary_layout, )?; Ok(BlobUpdateBatch { field: field.clone(), @@ -142,7 +140,6 @@ pub(super) fn delta_batch( matches: &[(usize, usize, usize)], row_count: usize, has_baseline: bool, - preserve_binary_layout: bool, ) -> Result<(RecordBatch, Option>)> { let length = if has_baseline { matches.last().map_or(0, |(offset, _, _)| offset + 1) @@ -152,10 +149,10 @@ pub(super) fn delta_batch( let target = update_arrow_type(field, batches)?; let mut arrays = vec![arrow_array::new_null_array(&target, 1)]; for batch in batches { - arrays.push(super::super::update_input::cast_row_update_value( + arrays.push(super::super::update_input::cast_update_value( &matched_column(batch, field.name())?, &target, - preserve_binary_layout, + super::super::update_input::CastMode::RowUpdate, )?); } let mut indices = vec![(0, 0); length]; diff --git a/crates/paimon/src/table/data_evolution_writer.rs b/crates/paimon/src/table/data_evolution_writer.rs index 96713d168..446f7f94e 100644 --- a/crates/paimon/src/table/data_evolution_writer.rs +++ b/crates/paimon/src/table/data_evolution_writer.rs @@ -85,7 +85,6 @@ pub struct DataEvolutionWriter { matched_batch_groups: Vec, next_group_id: usize, blob_uri_reader_factory: Option>, - preserve_binary_layout: bool, } impl DataEvolutionWriter { @@ -177,7 +176,6 @@ impl DataEvolutionWriter { matched_batch_groups: Vec::new(), next_group_id: 1, blob_uri_reader_factory: None, - preserve_binary_layout: false, }) } @@ -188,10 +186,6 @@ impl DataEvolutionWriter { self.blob_uri_reader_factory = factory; } - pub(super) fn preserve_binary_layout(&mut self) { - self.preserve_binary_layout = true; - } - fn validate_partition_values( &self, files: &[FileRowRange], @@ -206,10 +200,10 @@ impl DataEvolutionWriter { .matched_batches .iter() .map(|batch| { - super::update_input::cast_row_update_value( + super::update_input::cast_update_value( &matched_column(batch, name)?, &target_type, - self.preserve_binary_layout, + super::update_input::CastMode::RowUpdate, ) }) .collect::>>()?; @@ -392,7 +386,6 @@ impl DataEvolutionWriter { &self.matched_batches, &sorted_matches, file_range, - self.preserve_binary_layout, ) .await?; if normal_fields.is_empty() { @@ -475,7 +468,6 @@ impl DataEvolutionWriter { &self.update_columns, &self.matched_batches, &sorted_matches, - self.preserve_binary_layout, ) }) .collect::>>()?; @@ -1127,16 +1119,9 @@ fn apply_field_updates( update_columns: &[String], matched_batches: &[RecordBatch], matches: &[(usize, usize, usize)], - preserve_binary_layout: bool, ) -> Result { if update_columns.iter().any(|column| column == path) { - let updated = interleave_updated_leaf( - path, - original, - matched_batches, - matches, - preserve_binary_layout, - )?; + let updated = interleave_updated_leaf(path, original, matched_batches, matches)?; // Readers allow runtime NULLs and relax nested field nullability. // Restore the declared write layout, validating only visible children, // before constructing a partial-column file with the table schema. @@ -1175,7 +1160,6 @@ fn apply_field_updates( update_columns, matched_batches, matches, - preserve_binary_layout, ) }) .collect::>>()?; @@ -1206,7 +1190,6 @@ fn interleave_updated_leaf( original: &ArrayRef, matched_batches: &[RecordBatch], matches: &[(usize, usize, usize)], - preserve_binary_layout: bool, ) -> Result { let mut batch_arrays = Vec::::new(); let mut batch_id_map = HashMap::::new(); @@ -1216,10 +1199,10 @@ fn interleave_updated_leaf( *index } else { let value = matched_column(&matched_batches[batch_idx], path)?; - let value = super::update_input::cast_row_update_value( + let value = super::update_input::cast_update_value( &value, original.data_type(), - preserve_binary_layout, + super::update_input::CastMode::RowUpdate, )?; let index = batch_arrays.len(); batch_arrays.push(value); @@ -2033,7 +2016,6 @@ mod tests { &["value".into()], &[updates], &[(0, 0, 0), (1, 0, 1)], - false, ) .unwrap(); let expected = crate::arrow::paimon_type_to_arrow(write_field.data_type()).unwrap(); @@ -2062,8 +2044,7 @@ mod tests { &original, &["value".into()], &[invalid], - &[(0, 0, 0)], - false, + &[(0, 0, 0)] ) .is_err()); } diff --git a/crates/paimon/src/table/table_update.rs b/crates/paimon/src/table/table_update.rs index c20aa50ca..fdaa256ef 100644 --- a/crates/paimon/src/table/table_update.rs +++ b/crates/paimon/src/table/table_update.rs @@ -245,8 +245,6 @@ impl TableUpdate { /// matched rows retain their values; on new rows they are absent from the file. /// After last-write-wins, matched values follow row-ID update conversion /// rules; new rows must satisfy the ordinary Arrow write type. - /// Binary and LargeBinary are not implicitly converted to each other, - /// including inside nested values. /// Input batches may have different non-key columns. After last-write-wins /// deduplication, matched rows must supply their update columns and appended /// rows in each partition must share a field set. diff --git a/crates/paimon/src/table/table_upsert.rs b/crates/paimon/src/table/table_upsert.rs index 7017375e7..779573b31 100644 --- a/crates/paimon/src/table/table_upsert.rs +++ b/crates/paimon/src/table/table_upsert.rs @@ -421,7 +421,6 @@ impl TableUpsert { &self.table, self.update_columns.clone(), )?; - update.preserve_binary_layout(); update.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone()); if let Some(snapshot_id) = plan.snapshot_id() { update.pin_read_snapshot(snapshot_id); @@ -667,7 +666,7 @@ mod tests { } #[tokio::test] - async fn upsert_update_casts_keep_binary_write_layouts() { + async fn upsert_matched_binary_values_use_existing_row_id_casts() { use arrow_array::{BinaryArray, LargeBinaryArray}; let schema = PaimonSchema::builder() .column("key", PaimonType::Int(IntType::new())) @@ -700,15 +699,34 @@ mod tests { let update = table.new_write_builder().new_update().unwrap(); let input = numeric_source( vec![1], - Arc::new(LargeBinaryArray::from(vec![b"new".as_slice()])), + Arc::new(LargeBinaryArray::from(vec![b"\xff\0new".as_slice()])), ); - let error = update + let messages = update .upsert_by_arrow_with_key(vec![input], vec!["key".into()]) .await - .unwrap_err(); - assert!(error - .to_string() - .contains("incompatible with table type Binary")); + .unwrap(); + TableCommit::new(table.clone(), "update".into()) + .commit(messages) + .await + .unwrap(); + let mut builder = table.new_read_builder(); + builder.with_projection(&["value"]).unwrap(); + let plan = builder.new_scan().plan().await.unwrap(); + let batches: Vec<_> = builder + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 1); + let values = batches[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.value(0), b"\xff\0new"); } #[tokio::test] diff --git a/crates/paimon/src/table/update_input.rs b/crates/paimon/src/table/update_input.rs index 2249c15c7..f6a809db9 100644 --- a/crates/paimon/src/table/update_input.rs +++ b/crates/paimon/src/table/update_input.rs @@ -131,66 +131,32 @@ pub(super) fn cast_update_value( mode: CastMode, ) -> crate::Result { if mode == CastMode::RowUpdate { - cast_row_update_value(array, target, false) + cast_values(array, target, CastMode::Assignment).or_else(|_| { + validate_map_constructor_input(array, target)?; + let decoded = decode_dictionary(array)?; + // Python converts lists of pairs to tuples only for a top-level MAP. + if let DataType::Map(field, sorted) = target { + if matches!( + decoded.data_type(), + DataType::List(_) + | DataType::LargeList(_) + | DataType::ListView(_) + | DataType::LargeListView(_) + | DataType::FixedSizeList(..) + ) { + return constructor_map(&decoded, field, *sorted); + } + } + cast_values(array, target, CastMode::Constructor) + }) } else { - cast_values(array, target, mode, false) + cast_values(array, target, mode) } } -/// Upserts preserve binary write layouts without disabling other row-ID casts. -/// Keep this policy through the whole-column constructor fallback and all -/// recursive conversions, including MAP/ROW constructor paths. -pub(super) fn cast_row_update_value( - array: &ArrayRef, - target: &DataType, - preserve_binary_layout: bool, -) -> crate::Result { - cast_values(array, target, CastMode::Assignment, preserve_binary_layout).or_else(|_| { - validate_map_constructor_input(array, target)?; - let decoded = decode_dictionary(array)?; - // Python converts lists of pairs to tuples only for a top-level MAP. - if let DataType::Map(field, sorted) = target { - if matches!( - decoded.data_type(), - DataType::List(_) - | DataType::LargeList(_) - | DataType::ListView(_) - | DataType::LargeListView(_) - | DataType::FixedSizeList(..) - ) { - return constructor_map(&decoded, field, *sorted, preserve_binary_layout); - } - } - cast_values(array, target, CastMode::Constructor, preserve_binary_layout) - }) -} - /// Cast logical values recursively, so dictionary encoding and nesting cannot /// bypass primitive validation. The final outer cast only converts the layout. -fn cast_values( - array: &ArrayRef, - target: &DataType, - mode: CastMode, - preserve_binary_layout: bool, -) -> crate::Result { - if preserve_binary_layout && matches!(array.data_type(), DataType::Dictionary(..)) { - return cast_values( - &decode_dictionary(array)?, - target, - mode, - preserve_binary_layout, - ); - } - if preserve_binary_layout - && matches!( - (array.data_type(), target), - (DataType::Binary, DataType::LargeBinary) | (DataType::LargeBinary, DataType::Binary) - ) - { - // Check before the constructor's all-NULL shortcut. A failed safe - // cast must not retry a prohibited width conversion via NULL values. - return super::write_batch_normalize::normalize_write_array(array, target); - } +fn cast_values(array: &ArrayRef, target: &DataType, mode: CastMode) -> crate::Result { if mode == CastMode::Constructor && array.null_count() == array.len() { return Ok(new_null_array(target, array.len())); } @@ -198,12 +164,7 @@ fn cast_values( return Ok(array.clone()); } if matches!(array.data_type(), DataType::Dictionary(..)) { - return cast_values( - &decode_dictionary(array)?, - target, - mode, - preserve_binary_layout, - ); + return cast_values(&decode_dictionary(array)?, target, mode); } if let DataType::ListView(field) | DataType::LargeListView(field) = array.data_type() { // Convert only the offsets/layout first. Casting children here would @@ -218,7 +179,7 @@ fn cast_values( }, ) .map_err(|error| invalid(error.to_string()))?; - return cast_values(&materialized, target, mode, preserve_binary_layout); + return cast_values(&materialized, target, mode); } if matches!(array.data_type(), DataType::RunEndEncoded(..)) { return Err(invalid( @@ -228,13 +189,12 @@ fn cast_values( if mode == CastMode::Constructor { match (array.data_type(), target) { (DataType::Struct(_), DataType::Map(field, sorted)) => { - return constructor_map(array, field, *sorted, preserve_binary_layout); + return constructor_map(array, field, *sorted); } (DataType::Map(..), DataType::Struct(fields)) => { return constructor_struct_from_map( array.as_any().downcast_ref::().unwrap(), fields, - preserve_binary_layout, ); } _ => {} @@ -251,7 +211,7 @@ fn cast_values( .map(|field| match input.column_by_name(field.name()) { Some(column) => { let column = constructor_child(column, input.nulls(), 1, mode)?; - cast_values(&column, field.data_type(), mode, preserve_binary_layout) + cast_values(&column, field.data_type(), mode) } None => Ok(new_null_array(field.data_type(), input.len())), }) @@ -274,19 +234,17 @@ fn cast_values( array.as_any().downcast_ref::().unwrap(), field, mode, - preserve_binary_layout, )?), DataType::LargeList(_) => Some(cast_list( array.as_any().downcast_ref::().unwrap(), field, mode, - preserve_binary_layout, )?), DataType::FixedSizeList(_, size) => { let input = array.as_any().downcast_ref::().unwrap(); let values = constructor_child(input.values(), input.nulls(), *size as usize, mode)?; - let values = cast_values(&values, field.data_type(), mode, preserve_binary_layout)?; + let values = cast_values(&values, field.data_type(), mode)?; Some(Arc::new( FixedSizeListArray::try_new_with_length( field.clone(), @@ -321,7 +279,7 @@ fn cast_values( .map_err(|error| invalid(error.to_string()))?, None => values.slice(start as usize, length), }; - cast_values(&values, field.data_type(), mode, preserve_binary_layout) + cast_values(&values, field.data_type(), mode) }) .collect::>>()?; let entries = StructArray::try_new_with_length(fields.clone(), columns, None, length) @@ -410,7 +368,6 @@ fn validate_struct_cast_fields( fn constructor_struct_from_map( input: &MapArray, fields: &arrow_schema::Fields, - preserve_binary_layout: bool, ) -> crate::Result { let keys = decode_dictionary(input.keys())?; let columns = fields @@ -455,12 +412,7 @@ fn constructor_struct_from_map( None, ) .map_err(|error| invalid(error.to_string()))?; - cast_values( - &values, - field.data_type(), - CastMode::Constructor, - preserve_binary_layout, - ) + cast_values(&values, field.data_type(), CastMode::Constructor) }) .collect::>>()?; Ok(Arc::new( @@ -478,7 +430,6 @@ fn constructor_map( input: &ArrayRef, entries_field: &arrow_schema::FieldRef, sorted: bool, - preserve_binary_layout: bool, ) -> crate::Result { let DataType::Struct(fields) = entries_field.data_type() else { return Err(invalid("Map entries must have a struct type")); @@ -501,12 +452,7 @@ fn constructor_map( .iter() .map(|column| { let visible = constructor_child(column, input.nulls(), 1, CastMode::Constructor)?; - cast_values( - &visible, - fields[1].data_type(), - CastMode::Constructor, - preserve_binary_layout, - ) + cast_values(&visible, fields[1].data_type(), CastMode::Constructor) }) .collect::>>()?; let values = if columns.is_empty() { @@ -543,16 +489,11 @@ fn constructor_map( offsets.push(i32::try_from(keys.len()).map_err(|_| invalid("Map offset overflow"))?); } ( - concat_constructor_values(&keys, fields[0].data_type(), preserve_binary_layout)?, - concat_constructor_values(&values, fields[1].data_type(), preserve_binary_layout)?, + concat_constructor_values(&keys, fields[0].data_type())?, + concat_constructor_values(&values, fields[1].data_type())?, ) }; - let keys = cast_values( - &keys, - fields[0].data_type(), - CastMode::Constructor, - preserve_binary_layout, - )?; + let keys = cast_values(&keys, fields[0].data_type(), CastMode::Constructor)?; let entries = StructArray::try_new(fields.clone(), vec![keys, values], None) .map_err(|error| invalid(error.to_string()))?; Ok(Arc::new( @@ -598,11 +539,7 @@ fn constructor_list_value(input: &ArrayRef, index: usize) -> crate::Result crate::Result { +fn concat_constructor_values(values: &[ArrayRef], target: &DataType) -> crate::Result { if values.is_empty() { return Ok(arrow_array::new_empty_array(target)); } @@ -612,19 +549,13 @@ fn concat_constructor_values( .collect::>(); let combined = arrow_select::concat::concat(&values).map_err(|error| invalid(error.to_string()))?; - cast_values( - &combined, - target, - CastMode::Constructor, - preserve_binary_layout, - ) + cast_values(&combined, target, CastMode::Constructor) } fn cast_list( input: &GenericListArray, field: &arrow_schema::FieldRef, mode: CastMode, - preserve_binary_layout: bool, ) -> crate::Result { let offsets = input.value_offsets(); let start = offsets[0]; @@ -639,7 +570,7 @@ fn cast_list( .values() .slice(start.as_usize(), (end - start).as_usize()), }; - let values = cast_values(&values, field.data_type(), mode, preserve_binary_layout)?; + let values = cast_values(&values, field.data_type(), mode)?; let offsets = OffsetBuffer::new(offsets.into()); Ok(Arc::new( GenericListArray::::try_new(field.clone(), offsets, values, input.nulls().cloned()) @@ -2084,60 +2015,26 @@ mod tests { } #[test] - fn preserved_binary_layouts_allow_text_numeric_and_null_update_casts() { + fn row_update_casts_allow_text_numeric_and_null_binary_values() { let text: ArrayRef = Arc::new(StringArray::from(vec![Some("42"), None])); - let binary = cast_row_update_value(&text, &DataType::Binary, true).unwrap(); - let integers = cast_row_update_value(&binary, &DataType::Int32, true).unwrap(); + let binary = cast_update_value(&text, &DataType::Binary, CastMode::RowUpdate).unwrap(); + let integers = cast_update_value(&binary, &DataType::Int32, CastMode::RowUpdate).unwrap(); assert_eq!( integers.to_data(), Int32Array::from(vec![Some(42), None]).to_data() ); assert_eq!( - cast_row_update_value(&new_null_array(&DataType::Null, 2), &DataType::Binary, true) - .unwrap() - .null_count(), + cast_update_value( + &new_null_array(&DataType::Null, 2), + &DataType::Binary, + CastMode::RowUpdate + ) + .unwrap() + .null_count(), 2 ); } - #[test] - fn preserved_binary_layouts_reject_width_changes_before_null_fallbacks() { - use arrow_array::{types::Int8Type, DictionaryArray, Int8Array}; - for (source, target) in [ - (DataType::Binary, DataType::LargeBinary), - (DataType::LargeBinary, DataType::Binary), - ] { - let array = new_null_array(&source, 1); - assert!(cast_row_update_value(&array, &target, true).is_err()); - // Other row-ID/MERGE callers keep their established cast contract. - assert!(cast_update_value(&array, &target, CastMode::RowUpdate).is_ok()); - let dictionary: ArrayRef = Arc::new( - DictionaryArray::::try_new(Int8Array::from(vec![Some(0)]), array.clone()) - .unwrap(), - ); - assert!(cast_row_update_value(&dictionary, &target, true).is_err()); - let field = Arc::new(arrow_schema::Field::new("data", source, true)); - let nested: ArrayRef = Arc::new( - StructArray::try_new( - vec![field].into(), - vec![array], - Some(arrow_buffer::NullBuffer::from(vec![false])), - ) - .unwrap(), - ); - let expected = - DataType::Struct(vec![arrow_schema::Field::new("data", target, true)].into()); - // A NULL container reconstructs no child payload and follows the - // existing whole-value constructor behavior. - assert_eq!( - cast_row_update_value(&nested, &expected, true) - .unwrap() - .null_count(), - 1 - ); - } - } - #[test] fn binary_assignments_parse_text_and_never_reinterpret_numeric_bytes() { let binary: ArrayRef = Arc::new(arrow_array::BinaryArray::from(vec![ diff --git a/crates/paimon/src/table/write_batch_normalize.rs b/crates/paimon/src/table/write_batch_normalize.rs index a3b0cc8c2..f2ef58b61 100644 --- a/crates/paimon/src/table/write_batch_normalize.rs +++ b/crates/paimon/src/table/write_batch_normalize.rs @@ -239,7 +239,11 @@ fn normalize_blob_update_key(array: &ArrayRef, expected: &DataType) -> Result Date: Sat, 17 Oct 2026 21:06:28 +0800 Subject: [PATCH 5/5] Share logical key normalization across update operations --- crates/paimon/src/table/key_normalization.rs | 44 +++++++------------ crates/paimon/src/table/table_upsert.rs | 2 +- .../paimon/src/table/write_batch_normalize.rs | 2 +- 3 files changed, 18 insertions(+), 30 deletions(-) diff --git a/crates/paimon/src/table/key_normalization.rs b/crates/paimon/src/table/key_normalization.rs index 863007c88..3ea04dc59 100644 --- a/crates/paimon/src/table/key_normalization.rs +++ b/crates/paimon/src/table/key_normalization.rs @@ -23,8 +23,10 @@ use arrow_schema::DataType; use super::update_input::{cast_update_value, decode_dictionary, CastMode}; use super::write_batch_normalize::normalize_write_array; -/// Dictionary indices are local to each batch, never row identity. NULL keys -/// retain their validity bitmap; the operation decides whether NULL matches. +/// Unify Arrow representations without changing key identity. Dictionary +/// indices are batch-local; integer widths, decimal scales and temporal units +/// may differ only when conversion is exact. Callers validate the logical type +/// families and decide whether NULL keys match. Payload write rules are separate. pub(super) fn normalize_match_key( array: &ArrayRef, expected: &DataType, @@ -33,18 +35,6 @@ pub(super) fn normalize_match_key( if decoded.data_type() == &DataType::Null { return Ok(new_null_array(expected, decoded.len())); } - normalize_write_array(&decoded, expected) -} - -/// Upsert matches logical values before deciding update versus append. Integer -/// widths, decimal scales and units within a temporal family can differ, but -/// conversion must be exact so distinct source keys never collapse. Payloads, -/// including the original key columns, keep their own action's write contract. -pub(super) fn normalize_upsert_key( - array: &ArrayRef, - expected: &DataType, -) -> crate::Result { - let decoded = decode_dictionary(array)?; let source = decoded.data_type(); if source.is_decimal() && matches!(expected, DataType::Decimal128(..)) { let source_scale = match source { @@ -91,7 +81,7 @@ pub(super) fn normalize_upsert_key( if exact_family { cast_update_value(&decoded, expected, CastMode::Assignment) } else { - normalize_match_key(&decoded, expected) + normalize_write_array(&decoded, expected) } } @@ -129,15 +119,15 @@ mod tests { None, Some(i32::MAX as i64), ])); - let output = normalize_upsert_key(&input, &DataType::Int32).unwrap(); + let output = normalize_match_key(&input, &DataType::Int32).unwrap(); assert_eq!( output.as_any().downcast_ref::().unwrap(), &Int32Array::from(vec![Some(-1), None, Some(i32::MAX)]) ); let overflow: ArrayRef = Arc::new(Int64Array::from(vec![i32::MAX as i64 + 1])); - assert!(normalize_upsert_key(&overflow, &DataType::Int32).is_err()); + assert!(normalize_match_key(&overflow, &DataType::Int32).is_err()); let text: ArrayRef = Arc::new(StringArray::from(vec!["42"])); - assert!(normalize_upsert_key(&text, &DataType::Int32).is_err()); + assert!(normalize_match_key(&text, &DataType::Int32).is_err()); } #[test] @@ -150,7 +140,7 @@ mod tests { ) as ArrayRef }; let expected = DataType::Decimal128(10, 2); - let output = normalize_upsert_key(&array(12500), &expected).unwrap(); + let output = normalize_match_key(&array(12500), &expected).unwrap(); assert_eq!( output .as_any() @@ -159,7 +149,7 @@ mod tests { .value(0), 125 ); - assert!(normalize_upsert_key(&array(12501), &expected).is_err()); + assert!(normalize_match_key(&array(12501), &expected).is_err()); } #[test] @@ -171,7 +161,7 @@ mod tests { .unwrap(), ); let expected = DataType::Decimal128(38, to); - let output = normalize_upsert_key(&input, &expected).unwrap(); + let output = normalize_match_key(&input, &expected).unwrap(); assert_eq!(output.data_type(), &expected); assert_eq!( output.as_any().downcast_ref::().unwrap(), @@ -184,7 +174,7 @@ mod tests { .with_precision_and_scale(38, from) .unwrap(), ); - assert!(normalize_upsert_key(&nonzero, &expected).is_err()); + assert!(normalize_match_key(&nonzero, &expected).is_err()); } } @@ -196,8 +186,7 @@ mod tests { Some(2000), ])); let output = - normalize_upsert_key(&input, &DataType::Timestamp(TimeUnit::Millisecond, None)) - .unwrap(); + normalize_match_key(&input, &DataType::Timestamp(TimeUnit::Millisecond, None)).unwrap(); assert_eq!( output .as_any() @@ -206,14 +195,14 @@ mod tests { &TimestampMillisecondArray::from(vec![Some(-1), None, Some(2)]) ); let fractional: ArrayRef = Arc::new(TimestampMicrosecondArray::from(vec![-1001])); - assert!(normalize_upsert_key( + assert!(normalize_match_key( &fractional, &DataType::Timestamp(TimeUnit::Millisecond, None) ) .is_err()); let local: ArrayRef = Arc::new(TimestampMicrosecondArray::from(vec![0]).with_timezone("America/New_York")); - let output = normalize_upsert_key( + let output = normalize_match_key( &local, &DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())), ) @@ -227,8 +216,7 @@ mod tests { 0 ); assert!( - normalize_upsert_key(&local, &DataType::Timestamp(TimeUnit::Microsecond, None)) - .is_err() + normalize_match_key(&local, &DataType::Timestamp(TimeUnit::Microsecond, None)).is_err() ); } } diff --git a/crates/paimon/src/table/table_upsert.rs b/crates/paimon/src/table/table_upsert.rs index 779573b31..4b51ff66b 100644 --- a/crates/paimon/src/table/table_upsert.rs +++ b/crates/paimon/src/table/table_upsert.rs @@ -270,7 +270,7 @@ impl TableUpsert { .fields() .iter() .map(|field| { - super::key_normalization::normalize_upsert_key( + super::key_normalization::normalize_match_key( batch.column_by_name(field.name()).unwrap(), field.data_type(), ) diff --git a/crates/paimon/src/table/write_batch_normalize.rs b/crates/paimon/src/table/write_batch_normalize.rs index f2ef58b61..44b689b0b 100644 --- a/crates/paimon/src/table/write_batch_normalize.rs +++ b/crates/paimon/src/table/write_batch_normalize.rs @@ -245,7 +245,7 @@ fn normalize_blob_update_key(array: &ArrayRef, expected: &DataType) -> Result