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 adc8717d [core] Extend Arrow input support for row-level updates
(#1149)
adc8717d is described below
commit adc8717d8e413c8f4a11c5f79a5265c365cbe769
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 17 22:59:39 2026 +0800
[core] Extend Arrow input support for row-level updates (#1149)
---
bindings/python/src/arrow_input.rs | 350 ++++++++++++
bindings/python/src/lib.rs | 1 +
bindings/python/src/update_assignment.rs | 10 +-
bindings/python/src/write.rs | 4 +-
crates/paimon/src/arrow/evolution_cast.rs | 77 +--
crates/paimon/src/arrow/mod.rs | 2 +-
crates/paimon/src/table/key_normalization.rs | 28 -
crates/paimon/src/table/update_input.rs | 199 +++++--
.../src/table/update_input/container_tests.rs | 240 +++++++++
crates/paimon/src/table/update_input/decimal.rs | 239 ++++++++
crates/paimon/src/table/update_input/encoded.rs | 599 +++++++++++++++++++++
.../paimon/src/table/update_input/scalar_tests.rs | 207 +++++++
12 files changed, 1832 insertions(+), 124 deletions(-)
diff --git a/bindings/python/src/arrow_input.rs
b/bindings/python/src/arrow_input.rs
new file mode 100644
index 00000000..18cb5427
--- /dev/null
+++ b/bindings/python/src/arrow_input.rs
@@ -0,0 +1,350 @@
+// 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.
+
+//! Preserve C Data offsets before Arrow Rust materializes sparse Union arrays.
+
+use std::sync::Arc;
+
+use arrow::array::{make_array, ArrayData, ArrayRef, RecordBatch,
RecordBatchOptions};
+use arrow::datatypes::{DataType, Schema, UnionMode};
+use arrow::error::ArrowError;
+use arrow::pyarrow::FromPyArrow;
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::*;
+
+fn contains_sparse_union(data_type: &DataType) -> bool {
+ match data_type {
+ DataType::Union(_, UnionMode::Sparse) => true,
+ DataType::Union(fields, _) => fields
+ .iter()
+ .any(|(_, f)| contains_sparse_union(f.data_type())),
+ DataType::Struct(fields) => fields.iter().any(|f|
contains_sparse_union(f.data_type())),
+ DataType::List(f)
+ | DataType::LargeList(f)
+ | DataType::FixedSizeList(f, _)
+ | DataType::ListView(f)
+ | DataType::LargeListView(f)
+ | DataType::Map(f, _) => contains_sparse_union(f.data_type()),
+ DataType::Dictionary(_, values) => contains_sparse_union(values),
+ DataType::RunEndEncoded(_, values) =>
contains_sparse_union(values.data_type()),
+ _ => false,
+ }
+}
+
+pub(crate) fn array_from_pyarrow(value: &Bound<'_, PyAny>) ->
PyResult<ArrayRef> {
+ let mut data = ArrayData::from_pyarrow_bound(value)?;
+ // Foreign buffers may be unaligned; normalize only after making them
+ // valid for Arrow Rust's typed access and ArrayData builder validation.
+ data.align_buffers();
+ let data = normalize_sparse_unions(data).map_err(|e|
PyValueError::new_err(e.to_string()))?;
+ Ok(make_array(data))
+}
+
+pub(crate) fn batch_from_pyarrow(value: &Bound<'_, PyAny>) ->
PyResult<RecordBatch> {
+ let schema =
Arc::new(Schema::from_pyarrow_bound(&value.getattr("schema")?)?);
+ if !schema
+ .fields()
+ .iter()
+ .any(|f| contains_sparse_union(f.data_type()))
+ {
+ return RecordBatch::from_pyarrow_bound(value);
+ }
+ // The standard batch importer materializes its children before we can fix
+ // Union offsets. Import each column's C Data representation first instead.
+ let columns = value
+ .getattr("columns")?
+ .try_iter()?
+ .map(|column| array_from_pyarrow(&column?))
+ .collect::<PyResult<Vec<_>>>()?;
+ let rows = value.getattr("num_rows")?.extract::<usize>()?;
+ RecordBatch::try_new_with_options(
+ schema,
+ columns,
+ &RecordBatchOptions::default().with_row_count(Some(rows)),
+ )
+ .map_err(|e| PyValueError::new_err(e.to_string()))
+}
+
+fn normalize_sparse_unions(data: ArrayData) -> Result<ArrayData, ArrowError> {
+ if !contains_sparse_union(data.data_type()) {
+ return Ok(data);
+ }
+ normalize_imported_data(data)
+}
+
+fn normalize_imported_data(data: ArrayData) -> Result<ArrayData, ArrowError> {
+ let len = data.len();
+ normalize_imported_slice(data, 0, len)
+}
+
+fn normalize_imported_slice(
+ data: ArrayData,
+ relative_offset: usize,
+ len: usize,
+) -> Result<ArrayData, ArrowError> {
+ // Once a parent is sliced, normalize every descendant, including ordinary
+ // STRUCT siblings and Union children. Arrow's STRUCT materializer uses
+ // ArrayData::slice, which can otherwise apply nested offsets twice.
+ let (data_type, _, nulls, offset, mut buffers, children) =
data.into_parts();
+ let nulls = nulls.map(|n| n.slice(relative_offset, len));
+ let mut offset = offset + relative_offset;
+ // Defer validation until both parent and children share their coordinates.
+ // An intermediate ArrayData slice would validate a sliced parent's NULLs
+ // against unsliced non-nullable children, or propagate STRUCT offsets
twice.
+ let children = match &data_type {
+ DataType::Struct(_) | DataType::Union(_, UnionMode::Sparse) => {
+ if matches!(data_type, DataType::Union(_, UnionMode::Sparse)) {
+ // UnionArray::from(ArrayData) applies the parent offset to
type
+ // IDs but not sparse children. Canonicalize both before that.
+ buffers[0] = buffers[0].slice_with_length(offset, len);
+ }
+ let children = children
+ .into_iter()
+ .map(|c| normalize_imported_slice(c, offset, len))
+ .collect::<Result<Vec<_>, _>>()?;
+ offset = 0;
+ children
+ }
+ DataType::FixedSizeList(_, width) => {
+ let width = *width as usize;
+ let children = children
+ .into_iter()
+ .map(|c| normalize_imported_slice(c, offset * width, len *
width))
+ .collect::<Result<Vec<_>, _>>()?;
+ offset = 0;
+ children
+ }
+ _ => children
+ .into_iter()
+ .map(normalize_imported_data)
+ .collect::<Result<Vec<_>, _>>()?,
+ };
+ ArrayData::builder(data_type)
+ .len(len)
+ .offset(offset)
+ .nulls(nulls)
+ .buffers(buffers)
+ .child_data(children)
+ .build()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::{
+ Array, FixedSizeListArray, Int32Array, StringArray, StructArray,
UnionArray,
+ };
+ use arrow::buffer::{NullBuffer, ScalarBuffer};
+ use arrow::datatypes::{Field, UnionFields};
+
+ fn input() -> ArrayData {
+ UnionArray::try_new(
+ UnionFields::try_new(
+ [7, 42],
+ [
+ Field::new("n", DataType::Int32, true),
+ Field::new("s", DataType::Utf8, true),
+ ],
+ )
+ .unwrap(),
+ ScalarBuffer::from(vec![42, 7, 7, 42]),
+ None,
+ vec![
+ Arc::new(Int32Array::from(vec![0, 1, 2, 0])),
+ Arc::new(StringArray::from(vec!["hidden"; 4])),
+ ],
+ )
+ .unwrap()
+ .to_data()
+ }
+
+ fn assert_values(data: ArrayData) {
+ let array = UnionArray::from(data);
+ assert_eq!(array.type_ids().as_ref(), &[7, 7]);
+ assert_eq!(
+ array.child(7).to_data(),
+ Int32Array::from(vec![1, 2]).to_data()
+ );
+ }
+
+ #[test]
+ fn c_data_sparse_union_slice_rebases_children_and_type_ids() {
+ let raw = input().into_builder().offset(1).len(2).build().unwrap();
+ assert_values(normalize_sparse_unions(raw).unwrap());
+ }
+
+ #[test]
+ fn c_data_parent_offsets_are_applied_before_union_materialization() {
+ let child = input();
+ let rows = ArrayData::builder(DataType::Struct(
+ vec![Field::new("u", child.data_type().clone(), true)].into(),
+ ))
+ .offset(1)
+ .len(2)
+ .child_data(vec![child.clone()])
+ .build()
+ .unwrap();
+ let rows = StructArray::from(normalize_sparse_unions(rows).unwrap());
+ assert_values(rows.column(0).to_data());
+ // Nested STRUCT children are still in the original C Data coordinate
+ // system; the outer slice must not be propagated into them twice.
+ let nested = ArrayData::builder(DataType::Struct(
+ vec![Field::new("u", child.data_type().clone(), true)].into(),
+ ))
+ .len(4)
+ .child_data(vec![child.clone()])
+ .build()
+ .unwrap();
+ let outer = ArrayData::builder(DataType::Struct(
+ vec![Field::new("row", nested.data_type().clone(), true)].into(),
+ ))
+ .offset(1)
+ .len(2)
+ .child_data(vec![nested])
+ .build()
+ .unwrap();
+ let outer = StructArray::from(normalize_sparse_unions(outer).unwrap());
+ let nested = outer
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_values(nested.column(0).to_data());
+ let lists = ArrayData::builder(DataType::FixedSizeList(
+ Arc::new(Field::new("item", child.data_type().clone(), true)),
+ 1,
+ ))
+ .offset(1)
+ .len(2)
+ .child_data(vec![child])
+ .build()
+ .unwrap();
+ let lists =
FixedSizeListArray::from(normalize_sparse_unions(lists).unwrap());
+ assert_values(lists.values().to_data());
+ }
+
+ #[test]
+ fn c_data_slices_normalize_ordinary_nested_union_children_and_siblings() {
+ let leaf = Int32Array::from(vec![0, 1, 2, 3]).to_data();
+ let inner = ArrayData::builder(DataType::Struct(
+ vec![Field::new("b", DataType::Int32, true)].into(),
+ ))
+ .len(4)
+ .child_data(vec![leaf])
+ .build()
+ .unwrap();
+ let ordinary = ArrayData::builder(DataType::Struct(
+ vec![Field::new("a", inner.data_type().clone(), true)].into(),
+ ))
+ .len(4)
+ .child_data(vec![inner])
+ .build()
+ .unwrap();
+ let sibling = ArrayData::builder(DataType::Struct(
+ vec![
+ Field::new("u", input().data_type().clone(), true),
+ Field::new("ordinary", ordinary.data_type().clone(), true),
+ ]
+ .into(),
+ ))
+ .len(2)
+ .offset(1)
+ .child_data(vec![input(), ordinary.clone()])
+ .build()
+ .unwrap();
+ let sibling =
StructArray::from(normalize_sparse_unions(sibling).unwrap());
+ let expected = Int32Array::from(vec![1, 2]).to_data();
+ let assert_ordinary = |data: ArrayData| {
+ let outer = StructArray::from(data);
+ let inner = outer
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(inner.column(0).to_data(), expected);
+ };
+ assert_ordinary(sibling.column(1).to_data());
+ let union = ArrayData::builder(DataType::Union(
+ UnionFields::try_new(
+ [7],
+ [Field::new("ordinary", ordinary.data_type().clone(), true)],
+ )
+ .unwrap(),
+ UnionMode::Sparse,
+ ))
+ .len(2)
+ .offset(1)
+ .buffers(vec![ScalarBuffer::from(vec![7_i8; 4]).into_inner()])
+ .child_data(vec![ordinary])
+ .build()
+ .unwrap();
+ let union = UnionArray::from(normalize_sparse_unions(union).unwrap());
+ assert_ordinary(union.child(7).to_data());
+ }
+
+ #[test]
+ fn c_data_nullability_is_validated_after_parent_and_children_are_sliced() {
+ let numbers = Int32Array::from(vec![None, Some(1), Some(2), None]);
+ let parents = NullBuffer::from(vec![false, true, true, false]);
+ let rows = ArrayData::builder(DataType::Struct(
+ vec![Field::new("n", DataType::Int32, false)].into(),
+ ))
+ .len(4)
+ .nulls(Some(parents.clone()))
+ .child_data(vec![numbers.to_data()])
+ .build()
+ .unwrap();
+ let lists = ArrayData::builder(DataType::FixedSizeList(
+ Arc::new(Field::new("item", DataType::Int32, false)),
+ 1,
+ ))
+ .len(4)
+ .nulls(Some(parents))
+ .child_data(vec![numbers.to_data()])
+ .build()
+ .unwrap();
+ for ordinary in [rows, lists] {
+ // Both children are initially valid in their complete C Data
+ // coordinates. The nested parent's NULL masks must move together.
+ let input = ArrayData::builder(DataType::Struct(
+ vec![
+ Field::new("u", input().data_type().clone(), true),
+ Field::new("ordinary", ordinary.data_type().clone(), true),
+ ]
+ .into(),
+ ))
+ .len(2)
+ .offset(1)
+ .child_data(vec![input(), ordinary])
+ .build()
+ .unwrap();
+ let result =
StructArray::from(normalize_sparse_unions(input).unwrap());
+ let ordinary = result.column(1);
+ assert_eq!(ordinary.null_count(), 0);
+ let values = if let Some(row) =
ordinary.as_any().downcast_ref::<StructArray>() {
+ row.column(0)
+ } else {
+ ordinary
+ .as_any()
+ .downcast_ref::<FixedSizeListArray>()
+ .unwrap()
+ .values()
+ };
+ assert_eq!(values.to_data(), Int32Array::from(vec![1,
2]).to_data());
+ }
+ }
+}
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 72d17d42..d579ee28 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -17,6 +17,7 @@
use pyo3::prelude::*;
+mod arrow_input;
mod blob;
mod blob_uri_reader;
mod catalog_version;
diff --git a/bindings/python/src/update_assignment.rs
b/bindings/python/src/update_assignment.rs
index d0cc3134..290d9560 100644
--- a/bindings/python/src/update_assignment.rs
+++ b/bindings/python/src/update_assignment.rs
@@ -17,8 +17,8 @@
//! Python object and callback conversion for core update assignments.
-use arrow::array::{make_array, ArrayData, ArrayRef};
-use arrow::pyarrow::{FromPyArrow, ToPyArrow};
+use arrow::array::ArrayRef;
+use arrow::pyarrow::ToPyArrow;
use paimon::spec::DataField;
use paimon::table::UpdateAssignment;
use pyo3::exceptions::PyValueError;
@@ -41,12 +41,12 @@ fn bridge_error(error: PyErr, target: &CallbackError) ->
paimon::Error {
fn arrow_arrays(py: Python<'_>, value: &Bound<'_, PyAny>) ->
PyResult<Vec<ArrayRef>> {
let pa = py.import("pyarrow")?;
if value.is_instance(&pa.getattr("Array")?)? {
- Ok(vec![make_array(ArrayData::from_pyarrow_bound(value)?)])
+ Ok(vec![crate::arrow_input::array_from_pyarrow(value)?])
} else if value.is_instance(&pa.getattr("ChunkedArray")?)? {
value
.getattr("chunks")?
.try_iter()?
- .map(|chunk|
Ok(make_array(ArrayData::from_pyarrow_bound(&chunk?)?)))
+ .map(|chunk| crate::arrow_input::array_from_pyarrow(&chunk?))
.collect()
} else {
Err(PyValueError::new_err(
@@ -121,7 +121,7 @@ pub(crate) fn from_python(
(PyList::new(py, [scalar])?,),
Some(&kwargs),
)?;
-
Ok(make_array(ArrayData::from_pyarrow_bound(&one)?))
+ crate::arrow_input::array_from_pyarrow(&one)
})
})
.map_err(|cause| bridge_error(cause, &error))
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 312990d0..f9a1716a 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -540,7 +540,7 @@ fn arrow_table_batches(table: &Bound<'_, PyAny>) ->
PyResult<Vec<RecordBatch>> {
let mut batches = table
.call_method0("to_batches")?
.try_iter()?
- .map(|batch| RecordBatch::from_pyarrow_bound(&batch?))
+ .map(|batch| crate::arrow_input::batch_from_pyarrow(&batch?))
.collect::<PyResult<Vec<_>>>()?;
// Preserve empty input schema so core validates columns even with no rows.
if batches.is_empty() {
@@ -559,7 +559,7 @@ fn arrow_upsert_batches(input: &Bound<'_, PyAny>) ->
PyResult<Vec<RecordBatch>>
}
input
.try_iter()?
- .map(|batch| RecordBatch::from_pyarrow_bound(&batch?))
+ .map(|batch| crate::arrow_input::batch_from_pyarrow(&batch?))
.collect()
}
diff --git a/crates/paimon/src/arrow/evolution_cast.rs
b/crates/paimon/src/arrow/evolution_cast.rs
index 5e85e110..991932f0 100644
--- a/crates/paimon/src/arrow/evolution_cast.rs
+++ b/crates/paimon/src/arrow/evolution_cast.rs
@@ -131,45 +131,56 @@ pub(crate) fn cast_numeric(
let strings = super::evolution_decimal::decimal_to_string(source)?;
return Ok(Some(super::apply_declared_width(&strings, to)));
}
- // DecimalUtils.doubleValue divides the compact long by its scale;
- // only wide decimals use BigDecimal.doubleValue. FLOAT rounds that
- // double rather than the decimal coefficient directly.
- let values: Vec<Option<f64>> = if decimal.precision() <= 18 {
- let divisor = 10_f64.powi(decimal.scale() as i32);
- source
- .as_primitive::<Decimal128Type>()
- .iter()
- .map(|value| value.map(|value| value as f64 / divisor))
- .collect()
- } else {
- let strings = super::evolution_decimal::decimal_to_string(source)?;
- strings
- .as_string::<i32>()
- .iter()
- .map(|value| {
- value
- .map(|value| {
- value
- .parse::<f64>()
- .map_err(|_| invalid_numeric("Invalid DECIMAL
value"))
- })
- .transpose()
- })
- .collect::<crate::Result<Vec<_>>>()?
- };
- return Ok(Some(match target {
- ArrowType::Float32 => Arc::new(Float32Array::from_iter(
- values.into_iter().map(|v| v.map(|v| v as f32)),
- )) as ArrayRef,
- ArrowType::Float64 => Arc::new(Float64Array::from(values)) as
ArrayRef,
- _ => unreachable!("validated DECIMAL numeric target"),
- }));
+ return cast_decimal_to_floating(source, decimal.precision(),
decimal.scale(), &target)
+ .map(Some);
}
arrow_cast::cast(source, &target)
.map(Some)
.map_err(|error| invalid_numeric(error.to_string()))
}
+/// Java DECIMAL conversion shared by schema reads and external update inputs.
+pub(crate) fn cast_decimal_to_floating(
+ source: &ArrayRef,
+ precision: u32,
+ scale: u32,
+ target: &ArrowType,
+) -> crate::Result<ArrayRef> {
+ // DecimalUtils.doubleValue divides the compact long by its scale;
+ // only wide decimals use BigDecimal.doubleValue. FLOAT rounds that
+ // double rather than the decimal coefficient directly.
+ let values: Vec<Option<f64>> = if precision <= 18 {
+ let divisor = 10_f64.powi(scale as i32);
+ source
+ .as_primitive::<Decimal128Type>()
+ .iter()
+ .map(|value| value.map(|value| value as f64 / divisor))
+ .collect()
+ } else {
+ let strings = super::evolution_decimal::decimal_to_string(source)?;
+ strings
+ .as_string::<i32>()
+ .iter()
+ .map(|value| {
+ value
+ .map(|value| {
+ value
+ .parse::<f64>()
+ .map_err(|_| invalid_numeric("Invalid DECIMAL
value"))
+ })
+ .transpose()
+ })
+ .collect::<crate::Result<Vec<_>>>()?
+ };
+ Ok(match target {
+ ArrowType::Float32 => Arc::new(Float32Array::from_iter(
+ values.into_iter().map(|v| v.map(|v| v as f32)),
+ )) as ArrayRef,
+ ArrowType::Float64 => Arc::new(Float64Array::from(values)) as ArrayRef,
+ _ => unreachable!("validated DECIMAL numeric target"),
+ })
+}
+
fn invalid_numeric(message: impl Into<String>) -> crate::Error {
crate::Error::DataInvalid {
message: message.into(),
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index 7a057c0a..fddf9f19 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -26,7 +26,7 @@ mod cast_completion_java_test;
mod cast_text;
pub(crate) use cast_text::parse_timestamp as parse_java_timestamp;
mod evolution_cast;
-pub(crate) use evolution_cast::{cast_numeric, supports_numeric_cast};
+pub(crate) use evolution_cast::{cast_decimal_to_floating, cast_numeric,
supports_numeric_cast};
mod evolution_decimal;
mod evolution_string;
mod evolution_temporal;
diff --git a/crates/paimon/src/table/key_normalization.rs
b/crates/paimon/src/table/key_normalization.rs
index 3ea04dc5..50b78dc8 100644
--- a/crates/paimon/src/table/key_normalization.rs
+++ b/crates/paimon/src/table/key_normalization.rs
@@ -36,34 +36,6 @@ pub(super) fn normalize_match_key(
return Ok(new_null_array(expected, decoded.len()));
}
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!(
diff --git a/crates/paimon/src/table/update_input.rs
b/crates/paimon/src/table/update_input.rs
index f6a809db..38f1fccc 100644
--- a/crates/paimon/src/table/update_input.rs
+++ b/crates/paimon/src/table/update_input.rs
@@ -28,6 +28,9 @@ use arrow_array::{
use arrow_buffer::OffsetBuffer;
use arrow_schema::{DataType, Schema, TimeUnit};
+mod decimal;
+mod encoded;
+
/// PyPaimon distinguishes evaluated assignments (safe casts) from row-ID
/// input (numeric coercion). Both use the same encodings, strings and nested
/// traversal; callers select the contract explicitly.
@@ -133,27 +136,37 @@ pub(super) fn cast_update_value(
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)
+ construct_update_value(array, target)
})
} else {
cast_values(array, target, mode)
}
}
+/// The root MAP constructor converts list pairs to tuples. Preserve that
+/// context through encoding wrappers, but not through ROW/ARRAY children.
+fn construct_update_value(array: &ArrayRef, target: &DataType) ->
crate::Result<ArrayRef> {
+ if matches!(array.data_type(), DataType::Dictionary(..)) {
+ return encoded::construct_dictionary(array, target,
construct_update_value);
+ }
+ if matches!(array.data_type(), DataType::Union(..)) {
+ return encoded::construct_union(array, target, construct_update_value);
+ }
+ if let DataType::Map(field, sorted) = target {
+ if matches!(
+ array.data_type(),
+ DataType::List(_)
+ | DataType::LargeList(_)
+ | DataType::ListView(_)
+ | DataType::LargeListView(_)
+ | DataType::FixedSizeList(..)
+ ) {
+ return constructor_map(array, field, *sorted);
+ }
+ }
+ cast_values(array, target, CastMode::Constructor)
+}
+
/// 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<ArrayRef> {
@@ -164,8 +177,18 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode:
CastMode) -> crate::Re
return Ok(array.clone());
}
if matches!(array.data_type(), DataType::Dictionary(..)) {
+ if mode == CastMode::Constructor {
+ return encoded::construct_dictionary(array, target, |array,
target| {
+ cast_values(array, target, CastMode::Constructor)
+ });
+ }
return cast_values(&decode_dictionary(array)?, target, mode);
}
+ if mode == CastMode::Constructor && matches!(array.data_type(),
DataType::Union(..)) {
+ return encoded::construct_union(array, target, |array, target| {
+ cast_values(array, target, CastMode::Constructor)
+ });
+ }
if let DataType::ListView(field) | DataType::LargeListView(field) =
array.data_type() {
// Convert only the offsets/layout first. Casting children here would
// bypass the recursive precision and formatting rules below.
@@ -209,10 +232,7 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode:
CastMode) -> crate::Re
let columns = fields
.iter()
.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)
- }
+ Some(column) => cast_child(column, field.data_type(),
input.nulls(), 1, mode),
None => Ok(new_null_array(field.data_type(), input.len())),
})
.collect::<crate::Result<Vec<_>>>()?;
@@ -242,9 +262,13 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode:
CastMode) -> crate::Re
)?),
DataType::FixedSizeList(_, size) => {
let input =
array.as_any().downcast_ref::<FixedSizeListArray>().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_child(
+ input.values(),
+ field.data_type(),
+ input.nulls(),
+ *size as usize,
+ mode,
+ )?;
Some(Arc::new(
FixedSizeListArray::try_new_with_length(
field.clone(),
@@ -256,35 +280,32 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode:
CastMode) -> crate::Re
.map_err(|error| invalid(error.to_string()))?,
))
}
+ DataType::Map(..) => {
+ let input = array.as_any().downcast_ref::<MapArray>().unwrap();
+ let DataType::Struct(fields) = field.data_type() else {
+ return Err(invalid(
+ "Map entries can only be cast to two-field ROW
elements",
+ ));
+ };
+ let (entries, offsets) = cast_map_entries(input, fields,
mode)?;
+ Some(Arc::new(
+ ListArray::try_new(
+ field.clone(),
+ offsets,
+ Arc::new(entries),
+ input.nulls().cloned(),
+ )
+ .map_err(|error| invalid(error.to_string()))?,
+ ))
+ }
_ => None,
},
(DataType::Map(_, _), DataType::Map(field, sorted)) => {
let input = array.as_any().downcast_ref::<MapArray>().unwrap();
- let offsets = input.value_offsets();
- let start = offsets[0];
- let length = offsets[offsets.len() - 1] - start;
- let (offsets, indices) = constructor_offsets(offsets,
input.nulls(), mode);
- let length = indices.as_ref().map_or(length as usize, |i| i.len());
let DataType::Struct(fields) = field.data_type() else {
return Err(invalid("Map entries must have a struct type"));
};
- // Map fields may use custom names. Key/value roles are positional,
- // unlike ROW fields, so never match these two children by name.
- let columns = [input.keys(), input.values()]
- .into_iter()
- .zip(fields)
- .map(|(values, field)| {
- let values = match &indices {
- Some(indices) =>
arrow_select::take::take(values.as_ref(), indices, None)
- .map_err(|error| invalid(error.to_string()))?,
- None => values.slice(start as usize, length),
- };
- cast_values(&values, field.data_type(), mode)
- })
- .collect::<crate::Result<Vec<_>>>()?;
- let entries = StructArray::try_new_with_length(fields.clone(),
columns, None, length)
- .map_err(|error| invalid(error.to_string()))?;
- let offsets = OffsetBuffer::new(offsets.into());
+ let (entries, offsets) = cast_map_entries(input, fields, mode)?;
Some(Arc::new(
MapArray::try_new(
field.clone(),
@@ -323,6 +344,41 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode:
CastMode) -> crate::Re
cast_primitive(array, target, mode)
}
+/// Arrow represents a MAP as a list of key/value ROWs. Adapt these children
+/// positionally for both MAP and ARRAY<ROW> inputs, retaining recursive value
+/// conversion and the whole-column constructor's visibility rules.
+fn cast_map_entries(
+ input: &MapArray,
+ fields: &arrow_schema::Fields,
+ mode: CastMode,
+) -> crate::Result<(StructArray, OffsetBuffer<i32>)> {
+ if fields.len() != 2 {
+ return Err(invalid("Map entries require exactly two ROW fields"));
+ }
+ let offsets = input.value_offsets();
+ let start = offsets[0];
+ let length = offsets[offsets.len() - 1] - start;
+ let (offsets, indices) = constructor_offsets(offsets, input.nulls(), mode);
+ let length = indices.as_ref().map_or(length as usize, |i| i.len());
+ // MAP field names are aliases. Their key/value roles do not depend on the
+ // source names or the target ROW names, unlike a normal ROW-to-ROW cast.
+ let columns = [input.keys(), input.values()]
+ .into_iter()
+ .zip(fields)
+ .map(|(values, field)| {
+ let values = match &indices {
+ Some(indices) => arrow_select::take::take(values.as_ref(),
indices, None)
+ .map_err(|error| invalid(error.to_string()))?,
+ None => values.slice(start as usize, length),
+ };
+ cast_values(&values, field.data_type(), mode)
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ let entries = StructArray::try_new_with_length(fields.clone(), columns,
None, length)
+ .map_err(|error| invalid(error.to_string()))?;
+ Ok((entries, OffsetBuffer::new(offsets.into())))
+}
+
// PyPaimon's row-ID fallback reconstructs Python dicts only after a
whole-column
// safe cast fails. At the top level it rejects ambiguous schema-less null
keys.
fn validate_map_constructor_input(array: &ArrayRef, target: &DataType) ->
crate::Result<()> {
@@ -406,13 +462,12 @@ fn constructor_struct_from_map(
.collect::<crate::Result<Vec<_>>>()?;
// Missing fields, NULL parents and trailing entries do not
participate
// in conversion, matching PyArrow's ordered pair constructor for
ROW.
- let values = arrow_select::take::take(
- input.values().as_ref(),
+ encoded::construct_selected(
+ input.values(),
&arrow_array::UInt64Array::from(indices),
- None,
+ field.data_type(),
+ |array, target| cast_values(array, target,
CastMode::Constructor),
)
- .map_err(|error| invalid(error.to_string()))?;
- cast_values(&values, field.data_type(), CastMode::Constructor)
})
.collect::<crate::Result<Vec<_>>>()?;
Ok(Arc::new(
@@ -451,8 +506,13 @@ fn constructor_map(
.columns()
.iter()
.map(|column| {
- let visible = constructor_child(column, input.nulls(), 1,
CastMode::Constructor)?;
- cast_values(&visible, fields[1].data_type(),
CastMode::Constructor)
+ cast_child(
+ column,
+ fields[1].data_type(),
+ input.nulls(),
+ 1,
+ CastMode::Constructor,
+ )
})
.collect::<crate::Result<Vec<_>>>()?;
let values = if columns.is_empty() {
@@ -580,21 +640,23 @@ fn cast_list<O: OffsetSizeTrait>(
// Rebuilding Python values discards children hidden by a null parent. Safe
// casts deliberately retain them, since Arrow C++ validates their values too.
-fn constructor_child(
+fn cast_child(
array: &ArrayRef,
+ target: &DataType,
nulls: Option<&arrow_buffer::NullBuffer>,
width: usize,
mode: CastMode,
) -> crate::Result<ArrayRef> {
if mode != CastMode::Constructor || nulls.is_none_or(|n| n.null_count() ==
0) {
- return Ok(array.clone());
+ return cast_values(array, target, mode);
}
let nulls = nulls.unwrap();
let indices = arrow_array::UInt64Array::from_iter(
(0..array.len()).map(|i| nulls.is_valid(i / width).then_some(i as
u64)),
);
- arrow_select::take::take(array.as_ref(), &indices, None)
- .map_err(|error| invalid(error.to_string()))
+ encoded::construct_selected(array, &indices, target, |array, target| {
+ cast_values(array, target, CastMode::Constructor)
+ })
}
fn constructor_offsets<O: OffsetSizeTrait>(
@@ -706,7 +768,22 @@ fn cast_to_string(array: &ArrayRef, target: &DataType) ->
crate::Result<ArrayRef
safe: false,
..Default::default()
};
+ if matches!(array.data_type(), DataType::FixedSizeBinary(_)) {
+ // Fixed and variable width byte arrays carry the same UTF-8 values.
+ // Arrow Rust lacks the direct fixed-binary to STRING kernel.
+ let binary =
super::write_batch_normalize::normalize_write_array(array, &DataType::Binary)?;
+ return cast_to_string(&binary, target);
+ }
let floats: Option<StringArray> = match array.data_type() {
+ DataType::Float16 => Some(
+ array
+ .as_any()
+ .downcast_ref::<arrow_array::Float16Array>()
+ .unwrap()
+ .iter()
+ .map(|value| value.map(|value| format_float(value.to_f64())))
+ .collect(),
+ ),
DataType::Float32 => Some(
array
.as_any()
@@ -1365,6 +1442,12 @@ fn cast_primitive(array: &ArrayRef, target: &DataType,
mode: CastMode) -> crate:
return cast_float_decimal(array, *precision, *scale);
}
}
+ if source.is_decimal() && matches!(target, DataType::Decimal128(..)) {
+ return decimal::cast_exact_decimal(array, target);
+ }
+ if source.is_decimal() && matches!(target, DataType::Float32 |
DataType::Float64) {
+ return decimal::cast_to_floating(array, target);
+ }
if mode == CastMode::Constructor {
if let Some(converted) = coerce_temporal(array, target)? {
return Ok(converted);
@@ -1467,6 +1550,12 @@ fn cast_primitive(array: &ArrayRef, target: &DataType,
mode: CastMode) -> crate:
Ok(converted)
}
+#[cfg(test)]
+mod container_tests;
+
+#[cfg(test)]
+mod scalar_tests;
+
#[cfg(test)]
mod tests {
use super::*;
diff --git a/crates/paimon/src/table/update_input/container_tests.rs
b/crates/paimon/src/table/update_input/container_tests.rs
new file mode 100644
index 00000000..85dd4186
--- /dev/null
+++ b/crates/paimon/src/table/update_input/container_tests.rs
@@ -0,0 +1,240 @@
+// 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.
+
+use super::*;
+use arrow_array::{types::Int8Type, DictionaryArray, Int32Array, Int8Array};
+use arrow_buffer::NullBuffer;
+use arrow_schema::Field;
+
+fn map(keys: ArrayRef, values: ArrayRef, lengths: &[usize], valid: &[bool]) ->
ArrayRef {
+ let entries = StructArray::from(vec![
+ (
+ Arc::new(Field::new("encoded_key", keys.data_type().clone(),
false)),
+ keys,
+ ),
+ (
+ Arc::new(Field::new(
+ "encoded_value",
+ values.data_type().clone(),
+ true,
+ )),
+ values,
+ ),
+ ]);
+ Arc::new(MapArray::new(
+ Arc::new(Field::new("pairs", entries.data_type().clone(), false)),
+ OffsetBuffer::from_lengths(lengths.iter().copied()),
+ entries,
+ Some(NullBuffer::from(valid.to_vec())),
+ false,
+ ))
+}
+
+fn pair_type(key: DataType, value: DataType) -> DataType {
+ DataType::Struct(
+ vec![
+ Field::new("label", key, true),
+ Field::new("amount", value, true),
+ ]
+ .into(),
+ )
+}
+
+#[test]
+fn map_to_array_rows_preserves_aliases_slices_empty_and_null_values() {
+ let input = map(
+ Arc::new(StringArray::from(vec!["unused", "a", "b", "z"])),
+ Arc::new(Int32Array::from(vec![Some(-1), Some(1), None, Some(4)])),
+ &[1, 2, 0, 0, 1],
+ &[true, true, false, true, true],
+ )
+ .slice(1, 4);
+ let fields = pair_type(DataType::Utf8, DataType::Int64);
+ let item = Arc::new(Field::new("element", fields.clone(), false));
+ for target in [
+ DataType::List(item.clone()),
+ DataType::LargeList(item.clone()),
+ ] {
+ let output = cast_update_value(&input, &target,
CastMode::Assignment).unwrap();
+ assert_eq!(output.data_type(), &target);
+ assert!(output.is_null(1));
+ // Normalize only the container to inspect both offset widths.
+ let output = arrow_cast::cast(output.as_ref(),
&DataType::List(item.clone())).unwrap();
+ let output = output.as_any().downcast_ref::<ListArray>().unwrap();
+ assert_eq!(output.value_offsets(), &[0, 2, 2, 2, 3]);
+ let entries = output
+ .values()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(entries.data_type(), &fields);
+ assert_eq!(
+ entries.column(0).to_data(),
+ StringArray::from(vec!["a", "b", "z"]).to_data()
+ );
+ assert_eq!(
+ entries.column(1).to_data(),
+ Int64Array::from(vec![Some(1), None, Some(4)]).to_data()
+ );
+ }
+}
+
+#[test]
+fn map_to_array_recurses_through_rows_lists_and_dictionaries() {
+ let dictionary: ArrayRef = Arc::new(
+ DictionaryArray::<Int8Type>::try_new(
+ Int8Array::from(vec![0, 1]),
+ Arc::new(StringArray::from(vec!["42", "7"])),
+ )
+ .unwrap(),
+ );
+ let input = map(
+ Arc::new(StringArray::from(vec!["a", "b"])),
+ dictionary,
+ &[1, 1],
+ &[true, true],
+ );
+ let target = DataType::List(Arc::new(Field::new(
+ "item",
+ pair_type(DataType::Utf8, DataType::Int32),
+ true,
+ )));
+ let expected = cast_update_value(&input, &target,
CastMode::Assignment).unwrap();
+ let outer_row: ArrayRef = Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("nested", input.data_type().clone(), true)),
+ input.clone(),
+ )]));
+ let result = cast_update_value(
+ &outer_row,
+ &DataType::Struct(vec![Field::new("nested", target.clone(),
true)].into()),
+ CastMode::Assignment,
+ )
+ .unwrap();
+ assert_eq!(
+ result
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap()
+ .column(0)
+ .to_data(),
+ expected.to_data()
+ );
+ let outer_list: ArrayRef = Arc::new(ListArray::new(
+ Arc::new(Field::new("item", input.data_type().clone(), true)),
+ OffsetBuffer::from_lengths([2]),
+ input.clone(),
+ None,
+ ));
+ let result = cast_update_value(
+ &outer_list,
+ &DataType::List(Arc::new(Field::new("item", target.clone(), true))),
+ CastMode::Assignment,
+ )
+ .unwrap();
+ assert_eq!(
+ result
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap()
+ .values()
+ .to_data(),
+ expected.to_data()
+ );
+ let encoded: ArrayRef =
+ Arc::new(DictionaryArray::<Int8Type>::try_new(Int8Array::from(vec![0,
1]), input).unwrap());
+ assert_eq!(
+ cast_update_value(&encoded, &target, CastMode::Assignment)
+ .unwrap()
+ .to_data(),
+ expected.to_data()
+ );
+}
+
+#[test]
+fn map_to_array_keeps_whole_column_constructor_visibility() {
+ let input = map(
+ Arc::new(StringArray::from(vec!["a", "hidden"])),
+ Arc::new(Int64Array::from(vec![7, i64::MAX])),
+ &[1, 1, 0],
+ &[true, false, true],
+ );
+ let target = DataType::List(Arc::new(Field::new(
+ "item",
+ pair_type(DataType::Utf8, DataType::Int32),
+ true,
+ )));
+ assert!(cast_update_value(&input, &target, CastMode::Assignment).is_err());
+ let output = cast_update_value(&input, &target,
CastMode::RowUpdate).unwrap();
+ let output = output.as_any().downcast_ref::<ListArray>().unwrap();
+ assert_eq!(output.value_offsets(), &[0, 1, 1, 1]);
+ assert!(output.is_null(1));
+ let entries = output
+ .values()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(
+ entries.column(1).to_data(),
+ Int32Array::from(vec![7]).to_data()
+ );
+ let visible = map(
+ Arc::new(StringArray::from(vec!["bad"])),
+ Arc::new(Int64Array::from(vec![i64::MAX])),
+ &[1],
+ &[true],
+ );
+ assert!(cast_update_value(&visible, &target,
CastMode::RowUpdate).is_err());
+}
+
+#[test]
+fn map_entries_require_two_fields_and_actual_child_nullability() {
+ let input = map(
+ Arc::new(StringArray::from(vec!["a"])),
+ Arc::new(Int32Array::from(vec![7])),
+ &[1],
+ &[true],
+ );
+ for size in [0, 1, 3] {
+ let fields = (0..size)
+ .map(|i| Field::new(format!("f{i}"), DataType::Utf8, true))
+ .collect::<Vec<_>>();
+ let target = DataType::List(Arc::new(Field::new(
+ "item",
+ DataType::Struct(fields.into()),
+ true,
+ )));
+ assert!(cast_update_value(&input, &target,
CastMode::Assignment).is_err());
+ }
+ let fields = vec![
+ Field::new("k", DataType::Utf8, false),
+ Field::new("v", DataType::Int32, false),
+ ];
+ let target = DataType::List(Arc::new(Field::new(
+ "item",
+ DataType::Struct(fields.into()),
+ false,
+ )));
+ // A nullable source value FIELD with non-null actual values can be cast.
+ assert!(cast_update_value(&input, &target, CastMode::Assignment).is_ok());
+ let null_value = map(
+ Arc::new(StringArray::from(vec!["a"])),
+ Arc::new(Int32Array::from(vec![None])),
+ &[1],
+ &[true],
+ );
+ assert!(cast_update_value(&null_value, &target,
CastMode::Assignment).is_err());
+}
diff --git a/crates/paimon/src/table/update_input/decimal.rs
b/crates/paimon/src/table/update_input/decimal.rs
new file mode 100644
index 00000000..609d47e7
--- /dev/null
+++ b/crates/paimon/src/table/update_input/decimal.rs
@@ -0,0 +1,239 @@
+// 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.
+
+//! Exact DECIMAL input rescaling shared by updates and logical match keys.
+
+use arrow_array::{cast::AsArray, ArrayRef, Decimal128Array, Float32Array,
Float64Array};
+use arrow_schema::DataType;
+use std::sync::Arc;
+
+use super::{invalid, parse_decimal_text};
+
+fn precision_and_scale(array: &ArrayRef) -> (u8, i8) {
+ match array.data_type() {
+ DataType::Decimal32(p, s)
+ | DataType::Decimal64(p, s)
+ | DataType::Decimal128(p, s)
+ | DataType::Decimal256(p, s) => (*p, *s),
+ _ => unreachable!("validated DECIMAL input"),
+ }
+}
+
+fn map_decimal<T>(
+ array: &ArrayRef,
+ parse: impl Fn(&str) -> crate::Result<T>,
+) -> crate::Result<Vec<Option<T>>> {
+ let (_, scale) = precision_and_scale(array);
+ // Preserve the full coefficient. A generic display can clip digits to the
+ // declared source precision for foreign buffers containing more digits.
+ macro_rules! values {
+ ($ty:ty) => {
+ array
+ .as_primitive::<$ty>()
+ .iter()
+ .map(|value| {
+ value
+ .map(|value| parse(&format!("{value}e{}",
-i16::from(scale))))
+ .transpose()
+ })
+ .collect()
+ };
+ }
+ match array.data_type() {
+ DataType::Decimal32(..) => values!(arrow_array::types::Decimal32Type),
+ DataType::Decimal64(..) => values!(arrow_array::types::Decimal64Type),
+ DataType::Decimal128(..) =>
values!(arrow_array::types::Decimal128Type),
+ DataType::Decimal256(..) =>
values!(arrow_array::types::Decimal256Type),
+ _ => unreachable!("validated DECIMAL input"),
+ }
+}
+
+pub(super) fn cast_to_floating(array: &ArrayRef, target: &DataType) ->
crate::Result<ArrayRef> {
+ let (precision, scale) = precision_and_scale(array);
+ if precision <= 38 && scale >= 0 && scale as u8 <= precision {
+ let canonical = cast_exact_decimal(array,
&DataType::Decimal128(precision, scale))?;
+ return crate::arrow::cast_decimal_to_floating(
+ &canonical,
+ u32::from(precision),
+ scale as u32,
+ target,
+ );
+ }
+ // External Arrow decimals can exceed Paimon's logical precision or use a
+ // negative scale. They still update FLOAT/DOUBLE columns via the exact
+ // decimal value, without first rounding an oversized coefficient to f64.
+ let values = map_decimal(array, |text| {
+ text.parse::<f64>()
+ .map_err(|_| invalid("Invalid decimal floating assignment"))
+ })?;
+ Ok(match target {
+ DataType::Float32 => Arc::new(Float32Array::from_iter(
+ values.into_iter().map(|v| v.map(|v| v as f32)),
+ )) as ArrayRef,
+ DataType::Float64 => Arc::new(Float64Array::from(values)),
+ _ => unreachable!("validated DECIMAL floating target"),
+ })
+}
+
+pub(super) fn cast_exact_decimal(array: &ArrayRef, target: &DataType) ->
crate::Result<ArrayRef> {
+ let (_, source_scale) = precision_and_scale(array);
+ let DataType::Decimal128(target_precision, target_scale) = target else {
+ unreachable!("validated DECIMAL target")
+ };
+ let text_fallback = || {
+ // A zero or NULL can cross scales beyond the rescaler's power table.
+ // The exact text parser still rejects fractional loss and overflow.
+ let values = map_decimal(array, |text| {
+ parse_decimal_text(text, *target_precision, *target_scale)
+ })?;
+ Decimal128Array::from(values)
+ .with_precision_and_scale(*target_precision, *target_scale)
+ .map(|array| Arc::new(array) as ArrayRef)
+ .map_err(|error| invalid(error.to_string()))
+ };
+ let difference = i16::from(*target_scale) - i16::from(source_scale);
+ if !(-38..=38).contains(&difference) {
+ // Arrow subtracts scales in i8 and uses a Decimal128 power table.
+ // Check before entering the kernel, including its reverse cast.
+ return text_fallback();
+ }
+ let options = arrow_cast::CastOptions {
+ safe: false,
+ ..Default::default()
+ };
+ let exact = || {
+ let converted = arrow_cast::cast_with_options(array.as_ref(), target,
&options)
+ .map_err(|error| invalid(error.to_string()))?;
+ let restored =
+ arrow_cast::cast_with_options(converted.as_ref(),
array.data_type(), &options)
+ .map_err(|error| invalid(error.to_string()))?;
+ if restored.to_data() != array.to_data() {
+ return Err(invalid("Decimal assignment would lose scale
precision"));
+ }
+ Ok(converted)
+ };
+ exact().or_else(|_| text_fallback())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::super::{cast_update_value, CastMode};
+ use super::*;
+ use arrow_array::{Array, Decimal128Array, Decimal256Array, Decimal32Array,
Decimal64Array};
+ use arrow_buffer::i256;
+ use std::sync::Arc;
+
+ fn decimal(data_type: &DataType, value: i128) -> ArrayRef {
+ match data_type {
+ DataType::Decimal32(p, s) => Arc::new(
+ Decimal32Array::from(vec![Some(value as i32), None])
+ .with_precision_and_scale(*p, *s)
+ .unwrap(),
+ ),
+ DataType::Decimal64(p, s) => Arc::new(
+ Decimal64Array::from(vec![Some(value as i64), None])
+ .with_precision_and_scale(*p, *s)
+ .unwrap(),
+ ),
+ DataType::Decimal128(p, s) => Arc::new(
+ Decimal128Array::from(vec![Some(value), None])
+ .with_precision_and_scale(*p, *s)
+ .unwrap(),
+ ),
+ DataType::Decimal256(p, s) => Arc::new(
+ Decimal256Array::from(vec![Some(i256::from_i128(value)), None])
+ .with_precision_and_scale(*p, *s)
+ .unwrap(),
+ ),
+ _ => unreachable!(),
+ }
+ }
+
+ #[test]
+ fn zero_and_null_cross_scale_and_width_boundaries_without_overflow() {
+ for (source, target) in [
+ (DataType::Decimal32(9, -128), DataType::Decimal128(38, 38)),
+ (DataType::Decimal64(18, -90), DataType::Decimal128(38, 38)),
+ (DataType::Decimal128(38, -2), DataType::Decimal128(38, 38)),
+ (DataType::Decimal128(38, 38), DataType::Decimal128(20, 0)),
+ (DataType::Decimal256(76, -128), DataType::Decimal128(20, 2)),
+ (DataType::Decimal256(60, -40), DataType::Decimal128(20, 2)),
+ (DataType::Decimal256(76, 76), DataType::Decimal128(38, 38)),
+ ] {
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ let output = cast_update_value(&decimal(&source, 0), &target,
mode).unwrap();
+ assert_eq!(
+ output.to_data(),
+ decimal(&target, 0).to_data(),
+ "{source:?} -> {target:?}"
+ );
+ assert!(
+ cast_update_value(&decimal(&source, 1), &target,
mode).is_err(),
+ "{source:?} -> {target:?}"
+ );
+ }
+ }
+ }
+
+ #[test]
+ fn exact_rescaling_preserves_precision_and_nulls_for_every_input_width() {
+ for source in [
+ DataType::Decimal32(9, 3),
+ DataType::Decimal64(18, 3),
+ DataType::Decimal128(38, 3),
+ DataType::Decimal256(76, 3),
+ ] {
+ let target = DataType::Decimal128(10, 2);
+ let output =
+ cast_update_value(&decimal(&source, -1250), &target,
CastMode::Assignment).unwrap();
+ assert_eq!(output.to_data(), decimal(&target, -125).to_data());
+ assert!(
+ cast_update_value(&decimal(&source, -1251), &target,
CastMode::Assignment).is_err()
+ );
+ assert!(cast_update_value(
+ &decimal(&source, 1250),
+ &DataType::Decimal128(2, 2),
+ CastMode::Assignment
+ )
+ .is_err());
+ }
+ }
+
+ #[test]
+ fn floating_input_keeps_coefficients_wider_than_decimal128() {
+ let coefficient = format!("1508079088755887087797858531882816082{}",
"0".repeat(38))
+ .parse::<i256>()
+ .unwrap();
+ let source: ArrayRef = Arc::new(
+ Decimal256Array::from(vec![Some(coefficient), None,
Some(-coefficient)])
+ .with_precision_and_scale(76, 39)
+ .unwrap(),
+ );
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ let result = cast_update_value(&source, &DataType::Float64,
mode).unwrap();
+ assert_eq!(
+ result.to_data(),
+ Float64Array::from(vec![
+ Some(1.508079088755887e35),
+ None,
+ Some(-1.508079088755887e35),
+ ])
+ .to_data()
+ );
+ }
+ }
+}
diff --git a/crates/paimon/src/table/update_input/encoded.rs
b/crates/paimon/src/table/update_input/encoded.rs
new file mode 100644
index 00000000..4bb99363
--- /dev/null
+++ b/crates/paimon/src/table/update_input/encoded.rs
@@ -0,0 +1,599 @@
+// 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.
+
+//! Rebuild selected encoded values after the whole-column safe cast has
failed.
+
+use arrow_array::cast::AsArray;
+use arrow_array::{Array, ArrayRef, UInt64Array, UnionArray};
+use arrow_schema::DataType;
+use indexmap::IndexMap;
+
+use super::invalid;
+
+/// Decode references before constructor conversion, then restore NULL keys.
+/// Unlike taking a nullable index directly from a Union, taking from the typed
+/// output can represent NULL without assigning it an unrelated active type.
+pub(super) fn construct_dictionary(
+ array: &ArrayRef,
+ target: &DataType,
+ convert: fn(&ArrayRef, &DataType) -> crate::Result<ArrayRef>,
+) -> crate::Result<ArrayRef> {
+ let dictionary = array.as_any_dictionary();
+ let keys = arrow_cast::cast(dictionary.keys(), &DataType::UInt64)
+ .map_err(|error| invalid(error.to_string()))?;
+ let keys = keys.as_any().downcast_ref::<UInt64Array>().unwrap();
+ construct_selected(dictionary.values(), keys, target, convert)
+}
+
+/// Restore missing values only after conversion to a type with a validity
+/// bitmap. Union inputs cannot represent a missing selection independently of
+/// their active child, including below NULL parents or absent ROW fields.
+pub(super) fn construct_selected(
+ array: &ArrayRef,
+ indices: &UInt64Array,
+ target: &DataType,
+ convert: fn(&ArrayRef, &DataType) -> crate::Result<ArrayRef>,
+) -> crate::Result<ArrayRef> {
+ let mut selected = Vec::new();
+ let restore = indices
+ .iter()
+ .map(|key| {
+ key.map(|key| {
+ let index = selected.len() as u64;
+ selected.push(key);
+ index
+ })
+ })
+ .collect::<Vec<_>>();
+ if selected.is_empty() {
+ return Ok(arrow_array::new_null_array(target, indices.len()));
+ }
+ let values = if let Some(union) =
array.as_any().downcast_ref::<UnionArray>() {
+ // Taking a sparse Union would also take every inactive child. Select
+ // its active values directly, including through dictionary wrappers.
+ construct_union_rows(
+ union,
+ selected.into_iter().map(|row| row as usize),
+ target,
+ convert,
+ )?
+ } else {
+ let values = arrow_select::take::take(array.as_ref(),
&UInt64Array::from(selected), None)
+ .map_err(|error| invalid(error.to_string()))?;
+ convert(&values, target)?
+ };
+ arrow_select::take::take(values.as_ref(), &UInt64Array::from(restore),
None)
+ .map_err(|error| invalid(error.to_string()))
+}
+
+/// Only active children participate in the Python-value constructor. A sparse
+/// Union can contain invalid values at inactive positions; a dense Union can
+/// select any child offset repeatedly or in a different order.
+pub(super) fn construct_union(
+ array: &ArrayRef,
+ target: &DataType,
+ convert: fn(&ArrayRef, &DataType) -> crate::Result<ArrayRef>,
+) -> crate::Result<ArrayRef> {
+ let input = array.as_any().downcast_ref::<UnionArray>().unwrap();
+ construct_union_rows(input, 0..input.len(), target, convert)
+}
+
+fn construct_union_rows(
+ input: &UnionArray,
+ rows: impl ExactSizeIterator<Item = usize>,
+ target: &DataType,
+ convert: fn(&ArrayRef, &DataType) -> crate::Result<ArrayRef>,
+) -> crate::Result<ArrayRef> {
+ if rows.len() == 0 {
+ return Ok(arrow_array::new_empty_array(target));
+ }
+ let mut selected = IndexMap::<i8, Vec<u64>>::new();
+ let mut positions = Vec::with_capacity(rows.len());
+ for row in rows {
+ let entry = selected.entry(input.type_id(row));
+ let child = entry.index();
+ let offsets = entry.or_default();
+ positions.push((child, offsets.len()));
+ offsets.push(input.value_offset(row) as u64);
+ }
+ let values = selected
+ .into_iter()
+ .map(|(id, offsets)| {
+ let selected = arrow_select::take::take(
+ input.child(id).as_ref(),
+ &UInt64Array::from(offsets),
+ None,
+ )
+ .map_err(|error| invalid(error.to_string()))?;
+ convert(&selected, target)
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ let values = values
+ .iter()
+ .map(|array| array.as_ref())
+ .collect::<Vec<_>>();
+ arrow_select::interleave::interleave(&values, &positions)
+ .map_err(|error| invalid(error.to_string()))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::super::{cast_update_value, CastMode};
+ use super::*;
+ use arrow_array::{
+ types::Int8Type, DictionaryArray, FixedSizeListArray, Float64Array,
Int32Array, Int64Array,
+ Int8Array, ListArray, MapArray, NullArray, StringArray, StructArray,
+ };
+ use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
+ use arrow_schema::{Field, UnionFields};
+ use std::sync::Arc;
+
+ fn numeric_union(dense: bool) -> ArrayRef {
+ let fields = UnionFields::try_new(
+ [7, 42, 99],
+ [
+ Field::new("small", DataType::Int32, true),
+ Field::new("large", DataType::Int64, true),
+ Field::new("unused", DataType::Utf8, true),
+ ],
+ )
+ .unwrap();
+ let (left, right, offsets) = if dense {
+ (
+ vec![Some(-1), None, Some(2)],
+ vec![Some(3), Some(1), Some(i64::MAX)],
+ Some(ScalarBuffer::from(vec![0, 1, 2, 0, 1, 2])),
+ )
+ } else {
+ (
+ vec![Some(-1), Some(0), Some(2), Some(0), None, Some(0)],
+ vec![
+ Some(i64::MAX),
+ Some(1),
+ Some(i64::MAX),
+ Some(3),
+ Some(i64::MAX),
+ Some(i64::MAX),
+ ],
+ None,
+ )
+ };
+ Arc::new(
+ UnionArray::try_new(
+ fields,
+ ScalarBuffer::from(vec![7, 42, 7, 42, 7, 42]),
+ offsets,
+ vec![
+ Arc::new(Int32Array::from(left)),
+ Arc::new(Int64Array::from(right)),
+ Arc::new(StringArray::from(vec![
+ "invalid";
+ if dense { 1 } else { 6 }
+ ])),
+ ],
+ )
+ .unwrap(),
+ )
+ }
+
+ #[test]
+ fn active_values_keep_slices_nulls_and_nonstandard_type_ids() {
+ for dense in [true, false] {
+ let input = numeric_union(dense);
+ let slice = input.slice(1, 4);
+ assert_eq!(slice.null_count(), 0);
+ assert!(cast_update_value(&slice, &DataType::Int32,
CastMode::Assignment).is_err());
+ let result = cast_update_value(&slice, &DataType::Int32,
CastMode::RowUpdate).unwrap();
+ assert_eq!(
+ result.to_data(),
+ Int32Array::from(vec![Some(1), Some(2), Some(3),
None]).to_data()
+ );
+ // The same wide value must fail as soon as its type/offset is
active.
+ assert!(cast_update_value(&input, &DataType::Int32,
CastMode::RowUpdate).is_err());
+ assert_eq!(
+ cast_update_value(&slice.slice(0, 0), &DataType::Int32,
CastMode::RowUpdate)
+ .unwrap()
+ .len(),
+ 0
+ );
+ }
+ }
+
+ #[test]
+ fn dense_offsets_can_repeat_and_reorder_child_values() {
+ let input = numeric_union(true);
+ let fields = match input.data_type() {
+ DataType::Union(fields, _) => fields.clone(),
+ _ => unreachable!(),
+ };
+ let input = input.as_any().downcast_ref::<UnionArray>().unwrap();
+ let reordered: ArrayRef = Arc::new(
+ UnionArray::try_new(
+ fields,
+ ScalarBuffer::from(vec![7, 7, 42, 7]),
+ Some(ScalarBuffer::from(vec![2, 0, 1, 2])),
+ vec![
+ input.child(7).clone(),
+ input.child(42).clone(),
+ input.child(99).clone(),
+ ],
+ )
+ .unwrap(),
+ );
+ assert_eq!(
+ cast_update_value(&reordered, &DataType::Int32,
CastMode::RowUpdate)
+ .unwrap()
+ .to_data(),
+ Int32Array::from(vec![2, -1, 1, 2]).to_data(),
+ );
+ }
+
+ #[test]
+ fn every_selected_child_uses_constructor_semantics() {
+ let fields = UnionFields::try_new(
+ [3, 8],
+ [
+ Field::new("number", DataType::Float64, true),
+ Field::new("text", DataType::Utf8, true),
+ ],
+ )
+ .unwrap();
+ let input: ArrayRef = Arc::new(
+ UnionArray::try_new(
+ fields,
+ ScalarBuffer::from(vec![3, 8]),
+ Some(ScalarBuffer::from(vec![0, 0])),
+ vec![
+ Arc::new(Float64Array::from(vec![1.5])),
+ Arc::new(StringArray::from(vec!["42"])),
+ ],
+ )
+ .unwrap(),
+ );
+ assert_eq!(
+ cast_update_value(&input.slice(0, 1), &DataType::Int32,
CastMode::RowUpdate)
+ .unwrap()
+ .to_data(),
+ Int32Array::from(vec![1]).to_data(),
+ );
+ // A numeric text child must not get a fresh safe-cast attempt after
+ // the Union's whole-column safe cast has already failed.
+ assert!(cast_update_value(&input, &DataType::Int32,
CastMode::RowUpdate).is_err());
+ assert!(
+ cast_update_value(&input.slice(0, 1), &DataType::Utf8,
CastMode::RowUpdate).is_err()
+ );
+ }
+
+ #[test]
+ fn constructors_recurse_through_rows_and_arrays() {
+ for dense in [true, false] {
+ let input = numeric_union(dense).slice(1, 4);
+ let rows: ArrayRef = Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("n", input.data_type().clone(), true)),
+ input.clone(),
+ )]));
+ let target = DataType::Struct(vec![Field::new("n",
DataType::Int32, true)].into());
+ let output = cast_update_value(&rows, &target,
CastMode::RowUpdate).unwrap();
+ let expected = Int32Array::from(vec![Some(1), Some(2), Some(3),
None]);
+ assert_eq!(
+ output
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap()
+ .column(0)
+ .to_data(),
+ expected.to_data()
+ );
+ let lists: ArrayRef = Arc::new(ListArray::new(
+ Arc::new(Field::new("item", input.data_type().clone(), true)),
+ OffsetBuffer::from_lengths([2, 2]),
+ input,
+ None,
+ ));
+ let target = DataType::List(Arc::new(Field::new("item",
DataType::Int32, true)));
+ let output = cast_update_value(&lists, &target,
CastMode::RowUpdate).unwrap();
+ assert_eq!(
+ output
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap()
+ .values()
+ .to_data(),
+ expected.to_data()
+ );
+ }
+ }
+
+ fn pair_union() -> ArrayRef {
+ let pairs: ArrayRef = Arc::new(ListArray::new(
+ Arc::new(Field::new("item", DataType::Int32, true)),
+ OffsetBuffer::from_lengths([2]),
+ Arc::new(Int32Array::from(vec![1, 2])),
+ None,
+ ));
+ let pairs: ArrayRef = Arc::new(ListArray::new(
+ Arc::new(Field::new("item", pairs.data_type().clone(), true)),
+ OffsetBuffer::from_lengths([1]),
+ pairs,
+ None,
+ ));
+ Arc::new(
+ UnionArray::try_new(
+ UnionFields::try_new(
+ [7, 42],
+ [
+ Field::new("pairs", pairs.data_type().clone(), true),
+ Field::new("empty", DataType::Null, true),
+ ],
+ )
+ .unwrap(),
+ ScalarBuffer::from(vec![7, 42]),
+ Some(ScalarBuffer::from(vec![0, 0])),
+ vec![pairs, Arc::new(NullArray::new(1))],
+ )
+ .unwrap(),
+ )
+ }
+
+ fn map_type(key: DataType) -> DataType {
+ DataType::Map(
+ Arc::new(Field::new(
+ "entries",
+ DataType::Struct(
+ vec![
+ Field::new("key", key, false),
+ Field::new("value", DataType::Int64, true),
+ ]
+ .into(),
+ ),
+ false,
+ )),
+ false,
+ )
+ }
+
+ #[test]
+ fn map_pair_conversion_is_root_only_through_encoding_wrappers() {
+ let input = pair_union();
+ let target = map_type(DataType::Int32);
+ let output = cast_update_value(&input, &target,
CastMode::RowUpdate).unwrap();
+ let output = output.as_any().downcast_ref::<MapArray>().unwrap();
+ assert_eq!(output.value_offsets(), &[0, 1, 1]);
+ assert!(output.is_null(1));
+ assert_eq!(
+ output.values().to_data(),
+ Int64Array::from(vec![2]).to_data()
+ );
+ let encoded: ArrayRef = Arc::new(
+ DictionaryArray::<Int8Type>::try_new(Int8Array::from(vec![0, 1]),
input.clone())
+ .unwrap(),
+ );
+ assert_eq!(
+ cast_update_value(&encoded, &target, CastMode::RowUpdate)
+ .unwrap()
+ .to_data(),
+ output.to_data()
+ );
+ let nested: ArrayRef = Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("m", input.data_type().clone(), true)),
+ input,
+ )]));
+ let target = DataType::Struct(vec![Field::new("m", target,
true)].into());
+ assert!(cast_update_value(&nested, &target,
CastMode::RowUpdate).is_err());
+ }
+
+ #[test]
+ fn dictionary_union_preserves_null_indices_and_active_null_values() {
+ for dense in [true, false] {
+ let values = numeric_union(dense).slice(1, 4);
+ let input: ArrayRef = Arc::new(
+ DictionaryArray::<Int8Type>::try_new(
+ Int8Array::from(vec![Some(3), None, Some(0), Some(1)]),
+ values.clone(),
+ )
+ .unwrap(),
+ );
+ let output = cast_update_value(&input, &DataType::Int32,
CastMode::RowUpdate).unwrap();
+ assert_eq!(
+ output.to_data(),
+ Int32Array::from(vec![None, None, Some(1), Some(2)]).to_data()
+ );
+ let input: ArrayRef = Arc::new(
+
DictionaryArray::<Int8Type>::try_new(Int8Array::from(vec![None, None]), values)
+ .unwrap(),
+ );
+ assert_eq!(
+ cast_update_value(&input, &DataType::Int32,
CastMode::RowUpdate)
+ .unwrap()
+ .null_count(),
+ 2
+ );
+ }
+ let input: ArrayRef = Arc::new(
+ DictionaryArray::<Int8Type>::try_new(
+ Int8Array::from(vec![None, Some(0), Some(1), None]),
+ pair_union(),
+ )
+ .unwrap(),
+ );
+ let output =
+ cast_update_value(&input, &map_type(DataType::Int32),
CastMode::RowUpdate).unwrap();
+ let output = output.as_any().downcast_ref::<MapArray>().unwrap();
+ assert_eq!(output.value_offsets(), &[0, 0, 1, 1, 1]);
+ assert_eq!(output.null_count(), 3);
+ }
+
+ #[test]
+ fn union_dict_values_allow_explicit_null_map_values() {
+ let row: ArrayRef = Arc::new(StructArray::from(vec![
+ (
+ Arc::new(Field::new("a", DataType::Int32, true)),
+ Arc::new(Int32Array::from(vec![Some(1)])) as ArrayRef,
+ ),
+ (
+ Arc::new(Field::new("b", DataType::Int32, true)),
+ Arc::new(Int32Array::from(vec![None])) as ArrayRef,
+ ),
+ ]));
+ let target = map_type(DataType::Utf8);
+ assert!(cast_update_value(&row, &target,
CastMode::RowUpdate).is_err());
+ let input: ArrayRef = Arc::new(
+ UnionArray::try_new(
+ UnionFields::try_new([42], [Field::new("row",
row.data_type().clone(), true)])
+ .unwrap(),
+ ScalarBuffer::from(vec![42]),
+ Some(ScalarBuffer::from(vec![0])),
+ vec![row],
+ )
+ .unwrap(),
+ );
+ let output = cast_update_value(&input, &target,
CastMode::RowUpdate).unwrap();
+ let output = output.as_any().downcast_ref::<MapArray>().unwrap();
+ assert_eq!(
+ output.keys().to_data(),
+ StringArray::from(vec!["a", "b"]).to_data()
+ );
+ assert_eq!(
+ output.values().to_data(),
+ Int64Array::from(vec![Some(1), None]).to_data()
+ );
+ }
+
+ #[test]
+ fn null_parents_hide_union_values_before_constructor_conversion() {
+ for dense in [false, true] {
+ let input: ArrayRef = Arc::new(
+ UnionArray::try_new(
+ UnionFields::try_new(
+ [7, 42],
+ [
+ Field::new("number", DataType::Int32, true),
+ Field::new("text", DataType::Utf8, true),
+ ],
+ )
+ .unwrap(),
+ ScalarBuffer::from(vec![42, 7, 7, 42]),
+ dense.then(|| ScalarBuffer::from(vec![0, 0, 1, 1])),
+ if dense {
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
+ Arc::new(StringArray::from(vec!["hidden", "also
hidden"])),
+ ]
+ } else {
+ vec![
+ Arc::new(Int32Array::from(vec![0, 1, 2, 0])) as
ArrayRef,
+ Arc::new(StringArray::from(vec!["hidden"; 4])),
+ ]
+ },
+ )
+ .unwrap(),
+ );
+ let dictionary: ArrayRef = Arc::new(
+ DictionaryArray::<Int8Type>::try_new(
+ Int8Array::from(vec![0, 1, 2, 3]),
+ input.clone(),
+ )
+ .unwrap(),
+ );
+ for input in [input, dictionary] {
+ let parents = NullBuffer::from(vec![false, true, true, false]);
+ let rows: ArrayRef = Arc::new(StructArray::new(
+ vec![Field::new("n", input.data_type().clone(),
true)].into(),
+ vec![input.clone()],
+ Some(parents.clone()),
+ ));
+ let target = DataType::Struct(vec![Field::new("n",
DataType::Int32, true)].into());
+ assert!(cast_update_value(&rows, &target,
CastMode::Assignment).is_err());
+ let output = cast_update_value(&rows, &target,
CastMode::RowUpdate).unwrap();
+ let output = output.as_struct();
+ assert_eq!(output.nulls(), Some(&parents));
+ assert_eq!(
+ output.column(0).to_data(),
+ Int32Array::from(vec![None, Some(1), Some(2),
None]).to_data()
+ );
+
+ let lists: ArrayRef = Arc::new(FixedSizeListArray::new(
+ Arc::new(Field::new("item", input.data_type().clone(),
true)),
+ 1,
+ input,
+ Some(parents.clone()),
+ ));
+ let target = DataType::List(Arc::new(Field::new("item",
DataType::Int32, true)));
+ let output = cast_update_value(&lists, &target,
CastMode::RowUpdate).unwrap();
+ let output = output.as_list::<i32>();
+ assert_eq!(output.nulls(), Some(&parents));
+ assert_eq!(
+ output.value(1).to_data(),
+ Int32Array::from(vec![1]).to_data()
+ );
+ assert_eq!(
+ output.value(2).to_data(),
+ Int32Array::from(vec![2]).to_data()
+ );
+ }
+ }
+ }
+
+ #[test]
+ fn map_union_values_leave_absent_row_fields_null() {
+ for dense in [false, true] {
+ let values: ArrayRef = Arc::new(
+ UnionArray::try_new(
+ UnionFields::try_new([7], [Field::new("number",
DataType::Int32, true)])
+ .unwrap(),
+ ScalarBuffer::from(vec![7; 4]),
+ dense.then(|| ScalarBuffer::from(vec![0, 1, 2, 3])),
+ vec![Arc::new(Int32Array::from(vec![1, 2, 3, 4]))],
+ )
+ .unwrap(),
+ );
+ let entries = StructArray::new(
+ vec![
+ Field::new("key", DataType::Utf8, false),
+ Field::new("value", values.data_type().clone(), true),
+ ]
+ .into(),
+ vec![Arc::new(StringArray::from(vec!["n"; 4])), values],
+ None,
+ );
+ let input: ArrayRef = Arc::new(MapArray::new(
+ Arc::new(Field::new("entries", entries.data_type().clone(),
false)),
+ OffsetBuffer::from_lengths([1; 4]),
+ entries,
+ None,
+ false,
+ ));
+ let target = DataType::Struct(
+ vec![
+ Field::new("n", DataType::Int32, true),
+ Field::new("v", DataType::Utf8, true),
+ ]
+ .into(),
+ );
+ let output = cast_update_value(&input, &target,
CastMode::RowUpdate).unwrap();
+ let output = output.as_struct();
+ assert_eq!(
+ output.column(0).to_data(),
+ Int32Array::from(vec![1, 2, 3, 4]).to_data()
+ );
+ assert_eq!(
+ output.column(1).to_data(),
+ StringArray::from(vec![None::<&str>; 4]).to_data()
+ );
+ }
+ }
+}
diff --git a/crates/paimon/src/table/update_input/scalar_tests.rs
b/crates/paimon/src/table/update_input/scalar_tests.rs
new file mode 100644
index 00000000..56f06deb
--- /dev/null
+++ b/crates/paimon/src/table/update_input/scalar_tests.rs
@@ -0,0 +1,207 @@
+// 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.
+
+use super::*;
+use arrow_array::{Decimal128Array, Decimal256Array, FixedSizeBinaryArray};
+use arrow_buffer::{Buffer, NullBuffer};
+use arrow_schema::Field;
+
+#[test]
+fn fixed_binary_strings_keep_utf8_nulls_slices_and_empty_values() {
+ let input: ArrayRef = Arc::new(
+ FixedSizeBinaryArray::try_from_sparse_iter_with_size(
+ [
+ Some(b"old".as_slice()),
+ Some(b"a\0b".as_slice()),
+ None,
+ Some("中".as_bytes()),
+ ]
+ .into_iter(),
+ 3,
+ )
+ .unwrap(),
+ );
+ let input = input.slice(1, 3);
+ for target in [DataType::Utf8, DataType::LargeUtf8] {
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ let actual = cast_update_value(&input, &target, mode).unwrap();
+ let expected = arrow_cast::cast(
+ &StringArray::from(vec![Some("a\0b"), None, Some("中")]),
+ &target,
+ )
+ .unwrap();
+ assert_eq!(actual.to_data(), expected.to_data());
+ }
+ }
+ let empty: ArrayRef = Arc::new(
+ FixedSizeBinaryArray::try_new(
+ 0,
+ Buffer::from_vec(Vec::<u8>::new()),
+ Some(NullBuffer::from(vec![true, false, true])),
+ )
+ .unwrap(),
+ );
+ assert_eq!(
+ cast_update_value(&empty, &DataType::Utf8, CastMode::Assignment)
+ .unwrap()
+ .to_data(),
+ StringArray::from(vec![Some(""), None, Some("")]).to_data()
+ );
+ let invalid: ArrayRef =
+
Arc::new(FixedSizeBinaryArray::try_from_iter([b"\xffab".as_slice()].into_iter()).unwrap());
+ assert!(cast_update_value(&invalid, &DataType::Utf8,
CastMode::RowUpdate).is_err());
+ // The adaptation is for text, not a new binary-to-numeric cast.
+ let digits: ArrayRef =
+
Arc::new(FixedSizeBinaryArray::try_from_iter([b"123".as_slice()].into_iter()).unwrap());
+ assert!(cast_update_value(&digits, &DataType::Int32,
CastMode::RowUpdate).is_err());
+}
+
+#[test]
+fn half_float_text_uses_all_significant_digits() {
+ let input = Float64Array::from(vec![
+ Some(0.1),
+ None,
+ Some(1.5),
+ Some(-3.25),
+ Some(65504.0),
+ Some(2_f64.powi(-24)),
+ ]);
+ let input = arrow_cast::cast(&input, &DataType::Float16).unwrap();
+ let expected = StringArray::from(vec![
+ Some("0.0999755859375"),
+ None,
+ Some("1.5"),
+ Some("-3.25"),
+ Some("65504"),
+ Some("5.960464477539063e-8"),
+ ]);
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ assert_eq!(
+ cast_update_value(&input, &DataType::Utf8, mode)
+ .unwrap()
+ .to_data(),
+ expected.to_data()
+ );
+ }
+}
+
+#[test]
+fn scalar_encodings_recurse_inside_containers() {
+ let input: ArrayRef = Arc::new(
+ FixedSizeBinaryArray::try_from_iter([b"abc".as_slice(),
"中".as_bytes()].into_iter())
+ .unwrap(),
+ );
+ let rows: ArrayRef = Arc::new(StructArray::from(vec![(
+ Arc::new(Field::new("s", input.data_type().clone(), true)),
+ input.clone(),
+ )]));
+ let target = DataType::Struct(vec![Field::new("s", DataType::Utf8,
true)].into());
+ let output = cast_update_value(&rows, &target,
CastMode::Assignment).unwrap();
+ let expected = StringArray::from(vec!["abc", "中"]);
+ assert_eq!(
+ output
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap()
+ .column(0)
+ .to_data(),
+ expected.to_data()
+ );
+ let lists: ArrayRef = Arc::new(ListArray::new(
+ Arc::new(Field::new("item", input.data_type().clone(), true)),
+ OffsetBuffer::from_lengths([2]),
+ input,
+ None,
+ ));
+ let target = DataType::List(Arc::new(Field::new("item", DataType::Utf8,
true)));
+ let output = cast_update_value(&lists, &target,
CastMode::Assignment).unwrap();
+ assert_eq!(
+ output
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap()
+ .values()
+ .to_data(),
+ expected.to_data()
+ );
+}
+
+#[test]
+fn wide_decimal_floating_values_use_java_big_decimal_rounding() {
+ let coefficient = "1508079088755887087797858531882816082"
+ .parse::<i128>()
+ .unwrap();
+ let source: ArrayRef = Arc::new(
+ Decimal128Array::from(vec![Some(coefficient), None,
Some(-coefficient)])
+ .with_precision_and_scale(38, 1)
+ .unwrap(),
+ );
+ let expected = [
+ Some(1.508079088755887e35),
+ None,
+ Some(-1.508079088755887e35),
+ ];
+ let mut sources = vec![source];
+ for precision in [38, 39, 76] {
+ let wide: ArrayRef = Arc::new(
+ Decimal256Array::from_iter(
+ [Some(coefficient), None, Some(-coefficient)]
+ .into_iter()
+ .map(|value| value.map(arrow_buffer::i256::from_i128)),
+ )
+ .with_precision_and_scale(precision, 1)
+ .unwrap(),
+ );
+ sources.push(wide);
+ }
+ for source in sources {
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ let output = cast_update_value(&source, &DataType::Float64,
mode).unwrap();
+ assert_eq!(
+ output.to_data(),
+ Float64Array::from(expected.to_vec()).to_data()
+ );
+ let output = cast_update_value(&source, &DataType::Float32,
mode).unwrap();
+ assert_eq!(
+ output.to_data(),
+ Float32Array::from_iter(expected.into_iter().map(|v| v.map(|v|
v as f32)))
+ .to_data()
+ );
+ }
+ }
+}
+
+#[test]
+fn compact_decimal_floating_values_keep_java_long_division() {
+ let coefficient = 9_007_199_254_740_993_i128;
+ let source: ArrayRef = Arc::new(
+ Decimal128Array::from(vec![Some(coefficient), None])
+ .with_precision_and_scale(18, 3)
+ .unwrap(),
+ );
+ let expected = coefficient as f64 / 1000.0;
+ let output = cast_update_value(&source, &DataType::Float64,
CastMode::Assignment).unwrap();
+ assert_eq!(
+ output.to_data(),
+ Float64Array::from(vec![Some(expected), None]).to_data()
+ );
+ let output = cast_update_value(&source, &DataType::Float32,
CastMode::Assignment).unwrap();
+ assert_eq!(
+ output.to_data(),
+ Float32Array::from(vec![Some(expected as f32), None]).to_data()
+ );
+}
