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 a914bc42 [core] Support additional Arrow inputs in row-level updates
(#1152)
a914bc42 is described below
commit a914bc42820b65f88e77a284edd293d347d2ebc4
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 18 10:36:20 2026 +0800
[core] Support additional Arrow inputs in row-level updates (#1152)
---
crates/paimon/src/table/key_normalization.rs | 5 +
.../paimon/src/table/key_normalization/numeric.rs | 6 +-
crates/paimon/src/table/update_input.rs | 106 +++++++++--
crates/paimon/src/table/update_input/binary.rs | 177 ++++++++++++++++++
.../src/table/update_input/container_tests.rs | 66 +++++++
crates/paimon/src/table/update_input/decimal.rs | 200 +++++++++++++++++++--
crates/paimon/src/table/write_batch_normalize.rs | 103 +++++++++--
7 files changed, 613 insertions(+), 50 deletions(-)
diff --git a/crates/paimon/src/table/key_normalization.rs
b/crates/paimon/src/table/key_normalization.rs
index 16fcef2f..1508a347 100644
--- a/crates/paimon/src/table/key_normalization.rs
+++ b/crates/paimon/src/table/key_normalization.rs
@@ -45,6 +45,11 @@ pub(super) fn normalize_match_key(
if source.is_numeric() && expected.is_numeric() {
return numeric::normalize(&decoded, expected);
}
+ if source == &DataType::LargeUtf8 && expected == &DataType::Utf8 {
+ return super::write_batch_normalize::normalize_large_string_key(
+ decoded.as_any().downcast_ref().unwrap(),
+ );
+ }
let binary = |t: &DataType| {
matches!(
t,
diff --git a/crates/paimon/src/table/key_normalization/numeric.rs
b/crates/paimon/src/table/key_normalization/numeric.rs
index 5b49bd2f..5ae73b97 100644
--- a/crates/paimon/src/table/key_normalization/numeric.rs
+++ b/crates/paimon/src/table/key_normalization/numeric.rs
@@ -69,11 +69,7 @@ pub(super) fn normalize(array: &ArrayRef, target: &DataType)
-> crate::Result<Ar
return decimal_to_float(array, target);
}
if target.is_integer() {
- // Canonicalize first to avoid scale arithmetic overflow in generic
- // Arrow integer casts, including external Decimal256 zero values.
- let integral =
- cast_update_value(array, &DataType::Decimal128(38, 0),
CastMode::Assignment)?;
- return cast_update_value(&integral, target, CastMode::Assignment);
+ return cast_update_value(array, target, CastMode::Assignment);
}
}
if source.is_floating() || target.is_floating() {
diff --git a/crates/paimon/src/table/update_input.rs
b/crates/paimon/src/table/update_input.rs
index 38f1fccc..5fa976fa 100644
--- a/crates/paimon/src/table/update_input.rs
+++ b/crates/paimon/src/table/update_input.rs
@@ -28,6 +28,7 @@ use arrow_array::{
use arrow_buffer::OffsetBuffer;
use arrow_schema::{DataType, Schema, TimeUnit};
+mod binary;
mod decimal;
mod encoded;
@@ -282,6 +283,14 @@ fn cast_values(array: &ArrayRef, target: &DataType, mode:
CastMode) -> crate::Re
}
DataType::Map(..) => {
let input = array.as_any().downcast_ref::<MapArray>().unwrap();
+ if mode == CastMode::Constructor
+ && matches!(
+ field.data_type(),
+ DataType::List(_) | DataType::LargeList(_) |
DataType::FixedSizeList(..)
+ )
+ {
+ return constructor_array_pairs(input, field, target);
+ }
let DataType::Struct(fields) = field.data_type() else {
return Err(invalid(
"Map entries can only be cast to two-field ROW
elements",
@@ -355,28 +364,83 @@ fn cast_map_entries(
if fields.len() != 2 {
return Err(invalid("Map entries require exactly two ROW fields"));
}
+ let (columns, offsets) =
+ cast_map_columns(input, [fields[0].data_type(),
fields[1].data_type()], mode)?;
+ let length = *offsets.last().unwrap() as usize;
+ let entries = StructArray::try_new_with_length(fields.clone(),
columns.into(), None, length)
+ .map_err(|error| invalid(error.to_string()))?;
+ Ok((entries, offsets))
+}
+
+/// Select the visible MAP entries before rebuilding Python constructor values.
+/// Both ROW pairs and ARRAY pairs share the same key/value selection rules.
+fn cast_map_columns(
+ input: &MapArray,
+ types: [&DataType; 2],
+ mode: CastMode,
+) -> crate::Result<([ArrayRef; 2], OffsetBuffer<i32>)> {
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)
+ let cast = |values: &ArrayRef, target: &DataType| {
+ 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 as usize),
+ };
+ cast_values(&values, target, mode)
+ };
+ Ok((
+ [
+ cast(input.keys(), types[0])?,
+ cast(input.values(), types[1])?,
+ ],
+ OffsetBuffer::new(offsets.into()),
+ ))
+}
+
+/// A Python MAP value is a sequence of key/value tuples. Rebuilding it as
+/// ARRAY<ARRAY<T>> turns each tuple into a two-element ARRAY; this is input
+/// construction, not a MAP-to-ARRAY schema-evolution cast.
+fn constructor_array_pairs(
+ input: &MapArray,
+ pair_field: &arrow_schema::FieldRef,
+ target: &DataType,
+) -> crate::Result<ArrayRef> {
+ let element = match pair_field.data_type() {
+ DataType::List(element)
+ | DataType::LargeList(element)
+ | DataType::FixedSizeList(element, _) => element,
+ _ => return Err(invalid("Map pairs require two-element ARRAY values")),
+ };
+ let ([keys, values], offsets) =
+ cast_map_columns(input, [element.data_type(); 2],
CastMode::Constructor)?;
+ let positions = (0..keys.len())
+ .flat_map(|row| [(0, row), (1, row)])
+ .collect::<Vec<_>>();
+ let values =
+ arrow_select::interleave::interleave(&[keys.as_ref(),
values.as_ref()], &positions)
+ .map_err(|error| invalid(error.to_string()))?;
+ let options = arrow_cast::CastOptions {
+ safe: false,
+ ..Default::default()
+ };
+ let pairs = if keys.is_empty() {
+ arrow_array::new_empty_array(pair_field.data_type())
+ } else {
+ let pairs =
+ FixedSizeListArray::try_new_with_length(element.clone(), 2,
values, None, keys.len())
+ .map_err(|error| invalid(error.to_string()))?;
+ arrow_cast::cast_with_options(&pairs, pair_field.data_type(), &options)
+ .map_err(|error| invalid(error.to_string()))?
+ };
+ let output = ListArray::try_new(pair_field.clone(), offsets, pairs,
input.nulls().cloned())
.map_err(|error| invalid(error.to_string()))?;
- Ok((entries, OffsetBuffer::new(offsets.into())))
+ arrow_cast::cast_with_options(&output, target, &options)
+ .map_err(|error| invalid(error.to_string()))
}
// PyPaimon's row-ID fallback reconstructs Python dicts only after a
whole-column
@@ -774,6 +838,12 @@ fn cast_to_string(array: &ArrayRef, target: &DataType) ->
crate::Result<ArrayRef
let binary =
super::write_batch_normalize::normalize_write_array(array, &DataType::Binary)?;
return cast_to_string(&binary, target);
}
+ if matches!(
+ array.data_type(),
+ DataType::Binary | DataType::LargeBinary | DataType::BinaryView
+ ) {
+ return binary::cast_to_string(array, target);
+ }
let floats: Option<StringArray> = match array.data_type() {
DataType::Float16 => Some(
array
@@ -1343,8 +1413,7 @@ fn cast_primitive(array: &ArrayRef, target: &DataType,
mode: CastMode) -> crate:
) && (target.is_numeric() || *target == DataType::Boolean)
{
// Binary numeric inputs contain text, never native-endian numeric
bytes.
- let text = arrow_cast::cast_with_options(array.as_ref(),
&DataType::Utf8, &options)
- .map_err(|error| invalid(error.to_string()))?;
+ let text = binary::cast_to_string(array, &DataType::Utf8)?;
return cast_primitive(&text, target, mode);
}
if matches!(source, DataType::Timestamp(..)) && matches!(target,
DataType::Timestamp(..)) {
@@ -1445,6 +1514,9 @@ fn cast_primitive(array: &ArrayRef, target: &DataType,
mode: CastMode) -> crate:
if source.is_decimal() && matches!(target, DataType::Decimal128(..)) {
return decimal::cast_exact_decimal(array, target);
}
+ if source.is_decimal() && target.is_integer() {
+ return decimal::cast_to_integer(array, target, mode ==
CastMode::Constructor);
+ }
if source.is_decimal() && matches!(target, DataType::Float32 |
DataType::Float64) {
return decimal::cast_to_floating(array, target);
}
diff --git a/crates/paimon/src/table/update_input/binary.rs
b/crates/paimon/src/table/update_input/binary.rs
new file mode 100644
index 00000000..872d6449
--- /dev/null
+++ b/crates/paimon/src/table/update_input/binary.rs
@@ -0,0 +1,177 @@
+// 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.
+
+//! Text input decoding respects logical NULLs, independently of their bytes.
+
+use arrow_array::{builder::LargeStringBuilder, cast::AsArray, ArrayRef};
+use arrow_schema::DataType;
+use std::sync::Arc;
+
+use super::invalid;
+
+pub(super) fn cast_to_string(array: &ArrayRef, target: &DataType) ->
crate::Result<ArrayRef> {
+ let options = arrow_cast::CastOptions {
+ safe: false,
+ ..Default::default()
+ };
+ let result = arrow_cast::cast_with_options(array.as_ref(), target,
&options);
+ match result {
+ Ok(array) => Ok(array),
+ Err(_) => {
+ // The zero-copy kernel validates the entire value buffer,
including
+ // NULL slots and bytes outside a slice. Arrow C++ skips those
bytes.
+ // Rebuild only after the ordinary cast fails,
+ // still rejecting every invalid visible value. Large offsets avoid
+ // overflow before the final checked conversion to the target
width.
+ let strings = match array.data_type() {
+ DataType::Binary =>
visible_text(array.as_binary::<i32>().iter(), target)?,
+ DataType::LargeBinary =>
visible_text(array.as_binary::<i64>().iter(), target)?,
+ DataType::BinaryView =>
visible_text(array.as_binary_view().iter(), target)?,
+ _ => unreachable!("binary text input was validated by the
caller"),
+ };
+ arrow_cast::cast_with_options(strings.as_ref(), target, &options)
+ .map_err(|error| invalid(error.to_string()))
+ }
+ }
+}
+
+fn visible_text<'a>(
+ values: impl Iterator<Item = Option<&'a [u8]>> + Clone,
+ target: &DataType,
+) -> crate::Result<ArrayRef> {
+ if target == &DataType::Utf8 {
+ values
+ .clone()
+ .flatten()
+ .try_fold(0_usize, |length, bytes| {
+ length
+ .checked_add(bytes.len())
+ .filter(|length| *length <= i32::MAX as usize)
+ .ok_or_else(|| invalid("Binary text exceeds STRING offset
range"))
+ })?;
+ }
+ let mut builder = LargeStringBuilder::new();
+ for value in values {
+ match value {
+ Some(bytes) => builder.append_value(
+ std::str::from_utf8(bytes).map_err(|error|
invalid(error.to_string()))?,
+ ),
+ None => builder.append_null(),
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::super::{cast_update_value, CastMode};
+ use super::*;
+ use arrow_array::{Array, BinaryArray, Int32Array, LargeBinaryArray,
StringArray};
+ use arrow_buffer::{Buffer, NullBuffer, OffsetBuffer};
+ use std::sync::Arc;
+
+ fn hidden_bytes(large: bool) -> ArrayRef {
+ let offsets = vec![0, 1, 3, 5, 6, 8, 9];
+ let values = Buffer::from(b"x\xff\xff42\xc3-7z".as_slice());
+ let nulls = Some(NullBuffer::from(vec![true, false, true, false, true,
true]));
+ let array: ArrayRef = if large {
+ Arc::new(LargeBinaryArray::new(
+ OffsetBuffer::new(offsets.into()),
+ values,
+ nulls,
+ ))
+ } else {
+ Arc::new(BinaryArray::new(
+ OffsetBuffer::new(
+ offsets
+ .into_iter()
+ .map(|i| i as i32)
+ .collect::<Vec<_>>()
+ .into(),
+ ),
+ values,
+ nulls,
+ ))
+ };
+ array.slice(1, 4)
+ }
+
+ #[test]
+ fn hidden_invalid_utf8_is_not_a_text_or_numeric_value() {
+ for large in [false, true] {
+ let source = hidden_bytes(large);
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ let text = cast_update_value(&source, &DataType::Utf8,
mode).unwrap();
+ assert_eq!(
+ text.to_data(),
+ StringArray::from(vec![None, Some("42"), None,
Some("-7")]).to_data()
+ );
+ let integers = cast_update_value(&source, &DataType::Int32,
mode).unwrap();
+ assert_eq!(
+ integers.to_data(),
+ Int32Array::from(vec![None, Some(42), None,
Some(-7)]).to_data()
+ );
+ // A batch boundary may retain only a valid value while its
+ // backing buffer still owns invalid bytes from other slots.
+ let valid_slice = source.slice(1, 1);
+ assert_eq!(valid_slice.null_count(), 0);
+ let text = cast_update_value(&valid_slice, &DataType::Utf8,
mode).unwrap();
+ assert_eq!(text.to_data(),
StringArray::from(vec!["42"]).to_data());
+ let null_slice = source.slice(0, 1);
+ let text = cast_update_value(&null_slice, &DataType::Utf8,
mode).unwrap();
+ assert_eq!(
+ text.to_data(),
+ StringArray::from(vec![None::<&str>]).to_data()
+ );
+ let empty = cast_update_value(&source.slice(1, 0),
&DataType::Utf8, mode).unwrap();
+ assert!(empty.is_empty());
+ }
+ }
+ }
+
+ #[test]
+ fn oversized_visible_text_is_rejected_before_copying_values() {
+ // Reuse one small allocation to represent a >2 GiB logical text input.
+ let bytes = vec![b'x'; 1 << 20];
+ let values = std::iter::repeat_n(Some(bytes.as_slice()), 2048);
+ assert!(visible_text(values, &DataType::Utf8).is_err());
+ }
+
+ #[test]
+ fn visible_invalid_utf8_still_rejects_the_whole_assignment() {
+ let source: ArrayRef = Arc::new(BinaryArray::from(vec![
+ Some(b"42".as_slice()),
+ None,
+ Some(b"\xff".as_slice()),
+ ]));
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ for target in [DataType::Utf8, DataType::Int32] {
+ assert!(cast_update_value(&source, &target, mode).is_err());
+ }
+ }
+ }
+
+ #[test]
+ fn ordinary_binary_text_keeps_its_shared_value_buffer() {
+ let source = Arc::new(BinaryArray::from(vec![Some(b"42".as_slice()),
None])) as ArrayRef;
+ let result = cast_to_string(&source, &DataType::Utf8).unwrap();
+ assert_eq!(
+ source.to_data().buffers()[1].as_ptr(),
+ result.to_data().buffers()[1].as_ptr()
+ );
+ }
+}
diff --git a/crates/paimon/src/table/update_input/container_tests.rs
b/crates/paimon/src/table/update_input/container_tests.rs
index 85dd4186..7f403e38 100644
--- a/crates/paimon/src/table/update_input/container_tests.rs
+++ b/crates/paimon/src/table/update_input/container_tests.rs
@@ -238,3 +238,69 @@ fn
map_entries_require_two_fields_and_actual_child_nullability() {
);
assert!(cast_update_value(&null_value, &target,
CastMode::Assignment).is_err());
}
+
+#[test]
+fn map_pairs_construct_array_elements_without_converting_hidden_entries() {
+ let input = map(
+ Arc::new(Int64Array::from(vec![i64::MAX, 1, 3, i64::MAX, 5])),
+ Arc::new(Float64Array::from(vec![
+ Some(f64::MAX),
+ Some(2.75),
+ None,
+ Some(f64::MAX),
+ Some(-6.75),
+ ])),
+ &[1, 2, 1, 0, 1],
+ &[true, true, false, true, true],
+ )
+ .slice(1, 4);
+ let element = Arc::new(Field::new("element", DataType::Int32, true));
+ for pair_type in [
+ DataType::List(element.clone()),
+ DataType::LargeList(element.clone()),
+ DataType::FixedSizeList(element.clone(), 2),
+ ] {
+ let target = DataType::List(Arc::new(Field::new("pair", pair_type,
false)));
+ assert!(cast_update_value(&input, &target,
CastMode::Assignment).is_err());
+ let result = cast_update_value(&input, &target,
CastMode::RowUpdate).unwrap();
+ let outer = result.as_list::<i32>();
+ assert_eq!(outer.value_offsets(), &[0, 2, 2, 2, 3]);
+ assert!(outer.is_null(1));
+ let pairs = arrow_cast::cast(
+ outer.values().as_ref(),
+ &DataType::FixedSizeList(element.clone(), 2),
+ )
+ .unwrap();
+ assert_eq!(
+ pairs.as_fixed_size_list().values().to_data(),
+ Int32Array::from(vec![Some(1), Some(2), Some(3), None, Some(5),
Some(-6)]).to_data()
+ );
+ }
+}
+
+#[test]
+fn empty_map_pairs_do_not_require_a_particular_vector_width() {
+ let input = map(
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int32Array::from(vec![2])),
+ &[0, 1],
+ &[true, false],
+ );
+ for size in [0, 1, 3] {
+ let target = DataType::List(Arc::new(Field::new(
+ "pair",
+ DataType::FixedSizeList(Arc::new(Field::new("item",
DataType::Int32, true)), size),
+ false,
+ )));
+ let result = cast_update_value(&input, &target,
CastMode::RowUpdate).unwrap();
+ assert_eq!(result.as_list::<i32>().value_offsets(), &[0, 0, 0]);
+ assert!(result.is_null(1));
+ let visible = map(
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int32Array::from(vec![2])),
+ &[1],
+ &[true],
+ );
+ assert!(cast_update_value(&visible, &target,
CastMode::RowUpdate).is_err());
+ }
+}
diff --git a/crates/paimon/src/table/update_input/decimal.rs
b/crates/paimon/src/table/update_input/decimal.rs
index 609d47e7..315b5565 100644
--- a/crates/paimon/src/table/update_input/decimal.rs
+++ b/crates/paimon/src/table/update_input/decimal.rs
@@ -17,7 +17,8 @@
//! Exact DECIMAL input rescaling shared by updates and logical match keys.
-use arrow_array::{cast::AsArray, ArrayRef, Decimal128Array, Float32Array,
Float64Array};
+use arrow_array::{cast::AsArray, Array, ArrayRef, Decimal128Array,
Float32Array, Float64Array};
+use arrow_buffer::i256;
use arrow_schema::DataType;
use std::sync::Arc;
@@ -38,26 +39,35 @@ fn map_decimal<T>(
parse: impl Fn(&str) -> crate::Result<T>,
) -> crate::Result<Vec<Option<T>>> {
let (_, scale) = precision_and_scale(array);
+ map_coefficients(array, |value| {
+ parse(&format!("{value}e{}", -i16::from(scale)))
+ })
+}
+
+fn map_coefficients<T>(
+ array: &ArrayRef,
+ convert: impl Fn(i256) -> crate::Result<T>,
+) -> crate::Result<Vec<Option<T>>> {
// 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) => {
+ ($ty:ty, $widen:expr) => {
array
.as_primitive::<$ty>()
.iter()
- .map(|value| {
- value
- .map(|value| parse(&format!("{value}e{}",
-i16::from(scale))))
- .transpose()
- })
+ .map(|value| value.map(|value|
convert(($widen)(value))).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),
+ DataType::Decimal32(..) => values!(arrow_array::types::Decimal32Type,
|v| i256::from_i128(
+ v as i128
+ )),
+ DataType::Decimal64(..) => values!(arrow_array::types::Decimal64Type,
|v| i256::from_i128(
+ v as i128
+ )),
+ DataType::Decimal128(..) =>
values!(arrow_array::types::Decimal128Type, i256::from_i128),
+ DataType::Decimal256(..) =>
values!(arrow_array::types::Decimal256Type, |v| v),
_ => unreachable!("validated DECIMAL input"),
}
}
@@ -89,6 +99,65 @@ pub(super) fn cast_to_floating(array: &ArrayRef, target:
&DataType) -> crate::Re
})
}
+/// Extract the integral value before narrowing its physical width. Generic
+/// Arrow kernels build 10^scale even for zero, overflowing at extreme external
+/// scales. Row-ID constructors truncate fractions; evaluated assignments and
+/// match keys require exact values. Both contracts still reject width
overflow.
+pub(super) fn cast_to_integer(
+ array: &ArrayRef,
+ target: &DataType,
+ truncate: bool,
+) -> crate::Result<ArrayRef> {
+ let (_, scale) = precision_and_scale(array);
+ // Compute the scale factor once per array, without formatting each value
+ // or overflowing the narrower DECIMAL32/64/128 physical coefficient.
+ let power =
i256::from_i128(10).checked_pow(u32::from(scale.unsigned_abs()));
+ let overflow = || invalid("Decimal assignment exceeds target integer
range");
+ let values = map_coefficients(array, |value| {
+ if value == i256::ZERO {
+ return Ok(0);
+ }
+ let integral = if scale > 0 {
+ let Some(power) = power else {
+ return if truncate {
+ Ok(0)
+ } else {
+ Err(invalid("Decimal assignment would lose scale
precision"))
+ };
+ };
+ if !truncate && value % power != i256::ZERO {
+ return Err(invalid("Decimal assignment would lose scale
precision"));
+ }
+ value / power
+ } else {
+ value
+ .checked_mul(power.ok_or_else(overflow)?)
+ .ok_or_else(overflow)?
+ };
+ let value = integral.to_i128().ok_or_else(overflow)?;
+ if value.unsigned_abs() >= 10_u128.pow(38) {
+ return Err(overflow());
+ }
+ Ok(value)
+ })?;
+ let integral = Decimal128Array::from(values)
+ .with_precision_and_scale(38, 0)
+ .map_err(|error| invalid(error.to_string()))?;
+ let options = arrow_cast::CastOptions {
+ safe: false,
+ ..Default::default()
+ };
+ let converted = arrow_cast::cast_with_options(&integral, target, &options)
+ .map_err(|error| invalid(error.to_string()))?;
+ let restored =
+ arrow_cast::cast_with_options(converted.as_ref(),
integral.data_type(), &options)
+ .map_err(|error| invalid(error.to_string()))?;
+ if restored.to_data() != integral.to_data() {
+ return Err(invalid("Decimal assignment exceeds target integer range"));
+ }
+ Ok(converted)
+}
+
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 {
@@ -236,4 +305,113 @@ mod tests {
);
}
}
+
+ #[test]
+ fn integer_zero_does_not_need_an_oversized_decimal_power() {
+ for source in [
+ DataType::Decimal32(9, -128),
+ DataType::Decimal64(18, -90),
+ DataType::Decimal128(38, -90),
+ DataType::Decimal256(76, -128),
+ ] {
+ for target in [
+ DataType::Int8,
+ DataType::Int32,
+ DataType::Int64,
+ DataType::UInt64,
+ ] {
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ let output = cast_update_value(&decimal(&source, 0),
&target, mode).unwrap();
+ let expected = arrow_cast::cast(
+ &arrow_array::Int64Array::from(vec![Some(0), None]),
+ &target,
+ )
+ .unwrap();
+ assert_eq!(
+ output.to_data(),
+ expected.to_data(),
+ "{source:?} -> {target:?}"
+ );
+ assert!(cast_update_value(&decimal(&source, 1), &target,
mode).is_err());
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn integer_width_bounds_apply_after_fraction_truncation() {
+ for source in [DataType::Decimal128(38, 2), DataType::Decimal256(76,
2)] {
+ for (target, minimum, maximum) in [
+ (DataType::Int8, i8::MIN as i128, i8::MAX as i128),
+ (DataType::Int16, i16::MIN as i128, i16::MAX as i128),
+ (DataType::Int32, i32::MIN as i128, i32::MAX as i128),
+ (DataType::Int64, i64::MIN as i128, i64::MAX as i128),
+ (DataType::UInt8, 0, u8::MAX as i128),
+ (DataType::UInt16, 0, u16::MAX as i128),
+ (DataType::UInt32, 0, u32::MAX as i128),
+ (DataType::UInt64, 0, u64::MAX as i128),
+ ] {
+ for value in [minimum, maximum] {
+ let input = decimal(&source, value * 100);
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ let output = cast_update_value(&input, &target,
mode).unwrap();
+ // The output must contain the integer boundary itself,
+ // rather than a wrapped value or a newly introduced
NULL.
+ let expected: ArrayRef = if target.is_signed_integer()
{
+ Arc::new(arrow_array::Int64Array::from(vec![
+ Some(value as i64),
+ None,
+ ]))
+ } else {
+ Arc::new(arrow_array::UInt64Array::from(vec![
+ Some(value as u64),
+ None,
+ ]))
+ };
+ let expected = arrow_cast::cast(&expected,
&target).unwrap();
+ assert_eq!(output.to_data(), expected.to_data());
+ }
+ }
+ for value in [minimum - 1, maximum + 1] {
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ assert!(
+ cast_update_value(&decimal(&source, value * 100),
&target, mode)
+ .is_err()
+ );
+ }
+ }
+ let fraction = decimal(&source, maximum * 100 + 99);
+ assert!(cast_update_value(&fraction, &target,
CastMode::Assignment).is_err());
+ assert!(cast_update_value(&fraction, &target,
CastMode::RowUpdate).is_ok());
+ }
+ }
+ }
+
+ #[test]
+ fn wide_decimal_coefficients_are_reduced_before_integer_narrowing() {
+ let coefficient = format!("123{}",
"0".repeat(60)).parse::<i256>().unwrap();
+ let input: ArrayRef = Arc::new(
+ Decimal256Array::from(vec![Some(coefficient), None,
Some(-coefficient)])
+ .with_precision_and_scale(76, 60)
+ .unwrap(),
+ );
+ let expected = arrow_array::Int32Array::from(vec![Some(123), None,
Some(-123)]);
+ for mode in [CastMode::Assignment, CastMode::RowUpdate] {
+ let output = cast_update_value(&input, &DataType::Int32,
mode).unwrap();
+ assert_eq!(output.to_data(), expected.to_data());
+ }
+ let fractional: ArrayRef = Arc::new(
+ Decimal256Array::from(vec![
+ Some(coefficient + i256::ONE),
+ None,
+ Some(-coefficient - i256::ONE),
+ ])
+ .with_precision_and_scale(76, 60)
+ .unwrap(),
+ );
+ assert!(cast_update_value(&fractional, &DataType::Int32,
CastMode::Assignment).is_err());
+ let truncated =
+ cast_update_value(&fractional, &DataType::Int32,
CastMode::RowUpdate).unwrap();
+ assert_eq!(truncated.to_data(), expected.to_data());
+ }
}
diff --git a/crates/paimon/src/table/write_batch_normalize.rs
b/crates/paimon/src/table/write_batch_normalize.rs
index d732215f..36bd939c 100644
--- a/crates/paimon/src/table/write_batch_normalize.rs
+++ b/crates/paimon/src/table/write_batch_normalize.rs
@@ -444,27 +444,59 @@ fn normalize_large_string(array: &LargeStringArray) ->
Result<ArrayRef> {
/// Offset adaptation for matching keys; ordinary append type validation
remains
/// separate so this does not broaden accepted write schemas.
pub(super) fn normalize_large_binary(array: &LargeBinaryArray) ->
Result<ArrayRef> {
- let source = array.value_offsets();
- if array.null_count() != 0 && source[source.len() - 1] - source[0] >
i64::from(i32::MAX) {
+ let (offsets, values) =
+ normalize_large_key_bytes(array.value_offsets(), array.values(),
array.nulls())?;
+ BinaryArray::try_new(offsets, values, array.nulls().cloned())
+ .map(|array| Arc::new(array) as ArrayRef)
+ .map_err(normalization_error)
+}
+
+/// Matching reads logical string values, including NULLs with hidden storage.
+/// Ordinary write normalization keeps its existing offset-limit validation.
+pub(super) fn normalize_large_string_key(array: &LargeStringArray) ->
Result<ArrayRef> {
+ let (offsets, values) =
+ normalize_large_key_bytes(array.value_offsets(), array.values(),
array.nulls())?;
+ StringArray::try_new(offsets, values, array.nulls().cloned())
+ .map(|array| Arc::new(array) as ArrayRef)
+ .map_err(normalization_error)
+}
+
+fn normalize_large_key_bytes(
+ source: &[i64],
+ values: &Buffer,
+ nulls: Option<&NullBuffer>,
+) -> Result<(OffsetBuffer<i32>, Buffer)> {
+ if let Some(nulls) = nulls.filter(|nulls| {
+ nulls.null_count() != 0 && source[source.len() - 1] - source[0] >
i64::from(i32::MAX)
+ }) {
// NULL slots can retain arbitrarily large, unobservable byte ranges.
// Compact only when they prevent narrowing; ordinary arrays continue
// to share the visible value buffer below.
- let visible_len = array
- .iter()
- .flatten()
- .try_fold(0_usize, |len, value| len.checked_add(value.len()));
- if visible_len.is_none_or(|len| len > i32::MAX as usize) {
+ let visible_len = source
+ .windows(2)
+ .enumerate()
+ .filter(|(row, _)| nulls.is_valid(*row))
+ .try_fold(0_i64, |len, (_, range)| {
+ len.checked_add(range[1] - range[0])
+ });
+ if visible_len.is_none_or(|len| len > i64::from(i32::MAX)) {
return Err(Error::DataInvalid {
message: "Arrow byte range exceeds 32-bit offsets".into(),
source: None,
});
}
- return Ok(Arc::new(BinaryArray::from_iter(array.iter())));
+ let mut compacted = Vec::with_capacity(visible_len.unwrap() as usize);
+ let mut offsets = Vec::with_capacity(source.len());
+ offsets.push(0);
+ for (row, range) in source.windows(2).enumerate() {
+ if nulls.is_valid(row) {
+ compacted.extend_from_slice(&values[range[0] as
usize..range[1] as usize]);
+ }
+ offsets.push(compacted.len() as i32);
+ }
+ return Ok((OffsetBuffer::new(offsets.into()),
Buffer::from(compacted)));
}
- let (offsets, values) = narrow_byte_range(array.value_offsets(),
array.values())?;
- BinaryArray::try_new(offsets, values, array.nulls().cloned())
- .map(|array| Arc::new(array) as ArrayRef)
- .map_err(normalization_error)
+ narrow_byte_range(source, values)
}
fn narrow_byte_range(source: &[i64], values: &Buffer) ->
Result<(OffsetBuffer<i32>, Buffer)> {
@@ -652,11 +684,8 @@ mod tests {
let input: ArrayRef = Arc::new(LargeBinaryArray::new(
OffsetBuffer::new(offsets.into()),
text.values().clone(),
- Some(NullBuffer::from(if mixed {
- vec![false, true]
- } else {
- vec![false]
- })),
+ // Retain a nonzero validity-bit offset through compaction.
+ Some(NullBuffer::from(vec![true, false, true]).slice(1, if
mixed { 2 } else { 1 })),
));
let output =
crate::table::key_normalization::normalize_match_key(&input,
&DataType::Binary)
@@ -667,6 +696,29 @@ mod tests {
vec![None]
};
assert_eq!(output.to_data(),
BinaryArray::from(expected).to_data());
+ let bytes =
input.as_any().downcast_ref::<LargeBinaryArray>().unwrap();
+ // SAFETY: the shared mapping contains valid UTF-8. Avoid scanning
+ // the hidden 2 GiB NULL range merely to construct the test input.
+ let strings: ArrayRef = Arc::new(unsafe {
+ LargeStringArray::new_unchecked(
+ bytes.offsets().clone(),
+ bytes.values().clone(),
+ bytes.nulls().cloned(),
+ )
+ });
+ let output =
+ crate::table::key_normalization::normalize_match_key(&strings,
&DataType::Utf8)
+ .unwrap();
+ assert_eq!(
+ output.to_data(),
+ StringArray::from(if mixed {
+ vec![None, Some("abc")]
+ } else {
+ vec![None]
+ })
+ .to_data(),
+ );
+ assert!(normalize_write_array(&strings, &DataType::Utf8).is_err());
}
let visible_overflow: ArrayRef = Arc::new(LargeBinaryArray::new(
OffsetBuffer::new(vec![0, start + 3, start + 3].into()),
@@ -678,6 +730,23 @@ mod tests {
&DataType::Binary,
)
.is_err());
+ let bytes = visible_overflow
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ // SAFETY: as above, both the zero-filled prefix and tail are UTF-8.
+ let visible_string_overflow: ArrayRef = Arc::new(unsafe {
+ LargeStringArray::new_unchecked(
+ bytes.offsets().clone(),
+ bytes.values().clone(),
+ bytes.nulls().cloned(),
+ )
+ });
+ assert!(crate::table::key_normalization::normalize_match_key(
+ &visible_string_overflow,
+ &DataType::Utf8,
+ )
+ .is_err());
// A visible range larger than i32::MAX still cannot be represented.
// SAFETY: the mapped bytes are zeroes followed by a valid UTF-8 tail;
// offsets and NULL bitmap are valid. Skip scanning the 2 GiB prefix.
