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 eaa64bc0 [core] Align numeric narrowing and timestamp schema evolution
with Java (#1102)
eaa64bc0 is described below
commit eaa64bc0974fb091bb7f00b5ff2eee08a7f1cae3
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 11 16:57:39 2026 +0800
[core] Align numeric narrowing and timestamp schema evolution with Java
(#1102)
---
crates/paimon/src/arrow/evolution_cast.rs | 426 ++++++++++++++++++++++
crates/paimon/src/arrow/mod.rs | 1 +
crates/paimon/src/arrow/nested_evolution.rs | 64 ++++
crates/paimon/src/file_index/evaluator.rs | 4 +-
crates/paimon/src/predicate_stats.rs | 117 ++++++
crates/paimon/src/table/data_file_reader.rs | 49 ++-
crates/paimon/tests/schema_evolution_cast_test.rs | 180 +++++++++
7 files changed, 826 insertions(+), 15 deletions(-)
diff --git a/crates/paimon/src/arrow/evolution_cast.rs
b/crates/paimon/src/arrow/evolution_cast.rs
new file mode 100644
index 00000000..5a980386
--- /dev/null
+++ b/crates/paimon/src/arrow/evolution_cast.rs
@@ -0,0 +1,426 @@
+// 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 conversions which Arrow's schema-evolution cast cannot represent.
+
+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_schema::{DataType as ArrowType, TimeUnit};
+
+use crate::spec::DataType;
+
+/// 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.
+pub(super) fn numeric_to_integer(source: &ArrayRef, target: &ArrowType) ->
crate::Result<ArrayRef> {
+ macro_rules! values {
+ ($ty:ty, $convert:expr) => {
+ source
+ .as_primitive::<$ty>()
+ .iter()
+ .map(|value| value.map($convert))
+ .collect::<Vec<Option<i64>>>()
+ };
+ }
+ let wide = matches!(target, ArrowType::Int64);
+ let values = match source.data_type() {
+ ArrowType::Int8 => values!(Int8Type, |v| v as i64),
+ ArrowType::Int16 => values!(Int16Type, |v| v as i64),
+ ArrowType::Int32 => values!(Int32Type, |v| v as i64),
+ ArrowType::Int64 => values!(Int64Type, |v| v),
+ ArrowType::Float32 => values!(Float32Type, |v| if wide {
+ v as i64
+ } else {
+ v as i32 as i64
+ }),
+ ArrowType::Float64 => values!(Float64Type, |v| if wide {
+ v as i64
+ } else {
+ v as i32 as i64
+ }),
+ ArrowType::Decimal128(_, scale) => {
+ let divisor = 10_i128.pow(*scale as u32);
+ values!(Decimal128Type, |v| (v / divisor) as i64)
+ }
+ other => {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Invalid numeric schema-evolution input
{other:?}"),
+ source: None,
+ })
+ }
+ };
+ let result: ArrayRef = match target {
+ ArrowType::Int8 => Arc::new(Int8Array::from_iter(
+ values.into_iter().map(|v| v.map(|v| v as i8)),
+ )),
+ ArrowType::Int16 => Arc::new(Int16Array::from_iter(
+ values.into_iter().map(|v| v.map(|v| v as i16)),
+ )),
+ ArrowType::Int32 => Arc::new(Int32Array::from_iter(
+ values.into_iter().map(|v| v.map(|v| v as i32)),
+ )),
+ ArrowType::Int64 => Arc::new(Int64Array::from(values)),
+ _ => unreachable!("validated integral target"),
+ };
+ Ok(result)
+}
+
+/// Match DateTimeUtils.truncate for the same timestamp kind. Work in i128 so
+/// flooring a pre-epoch value cannot overflow the source's i64 unit before
+/// converting to a coarser target unit. Cross-kind timezone casts are handled
+/// by the caller's existing cast path.
+pub(super) fn timestamp_precision(
+ source: &ArrayRef,
+ source_type: &DataType,
+ target_type: &DataType,
+ target_arrow: &ArrowType,
+) -> crate::Result<Option<ArrayRef>> {
+ let precision = match (source_type, target_type) {
+ (DataType::Timestamp(from), DataType::Timestamp(to))
+ if from.precision() > to.precision() =>
+ {
+ to.precision()
+ }
+ (DataType::LocalZonedTimestamp(from),
DataType::LocalZonedTimestamp(to))
+ if from.precision() > to.precision() =>
+ {
+ to.precision()
+ }
+ _ => return Ok(None),
+ };
+ let nanos_per_tick = |unit: &TimeUnit| -> i128 {
+ match unit {
+ TimeUnit::Second => 1_000_000_000,
+ TimeUnit::Millisecond => 1_000_000,
+ TimeUnit::Microsecond => 1_000,
+ TimeUnit::Nanosecond => 1,
+ }
+ };
+ let ArrowType::Timestamp(source_unit, _) = source.data_type() else {
+ return Err(crate::Error::DataInvalid {
+ message: "Expected timestamp schema-evolution input".into(),
+ source: None,
+ });
+ };
+ let ArrowType::Timestamp(target_unit, _) = target_arrow else {
+ unreachable!()
+ };
+ let raw = arrow_cast::cast(source, &ArrowType::Int64).map_err(cast_error)?;
+ let quantum = 10_i128.pow(9 - precision);
+ let ticks =
Int64Array::from_iter(raw.as_primitive::<Int64Type>().iter().map(|value| {
+ value.and_then(|value| {
+ let nanos = i128::from(value) * nanos_per_tick(source_unit);
+ let truncated = nanos.div_euclid(quantum) * quantum;
+ i64::try_from(truncated / nanos_per_tick(target_unit)).ok()
+ })
+ }));
+ Ok(Some(
+ arrow_cast::cast(&(Arc::new(ticks) as ArrayRef),
target_arrow).map_err(cast_error)?,
+ ))
+}
+
+fn cast_error(error: arrow_schema::ArrowError) -> crate::Error {
+ crate::Error::UnexpectedError {
+ message: "Failed to cast timestamp during schema evolution".into(),
+ source: Some(Box::new(error)),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{LocalZonedTimestampType, TimestampType};
+ use arrow_array::{Decimal128Array, Float32Array, Float64Array};
+
+ #[test]
+ fn floating_point_narrowing_saturates_before_retaining_low_bits() {
+ let values = vec![
+ Some(128.9),
+ Some(-129.9),
+ Some(32768.9),
+ Some(-32769.9),
+ Some(1e30),
+ Some(-1e30),
+ None,
+ ];
+ for source in [
+ Arc::new(Float64Array::from(values.clone())) as ArrayRef,
+ Arc::new(Float32Array::from_iter(
+ values.iter().map(|v| v.map(|v| v as f32)),
+ )) as ArrayRef,
+ ] {
+ for (target, expected) in [
+ (
+ ArrowType::Int8,
+ vec![
+ Some(-128),
+ Some(127),
+ Some(0),
+ Some(-1),
+ Some(-1),
+ Some(0),
+ None,
+ ],
+ ),
+ (
+ ArrowType::Int16,
+ vec![
+ Some(128),
+ Some(-129),
+ Some(-32768),
+ Some(32767),
+ Some(-1),
+ Some(0),
+ None,
+ ],
+ ),
+ (
+ ArrowType::Int32,
+ vec![
+ Some(128),
+ Some(-129),
+ Some(32768),
+ Some(-32769),
+ Some(i32::MAX as i64),
+ Some(i32::MIN as i64),
+ None,
+ ],
+ ),
+ (
+ ArrowType::Int64,
+ vec![
+ Some(128),
+ Some(-129),
+ Some(32768),
+ Some(-32769),
+ Some(i64::MAX),
+ Some(i64::MIN),
+ None,
+ ],
+ ),
+ ] {
+ let output = numeric_to_integer(&source.slice(0,
values.len()), &target).unwrap();
+ let widened = arrow_cast::cast(&output,
&ArrowType::Int64).unwrap();
+ assert_eq!(
+ widened.as_primitive::<Int64Type>(),
+ &Int64Array::from(expected)
+ );
+ }
+ }
+ }
+
+ #[test]
+ fn decimal_integrals_truncate_then_wrap_at_each_target_width() {
+ let source: ArrayRef = Arc::new(
+ Decimal128Array::from(vec![
+ Some(12_899),
+ Some(-12_999),
+ Some(922_337_203_685_477_580_800),
+ Some(-922_337_203_685_477_580_900),
+ Some(99_999_999_999_999_999_999_999_999_999_999_999_999),
+ None,
+ ])
+ .with_precision_and_scale(38, 2)
+ .unwrap(),
+ );
+ // BigDecimal.longValue of the largest DECIMAL(38,2), after truncation.
+ let values = [
+ Some(128_i64),
+ Some(-129),
+ Some(i64::MIN),
+ Some(i64::MAX),
+ Some(-5_527_149_226_598_858_753),
+ None,
+ ];
+ for target in [
+ ArrowType::Int8,
+ ArrowType::Int16,
+ ArrowType::Int32,
+ ArrowType::Int64,
+ ] {
+ let output = numeric_to_integer(&source, &target).unwrap();
+ let expected = values
+ .iter()
+ .map(|value| {
+ value.map(|value| match target {
+ ArrowType::Int8 => value as i8 as i64,
+ ArrowType::Int16 => value as i16 as i64,
+ ArrowType::Int32 => value as i32 as i64,
+ _ => value,
+ })
+ })
+ .collect::<Vec<_>>();
+ let widened = arrow_cast::cast(&output,
&ArrowType::Int64).unwrap();
+ assert_eq!(
+ widened.as_primitive::<Int64Type>(),
+ &Int64Array::from(expected)
+ );
+ }
+ }
+
+ #[test]
+ fn timestamp_precision_covers_every_unit_and_pre_epoch_flooring() {
+ for zoned in [false, true] {
+ let paimon_type = |precision| {
+ if zoned {
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(precision).unwrap())
+ } else {
+ DataType::Timestamp(TimestampType::new(precision).unwrap())
+ }
+ };
+ let source_type = paimon_type(9);
+ let source_arrow =
crate::arrow::paimon_type_to_arrow(&source_type).unwrap();
+ let raw: ArrayRef = Arc::new(Int64Array::from(vec![
+ Some(123_456_789),
+ Some(-1),
+ Some(-123_456_789),
+ Some(i64::MIN),
+ None,
+ ]));
+ let source = arrow_cast::cast(&raw, &source_arrow).unwrap();
+ for precision in 0..9 {
+ let target = paimon_type(precision);
+ let arrow_type =
crate::arrow::paimon_type_to_arrow(&target).unwrap();
+ let output = timestamp_precision(&source, &source_type,
&target, &arrow_type)
+ .unwrap()
+ .unwrap();
+ assert_eq!(output.data_type(), &arrow_type);
+ let divisor = 10_i128.pow(9 - precision);
+ let quantum_in_target = match precision {
+ 0 => 1,
+ 1..=3 => 10_i128.pow(3 - precision),
+ 4..=6 => 10_i128.pow(6 - precision),
+ _ => 10_i128.pow(9 - precision),
+ };
+ let expected = raw
+ .as_primitive::<Int64Type>()
+ .iter()
+ .map(|v| {
+ v.and_then(|v| {
+ i64::try_from(i128::from(v).div_euclid(divisor) *
quantum_in_target)
+ .ok()
+ })
+ })
+ .collect::<Vec<_>>();
+ let widened = arrow_cast::cast(&output,
&ArrowType::Int64).unwrap();
+ assert_eq!(
+ widened.as_primitive::<Int64Type>(),
+ &Int64Array::from(expected),
+ "precision {precision}"
+ );
+ }
+ assert!(
+ timestamp_precision(&source, &source_type, &source_type,
&source_arrow)
+ .unwrap()
+ .is_none()
+ );
+ }
+ }
+}
+
+#[cfg(test)]
+mod integer_tests {
+ use super::*;
+
+ #[test]
+ fn every_signed_integer_width_preserves_nulls_and_java_narrowing() {
+ let sources: Vec<ArrayRef> = vec![
+ Arc::new(Int8Array::from(vec![
+ Some(i8::MIN),
+ Some(i8::MAX),
+ Some(-1),
+ None,
+ ])),
+ Arc::new(Int16Array::from(vec![
+ Some(i16::MIN),
+ Some(i16::MAX),
+ Some(-129),
+ None,
+ ])),
+ Arc::new(Int32Array::from(vec![
+ Some(i32::MIN),
+ Some(i32::MAX),
+ Some(32768),
+ None,
+ ])),
+ Arc::new(Int64Array::from(vec![
+ Some(i64::MIN),
+ Some(i64::MAX),
+ Some(2147483648),
+ None,
+ ])),
+ ];
+ let expected: [[Vec<Option<i64>>; 4]; 4] = [
+ [
+ vec![Some(-128), Some(127), Some(-1), None],
+ vec![Some(-128), Some(127), Some(-1), None],
+ vec![Some(-128), Some(127), Some(-1), None],
+ vec![Some(-128), Some(127), Some(-1), None],
+ ],
+ [
+ vec![Some(0), Some(-1), Some(127), None],
+ vec![Some(-32768), Some(32767), Some(-129), None],
+ vec![Some(-32768), Some(32767), Some(-129), None],
+ vec![Some(-32768), Some(32767), Some(-129), None],
+ ],
+ [
+ vec![Some(0), Some(-1), Some(0), None],
+ vec![Some(0), Some(-1), Some(-32768), None],
+ vec![
+ Some(i32::MIN as i64),
+ Some(i32::MAX as i64),
+ Some(32768),
+ None,
+ ],
+ vec![
+ Some(i32::MIN as i64),
+ Some(i32::MAX as i64),
+ Some(32768),
+ None,
+ ],
+ ],
+ [
+ vec![Some(0), Some(-1), Some(0), None],
+ vec![Some(0), Some(-1), Some(0), None],
+ vec![Some(0), Some(-1), Some(i32::MIN as i64), None],
+ vec![Some(i64::MIN), Some(i64::MAX), Some(2147483648), None],
+ ],
+ ];
+ for (source, expected) in sources.into_iter().zip(expected) {
+ for (target, expected) in [
+ ArrowType::Int8,
+ ArrowType::Int16,
+ ArrowType::Int32,
+ ArrowType::Int64,
+ ]
+ .iter()
+ .zip(expected)
+ {
+ let output = numeric_to_integer(&source, target).unwrap();
+ let widened = arrow_cast::cast(&output,
&ArrowType::Int64).unwrap();
+ assert_eq!(
+ widened.as_primitive::<Int64Type>(),
+ &Int64Array::from(expected)
+ );
+ }
+ }
+ }
+}
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index 286580e5..2ef79ced 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+mod evolution_cast;
pub(crate) mod filtering;
pub(crate) mod format;
pub(crate) mod nested_evolution;
diff --git a/crates/paimon/src/arrow/nested_evolution.rs
b/crates/paimon/src/arrow/nested_evolution.rs
index ffcc7bae..e3fad1d3 100644
--- a/crates/paimon/src/arrow/nested_evolution.rs
+++ b/crates/paimon/src/arrow/nested_evolution.rs
@@ -221,6 +221,27 @@ pub(crate) fn evolve_column(
}
}
+ if matches!(
+ target_type,
+ DataType::TinyInt(_) | DataType::SmallInt(_) | DataType::Int(_) |
DataType::BigInt(_)
+ ) && matches!(
+ source_type,
+ DataType::TinyInt(_)
+ | DataType::SmallInt(_)
+ | DataType::Int(_)
+ | DataType::BigInt(_)
+ | DataType::Float(_)
+ | DataType::Double(_)
+ | DataType::Decimal(_)
+ ) {
+ return super::evolution_cast::numeric_to_integer(source,
&target_arrow);
+ }
+ if let Some(result) =
+ super::evolution_cast::timestamp_precision(source, source_type,
target_type, &target_arrow)?
+ {
+ return Ok(result);
+ }
+
cast(source, &target_arrow).map_err(|e| crate::Error::UnexpectedError {
message: format!(
"failed to cast nested value from {:?} to {:?} during schema
evolution",
@@ -1033,6 +1054,49 @@ mod tests {
assert_eq!(strings(&out).value(0), "{k -> 7}");
}
+ #[test]
+ fn narrowing_integers_keeps_low_bits_like_java() {
+ let source: ArrayRef = Arc::new(Int64Array::from(vec![
+ Some(128),
+ Some(-129),
+ Some(i64::MAX),
+ Some(i64::MIN),
+ None,
+ ]));
+ let target = DataType::TinyInt(crate::spec::TinyIntType::new());
+ let output = evolve_column(&source,
&DataType::BigInt(BigIntType::new()), &target).unwrap();
+ assert_eq!(
+ output
+ .as_any()
+ .downcast_ref::<arrow_array::Int8Array>()
+ .unwrap(),
+ &arrow_array::Int8Array::from(vec![Some(-128), Some(127),
Some(-1), Some(0), None]),
+ );
+ }
+
+ #[test]
+ fn timestamp_precision_is_truncated_within_the_same_arrow_unit() {
+ let source: ArrayRef = Arc::new(TimestampMillisecondArray::from(vec![
+ Some(123),
+ Some(-1),
+ Some(-123),
+ None,
+ ]));
+ let output = evolve_column(
+ &source,
+ &DataType::Timestamp(TimestampType::new(3).unwrap()),
+ &DataType::Timestamp(TimestampType::new(1).unwrap()),
+ )
+ .unwrap();
+ assert_eq!(
+ output
+ .as_any()
+ .downcast_ref::<TimestampMillisecondArray>()
+ .unwrap(),
+ &TimestampMillisecondArray::from(vec![Some(100), Some(-100),
Some(-200), None]),
+ );
+ }
+
#[test]
fn reducing_decimal_scale_rounds_half_up_for_both_signs() {
let source: ArrayRef = Arc::new(
diff --git a/crates/paimon/src/file_index/evaluator.rs
b/crates/paimon/src/file_index/evaluator.rs
index 51c5e82a..8fdd9bfc 100644
--- a/crates/paimon/src/file_index/evaluator.rs
+++ b/crates/paimon/src/file_index/evaluator.rs
@@ -186,8 +186,8 @@ fn devolve_literals(
if same_type_ignoring_nullability(table_type, data_type) {
return Some(literals.to_vec());
}
- // Narrowing can turn non-null file values into NULL, so reject it even for
- // predicates without literals, such as IS NULL and IS NOT NULL.
+ // Integer narrowing can wrap values, so it is outside Java's evolution
+ // whitelist, including predicates without literals such as IS NULL.
if integer_width(table_type)? <= integer_width(data_type)? {
return None;
}
diff --git a/crates/paimon/src/predicate_stats.rs
b/crates/paimon/src/predicate_stats.rs
index a111a658..49e86ac6 100644
--- a/crates/paimon/src/predicate_stats.rs
+++ b/crates/paimon/src/predicate_stats.rs
@@ -53,6 +53,9 @@ pub(crate) fn data_leaf_may_match<T: StatsAccessor>(
if row_count <= 0 {
return false;
}
+ if !stats_evolution_is_safe(stats_data_type, predicate_data_type) {
+ return true;
+ }
let null_count = stats.null_count(index);
let all_null = null_count.map(|count| count == row_count);
@@ -272,6 +275,9 @@ pub(crate) fn data_leaf_must_match<T: StatsAccessor>(
if row_count <= 0 {
return false;
}
+ if !stats_evolution_is_safe(stats_data_type, predicate_data_type) {
+ return false;
+ }
let null_count = stats.null_count(index);
match op {
@@ -568,6 +574,24 @@ fn predicate_must_match_with_schema<T: StatsAccessor>(
}
}
+fn stats_evolution_is_safe(source: &DataType, target: &DataType) -> bool {
+ // Java's CastExecutors.castLiteralsWithEvolution permits unchanged types
+ // and integer widening. Narrowing can wrap, and timestamp precision
changes
+ // alter values even when their Arrow storage types are identical. Neither
+ // the original bounds nor null counts can prove a match after unsafe
casts.
+ source.equals_ignore_nullable(target)
+ || matches!(
+ (source, target),
+ (
+ DataType::TinyInt(_),
+ DataType::SmallInt(_) | DataType::Int(_) | DataType::BigInt(_)
+ ) | (
+ DataType::SmallInt(_),
+ DataType::Int(_) | DataType::BigInt(_)
+ ) | (DataType::Int(_), DataType::BigInt(_))
+ )
+}
+
fn coerce_stats_datum_for_predicate(datum: Datum, predicate_data_type:
&DataType) -> Option<Datum> {
match (datum, predicate_data_type) {
(datum @ Datum::Bool(_), DataType::Boolean(_))
@@ -783,6 +807,99 @@ mod tests {
nullable_int_stats(Some(min), Some(max), Some(0))
}
+ #[test]
+ fn timestamp_precision_changes_make_bounds_and_null_counts_unknown() {
+ use crate::spec::{LocalZonedTimestampType, TimestampType};
+
+ for (source, target, value) in [
+ (
+ DataType::Timestamp(TimestampType::new(3).unwrap()),
+ DataType::Timestamp(TimestampType::new(1).unwrap()),
+ Datum::Timestamp {
+ millis: 123,
+ nanos: 0,
+ },
+ ),
+ (
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()),
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(1).unwrap()),
+ Datum::LocalZonedTimestamp {
+ millis: 123,
+ nanos: 0,
+ },
+ ),
+ ] {
+ let stats = MockStats {
+ row_count: 1,
+ null_count: Some(0),
+ min: Some(value.clone()),
+ max: Some(value.clone()),
+ };
+ // Even a predicate which matches the *old* bounds cannot establish
+ // must-match: NOT would then wrongly prune the converted value.
+ for op in [
+ PredicateOperator::Eq,
+ PredicateOperator::NotEq,
+ PredicateOperator::Lt,
+ PredicateOperator::Gt,
+ PredicateOperator::Between,
+ PredicateOperator::NotBetween,
+ PredicateOperator::IsNull,
+ PredicateOperator::IsNotNull,
+ ] {
+ let literals = [value.clone(), value.clone()];
+ assert!(data_leaf_may_match(
+ 0, &source, &target, op, &literals, &stats
+ ));
+ assert!(!data_leaf_must_match(
+ 0, &source, &target, op, &literals, &stats
+ ));
+ }
+ }
+ }
+
+ #[test]
+ fn integer_widening_preserves_stats_pruning_but_narrowing_is_unknown() {
+ use crate::spec::{BigIntType, TinyIntType};
+
+ let source = DataType::Int(IntType::new());
+ let stats = int_stats(128, 128);
+ let widened = DataType::BigInt(BigIntType::new());
+ assert!(!data_leaf_may_match(
+ 0,
+ &source,
+ &widened,
+ PredicateOperator::Eq,
+ &[Datum::Long(100)],
+ &stats
+ ));
+ assert!(data_leaf_must_match(
+ 0,
+ &source,
+ &widened,
+ PredicateOperator::Eq,
+ &[Datum::Long(128)],
+ &stats
+ ));
+ let narrowed = DataType::TinyInt(TinyIntType::new());
+ assert!(data_leaf_may_match(
+ 0,
+ &source,
+ &narrowed,
+ PredicateOperator::Eq,
+ &[Datum::TinyInt(-128)],
+ &stats
+ ));
+ assert!(!data_leaf_must_match(
+ 0,
+ &source,
+ &narrowed,
+ PredicateOperator::IsNotNull,
+ &[],
+ &stats
+ ));
+ }
+
fn nullable_int_stats(
min: Option<i32>,
max: Option<i32>,
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 1b4ac987..d1425d16 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -539,18 +539,40 @@ impl DataFileReader {
for predicate in &predicates {
crate::arrow::residual::collect_predicate_leaf_refs(predicate,
&mut refs);
}
- let metadata_residual = (projects_row_id
- && predicates
- .iter()
- .any(|predicate| !matches!(predicate, Predicate::AlwaysTrue)))
+ // A predicate's literals use the current table type. Casting a
physical
+ // old-file column inside the decoder can change Java narrowing or
+ // timestamp precision semantics, and filtering there loses those rows
+ // before schema reconciliation. Evaluate the complete predicate after
+ // reconciliation, as Java's schema-evolution reader does.
+ let evolved_predicate = data_fields.as_deref().is_some_and(|fields| {
+ refs.iter().any(|(name, index)| {
+ if crate::spec::is_row_tracking_column(name) {
+ return false;
+ }
+ table_fields.get(*index).is_some_and(|target| {
+ fields
+ .iter()
+ .find(|source| source.id() == target.id())
+ .is_none_or(|source| {
+ !source
+ .data_type()
+ .equals_ignore_nullable(target.data_type())
+ })
+ })
+ })
+ });
+ let post_read_filter = evolved_predicate
+ || (projects_row_id
+ && predicates
+ .iter()
+ .any(|predicate| !matches!(predicate,
Predicate::AlwaysTrue)))
|| (row_tracking_enabled
&& refs
.iter()
.any(|(name, _)| *name ==
crate::spec::SEQUENCE_NUMBER_FIELD_NAME));
- if metadata_residual {
- // Decoder-side filtering can drop rows before positional `_ROW_ID`
- // attachment. Read predicate-only columns without pushing the
filter
- // down, attach the original ids, then evaluate the exact
predicate.
+ if post_read_filter {
+ // Read predicate-only columns without decoder filtering. Reconcile
+ // old-file types and attach original row IDs before exact
filtering.
for (name, index) in refs {
if name == crate::spec::SEQUENCE_NUMBER_FIELD_NAME {
crate::arrow::residual::push_unique_scan_field(
@@ -572,7 +594,7 @@ impl DataFileReader {
let projects_row_id = read_type
.iter()
.any(|field| field.name() == ROW_ID_FIELD_NAME);
- let residual_predicates = metadata_residual.then(||
crate::arrow::format::FilePredicates {
+ let residual_predicates = post_read_filter.then(||
crate::arrow::format::FilePredicates {
predicates: predicates.clone(),
row_filter_factory: None,
file_fields: table_fields.clone(),
@@ -582,7 +604,7 @@ impl DataFileReader {
// post-filter until expression adaptation is proven for that path.
// Positional `_ROW_ID` materialization must see the unfiltered row
// stream, so the engine hook stays disabled for this projection.
- let row_filter_factory = (data_fields.is_none() && !projects_row_id &&
!metadata_residual)
+ let row_filter_factory = (data_fields.is_none() && !projects_row_id &&
!post_read_filter)
.then(|| self.row_filter_factory.clone())
.flatten();
let file_io = self.file_io.clone();
@@ -629,7 +651,7 @@ impl DataFileReader {
mosaic_prefetch,
)?;
// Remap predicates from table-level to file-level indices.
- let file_predicates = if metadata_residual {
+ let file_predicates = if post_read_filter {
None
} else {
let remapped = crate::arrow::filtering::remap_predicates_to_file(
@@ -3297,11 +3319,12 @@ mod tests {
)
.unwrap();
for (case, values, expected_values) in [
- ("overflow_only", vec![Some(383)], vec![None]),
+ // Java's Number.byteValue keeps the low bits: 383 becomes 127.
+ ("overflow_only", vec![Some(383)], vec![Some(127)]),
(
"mixed",
vec![Some(127), Some(383), None],
- vec![Some(127), None, None],
+ vec![Some(127), Some(127), None],
),
] {
let mut avro_writer = apache_avro::Writer::new(&avro_schema,
Vec::new());
diff --git a/crates/paimon/tests/schema_evolution_cast_test.rs
b/crates/paimon/tests/schema_evolution_cast_test.rs
new file mode 100644
index 00000000..6b3b63f5
--- /dev/null
+++ b/crates/paimon/tests/schema_evolution_cast_test.rs
@@ -0,0 +1,180 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use std::sync::Arc;
+
+use arrow_array::{
+ ArrayRef, Int32Array, Int64Array, Int8Array, RecordBatch,
TimestampMillisecondArray,
+};
+use futures::TryStreamExt;
+use paimon::spec::{
+ BigIntType, DataType, Datum, IntType, Predicate, PredicateBuilder, Schema,
SchemaChange,
+ TableSchema, TimestampType, TinyIntType,
+};
+use paimon::table::Table;
+
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs, write_batch};
+
+async fn ids(table: &Table, predicate: Predicate) -> Vec<i32> {
+ let mut builder = table.new_read_builder();
+ builder
+ .with_filter(predicate)
+ .with_projection(&["id"])
+ .unwrap();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let mut ids = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .iter()
+ .copied()
+ })
+ .collect::<Vec<_>>();
+ ids.sort_unstable();
+ ids
+}
+
+#[tokio::test]
+async fn
converted_predicates_filter_after_schema_evolution_for_every_table_mode() {
+ for mode in ["append", "pk", "de"] {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("v", DataType::BigInt(BigIntType::new()))
+ .column("t", DataType::Timestamp(TimestampType::new(3).unwrap()))
+ .option("read.batch-size", "1");
+ if mode == "pk" {
+ schema = schema.primary_key(["id"]).option("bucket", "1");
+ }
+ if mode == "de" {
+ schema = schema
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ let (io, old) = memory_table(
+ &format!("memory:/evolution_cast_{mode}"),
+ TableSchema::new(0, &schema.build().unwrap()),
+ );
+ setup_dirs(&io, old.location()).await;
+ persist_table_schema(&io, old.location(), old.schema()).await;
+ write_batch(
+ &old,
+ &RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef),
+ (
+ "v",
+ Arc::new(Int64Array::from(vec![Some(128), Some(-129),
None])) as ArrayRef,
+ ),
+ (
+ "t",
+ Arc::new(TimestampMillisecondArray::from(vec![
+ Some(123),
+ Some(-1),
+ None,
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await;
+ // A single-value file's bounds must not prune values after truncation
+ // or wrapping. A wide old-file range can hide this stats regression.
+ write_batch(
+ &old,
+ &RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![5])) as ArrayRef),
+ ("v", Arc::new(Int64Array::from(vec![128])) as ArrayRef),
+ (
+ "t",
+ Arc::new(TimestampMillisecondArray::from(vec![123])) as
ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await;
+ let schema = old
+ .schema()
+ .apply_changes(vec![
+ SchemaChange::update_column_type("v".into(),
DataType::TinyInt(TinyIntType::new())),
+ SchemaChange::update_column_type(
+ "t".into(),
+ DataType::Timestamp(TimestampType::new(1).unwrap()),
+ ),
+ ])
+ .unwrap();
+ persist_table_schema(&io, old.location(), &schema).await;
+ let current = old.copy_with_resolved_schema(schema, "main").unwrap();
+ write_batch(
+ ¤t,
+ &RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![4])) as ArrayRef),
+ ("v", Arc::new(Int8Array::from(vec![7])) as ArrayRef),
+ (
+ "t",
+ Arc::new(TimestampMillisecondArray::from(vec![500])) as
ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await;
+ let pb = PredicateBuilder::new(current.schema().fields());
+ let narrowed = pb.equal("v", Datum::TinyInt(-128)).unwrap();
+ let truncated = pb
+ .equal(
+ "t",
+ Datum::Timestamp {
+ millis: 100,
+ nanos: 0,
+ },
+ )
+ .unwrap();
+ assert_eq!(
+ ids(¤t, narrowed.clone()).await,
+ vec![1, 5],
+ "{mode} numeric"
+ );
+ assert_eq!(
+ ids(¤t, truncated.clone()).await,
+ vec![1, 5],
+ "{mode} timestamp"
+ );
+ assert_eq!(
+ ids(¤t, Predicate::and(vec![narrowed, truncated])).await,
+ vec![1, 5],
+ "{mode} compound"
+ );
+ assert_eq!(
+ ids(¤t, pb.is_null("v").unwrap()).await,
+ vec![3],
+ "{mode} NULL"
+ );
+ }
+}
