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 95d0e110 Complete Java-compatible schema evolution for native reads
(#1128)
95d0e110 is described below
commit 95d0e110abb230920f44e330be73d5bf4c8cb1ae
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 15 10:16:59 2026 +0800
Complete Java-compatible schema evolution for native reads (#1128)
---
Cargo.lock | 4 +-
DEPENDENCIES.rust.tsv | 2 +-
benchmarks/tpcds/DEPENDENCIES.rust.tsv | 2 +-
bindings/c/DEPENDENCIES.rust.tsv | 2 +-
bindings/go/DEPENDENCIES.rust.tsv | 2 +-
bindings/python/DEPENDENCIES.rust.tsv | 2 +-
bindings/python/python/pypaimon_rust/data.pyi | 14 +
bindings/python/src/evolution.rs | 83 +++
bindings/python/src/lib.rs | 2 +
crates/integration_tests/DEPENDENCIES.rust.tsv | 2 +-
.../integrations/datafusion/DEPENDENCIES.rust.tsv | 2 +-
crates/integrations/datafusion/src/table/mod.rs | 24 +-
.../datafusion/src/variant_pushdown.rs | 2 +-
crates/paimon-rest-server/DEPENDENCIES.rust.tsv | 2 +-
crates/paimon/DEPENDENCIES.rust.tsv | 2 +-
.../paimon/src/arrow/cast_completion_java_test.rs | 90 ++-
crates/paimon/src/arrow/cast_executor.rs | 8 +-
crates/paimon/src/arrow/evolution_cast.rs | 6 +-
crates/paimon/src/arrow/evolution_temporal.rs | 15 +-
crates/paimon/src/arrow/mod.rs | 85 +++
crates/paimon/src/arrow/nested_evolution.rs | 738 ++++++++++++++++++---
crates/paimon/src/arrow/schema_evolution.rs | 14 +
crates/paimon/src/spec/types.rs | 62 +-
crates/paimon/src/table/audit_log_table/merge.rs | 3 +-
crates/paimon/src/table/audit_log_table/read.rs | 3 +-
crates/paimon/src/table/data_evolution_reader.rs | 13 +-
.../table/data_evolution_reader/blob_fallback.rs | 3 +-
crates/paimon/src/table/data_file_reader.rs | 5 +-
crates/paimon/src/table/format_table_read.rs | 4 +-
crates/paimon/src/table/kv_file_reader.rs | 13 +-
crates/paimon/src/table/mod.rs | 1 +
crates/paimon/src/table/query_auth/rules.rs | 17 +-
crates/paimon/src/table/table_read.rs | 5 +-
crates/paimon/src/table/write_batch_normalize.rs | 4 +-
34 files changed, 1079 insertions(+), 157 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 69911e8b..22ea40a7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6046,9 +6046,9 @@ dependencies = [
[[package]]
name = "rustls"
-version = "0.23.45"
+version = "0.23.46"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634"
+checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf"
dependencies = [
"aws-lc-rs",
"once_cell",
diff --git a/DEPENDENCIES.rust.tsv b/DEPENDENCIES.rust.tsv
index f266a0a4..2313f8f4 100644
--- a/DEPENDENCIES.rust.tsv
+++ b/DEPENDENCIES.rust.tsv
@@ -504,7 +504,7 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
diff --git a/benchmarks/tpcds/DEPENDENCIES.rust.tsv
b/benchmarks/tpcds/DEPENDENCIES.rust.tsv
index 6e030d48..24915be4 100644
--- a/benchmarks/tpcds/DEPENDENCIES.rust.tsv
+++ b/benchmarks/tpcds/DEPENDENCIES.rust.tsv
@@ -328,7 +328,7 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
diff --git a/bindings/c/DEPENDENCIES.rust.tsv b/bindings/c/DEPENDENCIES.rust.tsv
index 8b8f6cea..d3a70ef8 100644
--- a/bindings/c/DEPENDENCIES.rust.tsv
+++ b/bindings/c/DEPENDENCIES.rust.tsv
@@ -287,7 +287,7 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
diff --git a/bindings/go/DEPENDENCIES.rust.tsv
b/bindings/go/DEPENDENCIES.rust.tsv
index 8b8f6cea..d3a70ef8 100644
--- a/bindings/go/DEPENDENCIES.rust.tsv
+++ b/bindings/go/DEPENDENCIES.rust.tsv
@@ -287,7 +287,7 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
diff --git a/bindings/python/DEPENDENCIES.rust.tsv
b/bindings/python/DEPENDENCIES.rust.tsv
index 9c4513eb..563899bf 100644
--- a/bindings/python/DEPENDENCIES.rust.tsv
+++ b/bindings/python/DEPENDENCIES.rust.tsv
@@ -428,7 +428,7 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
diff --git a/bindings/python/python/pypaimon_rust/data.pyi
b/bindings/python/python/pypaimon_rust/data.pyi
index c46aa2bb..9b50d350 100644
--- a/bindings/python/python/pypaimon_rust/data.pyi
+++ b/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:
diff --git a/bindings/python/src/evolution.rs b/bindings/python/src/evolution.rs
new file mode 100644
index 00000000..547c14cb
--- /dev/null
+++ b/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)?)
+}
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 6b2fcc54..a15c7600 100644
--- a/bindings/python/src/lib.rs
+++ b/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(())
}
diff --git a/crates/integration_tests/DEPENDENCIES.rust.tsv
b/crates/integration_tests/DEPENDENCIES.rust.tsv
index c8f095a7..3c350db3 100644
--- a/crates/integration_tests/DEPENDENCIES.rust.tsv
+++ b/crates/integration_tests/DEPENDENCIES.rust.tsv
@@ -257,7 +257,7 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
diff --git a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
index 88bc626b..091e6a23 100644
--- a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
+++ b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
@@ -423,7 +423,7 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
diff --git a/crates/integrations/datafusion/src/table/mod.rs
b/crates/integrations/datafusion/src/table/mod.rs
index ad74a594..9a4727c5 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/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()
diff --git a/crates/integrations/datafusion/src/variant_pushdown.rs
b/crates/integrations/datafusion/src/variant_pushdown.rs
index 6013c504..4955b302 100644
--- a/crates/integrations/datafusion/src/variant_pushdown.rs
+++ b/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(),
diff --git a/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
b/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
index aa748ee1..55d7caf6 100644
--- a/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
+++ b/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
@@ -261,7 +261,7 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
diff --git a/crates/paimon/DEPENDENCIES.rust.tsv
b/crates/paimon/DEPENDENCIES.rust.tsv
index 94caabea..3a0678c2 100644
--- a/crates/paimon/DEPENDENCIES.rust.tsv
+++ b/crates/paimon/DEPENDENCIES.rust.tsv
@@ -420,7 +420,7 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
diff --git a/crates/paimon/src/arrow/cast_completion_java_test.rs
b/crates/paimon/src/arrow/cast_completion_java_test.rs
index dee9fb83..56422f19 100644
--- a/crates/paimon/src/arrow/cast_completion_java_test.rs
+++ b/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",
diff --git a/crates/paimon/src/arrow/cast_executor.rs
b/crates/paimon/src/arrow/cast_executor.rs
index 4ce174b3..4d5adbf7 100644
--- a/crates/paimon/src/arrow/cast_executor.rs
+++ b/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)?;
diff --git a/crates/paimon/src/arrow/evolution_cast.rs
b/crates/paimon/src/arrow/evolution_cast.rs
index c0bd85d1..5e85e110 100644
--- a/crates/paimon/src/arrow/evolution_cast.rs
+++ b/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(
diff --git a/crates/paimon/src/arrow/evolution_temporal.rs
b/crates/paimon/src/arrow/evolution_temporal.rs
index a4349099..cd99a702 100644
--- a/crates/paimon/src/arrow/evolution_temporal.rs
+++ b/crates/paimon/src/arrow/evolution_temporal.rs
@@ -44,7 +44,7 @@ pub(super) fn cast_temporal(
DataType::LocalZonedTimestamp(_),
DataType::Timestamp(_) | DataType::Date(_) | DataType::Time(_)
) | (
- DataType::Date(_) | DataType::BigInt(_),
+ DataType::Date(_) | DataType::Int(_) | DataType::BigInt(_),
DataType::LocalZonedTimestamp(_)
)
) {
@@ -96,12 +96,15 @@ fn cast_in_zone_with_storage(
);
let numeric = matches!(
(from, to),
- (DataType::BigInt(_), DataType::LocalZonedTimestamp(_))
+ (
+ DataType::Int(_) | DataType::BigInt(_),
+ DataType::LocalZonedTimestamp(_)
+ )
);
- if !cross
- && !date
- && !numeric
- && !(timestamp && matches!(to, DataType::Date(_) | DataType::Time(_)))
+ if !(cross
+ || date
+ || numeric
+ || timestamp && matches!(to, DataType::Date(_) | DataType::Time(_)))
{
return Ok(None);
}
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index 72101876..a7119809 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -39,6 +39,7 @@ mod row_filter;
mod string_transform;
pub(crate) use string_transform::evaluate_string_transform;
pub(crate) mod schema_evolution;
+pub use schema_evolution::cast_array_for_schema_evolution;
pub(crate) mod shredding;
mod variant;
@@ -380,6 +381,90 @@ pub fn build_target_arrow_schema(fields: &[DataField]) ->
crate::Result<Arc<Arro
Ok(Arc::new(ArrowSchema::new(arrow_fields)))
}
+/// Read results can contain NULL after a Java cast overflows or a historical
+/// file lacks a requested field, even when the current logical type is NOT
NULL.
+/// Keep those runtime values representable without relaxing the writer schema.
+pub(crate) fn paimon_type_to_arrow_for_read(dt: &PaimonDataType) ->
crate::Result<ArrowDataType> {
+ Ok(match dt {
+ PaimonDataType::Row(row) => ArrowDataType::Struct(
+ row.fields()
+ .iter()
+ .map(|field| {
+ Ok(ArrowField::new(
+ field.name(),
+ paimon_type_to_arrow_for_read(field.data_type())?,
+ true,
+ ))
+ })
+ .collect::<crate::Result<Vec<_>>>()?
+ .into(),
+ ),
+ PaimonDataType::Array(array) =>
ArrowDataType::List(Arc::new(ArrowField::new(
+ "element",
+ paimon_type_to_arrow_for_read(array.element_type())?,
+ true,
+ ))),
+ PaimonDataType::Map(map) => ArrowDataType::Map(
+ Arc::new(ArrowField::new(
+ "entries",
+ ArrowDataType::Struct(
+ vec![
+ ArrowField::new(
+ "key",
+ paimon_type_to_arrow_for_read(map.key_type())?,
+ false,
+ ),
+ ArrowField::new(
+ "value",
+ paimon_type_to_arrow_for_read(map.value_type())?,
+ true,
+ ),
+ ]
+ .into(),
+ ),
+ false,
+ )),
+ false,
+ ),
+ PaimonDataType::Multiset(multiset) => ArrowDataType::Map(
+ Arc::new(ArrowField::new(
+ "entries",
+ ArrowDataType::Struct(
+ vec![
+ ArrowField::new(
+ "key",
+
paimon_type_to_arrow_for_read(multiset.element_type())?,
+ multiset.element_type().is_nullable(),
+ ),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ]
+ .into(),
+ ),
+ false,
+ )),
+ false,
+ ),
+ _ => paimon_type_to_arrow(dt)?,
+ })
+}
+
+/// Build the nullable runtime schema used by table readers. Logical table
+/// constraints and the schema used to validate/write new records are
unchanged.
+pub fn build_read_arrow_schema(fields: &[DataField]) ->
crate::Result<Arc<ArrowSchema>> {
+ let fields = fields
+ .iter()
+ .map(|field| {
+ Ok(arrow_field_with_paimon_id(
+ field.name(),
+ paimon_type_to_arrow_for_read(field.data_type())?,
+ true,
+ field.id(),
+ ))
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ Ok(Arc::new(ArrowSchema::new(fields)))
+}
+
#[cfg(test)]
mod tests {
use super::*;
diff --git a/crates/paimon/src/arrow/nested_evolution.rs
b/crates/paimon/src/arrow/nested_evolution.rs
index 09fc1288..2e511139 100644
--- a/crates/paimon/src/arrow/nested_evolution.rs
+++ b/crates/paimon/src/arrow/nested_evolution.rs
@@ -34,12 +34,14 @@
use std::sync::Arc;
use arrow_array::{
- new_null_array, Array, ArrayRef, ListArray, MapArray, StringArray,
StructArray, UInt32Array,
+ new_null_array, Array, ArrayRef, BooleanArray, GenericListArray,
ListArray, MapArray,
+ OffsetSizeTrait, StringArray, StructArray, UInt32Array,
};
+use arrow_buffer::{NullBuffer, OffsetBuffer};
use arrow_cast::cast;
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Fields};
-use crate::arrow::paimon_type_to_arrow;
+use crate::arrow::paimon_type_to_arrow_for_read;
use crate::spec::{
is_variant_extraction_row_type, map_selected_keys, DataField, DataType,
MapType, RowType,
};
@@ -62,13 +64,26 @@ pub(crate) fn evolve_field(
if is_variant_extraction_row_type(target.data_type()) {
// A format read plan may already have assembled the synthetic
// ROW while its field context still describes the VARIANT.
- if source.data_type() ==
&paimon_type_to_arrow(target.data_type())? {
- return Ok(source.clone());
+ if source
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .is_some_and(|rows| {
+ rows.column_names()
+ == row
+ .fields()
+ .iter()
+ .map(|field| field.name())
+ .collect::<Vec<_>>()
+ })
+ {
+ return evolve_column(source, target.data_type(),
target.data_type());
}
- return
crate::arrow::shredding::variant::assemble_variant_extraction_array(
- source.as_ref(),
- row,
- );
+ let assembled =
+
crate::arrow::shredding::variant::assemble_variant_extraction_array(
+ source.as_ref(),
+ row,
+ )?;
+ return evolve_column(&assembled, target.data_type(),
target.data_type());
}
}
}
@@ -89,8 +104,20 @@ pub(crate) fn selected_map_to_row(
message: "Selected MAP input must be a MapArray".into(),
source: None,
})?;
- let names = map
- .keys()
+ if !matches!(
+ map_type.key_type(),
+ DataType::Char(_) | DataType::VarChar(_)
+ ) {
+ return Err(crate::Error::Unsupported {
+ message: "Selected MAP keys must have a character string
type".into(),
+ });
+ }
+ let normalized_keys = evolve_column(
+ map.keys(),
+ map_type.key_type(),
+ &DataType::VarChar(crate::spec::VarCharType::string_type()),
+ )?;
+ let names = normalized_keys
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| crate::Error::DataInvalid {
@@ -123,7 +150,8 @@ pub(crate) fn selected_map_to_row(
field.data_type(),
)?);
}
- let ArrowDataType::Struct(fields) =
paimon_type_to_arrow(&DataType::Row(row_type.clone()))?
+ let ArrowDataType::Struct(fields) =
+ paimon_type_to_arrow_for_read(&DataType::Row(row_type.clone()))?
else {
unreachable!()
};
@@ -141,14 +169,35 @@ pub(crate) fn selected_map_to_row(
/// actually holds) with `target_type` (the type the read schema wants).
///
/// Recurses through ROW so an added, dropped, renamed or promoted nested field
-/// is handled at any depth. Non-nested mismatches fall back to Arrow `cast`,
-/// mirroring the top-level promotion path.
+/// is handled at any depth. Leaf conversions use the Java-compatible scalar
+/// executors also used by top-level old-file reads.
pub(crate) fn evolve_column(
source: &ArrayRef,
source_type: &DataType,
target_type: &DataType,
) -> crate::Result<ArrayRef> {
- let target_arrow = paimon_type_to_arrow(target_type)?;
+ // DECIMAL has at most 38 digits in Paimon, but Arrow metadata may use the
+ // wider Decimal256 storage for the same logical type. Normalize storage
+ // before dispatching Java numeric executors, preserving precision/scale.
+ if matches!(source_type, DataType::Decimal(_))
+ && matches!(source.data_type(), ArrowDataType::Decimal256(_, _))
+ {
+ let source_arrow = paimon_type_to_arrow_for_read(source_type)?;
+ let normalized = arrow_cast::cast_with_options(
+ source,
+ &source_arrow,
+ &arrow_cast::CastOptions {
+ safe: false,
+ ..Default::default()
+ },
+ )
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("failed to normalize DECIMAL storage: {error}"),
+ source: None,
+ })?;
+ return evolve_column(&normalized, source_type, target_type);
+ }
+ let target_arrow = paimon_type_to_arrow_for_read(target_type)?;
// Arrow equality alone is NOT enough to skip the walk:
`paimon_type_to_arrow`
// drops field ids, so a nested child that was dropped and re-added under
the
// same name and type produces an identical Arrow struct while carrying a
@@ -160,6 +209,25 @@ pub(crate) fn evolve_column(
if source_type.equals_ignore_nullable(target_type) && source.data_type()
== &target_arrow {
return Ok(source.clone());
}
+ if source_type.equals_ignore_nullable(target_type)
+ && !matches!(
+ source_type,
+ DataType::Row(_) | DataType::Array(_) | DataType::Map(_) |
DataType::Multiset(_)
+ )
+ {
+ return arrow_cast::cast_with_options(
+ source,
+ &target_arrow,
+ &arrow_cast::CastOptions {
+ safe: false,
+ ..Default::default()
+ },
+ )
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("failed to normalize schema-evolution storage:
{error}"),
+ source: None,
+ });
+ }
if matches!(target_type, DataType::VarChar(_) | DataType::Char(_))
&& matches!(
@@ -180,7 +248,21 @@ pub(crate) fn evolve_column(
// nothing to evolve — leave it to the cast below, as before.
(DataType::Row(_), DataType::Row(_))
if is_variant_extraction_row_type(target_type)
- || is_variant_extraction_row_type(source_type) => {}
+ || is_variant_extraction_row_type(source_type) =>
+ {
+ return arrow_cast::cast_with_options(
+ source,
+ &target_arrow,
+ &arrow_cast::CastOptions {
+ safe: false,
+ ..Default::default()
+ },
+ )
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("failed to normalize variant-extraction ROW
storage: {error}"),
+ source: None,
+ });
+ }
(DataType::Row(target_row), DataType::Row(source_row)) => {
return evolve_struct(source, source_row, target_row)
}
@@ -245,13 +327,21 @@ pub(crate) fn evolve_column(
return Ok(result);
}
- cast(source, &target_arrow).map_err(|e| crate::Error::UnexpectedError {
+ // Old-file reads and authorization use the same Java executor for lexical
+ // parsing, numeric text and character/binary widths. Arrow's default cast
+ // can otherwise silently replace invalid strings with NULL.
+ if super::CastExecutor::supports(source_type, target_type) {
+ return super::CastExecutor::new(
+ source_type.clone(),
+ source_type.clone(),
+ target_type.clone(),
+ )?
+ .cast(source);
+ }
+ Err(crate::Error::Unsupported {
message: format!(
- "failed to cast nested value from {:?} to {:?} during schema
evolution",
- source.data_type(),
- target_arrow
+ "Cannot cast from type {source_type} to type {target_type} during
schema evolution"
),
- source: Some(Box::new(e)),
})
}
@@ -273,8 +363,18 @@ fn render_string_values(
let children = row_type
.fields()
.iter()
- .enumerate()
- .map(|(index, field)| render_string_values(rows.column(index),
field.data_type()))
+ .map(|field| {
+ let column =
rows.column_by_name(field.name()).ok_or_else(|| {
+ crate::Error::DataInvalid {
+ message: format!(
+ "missing historical ROW field '{}' during
string conversion",
+ field.name()
+ ),
+ source: None,
+ }
+ })?;
+ render_string_values(&visible_struct_child(column, rows)?,
field.data_type())
+ })
.collect::<crate::Result<Vec<_>>>()?;
Ok((0..rows.len())
.map(|index| {
@@ -292,12 +392,7 @@ fn render_string_values(
.collect())
}
DataType::Array(array_type) => {
- let rows =
source.as_any().downcast_ref::<ListArray>().ok_or_else(|| {
- crate::Error::DataInvalid {
- message: format!("expected ARRAY array, got {:?}",
source.data_type()),
- source: None,
- }
- })?;
+ let rows = visible_list(source)?;
let values = render_string_values(rows.values(),
array_type.element_type())?;
let offsets = rows.value_offsets();
Ok((0..rows.len())
@@ -316,12 +411,7 @@ fn render_string_values(
.collect())
}
DataType::Map(map_type) => {
- let rows =
source.as_any().downcast_ref::<MapArray>().ok_or_else(|| {
- crate::Error::DataInvalid {
- message: format!("expected MAP array, got {:?}",
source.data_type()),
- source: None,
- }
- })?;
+ let rows = visible_map(as_map_array(source, "MAP")?)?;
let keys = render_string_values(rows.keys(), map_type.key_type())?;
let values = render_string_values(rows.values(),
map_type.value_type())?;
let offsets = rows.value_offsets();
@@ -396,11 +486,11 @@ fn evolve_struct(
let mut fields: Vec<Arc<ArrowField>> =
Vec::with_capacity(target_row.fields().len());
let mut arrays: Vec<ArrayRef> =
Vec::with_capacity(target_row.fields().len());
for target_field in target_row.fields() {
- let target_arrow = paimon_type_to_arrow(target_field.data_type())?;
+ let target_arrow =
paimon_type_to_arrow_for_read(target_field.data_type())?;
fields.push(Arc::new(ArrowField::new(
target_field.name(),
target_arrow.clone(),
- target_field.data_type().is_nullable(),
+ true,
)));
// Pair by id (Java: `createIndexMapping` over the ROW's fields).
@@ -431,7 +521,11 @@ fn evolve_struct(
),
source: None,
})?;
- arrays.push(evolve_field(column, source_field, target_field)?)
+ arrays.push(evolve_field(
+ &visible_struct_child(column, source_struct)?,
+ source_field,
+ target_field,
+ )?)
}
}
}
@@ -452,45 +546,13 @@ fn evolve_list(
source_element: &DataType,
target_element: &DataType,
) -> crate::Result<ArrayRef> {
- // A Paimon ARRAY converts to an Arrow `List`, but a decoded column can
arrive
- // in another list layout — an external Parquet file's embedded Arrow
schema
- // may yield `LargeList` or `FixedSizeList`. Normalize the layout with
Arrow's
- // own cast (which range-checks large offsets) before reconciling the
element,
- // so those files stay readable.
- let normalized;
- let list = match source.as_any().downcast_ref::<ListArray>() {
- Some(list) => list,
- None => {
- let source_layout = ArrowDataType::List(Arc::new(ArrowField::new(
- "element",
- paimon_type_to_arrow(source_element)?,
- source_element.is_nullable(),
- )));
- normalized = cast(source, &source_layout).map_err(|e|
crate::Error::DataInvalid {
- message: format!(
- "expected a list array for ARRAY schema evolution, got
{:?}, which does not convert to {source_layout:?}: {e}",
- source.data_type()
- ),
- source: None,
- })?;
- normalized
- .as_any()
- .downcast_ref::<ListArray>()
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!(
- "normalizing {:?} did not yield a list array",
- source.data_type()
- ),
- source: None,
- })?
- }
- };
+ let list = visible_list(source)?;
let values = evolve_column(list.values(), source_element, target_element)?;
let element_field = Arc::new(ArrowField::new(
"element",
- paimon_type_to_arrow(target_element)?,
- target_element.is_nullable(),
+ paimon_type_to_arrow_for_read(target_element)?,
+ true,
));
let evolved = ListArray::try_new(
element_field,
@@ -505,6 +567,166 @@ fn evolve_list(
Ok(Arc::new(evolved))
}
+/// Normalize physical offsets without casting children before field-id
evolution.
+/// Arrow metadata can store logical ARRAYs as LargeList or FixedSizeList.
Their
+/// children still describe the historical file schema and must remain intact.
+fn normalize_list(source: &ArrayRef) -> crate::Result<ListArray> {
+ if let Some(list) = source.as_any().downcast_ref::<ListArray>() {
+ return Ok(list.clone());
+ }
+ let child = match source.data_type() {
+ ArrowDataType::LargeList(child) | ArrowDataType::FixedSizeList(child,
_) => child,
+ _ => {
+ return Err(crate::Error::DataInvalid {
+ message: format!("expected ARRAY array, got {:?}",
source.data_type()),
+ source: None,
+ });
+ }
+ };
+ let normalized = cast(source,
&ArrowDataType::List(child.clone())).map_err(|e| {
+ crate::Error::DataInvalid {
+ message: format!("failed to normalize ARRAY offsets: {e}"),
+ source: None,
+ }
+ })?;
+ normalized
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .cloned()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "normalizing ARRAY offsets did not yield a
ListArray".into(),
+ source: None,
+ })
+}
+
+fn visible_struct_child(column: &ArrayRef, rows: &StructArray) ->
crate::Result<ArrayRef> {
+ if rows.null_count() == 0 {
+ return Ok(column.clone());
+ }
+ let hidden = BooleanArray::from(
+ (0..rows.len())
+ .map(|row| rows.is_null(row))
+ .collect::<Vec<_>>(),
+ );
+ arrow_select::nullif::nullif(column.as_ref(), &hidden).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!("failed to mask a NULL ROW child: {error}"),
+ source: None,
+ }
+ })
+}
+
+/// Only convert elements the visible containers can observe. Slices may retain
+/// unrelated prefix/suffix values; NULL containers can also own physical
slots.
+/// Java's array/map getters never evaluate those slots.
+fn visible_entries<O: OffsetSizeTrait>(
+ offsets: &[O],
+ nulls: Option<&NullBuffer>,
+ values: &ArrayRef,
+) -> crate::Result<(OffsetBuffer<i32>, ArrayRef)> {
+ let start = offsets[0].as_usize();
+ let end = offsets[offsets.len() - 1].as_usize();
+ let has_hidden_entries = (0..offsets.len() - 1).any(|row| {
+ nulls.is_some_and(|nulls| nulls.is_null(row)) && offsets[row] !=
offsets[row + 1]
+ });
+ let mut result_offsets = Vec::with_capacity(offsets.len());
+ result_offsets.push(0);
+ let selected = if has_hidden_entries {
+ let mut pieces = Vec::new();
+ let mut count = 0usize;
+ for row in 0..offsets.len() - 1 {
+ let from = offsets[row].as_usize();
+ let to = offsets[row + 1].as_usize();
+ if nulls.is_none_or(|nulls| nulls.is_valid(row)) && from != to {
+ pieces.push(values.slice(from, to - from));
+ count += to - from;
+ }
+ result_offsets.push(i32::try_from(count).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!(
+ "ARRAY/MAP visible offsets exceed Arrow List storage:
{error}"
+ ),
+ source: None,
+ }
+ })?);
+ }
+ if pieces.is_empty() {
+ values.slice(start, 0)
+ } else {
+ arrow_select::concat::concat(
+ &pieces.iter().map(|part| part.as_ref()).collect::<Vec<_>>(),
+ )
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("failed to gather visible ARRAY/MAP entries:
{error}"),
+ source: None,
+ })?
+ }
+ } else {
+ for offset in &offsets[1..] {
+ result_offsets.push(i32::try_from(offset.as_usize() -
start).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!(
+ "ARRAY/MAP visible offsets exceed Arrow List storage:
{error}"
+ ),
+ source: None,
+ }
+ })?);
+ }
+ values.slice(start, end - start)
+ };
+ Ok((OffsetBuffer::new(result_offsets.into()), selected))
+}
+
+fn visible_generic_list<O: OffsetSizeTrait>(
+ list: &GenericListArray<O>,
+) -> crate::Result<ListArray> {
+ let (offsets, values) = visible_entries(list.value_offsets(),
list.nulls(), list.values())?;
+ let child = match list.data_type() {
+ ArrowDataType::List(child) | ArrowDataType::LargeList(child) =>
child.clone(),
+ _ => unreachable!(),
+ };
+ ListArray::try_new(child, offsets, values,
list.nulls().cloned()).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!("failed to build visible ARRAY: {error}"),
+ source: None,
+ }
+ })
+}
+
+fn visible_list(source: &ArrayRef) -> crate::Result<ListArray> {
+ if let Some(list) = source
+ .as_any()
+ .downcast_ref::<arrow_array::LargeListArray>()
+ {
+ return visible_generic_list(list);
+ }
+ visible_generic_list(&normalize_list(source)?)
+}
+
+fn visible_map(map: &MapArray) -> crate::Result<MapArray> {
+ let entries: ArrayRef = Arc::new(map.entries().clone());
+ let (offsets, entries) = visible_entries(map.value_offsets(), map.nulls(),
&entries)?;
+ let entries = entries
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap()
+ .clone();
+ let ArrowDataType::Map(field, sorted) = map.data_type() else {
+ unreachable!()
+ };
+ MapArray::try_new(
+ field.clone(),
+ offsets,
+ entries,
+ map.nulls().cloned(),
+ *sorted,
+ )
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("failed to build visible MAP: {error}"),
+ source: None,
+ })
+}
+
/// Rebuild a MAP whose key or value type evolved, reusing the source offsets
and
/// validity. Mirrors Java `SchemaEvolutionUtil.createMapCastExecutor`.
fn evolve_map(
@@ -512,7 +734,7 @@ fn evolve_map(
source_map: &MapType,
target_map: &MapType,
) -> crate::Result<ArrayRef> {
- let map = as_map_array(source, "MAP")?;
+ let map = visible_map(as_map_array(source, "MAP")?)?;
let keys = evolve_column(map.keys(), source_map.key_type(),
target_map.key_type())?;
let values = evolve_column(
map.values(),
@@ -520,14 +742,18 @@ fn evolve_map(
target_map.value_type(),
)?;
rebuild_map(
- map,
+ &map,
keys,
values,
- ArrowField::new("key", paimon_type_to_arrow(target_map.key_type())?,
false),
+ ArrowField::new(
+ "key",
+ paimon_type_to_arrow_for_read(target_map.key_type())?,
+ false,
+ ),
ArrowField::new(
"value",
- paimon_type_to_arrow(target_map.value_type())?,
- target_map.value_type().is_nullable(),
+ paimon_type_to_arrow_for_read(target_map.value_type())?,
+ true,
),
)
}
@@ -546,16 +772,16 @@ fn evolve_multiset(
source_element: &DataType,
target_element: &DataType,
) -> crate::Result<ArrayRef> {
- let map = as_map_array(source, "MULTISET")?;
+ let map = visible_map(as_map_array(source, "MULTISET")?)?;
let elements = evolve_column(map.keys(), source_element, target_element)?;
let counts = map.values().clone();
rebuild_map(
- map,
+ &map,
elements,
counts,
ArrowField::new(
"key",
- paimon_type_to_arrow(target_element)?,
+ paimon_type_to_arrow_for_read(target_element)?,
target_element.is_nullable(),
),
ArrowField::new("value", ArrowDataType::Int32, false),
@@ -718,6 +944,42 @@ mod tests {
assert_eq!(output.column(2).null_count(), 3);
}
+ #[test]
+ fn selected_map_normalizes_large_string_keys_in_sliced_input() {
+ use arrow_array::builder::{Int32Builder, LargeStringBuilder,
MapBuilder};
+ let mut builder = MapBuilder::new(None, LargeStringBuilder::new(),
Int32Builder::new());
+ for (key, value) in [("prefix", 99), ("key", 7), ("key", 9),
("suffix", 88)] {
+ builder.keys().append_value(key);
+ builder.values().append_value(value);
+ if key != "key" || value == 9 {
+ builder.append(true).unwrap();
+ }
+ }
+ let source: ArrayRef = Arc::new(builder.finish());
+ let source_field = field(
+ 10,
+ "attrs",
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::new()),
+ )),
+ );
+ let requested =
+ crate::spec::map_selected_keys_field(&source_field,
&["key".into(), "prefix".into()])
+ .unwrap();
+ let output = evolve_field(&source.slice(1, 1), &source_field,
&requested).unwrap();
+ let output = output.as_any().downcast_ref::<StructArray>().unwrap();
+ assert_eq!(
+ output
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap(),
+ &Int32Array::from(vec![7])
+ );
+ assert_eq!(output.column(1).null_count(), 1);
+ }
+
fn field(id: i32, name: &str, dt: DataType) -> DataField {
DataField::new(id, name.to_string(), dt)
}
@@ -784,7 +1046,10 @@ mod tests {
// The rebuilt struct must match the type the read schema declares, or
// RecordBatch assembly would reject it.
- assert_eq!(out.data_type(), &paimon_type_to_arrow(&target).unwrap());
+ assert_eq!(
+ out.data_type(),
+ &paimon_type_to_arrow_for_read(&target).unwrap()
+ );
let evolved = as_struct(&out);
assert_eq!(
evolved.column_names(),
@@ -901,7 +1166,10 @@ mod tests {
let ts6 = DataType::Timestamp(TimestampType::new(6).unwrap());
let micros = evolve_column(&source, &source_type, &ts6).unwrap();
- assert_eq!(micros.data_type(), &paimon_type_to_arrow(&ts6).unwrap());
+ assert_eq!(
+ micros.data_type(),
+ &paimon_type_to_arrow_for_read(&ts6).unwrap()
+ );
let micros = micros
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
@@ -947,7 +1215,7 @@ mod tests {
Some((i64::from(i32::MAX) + 3) * ticks),
None,
]));
- let source = cast(&raw,
&paimon_type_to_arrow(&source_type).unwrap())
+ let source = cast(&raw,
&paimon_type_to_arrow_for_read(&source_type).unwrap())
.unwrap()
.slice(1, 6);
let out =
@@ -1034,7 +1302,7 @@ mod tests {
};
let raw: ArrayRef =
Arc::new(Int64Array::from(vec![Some(i64::MAX), Some(i64::MIN),
None]));
- let source = cast(&raw,
&paimon_type_to_arrow(&source_type).unwrap()).unwrap();
+ let source = cast(&raw,
&paimon_type_to_arrow_for_read(&source_type).unwrap()).unwrap();
let output =
evolve_column(&source, &source_type,
&DataType::BigInt(BigIntType::new())).unwrap();
assert_eq!(
@@ -1117,7 +1385,7 @@ mod tests {
)
};
let raw: ArrayRef = Arc::new(Int64Array::from(raw));
- let leaf = cast(&raw,
&paimon_type_to_arrow(&source_type).unwrap()).unwrap();
+ let leaf = cast(&raw,
&paimon_type_to_arrow_for_read(&source_type).unwrap()).unwrap();
let validity = Some(NullBuffer::from(vec![true, false, true]));
let values = |array: &ArrayRef| {
cast(array, &ArrowDataType::Int64)
@@ -1131,7 +1399,8 @@ mod tests {
let source_row = row(vec![field(11, "v", source_type.clone())]);
let target_row = row(vec![field(11, "v", target_type.clone())]);
- let ArrowDataType::Struct(fields) =
paimon_type_to_arrow(&source_row).unwrap() else {
+ let ArrowDataType::Struct(fields) =
paimon_type_to_arrow_for_read(&source_row).unwrap()
+ else {
panic!()
};
let source: ArrayRef = Arc::new(
@@ -1140,7 +1409,11 @@ mod tests {
let out = evolve_column(&source, &source_row,
&target_row).unwrap();
let out = as_struct(&out);
assert_eq!(out.nulls(), source.nulls());
- assert_eq!(values(out.column(0)), expected, "ROW
reverse={reverse}");
+ assert_eq!(
+ values(out.column(0)),
+ vec![expected[0], None, expected[2]],
+ "ROW reverse={reverse}"
+ );
let source: ArrayRef = Arc::new(
ListArray::try_new(
@@ -1483,7 +1756,7 @@ mod tests {
assert_eq!(
out.data_type(),
- &paimon_type_to_arrow(&target_type).unwrap()
+ &paimon_type_to_arrow_for_read(&target_type).unwrap()
);
let list = out
.as_any()
@@ -1551,7 +1824,7 @@ mod tests {
assert_eq!(
out.data_type(),
- &paimon_type_to_arrow(&target_type).unwrap()
+ &paimon_type_to_arrow_for_read(&target_type).unwrap()
);
let map = out
.as_any()
@@ -1685,6 +1958,289 @@ mod tests {
);
}
+ #[test]
+ fn casts_only_visible_list_elements_including_null_container_slots() {
+ let values: ArrayRef = Arc::new(StringArray::from(vec![
+ "bad-prefix",
+ "1.9",
+ "bad-null",
+ "-2.5",
+ "bad-suffix",
+ ]));
+ let array: ArrayRef = Arc::new(
+ ListArray::try_new(
+ Arc::new(ArrowField::new("item", ArrowDataType::Utf8, true)),
+ OffsetBuffer::new(vec![0i32, 1, 2, 3, 4, 5].into()),
+ values,
+ Some(NullBuffer::from(vec![true, true, false, true, true])),
+ )
+ .unwrap(),
+ );
+ let source_type = DataType::Array(ArrayType::new(DataType::VarChar(
+ VarCharType::string_type(),
+ )));
+ let target_type =
DataType::Array(ArrayType::new(DataType::Int(IntType::new())));
+ let output = evolve_column(&array.slice(1, 3), &source_type,
&target_type).unwrap();
+ let rows = output.as_any().downcast_ref::<ListArray>().unwrap();
+ assert_eq!(rows.value_offsets(), &[0, 1, 1, 2]);
+ assert_eq!(
+ rows.values().as_any().downcast_ref::<Int32Array>().unwrap(),
+ &Int32Array::from(vec![1, -2])
+ );
+ assert!(rows.is_null(1));
+ assert!(evolve_column(&array.slice(0, 1), &source_type,
&target_type).is_err());
+ }
+
+ #[test]
+ fn casts_only_children_of_non_null_rows() {
+ let child: ArrayRef = Arc::new(StringArray::from(vec!["1.9", "bad",
"-2.5"]));
+ let array: ArrayRef = Arc::new(
+ StructArray::try_new(
+ Fields::from(vec![ArrowField::new("old", ArrowDataType::Utf8,
true)]),
+ vec![child],
+ Some(NullBuffer::from(vec![true, false, true])),
+ )
+ .unwrap(),
+ );
+ let output = evolve_column(
+ &array,
+ &row(vec![field(1, "old", string_type())]),
+ &row(vec![field(1, "renamed", DataType::Int(IntType::new()))]),
+ )
+ .unwrap();
+ let rows = as_struct(&output);
+ assert_eq!(
+ rows.column_by_name("renamed")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap(),
+ &Int32Array::from(vec![Some(1), None, Some(-2)])
+ );
+ assert!(rows.is_null(1));
+ }
+
+ #[test]
+ fn read_schema_allows_decimal_overflow_nulls_without_relaxing_writes() {
+ let source_type = DataType::Array(ArrayType::new(DataType::Decimal(
+ DecimalType::with_nullable(false, 6, 3).unwrap(),
+ )));
+ let target_type = DataType::Array(ArrayType::new(DataType::Decimal(
+ DecimalType::with_nullable(false, 3, 2).unwrap(),
+ )));
+ let source: ArrayRef = Arc::new(
+ ListArray::try_new(
+ Arc::new(ArrowField::new(
+ "element",
+ ArrowDataType::Decimal128(6, 3),
+ false,
+ )),
+ OffsetBuffer::new(vec![0, 2].into()),
+ Arc::new(
+ Decimal128Array::from(vec![9995, 1000])
+ .with_precision_and_scale(6, 3)
+ .unwrap(),
+ ),
+ None,
+ )
+ .unwrap(),
+ );
+ let output = evolve_column(&source, &source_type,
&target_type).unwrap();
+ output.to_data().validate_full().unwrap();
+ let list = output.as_any().downcast_ref::<ListArray>().unwrap();
+ assert_eq!(
+ list.values()
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>(),
+ vec![None, Some(100)]
+ );
+ let fields = vec![field(1, "value", target_type)];
+ let read_schema =
crate::arrow::build_read_arrow_schema(&fields).unwrap();
+ arrow_array::RecordBatch::try_new(read_schema, vec![output]).unwrap();
+ let write_schema =
crate::arrow::build_target_arrow_schema(&fields).unwrap();
+ let ArrowDataType::List(child) = write_schema.field(0).data_type()
else {
+ panic!()
+ };
+ assert!(!child.is_nullable());
+ }
+
+ #[test]
+ fn
assembled_variant_rows_normalize_runtime_nullability_and_large_strings() {
+ let row = crate::spec::variant_extraction_row(
+ true,
+ vec![
+ (
+ DataType::Int(IntType::with_nullable(false)),
+ "$.x".into(),
+ false,
+ "UTC".into(),
+ ),
+ (
+ DataType::VarChar(VarCharType::string_type()),
+ "$.s".into(),
+ false,
+ "UTC".into(),
+ ),
+ ],
+ )
+ .unwrap();
+ let target = field(1, "payload", DataType::Row(row));
+ let source_field = field(
+ 1,
+ "payload",
+ DataType::Variant(crate::spec::VariantType::new()),
+ );
+ let source: ArrayRef = Arc::new(
+ StructArray::try_new(
+ vec![
+ ArrowField::new("0", ArrowDataType::Int32, false),
+ ArrowField::new("1", ArrowDataType::LargeUtf8, true),
+ ]
+ .into(),
+ vec![
+ Arc::new(Int32Array::from(vec![7])),
+
Arc::new(arrow_array::LargeStringArray::from(vec!["text"])),
+ ],
+ None,
+ )
+ .unwrap(),
+ );
+ for output in [
+ evolve_column(&source, target.data_type(),
target.data_type()).unwrap(),
+ evolve_field(&source, &source_field, &target).unwrap(),
+ ] {
+ assert_eq!(
+ output.data_type(),
+ &paimon_type_to_arrow_for_read(target.data_type()).unwrap()
+ );
+ let rows = as_struct(&output);
+ assert_eq!(
+ rows.column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 7
+ );
+ assert_eq!(
+ rows.column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "text"
+ );
+ }
+ let variant =
crate::variant::GenericVariant::parse_json(r#"{"x":7,"s":"text"}"#).unwrap();
+ let ArrowDataType::Struct(fields) = crate::arrow::variant_arrow_type()
else {
+ panic!()
+ };
+ let raw: ArrayRef = Arc::new(
+ StructArray::try_new(
+ fields,
+ vec![
+
Arc::new(arrow_array::BinaryArray::from(vec![Some(variant.value())])),
+ Arc::new(arrow_array::BinaryArray::from(vec![Some(
+ variant.metadata(),
+ )])),
+ ],
+ None,
+ )
+ .unwrap(),
+ );
+ let fresh = evolve_field(&raw, &source_field, &target).unwrap();
+ assert_eq!(
+ fresh.data_type(),
+ &paimon_type_to_arrow_for_read(target.data_type()).unwrap()
+ );
+ }
+
+ #[test]
+ fn casts_only_entries_of_non_null_maps() {
+ let fields = Fields::from(vec![
+ ArrowField::new("key", ArrowDataType::Utf8, false),
+ ArrowField::new("value", ArrowDataType::Utf8, true),
+ ]);
+ let entries = StructArray::new(
+ fields.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["one", "hidden", "two"])),
+ Arc::new(StringArray::from(vec!["1.9", "bad", "-2.5"])),
+ ],
+ None,
+ );
+ let array: ArrayRef = Arc::new(
+ MapArray::try_new(
+ Arc::new(ArrowField::new(
+ "entries",
+ ArrowDataType::Struct(fields),
+ false,
+ )),
+ OffsetBuffer::new(vec![0i32, 1, 2, 3].into()),
+ entries,
+ Some(NullBuffer::from(vec![true, false, true])),
+ false,
+ )
+ .unwrap(),
+ );
+ let from = DataType::Map(MapType::new(string_type(), string_type()));
+ let to = DataType::Map(MapType::new(string_type(),
DataType::Int(IntType::new())));
+ let output = evolve_column(&array, &from, &to).unwrap();
+ let rows = output.as_any().downcast_ref::<MapArray>().unwrap();
+ assert_eq!(rows.value_offsets(), &[0, 1, 1, 2]);
+ assert_eq!(
+ rows.values().as_any().downcast_ref::<Int32Array>().unwrap(),
+ &Int32Array::from(vec![1, -2])
+ );
+ assert!(rows.is_null(1));
+ }
+
+ #[test]
+ fn renders_large_list_strings_in_sliced_containers() {
+ let array: ArrayRef = Arc::new(
+ arrow_array::LargeListArray::try_new(
+ Arc::new(ArrowField::new("item", ArrowDataType::Int32, true)),
+ arrow_buffer::OffsetBuffer::new(vec![0i64, 1, 3, 3, 4,
5].into()),
+ Arc::new(Int32Array::from(vec![
+ Some(9),
+ Some(1),
+ None,
+ Some(7),
+ Some(8),
+ ])),
+ Some(arrow_buffer::NullBuffer::from(vec![
+ true, true, true, false, true,
+ ])),
+ )
+ .unwrap(),
+ );
+ let array_type =
DataType::Array(ArrayType::new(DataType::Int(IntType::new())));
+ let output = evolve_column(&array.slice(1, 3), &array_type,
&string_type()).unwrap();
+ assert_eq!(
+ strings(&output),
+ StringArray::from(vec![Some("[1, null]"), Some("[]"), None])
+ );
+
+ let source: ArrayRef = Arc::new(StructArray::new(
+ Fields::from(vec![ArrowField::new("a", array.data_type().clone(),
true)]),
+ vec![array],
+ None,
+ ));
+ let output = evolve_column(
+ &source.slice(1, 3),
+ &row(vec![field(4, "a", array_type)]),
+ &string_type(),
+ )
+ .unwrap();
+ assert_eq!(
+ strings(&output),
+ StringArray::from(vec!["{[1, null]}", "{[]}", "{null}"])
+ );
+ }
+
#[test]
fn evolves_a_large_list_element() {
// A format table's embedded Arrow schema can decode a Paimon ARRAY as
a
@@ -1721,7 +2277,7 @@ mod tests {
assert_eq!(
out.data_type(),
- &paimon_type_to_arrow(&target_type).unwrap()
+ &paimon_type_to_arrow_for_read(&target_type).unwrap()
);
let list = out
.as_any()
@@ -1761,7 +2317,7 @@ mod tests {
assert_eq!(
out.data_type(),
- &paimon_type_to_arrow(&source_type).unwrap(),
+ &paimon_type_to_arrow_for_read(&source_type).unwrap(),
"the layout must be normalized to the Paimon ARRAY's Arrow type"
);
assert_eq!(out.len(), 2);
diff --git a/crates/paimon/src/arrow/schema_evolution.rs
b/crates/paimon/src/arrow/schema_evolution.rs
index ea9450b3..5a6cff05 100644
--- a/crates/paimon/src/arrow/schema_evolution.rs
+++ b/crates/paimon/src/arrow/schema_evolution.rs
@@ -22,6 +22,20 @@
use crate::spec::DataField;
use std::collections::HashMap;
+/// Cast an Arrow column from its historical Paimon type into a requested type.
+/// Nested fields are paired by field ID; scalar conversions use Java rules.
+/// Physical Arrow layouts, NULL containers and sliced values are reconciled
+/// by the same implementation used by the table readers.
+/// Nested output fields permit runtime NULLs from casts or absent historical
+/// fields; the logical target type remains unchanged.
+pub fn cast_array_for_schema_evolution(
+ array: &arrow_array::ArrayRef,
+ source_type: &crate::spec::DataType,
+ target_type: &crate::spec::DataType,
+) -> crate::Result<arrow_array::ArrayRef> {
+ super::nested_evolution::evolve_column(array, source_type, target_type)
+}
+
/// Sentinel value indicating a field does not exist in the data schema.
pub const NULL_FIELD_INDEX: i32 = -1;
diff --git a/crates/paimon/src/spec/types.rs b/crates/paimon/src/spec/types.rs
index 0ff00663..2fdee4dd 100644
--- a/crates/paimon/src/spec/types.rs
+++ b/crates/paimon/src/spec/types.rs
@@ -119,15 +119,10 @@ fn type_base_name(s: &str) -> &str {
/// Nullability of a plain keyword type (`NAME` or `NAME NOT NULL`), `None`
when `s` is neither.
fn plain_type_nullable(s: &str, name: &str) -> Option<bool> {
- let (base, rest) = s.split_once(' ').unwrap_or((s, ""));
- if base != name {
- None
- } else if rest.is_empty() {
- Some(true)
- } else if rest.contains("NOT NULL") {
- Some(false)
- } else {
- None
+ match s.strip_prefix(name)?.trim() {
+ "" => Some(true),
+ "NOT NULL" => Some(false),
+ _ => None,
}
}
@@ -199,6 +194,7 @@ impl DataType {
"TINYINT" =>
DataType::TinyInt(TinyIntType::with_nullable(plain("TINYINT")?)),
"SMALLINT" =>
DataType::SmallInt(SmallIntType::with_nullable(plain("SMALLINT")?)),
"INT" => DataType::Int(IntType::with_nullable(plain("INT")?)),
+ "INTEGER" =>
DataType::Int(IntType::with_nullable(plain("INTEGER")?)),
"BIGINT" =>
DataType::BigInt(BigIntType::with_nullable(plain("BIGINT")?)),
"FLOAT" =>
DataType::Float(FloatType::with_nullable(plain("FLOAT")?)),
"DOUBLE" =>
DataType::Double(DoubleType::with_nullable(plain("DOUBLE")?)),
@@ -1014,6 +1010,9 @@ impl FromStr for DecimalType {
type Err = Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
+ if let Some(nullable) = plain_type_nullable(s, "DECIMAL") {
+ return Self::with_nullable(nullable, Self::DEFAULT_PRECISION,
Self::DEFAULT_SCALE);
+ }
if !s.starts_with(serde_utils::DECIMAL::NAME) {
return DataTypeInvalidSnafu {
message: "Invalid DECIMAL type. Expected string to start with
'DECIMAL'.",
@@ -1289,6 +1288,11 @@ impl FromStr for LocalZonedTimestampType {
type Err = Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
+ if let Some(nullable) = plain_type_nullable(s, "TIMESTAMP_LTZ")
+ .or_else(|| plain_type_nullable(s, "TIMESTAMP WITH LOCAL TIME
ZONE"))
+ {
+ return Self::with_nullable(nullable, Self::DEFAULT_PRECISION);
+ }
// PyPaimon persists the TIMESTAMP_LTZ(p) alias for this type.
let alias = s.starts_with("TIMESTAMP_LTZ(");
if !alias && (!s.starts_with(serde_utils::LocalZonedTimestamp::NAME)
|| !s.contains("WITH"))
@@ -1442,6 +1446,9 @@ impl FromStr for TimeType {
type Err = Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
+ if let Some(nullable) = plain_type_nullable(s, "TIME") {
+ return Self::with_nullable(nullable, Self::DEFAULT_PRECISION);
+ }
if !s.starts_with(serde_utils::TIME::NAME) || s.contains("STAMP") {
return DataTypeInvalidSnafu {
message: "Invalid TIME type. Expected string to start with
'TIME'.",
@@ -1539,6 +1546,9 @@ impl FromStr for TimestampType {
type Err = Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
+ if let Some(nullable) = plain_type_nullable(s, "TIMESTAMP") {
+ return Self::with_nullable(nullable, Self::DEFAULT_PRECISION);
+ }
if !s.starts_with(serde_utils::TIMESTAMP::NAME)
|| s.starts_with("TIMESTAMP_LTZ")
|| s.contains("WITH LOCAL TIME ZONE")
@@ -2921,6 +2931,40 @@ mod tests {
}
}
+ #[test]
+ fn test_atomic_defaults_and_integer_alias_match_java() {
+ let cases = [
+ ("INTEGER", DataType::Int(IntType::new())),
+ ("DECIMAL", DataType::Decimal(DecimalType::default())),
+ ("TIME", DataType::Time(TimeType::default())),
+ ("TIMESTAMP", DataType::Timestamp(TimestampType::default())),
+ (
+ "TIMESTAMP_LTZ",
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::default()),
+ ),
+ (
+ "TIMESTAMP WITH LOCAL TIME ZONE",
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::default()),
+ ),
+ ];
+ for (name, expected) in cases {
+ for nullable in [true, false] {
+ let input = format!("{name}{}", if nullable { "" } else { "
NOT NULL" });
+ let parsed: DataType =
serde_json::from_value(serde_json::json!(input)).unwrap();
+ assert_eq!(
+ parsed,
+ expected.copy_with_nullable(nullable).unwrap(),
+ "{input}"
+ );
+ assert_eq!(
+
serde_json::from_str::<DataType>(&serde_json::to_string(&parsed).unwrap())
+ .unwrap(),
+ parsed
+ );
+ }
+ }
+ }
+
#[test]
fn test_string_type_serializes_like_java() {
let nullable = DataType::VarChar(VarCharType::string_type());
diff --git a/crates/paimon/src/table/audit_log_table/merge.rs
b/crates/paimon/src/table/audit_log_table/merge.rs
index 801fa5cc..b036124c 100644
--- a/crates/paimon/src/table/audit_log_table/merge.rs
+++ b/crates/paimon/src/table/audit_log_table/merge.rs
@@ -21,7 +21,6 @@ use super::super::sort_merge::{
compare_sequence_order, AggregateMergeFunction, BufferedBatch,
MergeFunction, MergeResult,
MergeRow, PartialUpdateMergeFunction,
};
-use crate::arrow::build_target_arrow_schema;
use crate::spec::{
CoreOptions, DataField, MergeEngine, RowKind, SEQUENCE_NUMBER_FIELD_ID,
VALUE_KIND_FIELD_ID,
};
@@ -76,7 +75,7 @@ pub(in crate::table) fn new_merge_function(
Ok(Box::new(AuditValueMergeFunction {
inner,
value_projection,
- value_schema: build_target_arrow_schema(&value_fields)?,
+ value_schema: crate::arrow::build_read_arrow_schema(&value_fields)?,
fields: fields.to_vec(),
}))
}
diff --git a/crates/paimon/src/table/audit_log_table/read.rs
b/crates/paimon/src/table/audit_log_table/read.rs
index 6efce91a..c4bf56b1 100644
--- a/crates/paimon/src/table/audit_log_table/read.rs
+++ b/crates/paimon/src/table/audit_log_table/read.rs
@@ -21,7 +21,6 @@ use super::{
audit_sequence_number_enabled, rowkind_array_from_column, PaimonTableRead,
TableRead,
TableReadKind, MAX_MERGE_INPUT_STREAMS,
};
-use crate::arrow::build_target_arrow_schema;
use crate::spec::{
DataField, DataType, MergeEngine, TinyIntType, ROW_KIND_FIELD_ID,
SEQUENCE_NUMBER_FIELD_ID,
VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME,
@@ -79,7 +78,7 @@ impl<'a> AuditLogRead<'a> {
source: None,
});
}
- let audit_schema = build_target_arrow_schema(&output_read_type)?;
+ let audit_schema =
crate::arrow::build_read_arrow_schema(&output_read_type)?;
let mut read_type: Vec<_> = output_read_type
.iter()
.filter(|field| field.id() != ROW_KIND_FIELD_ID)
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index a29c9241..411bcb57 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -26,7 +26,7 @@ use super::read_limit::take_limited_batch;
use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::format::FilePredicates;
use crate::arrow::format::MosaicPrefetchOptions;
-use crate::arrow::{build_target_arrow_schema, ReadBudget};
+use crate::arrow::ReadBudget;
use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
use crate::file_index::evaluator::evaluate_file_index;
use crate::file_index::file_index_result::FileIndexResult;
@@ -163,7 +163,7 @@ impl DataEvolutionReader {
.filter(|f| f.name() != ROW_ID_FIELD_NAME)
.cloned()
.collect();
- let output_schema = build_target_arrow_schema(&read_type)?;
+ let output_schema = crate::arrow::build_read_arrow_schema(&read_type)?;
// Widen the file read set with predicate columns not already projected
// so the residual filter can evaluate every leaf. Extras land strictly
@@ -197,7 +197,7 @@ impl DataEvolutionReader {
}
None => None,
};
- let wide_output_schema = build_target_arrow_schema(&wide_read_type)?;
+ let wide_output_schema =
crate::arrow::build_read_arrow_schema(&wide_read_type)?;
Ok(Self {
file_io,
@@ -996,7 +996,7 @@ impl DataEvolutionReader {
// Match the input cap so a LIMIT cannot resolve payloads for rows that
// are only going to be discarded from the merge output.
let merge_batch_size = batch_size.unwrap_or(1024).clamp(1, 1024);
- let target_schema = build_target_arrow_schema(&read_type)?;
+ let target_schema = crate::arrow::build_read_arrow_schema(&read_type)?;
Ok(try_stream! {
let (file_infos, file_schema_nested_enabled) = load_file_infos(
@@ -1931,7 +1931,7 @@ fn read_vector_bunch_with_null_gaps_stream(
}
}
- let target_schema = build_target_arrow_schema(read_fields)?;
+ let target_schema = crate::arrow::build_read_arrow_schema(read_fields)?;
let null_batch_size = batch_size.unwrap_or(1024).max(1);
let split = split.clone();
@@ -7744,7 +7744,8 @@ mod tests {
None,
);
- let row_schema =
build_target_arrow_schema(std::slice::from_ref(&id_field)).unwrap();
+ let row_schema =
+
crate::arrow::build_target_arrow_schema(std::slice::from_ref(&id_field)).unwrap();
let batch = RecordBatch::try_new(
row_schema.clone(),
vec![Arc::new(Int32Array::from(vec![1, 2]))],
diff --git a/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
b/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
index a650ab20..5b71a91d 100644
--- a/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
+++ b/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
@@ -19,6 +19,7 @@ use super::{
blob_file_row_range, row_range_overlaps_any,
selected_absolute_row_ranges_for_file, BlobBunch,
DeletionVectorContext,
};
+#[cfg(test)]
use crate::arrow::build_target_arrow_schema;
use crate::arrow::format::blob::{
build_blob_array_batch, build_blob_batch, build_blob_map_batch,
BlobFieldKind, BlobReadValue,
@@ -164,7 +165,7 @@ pub(super) fn read(
});
}
- let target_schema = build_target_arrow_schema(&read_fields)?;
+ let target_schema = crate::arrow::build_read_arrow_schema(&read_fields)?;
let field_kind = match read_fields[0].data_type() {
DataType::Blob(_) => BlobFieldKind::Scalar,
DataType::Array(_) => BlobFieldKind::Array,
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index d1425d16..b97f051a 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+#[cfg(test)]
use crate::arrow::build_target_arrow_schema;
use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::format::{
@@ -617,7 +618,7 @@ impl DataFileReader {
let mosaic_prefetch = self.mosaic_prefetch;
let read_timing = self.read_timing.clone();
- let target_schema = build_target_arrow_schema(&read_type)?;
+ let target_schema = crate::arrow::build_read_arrow_schema(&read_type)?;
let file_fields = data_fields.clone().unwrap_or_else(||
table_fields.clone());
let data_schema_fields = data_schema_fields_for_file(
&file_fields,
@@ -954,7 +955,7 @@ impl DataFileReader {
let table_options = Arc::clone(&self.table_options);
let mosaic_prefetch = self.mosaic_prefetch;
- let target_schema = build_target_arrow_schema(&read_type)?;
+ let target_schema = crate::arrow::build_read_arrow_schema(&read_type)?;
let file_fields = data_fields.clone().unwrap_or_else(||
table_fields.clone());
let data_schema_fields =
data_schema_fields_for_file(&file_fields,
file_meta.write_cols.as_deref(), None)?;
diff --git a/crates/paimon/src/table/format_table_read.rs
b/crates/paimon/src/table/format_table_read.rs
index 3eb95274..e8a0fea4 100644
--- a/crates/paimon/src/table/format_table_read.rs
+++ b/crates/paimon/src/table/format_table_read.rs
@@ -23,7 +23,7 @@ use super::table_read::{configured_mosaic_prefetch,
configured_parquet_read_budg
use super::{ArrowRecordBatchStream, Table};
use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::partition::partition_array;
-use crate::arrow::{build_target_arrow_schema, ReadBudget};
+use crate::arrow::ReadBudget;
use crate::resource::ResourceContext;
use crate::spec::{DataField, Predicate};
use crate::{DataSplit, Error};
@@ -132,7 +132,7 @@ impl<'a> FormatTableRead<'a> {
// `TableRead` is public and can be constructed and filtered directly.
super::row_id_predicate::reject_row_id_filter(&self.data_predicates,
"format tables")?;
let read_type = self.read_type.clone();
- let output_schema = build_target_arrow_schema(&read_type)?;
+ let output_schema = crate::arrow::build_read_arrow_schema(&read_type)?;
let partition_keys = self.table.schema().partition_keys().to_vec();
let partition_fields = self.table.schema().partition_fields();
let (data_read_type, data_columns, partition_columns) =
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index f50585d6..44c9a07a 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -32,7 +32,7 @@ use super::sort_merge::{
};
use crate::arrow::format::MosaicPrefetchOptions;
use crate::arrow::shredding::variant::assemble_variant_extraction_array;
-use crate::arrow::{build_target_arrow_schema, ReadBudget};
+use crate::arrow::ReadBudget;
use crate::deletion_vector::DeletionVectorFactory;
use crate::file_index::evaluator::evaluate_file_index;
use crate::file_index::file_index_result::FileIndexResult;
@@ -524,10 +524,10 @@ impl KeyValueFileReader {
internal_read_type.push(value_kind_field);
internal_read_type.extend(user_fields.clone());
- let internal_schema = build_target_arrow_schema(&internal_read_type)?;
+ let internal_schema =
crate::arrow::build_read_arrow_schema(&internal_read_type)?;
// Output schema: user's read_type order
- let output_schema = build_target_arrow_schema(&self.config.read_type)?;
+ let output_schema =
crate::arrow::build_read_arrow_schema(&self.config.read_type)?;
// Indices within internal_schema (offset 2 for _SEQ and _VK).
let seq_index = 0;
@@ -628,7 +628,7 @@ impl KeyValueFileReader {
let mut merge_output_fields: Vec<DataField> = Vec::new();
merge_output_fields.extend(key_fields);
merge_output_fields.extend(value_fields);
- let merge_output_schema =
build_target_arrow_schema(&merge_output_fields)?;
+ let merge_output_schema =
crate::arrow::build_read_arrow_schema(&merge_output_fields)?;
Ok(try_stream! {
for split_group in &split_groups {
@@ -838,7 +838,10 @@ impl KeyValueFileReader {
match config.read_type[out_idx].data_type() {
PaimonDataType::Row(row) if
is_variant_extraction_row(row) =>
{
-
assemble_variant_extraction_array(column.as_ref(), row)
+ let assembled =
assemble_variant_extraction_array(column.as_ref(), row)?;
+
crate::arrow::nested_evolution::evolve_column(
+ &assembled,
config.read_type[out_idx].data_type(), config.read_type[out_idx].data_type(),
+ )
}
_ => {
let source =
merge_read_type.iter().find(|field| field.id() ==
config.read_type[out_idx].id()).unwrap();
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index bd8f2cd5..92d5b7c9 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -175,6 +175,7 @@ pub(crate) mod vector_search_result;
mod vector_search_test_utils;
mod vindex_index_build_builder;
mod write_batch_normalize;
+pub use write_batch_normalize::normalize_write_array;
mod write_builder;
use crate::Result;
diff --git a/crates/paimon/src/table/query_auth/rules.rs
b/crates/paimon/src/table/query_auth/rules.rs
index 98187099..83387330 100644
--- a/crates/paimon/src/table/query_auth/rules.rs
+++ b/crates/paimon/src/table/query_auth/rules.rs
@@ -1475,18 +1475,15 @@ mod tests {
field(4, "b",
DataType::VarBinary(VarBinaryType::new(10).unwrap())),
field(5, "i", DataType::Int(IntType::new())),
];
- for (target, mask) in [
- ("ts", cast("i", "TIMESTAMP(3)")),
- ("d", cast("b", "DOUBLE")),
- ] {
- let message = refusal(parse(&[], &[(target, &mask)], &fields));
- assert!(
- message.contains("existing Python/Java scalar"),
- "{mask}: {message}"
- );
- }
+ let mask = cast("b", "DOUBLE");
+ let message = refusal(parse(&[], &[("d", &mask)], &fields));
+ assert!(
+ message.contains("existing Python/Java scalar"),
+ "{mask}: {message}"
+ );
for (target, mask) in [
("n", cast("s", "BIGINT")),
+ ("ts", cast("i", "TIMESTAMP(3)")),
("ts", cast("n", "TIMESTAMP(3)")),
("s", cast("d", "VARCHAR(10)")),
("s", cast("b", "VARCHAR(10)")),
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 04d74400..6242a905 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -22,7 +22,6 @@ use super::incremental_scan::{IncrementalPlan,
IncrementalScanMode, IncrementalS
use super::kv_file_reader::{KeyValueFileReader, KeyValueReadConfig};
use super::read_builder::split_scan_predicates;
use super::{ArrowRecordBatchStream, Table};
-use crate::arrow::build_target_arrow_schema;
use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
use crate::arrow::format::MosaicPrefetchOptions;
use crate::arrow::ReadBudget;
@@ -852,7 +851,7 @@ impl<'a> PaimonTableRead<'a> {
let key_indices = primary_key_indices(self.table, &diff_read_type)?;
ensure_diff_supported_read_type(&diff_read_type, &key_indices)?;
let value_indices = value_indices_for_diff(self.table,
&diff_read_type);
- let output_schema = build_target_arrow_schema(&self.read_type)?;
+ let output_schema =
crate::arrow::build_read_arrow_schema(&self.read_type)?;
let output_col_indices = self
.read_type
.iter()
@@ -1762,7 +1761,7 @@ fn audit_schema_for_read_type(
));
}
fields.extend(read_type.iter().cloned());
- build_target_arrow_schema(&fields)
+ crate::arrow::build_read_arrow_schema(&fields)
}
fn audit_sequence_number_enabled(table: &Table) -> bool {
diff --git a/crates/paimon/src/table/write_batch_normalize.rs
b/crates/paimon/src/table/write_batch_normalize.rs
index 57fce025..ad5966cc 100644
--- a/crates/paimon/src/table/write_batch_normalize.rs
+++ b/crates/paimon/src/table/write_batch_normalize.rs
@@ -32,7 +32,9 @@ use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{DataType, Field};
use std::sync::Arc;
-pub(super) fn normalize_write_array(array: &ArrayRef, expected: &DataType) ->
Result<ArrayRef> {
+/// Normalize equivalent Arrow layouts for writing, validating NOT NULL
children
+/// only in values visible through non-NULL parents.
+pub fn normalize_write_array(array: &ArrayRef, expected: &DataType) ->
Result<ArrayRef> {
normalize_visible_array(array, expected, None)
}
