Complete Java-compatible schema evolution for native reads

bindings/python/python/pypaimon_rust/data.pyi

@@ -19,6 +19,20 @@ from typing import Sequence
 
 import pyarrow
 
+def read_arrow_schema(fields: str) -> pyarrow.Schema:
+    """Build the runtime read schema from Paimon fields (JSON)."""
+    ...
+
+def normalize_array_for_write(column: pyarrow.Array, target_type: pyarrow.DataType) -> pyarrow.Array:
+    """Normalize a write layout and validate visible NOT NULL children in core."""
+    ...
+
+def cast_array_for_schema_evolution(
+    column: pyarrow.Array, source_type: str, target_type: str
+) -> pyarrow.Array:
+    """Cast historical Paimon types (JSON) using the core table-reader rules."""
+    ...
+
 def variant_get_numeric_fields(
     column: pyarrow.Array, fields: Sequence[str]
 ) -> pyarrow.FixedSizeListArray:

bindings/python/src/evolution.rs

@@ -0,0 +1,83 @@
+// 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 arrow::array::{make_array, Array, ArrayData};
+use arrow::pyarrow::{FromPyArrow, ToPyArrow};
+use paimon::spec::{DataField, DataType};
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::*;
+
+use crate::error::to_py_err;
+
+/// Arrow and logical-type transport only; evolution belongs to Rust core.
+#[pyfunction]
+fn cast_array_for_schema_evolution(
+    py: Python<'_>,
+    column: &Bound<'_, PyAny>,
+    source_type: &str,
+    target_type: &str,
+) -> PyResult<Py<PyAny>> {
+    let parse = |value| -> PyResult<DataType> {
+        serde_json::from_str(value)
+            .map_err(|error| PyValueError::new_err(format!("Invalid Paimon type: {error}")))
+    };
+    let source = parse(source_type)?;
+    let target = parse(target_type)?;
+    let data = ArrayData::from_pyarrow_bound(column)?;
+    data.validate_full().map_err(|error| {
+        PyValueError::new_err(format!("Invalid Arrow evolution input: {error}"))
+    })?;
+    let column = make_array(data);
+    let output = py
+        .detach(|| paimon::arrow::cast_array_for_schema_evolution(&column, &source, &target))
+        .map_err(to_py_err)?;
+    Ok(output.to_data().to_pyarrow(py)?.unbind())
+}
+
+/// Transport the core runtime schema; Python does not duplicate read nullability.
+#[pyfunction]
+fn read_arrow_schema(py: Python<'_>, fields: &str) -> PyResult<Py<PyAny>> {
+    let fields: Vec<DataField> = serde_json::from_str(fields)
+        .map_err(|error| PyValueError::new_err(format!("Invalid Paimon fields: {error}")))?;
+    let schema = paimon::arrow::build_read_arrow_schema(&fields).map_err(to_py_err)?;
+    Ok(schema.as_ref().to_pyarrow(py)?.unbind())
+}
+
+/// Reuse the writer's existing normalization and value-sensitive constraints.
+#[pyfunction]
+fn normalize_array_for_write(
+    py: Python<'_>,
+    column: &Bound<'_, PyAny>,
+    target_type: &Bound<'_, PyAny>,
+) -> PyResult<Py<PyAny>> {
+    let target = arrow::datatypes::DataType::from_pyarrow_bound(target_type)?;
+    let data = ArrayData::from_pyarrow_bound(column)?;
+    data.validate_full()
+        .map_err(|error| PyValueError::new_err(format!("Invalid Arrow write input: {error}")))?;
+    let column = make_array(data);
+    let output = py
+        .detach(|| paimon::table::normalize_write_array(&column, &target))
+        .map_err(to_py_err)?;
+    Ok(output.to_data().to_pyarrow(py)?.unbind())
+}
+
+pub(crate) fn register_module(m: &Bound<'_, PyModule>) -> PyResult<()> {
+    let data = m.getattr("data")?.cast_into::<PyModule>()?;
+    data.add_function(wrap_pyfunction!(cast_array_for_schema_evolution, &data)?)?;
+    data.add_function(wrap_pyfunction!(read_arrow_schema, &data)?)?;
+    data.add_function(wrap_pyfunction!(normalize_array_for_write, &data)?)
+}

bindings/python/src/lib.rs

@@ -22,6 +22,7 @@ mod blob_uri_reader;
 mod commit_callback;
 mod context;
 mod error;
+mod evolution;
 mod full_text_search;
 mod hybrid_search;
 mod index_build;
@@ -47,5 +48,6 @@ fn pypaimon_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
     error::register_module(py, m)?;
     context::register_module(py, m)?;
     variant::register_module(py, m)?;
+    evolution::register_module(m)?;
     Ok(())
 }

crates/integrations/datafusion/src/table/mod.rs

@@ -68,7 +68,7 @@ pub(crate) fn datafusion_arrow_schema(
     schema_force_view_types: bool,
 ) -> DFResult<ArrowSchemaRef> {
     let paimon_schema =
-        paimon::arrow::build_target_arrow_schema(fields).map_err(to_datafusion_error)?;
+        paimon::arrow::build_read_arrow_schema(fields).map_err(to_datafusion_error)?;
     let fields = paimon_schema
         .fields()
         .iter()
@@ -902,6 +902,28 @@ mod tests {
         assert_eq!(row_fields[0].data_type(), &ArrowDataType::Utf8);
     }
 
+    #[test]
+    fn test_datafusion_read_schema_keeps_cast_nulls_visible_to_optimizers() {
+        let decimal =
+            DataType::Decimal(paimon::spec::DecimalType::with_nullable(false, 3, 2).unwrap());
+        let fields = vec![
+            DataField::new(0, "value".into(), decimal.clone()),
+            DataField::new(1, "values".into(), DataType::Array(ArrayType::new(decimal))),
+        ];
+        let schema = datafusion_arrow_schema(&fields, false).unwrap();
+        assert!(schema.field(0).is_nullable());
+        let ArrowDataType::List(child) = schema.field(1).data_type() else {
+            panic!()
+        };
+        assert!(child.is_nullable());
+        let write_schema = paimon::arrow::build_target_arrow_schema(&fields).unwrap();
+        assert!(!write_schema.field(0).is_nullable());
+        let ArrowDataType::List(child) = write_schema.field(1).data_type() else {
+            panic!()
+        };
+        assert!(!child.is_nullable());
+    }
+
     fn planned_file_names(scan: &PaimonTableScan) -> Vec<String> {
         let mut names = scan
             .planned_partitions()

crates/integrations/datafusion/src/variant_pushdown.rs

@@ -389,7 +389,7 @@ fn build_projection_rewrite(
         .transpose()?;
 
     let arrow_schema =
-        paimon::arrow::build_target_arrow_schema(&read_type).map_err(to_datafusion_error)?;
+        paimon::arrow::build_read_arrow_schema(&read_type).map_err(to_datafusion_error)?;
     let schema = Arc::new(DFSchema::try_from_qualified_schema(
         scan.table_name.clone(),
         arrow_schema.as_ref(),

crates/paimon/src/arrow/cast_completion_java_test.rs

@@ -297,6 +297,94 @@ fn unrepresentable(expected: &Value, target: &ArrowType) -> bool {
     full % divisor != 0 || i64::try_from(full / divisor).is_err()
 }
 
+#[test]
+fn schema_evolution_scalar_casts_match_actual_java() {
+    let zone = jiff::tz::TimeZone::system();
+    let zone = match zone.iana_name() {
+        Some("Asia/Shanghai") => "Asia/Shanghai",
+        Some("America/New_York") => "America/New_York",
+        Some("UTC" | "Etc/UTC" | "Etc/GMT" | "GMT") => "UTC",
+        _ => return,
+    };
+    let mut tested = 0;
+    let mut failures = Vec::new();
+    for line in CASES.lines() {
+        let case: Value = serde_json::from_str(line).unwrap();
+        let from: DataType = serde_json::from_value(case["fields"][0]["type"].clone()).unwrap();
+        let original: DataType =
+            serde_json::from_value(case["transform"]["fieldRef"]["type"].clone()).unwrap();
+        if from != original {
+            continue;
+        }
+        let target: DataType = serde_json::from_value(case["transform"]["type"].clone()).unwrap();
+        let target_arrow = paimon_type_to_arrow(&target).unwrap();
+        for (row, expected) in case["rows"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .zip(case["expected"][zone]["results"].as_array().unwrap())
+        {
+            let Some(source) = source(&row[0], &from) else {
+                continue;
+            };
+            // These outputs cannot be represented by Arrow, independently of
+            // Java's executor. Storage-boundary behavior has dedicated tests.
+            if unrepresentable(expected, &paimon_type_to_arrow(&target).unwrap()) {
+                continue;
+            }
+            if matches!(source.data_type(), ArrowType::Binary)
+                && row[0].as_array().is_some_and(|bytes| {
+                    std::str::from_utf8(
+                        &bytes
+                            .iter()
+                            .map(|v| v.as_u64().unwrap() as u8)
+                            .collect::<Vec<_>>(),
+                    )
+                    .is_err()
+                })
+            {
+                continue;
+            }
+            let actual = super::nested_evolution::evolve_column(&source, &from, &target);
+            tested += 1;
+            if expected.get("error").is_some() {
+                if actual.is_ok() {
+                    failures.push(format!(
+                        "{} input {} expected an error, got {:?}",
+                        case["name"],
+                        row[0],
+                        actual.map(|v| external(&v))
+                    ));
+                }
+            } else if !actual.as_ref().is_ok_and(|value| {
+                let value = external(value);
+                if expected.is_number()
+                    && matches!(target_arrow, ArrowType::Float32 | ArrowType::Float64)
+                {
+                    value.as_f64() == expected.as_f64()
+                } else {
+                    value == *expected
+                }
+            }) {
+                failures.push(format!(
+                    "{} input {} expected {}, got {:?}",
+                    case["name"],
+                    row[0],
+                    expected,
+                    actual.map(|v| external(&v))
+                ));
+            }
+        }
+    }
+    assert!(tested > 500, "too few Java cases: {tested}");
+    assert!(
+        failures.is_empty(),
+        "{} mismatches in {tested} Java cases: {:#?}",
+        failures.len(),
+        &failures[..failures.len().min(30)]
+    );
+}
+
 #[test]
 fn cast_completion_matches_actual_java_in_three_zones() {
     let mut tested = 0;
@@ -389,7 +477,7 @@ fn existing_python_java_capability_matrix_is_complete() {
         "01111110111000000000",
         "11111111111000000000",
         "11111111111000000000",
-        "11111111111000000000",
+        "11111111111000001111",
         "11111111111000001111",
         "01111111111000000000",
         "01111111111000000000",

crates/paimon/src/arrow/cast_executor.rs

@@ -96,7 +96,10 @@ impl CastExecutor {
                     to,
                     DataType::Float(_) | DataType::Double(_) | DataType::Date(_)
                 ) || timestamp(to))
-            || matches!(from, DataType::BigInt(_) | DataType::Date(_)) && timestamp(to)
+            || matches!(
+                from,
+                DataType::Int(_) | DataType::BigInt(_) | DataType::Date(_)
+            ) && timestamp(to)
             || timestamp(from)
                 && (timestamp(to)
                     || matches!(
@@ -277,7 +280,8 @@ impl CastExecutor {
                     .collect::<crate::Result<Vec<_>>>()?;
                 strict_cast(&(Arc::new(Int64Array::from(values)) as ArrayRef), &target)?
             }
-        } else if matches!(self.from, DataType::BigInt(_)) && timestamp(&self.to) {
+        } else if matches!(self.from, DataType::Int(_) | DataType::BigInt(_)) && timestamp(&self.to)
+        {
             self.require(number(&self.current))?;
             let (source, _) = self.number_input(&source)?;
             let long = super::evolution_cast::numeric_to_integer(&source, &ArrowType::Int64)?;

crates/paimon/src/arrow/evolution_cast.rs

@@ -342,7 +342,11 @@ pub(super) fn timestamp_precision(
         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()
+            // Java Timestamp keeps signed-long milliseconds. Flooring below
+            // i64::MIN wraps that millisecond value before Arrow unit scaling.
+            let millis = truncated.div_euclid(1_000_000) as i64;
+            let nanos = i128::from(millis) * 1_000_000 + truncated.rem_euclid(1_000_000);
+            i64::try_from(nanos / nanos_per_tick(target_unit)).ok()
         })
     }));
     Ok(Some(