From 2adf2dbd12a8060e075b335530283bf7d5869c48 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Wed, 14 Oct 2026 11:09:02 +0800 Subject: [PATCH 1/3] [read] Support Java numeric REST authorization casts --- crates/paimon/src/arrow/cast_width.rs | 74 +++++ crates/paimon/src/arrow/evolution_cast.rs | 219 +++++++++++++- crates/paimon/src/arrow/evolution_decimal.rs | 66 ++++- crates/paimon/src/arrow/mod.rs | 3 + crates/paimon/src/arrow/nested_evolution.rs | 18 +- crates/paimon/src/arrow/string_transform.rs | 14 +- crates/paimon/src/common/number.rs | 56 ++++ .../paimon/src/table/format_table_defaults.rs | 33 +-- crates/paimon/src/table/query_auth.rs | 2 + .../query_auth/numeric_cast_java_test.rs | 278 ++++++++++++++++++ .../paimon/src/table/query_auth/row_filter.rs | 32 +- crates/paimon/src/table/query_auth/rules.rs | 197 +++++++++---- 12 files changed, 850 insertions(+), 142 deletions(-) create mode 100644 crates/paimon/src/arrow/cast_width.rs create mode 100644 crates/paimon/src/table/query_auth/numeric_cast_java_test.rs diff --git a/crates/paimon/src/arrow/cast_width.rs b/crates/paimon/src/arrow/cast_width.rs new file mode 100644 index 000000000..940e6baa2 --- /dev/null +++ b/crates/paimon/src/arrow/cast_width.rs @@ -0,0 +1,74 @@ +// 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. + +//! Java character/byte widths shared by casts and authorization transforms. + +use arrow_array::{Array, ArrayRef}; +use std::sync::Arc; + +pub(crate) fn apply_declared_width(array: &ArrayRef, to: &crate::spec::DataType) -> ArrayRef { + use crate::spec::DataType as Paimon; + match to { + Paimon::VarChar(t) => fit_strings(array, t.length() as usize, false), + Paimon::Char(t) => fit_strings(array, t.length(), true), + Paimon::VarBinary(t) => fit_bytes(array, t.length() as usize, false), + Paimon::Binary(t) => fit_bytes(array, t.length(), true), + _ => Arc::clone(array), + } +} + +/// Counts characters, as Java does. +fn fit_strings(array: &ArrayRef, length: usize, pad: bool) -> ArrayRef { + let Some(strings) = array.as_any().downcast_ref::() else { + return Arc::clone(array); + }; + let fitted: arrow_array::StringArray = strings + .iter() + .map(|value| { + value.map(|value| { + let chars = value.chars().count(); + if chars > length { + value.chars().take(length).collect::() + } else if pad && chars < length { + format!("{value}{}", " ".repeat(length - chars)) + } else { + value.to_string() + } + }) + }) + .collect(); + Arc::new(fitted) +} + +fn fit_bytes(array: &ArrayRef, length: usize, pad: bool) -> ArrayRef { + let Some(bytes) = array.as_any().downcast_ref::() else { + return Arc::clone(array); + }; + let fitted: arrow_array::BinaryArray = bytes + .iter() + .map(|value| { + value.map(|value| { + let mut value = value.to_vec(); + if value.len() > length || pad { + value.resize(length, 0); + } + value + }) + }) + .collect(); + Arc::new(fitted) +} diff --git a/crates/paimon/src/arrow/evolution_cast.rs b/crates/paimon/src/arrow/evolution_cast.rs index 4648dd6fa..c0bd85d1b 100644 --- a/crates/paimon/src/arrow/evolution_cast.rs +++ b/crates/paimon/src/arrow/evolution_cast.rs @@ -22,11 +22,161 @@ use std::sync::Arc; use arrow_array::types::{ Decimal128Type, Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type, }; -use arrow_array::{cast::AsArray, ArrayRef, Int16Array, Int32Array, Int64Array, Int8Array}; +use arrow_array::{ + cast::AsArray, ArrayRef, BooleanArray, Float32Array, Float64Array, Int16Array, Int32Array, + Int64Array, Int8Array, +}; use arrow_schema::{DataType as ArrowType, TimeUnit}; use crate::spec::DataType; +/// Existing Python-supported numeric/Boolean CAST rules whose Java semantics +/// differ from generic Arrow casts. Shared by authorization and old-file reads. +pub(crate) fn supports_numeric_cast(from: &DataType, to: &DataType) -> bool { + let integer = |t: &DataType| { + matches!( + t, + DataType::TinyInt(_) | DataType::SmallInt(_) | DataType::Int(_) | DataType::BigInt(_) + ) + }; + let numeric = |t: &DataType| { + integer(t) + || matches!( + t, + DataType::Float(_) | DataType::Double(_) | DataType::Decimal(_) + ) + }; + let text = |t: &DataType| matches!(t, DataType::Char(_) | DataType::VarChar(_)); + numeric(from) && numeric(to) + || matches!(from, DataType::Boolean(_)) + && numeric(to) + && !matches!(to, DataType::Decimal(_)) + || text(from) && (integer(to) || matches!(to, DataType::Boolean(_) | DataType::Decimal(_))) + || integer(from) && matches!(to, DataType::Boolean(_)) + || matches!(from, DataType::Decimal(_)) && text(to) +} + +pub(crate) fn cast_numeric( + source: &ArrayRef, + from: &DataType, + to: &DataType, +) -> crate::Result> { + if !supports_numeric_cast(from, to) { + return Ok(None); + } + // Old files can store a logical STRING as LargeUtf8 or Utf8View. Resolve + // that physical representation before typed access instead of panicking. + let normalized; + let source = if matches!(from, DataType::Char(_) | DataType::VarChar(_)) + && source.data_type() != &ArrowType::Utf8 + { + normalized = arrow_cast::cast(source, &ArrowType::Utf8) + .map_err(|error| invalid_numeric(error.to_string()))?; + &normalized + } else { + source + }; + if let Some(decimal) = super::evolution_decimal::cast_decimal(source, from, to)? { + return Ok(Some(decimal)); + } + let target = super::paimon_type_to_arrow(to)?; + let text = matches!(from, DataType::Char(_) | DataType::VarChar(_)); + if text && matches!(to, DataType::Boolean(_)) { + let values = source + .as_string::() + .iter() + .map(|value| { + value + .map(|value| match value.to_ascii_lowercase().as_str() { + "t" | "true" | "y" | "yes" | "1" => Ok(true), + "f" | "false" | "n" | "no" | "0" => Ok(false), + _ => Err(invalid_numeric(format!( + "Cannot parse '{value}' as BOOLEAN" + ))), + }) + .transpose() + }) + .collect::>>()?; + return Ok(Some(Arc::new(BooleanArray::from(values)))); + } + let integral = matches!( + target, + ArrowType::Int8 | ArrowType::Int16 | ArrowType::Int32 | ArrowType::Int64 + ); + if text && integral { + let (min, max) = match target { + ArrowType::Int8 => (i64::from(i8::MIN), i64::from(i8::MAX)), + ArrowType::Int16 => (i64::from(i16::MIN), i64::from(i16::MAX)), + ArrowType::Int32 => (i64::from(i32::MIN), i64::from(i32::MAX)), + _ => (i64::MIN, i64::MAX), + }; + let values = source + .as_string::() + .iter() + .map(|value| { + value + .map(|value| crate::common::number::parse_java_integer(value, min, max)) + .transpose() + }) + .collect::>>()?; + return arrow_cast::cast(&(Arc::new(Int64Array::from(values)) as ArrayRef), &target) + .map(Some) + .map_err(|error| invalid_numeric(error.to_string())); + } + if integral && !matches!(from, DataType::Boolean(_)) { + return numeric_to_integer(source, &target).map(Some); + } + if let DataType::Decimal(decimal) = from { + if matches!(to, DataType::Char(_) | DataType::VarChar(_)) { + 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> = if decimal.precision() <= 18 { + let divisor = 10_f64.powi(decimal.scale() as i32); + source + .as_primitive::() + .iter() + .map(|value| value.map(|value| value as f64 / divisor)) + .collect() + } else { + let strings = super::evolution_decimal::decimal_to_string(source)?; + strings + .as_string::() + .iter() + .map(|value| { + value + .map(|value| { + value + .parse::() + .map_err(|_| invalid_numeric("Invalid DECIMAL value")) + }) + .transpose() + }) + .collect::>>()? + }; + 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"), + })); + } + arrow_cast::cast(source, &target) + .map(Some) + .map_err(|error| invalid_numeric(error.to_string())) +} + +fn invalid_numeric(message: impl Into) -> crate::Error { + crate::Error::DataInvalid { + message: message.into(), + source: None, + } +} + /// Java Number narrowing keeps low bits for integers. Floating-point values /// first truncate and saturate to int/long, then narrow to byte/short. DECIMAL /// truncates its fraction before BigDecimal.longValue keeps the low 64 bits. @@ -213,6 +363,73 @@ mod tests { use crate::spec::{LocalZonedTimestampType, TimestampType}; use arrow_array::{Decimal128Array, Float32Array, Float64Array}; + #[test] + fn physical_string_encodings_are_normalized_before_numeric_typed_access() { + use arrow_array::{LargeStringArray, StringViewArray}; + let from: DataType = serde_json::from_str("\"STRING\"").unwrap(); + for (target, values, expected) in [ + ( + "INT", + vec![ + Some("bad-prefix"), + Some("1.9"), + Some("-2.5"), + None, + Some("bad-suffix"), + ], + Arc::new(Int32Array::from(vec![Some(1), Some(-2), None])) as ArrayRef, + ), + ( + "BOOLEAN", + vec![ + Some("bad-prefix"), + Some("YES"), + Some("0"), + None, + Some("bad-suffix"), + ], + Arc::new(BooleanArray::from(vec![Some(true), Some(false), None])) as ArrayRef, + ), + ( + "DECIMAL(4,2)", + vec![ + Some("bad-prefix"), + Some("1.005"), + Some("99.995"), + None, + Some("bad-suffix"), + ], + Arc::new( + Decimal128Array::from(vec![Some(101), None, None]) + .with_precision_and_scale(4, 2) + .unwrap(), + ) as ArrayRef, + ), + ] { + let to: DataType = serde_json::from_value(serde_json::json!(target)).unwrap(); + for source in [ + Arc::new(LargeStringArray::from(values.clone())) as ArrayRef, + Arc::new(StringViewArray::from(values.clone())) as ArrayRef, + ] { + let result = + crate::arrow::nested_evolution::evolve_column(&source.slice(1, 3), &from, &to) + .unwrap(); + assert_eq!( + result.to_data(), + expected.to_data(), + "{target}: {:?}", + source.data_type() + ); + assert!(crate::arrow::nested_evolution::evolve_column( + &source.slice(0, 1), + &from, + &to + ) + .is_err()); + } + } + } + #[test] fn floating_point_narrowing_saturates_before_retaining_low_bits() { let values = vec![ diff --git a/crates/paimon/src/arrow/evolution_decimal.rs b/crates/paimon/src/arrow/evolution_decimal.rs index 548bf37a6..67c274611 100644 --- a/crates/paimon/src/arrow/evolution_decimal.rs +++ b/crates/paimon/src/arrow/evolution_decimal.rs @@ -21,8 +21,8 @@ use std::sync::Arc; use arrow_array::{ cast::AsArray, - types::{Float32Type, Float64Type}, - ArrayRef, Decimal128Array, + types::{Float32Type, Float64Type, Int64Type}, + ArrayRef, Decimal128Array, StringArray, }; use crate::common::number::java_double_decimal_text; @@ -61,6 +61,23 @@ pub(super) fn cast_decimal( .iter() .map(|v| v.map(|v| decimal_text(v, precision, scale)).transpose()) .collect::>>()?, + DataType::Decimal(_) => decimal_to_string(source)? + .as_string::() + .iter() + .map(|v| v.map(|v| decimal_text(v, precision, scale)).transpose()) + .collect::>>()?, + DataType::TinyInt(_) | DataType::SmallInt(_) | DataType::Int(_) | DataType::BigInt(_) => { + let integers = arrow_cast::cast(source, &arrow_schema::DataType::Int64) + .map_err(|error| invalid(error.to_string()))?; + integers + .as_primitive::() + .iter() + .map(|v| { + v.map(|v| decimal_text(&v.to_string(), precision, scale)) + .transpose() + }) + .collect::>>()? + } _ => return Ok(None), }; let values = values.into_iter().map(Option::flatten); @@ -70,6 +87,36 @@ pub(super) fn cast_decimal( Ok(Some(Arc::new(result))) } +/// Decimal.toString delegates to BigDecimal.toPlainString, including scaled zero. +pub(super) fn decimal_to_string(source: &ArrayRef) -> crate::Result { + let arrow_schema::DataType::Decimal128(_, scale) = source.data_type() else { + return Err(invalid("Expected DECIMAL array")); + }; + let scale = u32::try_from(*scale).map_err(|_| invalid("Invalid DECIMAL scale"))?; + let divisor = 10_u128.pow(scale); + Ok(Arc::new(StringArray::from_iter( + source + .as_primitive::() + .iter() + .map(|value| { + value.map(|value| { + let negative = if value < 0 { "-" } else { "" }; + let value = value.unsigned_abs(); + if scale == 0 { + format!("{negative}{value}") + } else { + format!( + "{negative}{}.{:0width$}", + value / divisor, + value % divisor, + width = scale as usize + ) + } + }) + }), + ))) +} + fn float_decimal(value: f64, precision: u32, scale: u32) -> crate::Result> { if !value.is_finite() { return Err(invalid("Non-finite decimal schema-evolution value")); @@ -81,6 +128,21 @@ fn float_decimal(value: f64, precision: u32, scale: u32) -> crate::Result