diff --git a/Cargo.lock b/Cargo.lock index 69911e8bc..22ea40a77 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 f266a0a4b..2313f8f4c 100644 --- a/DEPENDENCIES.rust.tsv +++ b/DEPENDENCIES.rust.tsv @@ -504,7 +504,7 @@ rust-stemmers@1.2.0 X X rustc-hash@2.1.3 X X rustc_version@0.4.1 X X rustix@1.1.4 X X X -rustls@0.23.45 X X X +rustls@0.23.46 X X X rustls-native-certs@0.8.4 X X X rustls-pki-types@1.15.0 X X rustls-platform-verifier@0.7.0 X X diff --git a/benchmarks/tpcds/DEPENDENCIES.rust.tsv b/benchmarks/tpcds/DEPENDENCIES.rust.tsv index 6e030d487..24915be44 100644 --- a/benchmarks/tpcds/DEPENDENCIES.rust.tsv +++ b/benchmarks/tpcds/DEPENDENCIES.rust.tsv @@ -328,7 +328,7 @@ roaring@0.11.4 X X rust-ini@0.21.3 X rustc_version@0.4.1 X X rustix@1.1.4 X X X -rustls@0.23.45 X X X +rustls@0.23.46 X X X rustls-native-certs@0.8.4 X X X rustls-pki-types@1.15.0 X X rustls-platform-verifier@0.7.0 X X diff --git a/bindings/c/DEPENDENCIES.rust.tsv b/bindings/c/DEPENDENCIES.rust.tsv index 8b8f6cea6..d3a70ef8c 100644 --- a/bindings/c/DEPENDENCIES.rust.tsv +++ b/bindings/c/DEPENDENCIES.rust.tsv @@ -287,7 +287,7 @@ rsa@0.9.10 X X rust-ini@0.21.3 X rustc_version@0.4.1 X X rustix@1.1.4 X X X -rustls@0.23.45 X X X +rustls@0.23.46 X X X rustls-native-certs@0.8.4 X X X rustls-pki-types@1.15.0 X X rustls-platform-verifier@0.7.0 X X diff --git a/bindings/go/DEPENDENCIES.rust.tsv b/bindings/go/DEPENDENCIES.rust.tsv index 8b8f6cea6..d3a70ef8c 100644 --- a/bindings/go/DEPENDENCIES.rust.tsv +++ b/bindings/go/DEPENDENCIES.rust.tsv @@ -287,7 +287,7 @@ rsa@0.9.10 X X rust-ini@0.21.3 X rustc_version@0.4.1 X X rustix@1.1.4 X X X -rustls@0.23.45 X X X +rustls@0.23.46 X X X rustls-native-certs@0.8.4 X X X rustls-pki-types@1.15.0 X X rustls-platform-verifier@0.7.0 X X diff --git a/bindings/python/DEPENDENCIES.rust.tsv b/bindings/python/DEPENDENCIES.rust.tsv index 9c4513eb4..563899bf6 100644 --- a/bindings/python/DEPENDENCIES.rust.tsv +++ b/bindings/python/DEPENDENCIES.rust.tsv @@ -428,7 +428,7 @@ rust-stemmers@1.2.0 X X rustc-hash@2.1.3 X X rustc_version@0.4.1 X X rustix@1.1.4 X X X -rustls@0.23.45 X X X +rustls@0.23.46 X X X rustls-native-certs@0.8.4 X X X rustls-pki-types@1.15.0 X X rustls-platform-verifier@0.7.0 X X diff --git a/bindings/python/python/pypaimon_rust/data.pyi b/bindings/python/python/pypaimon_rust/data.pyi index c46aa2bbf..9b50d3505 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 000000000..547c14cb3 --- /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> { + let parse = |value| -> PyResult { + 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> { + let fields: Vec = 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> { + 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::()?; + 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 6b2fcc543..a15c7600d 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 c8f095a76..3c350db3d 100644 --- a/crates/integration_tests/DEPENDENCIES.rust.tsv +++ b/crates/integration_tests/DEPENDENCIES.rust.tsv @@ -257,7 +257,7 @@ roaring@0.11.4 X X rust-ini@0.21.3 X rustc_version@0.4.1 X X rustix@1.1.4 X X X -rustls@0.23.45 X X X +rustls@0.23.46 X X X rustls-native-certs@0.8.4 X X X rustls-pki-types@1.15.0 X X rustls-platform-verifier@0.7.0 X X diff --git a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv index 88bc626bf..091e6a231 100644 --- a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv +++ b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv @@ -423,7 +423,7 @@ rust-stemmers@1.2.0 X X rustc-hash@2.1.3 X X rustc_version@0.4.1 X X rustix@1.1.4 X X X -rustls@0.23.45 X X X +rustls@0.23.46 X X X rustls-native-certs@0.8.4 X X X rustls-pki-types@1.15.0 X X rustls-platform-verifier@0.7.0 X X diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index ad74a5949..9a4727c5b 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 { 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 { 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 6013c504a..4955b302c 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 aa748ee16..55d7caf6d 100644 --- a/crates/paimon-rest-server/DEPENDENCIES.rust.tsv +++ b/crates/paimon-rest-server/DEPENDENCIES.rust.tsv @@ -261,7 +261,7 @@ roaring@0.11.4 X X rust-ini@0.21.3 X rustc_version@0.4.1 X X rustix@1.1.4 X X X -rustls@0.23.45 X X X +rustls@0.23.46 X X X rustls-native-certs@0.8.4 X X X rustls-pki-types@1.15.0 X X rustls-platform-verifier@0.7.0 X X diff --git a/crates/paimon/DEPENDENCIES.rust.tsv b/crates/paimon/DEPENDENCIES.rust.tsv index 94caabead..3a0678c2b 100644 --- a/crates/paimon/DEPENDENCIES.rust.tsv +++ b/crates/paimon/DEPENDENCIES.rust.tsv @@ -420,7 +420,7 @@ rust-stemmers@1.2.0 X X rustc-hash@2.1.3 X X rustc_version@0.4.1 X X rustix@1.1.4 X X X -rustls@0.23.45 X X X +rustls@0.23.46 X X X rustls-native-certs@0.8.4 X X X rustls-pki-types@1.15.0 X X rustls-platform-verifier@0.7.0 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 dee9fb83c..56422f192 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::>(), + ) + .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 4ce174b3f..4d5adbf72 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::>>()?; 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 c0bd85d1b..5e85e1102 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 a43490999..cd99a7024 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 721018768..a71198099 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 crate::Result { + 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::>>()? + .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> { + 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::>>()?; + 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 09fc1288b..2e511139e 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::() + .is_some_and(|rows| { + rows.column_names() + == row + .fields() + .iter() + .map(|field| field.name()) + .collect::>() + }) + { + 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::() .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 { - 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::>>()?; 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::().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::().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> = Vec::with_capacity(target_row.fields().len()); let mut arrays: Vec = 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 { - // 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::() { - 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::() - .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 { + if let Some(list) = source.as_any().downcast_ref::() { + 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::() + .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 { + if rows.null_count() == 0 { + return Ok(column.clone()); + } + let hidden = BooleanArray::from( + (0..rows.len()) + .map(|row| rows.is_null(row)) + .collect::>(), + ); + 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( + offsets: &[O], + nulls: Option<&NullBuffer>, + values: &ArrayRef, +) -> crate::Result<(OffsetBuffer, 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::>(), + ) + .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( + list: &GenericListArray, +) -> crate::Result { + 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 { + if let Some(list) = source + .as_any() + .downcast_ref::() + { + return visible_generic_list(list); + } + visible_generic_list(&normalize_list(source)?) +} + +fn visible_map(map: &MapArray) -> crate::Result { + 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::() + .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 { - 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 { - 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::().unwrap(); + assert_eq!( + output + .column(0) + .as_any() + .downcast_ref::() + .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::() @@ -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::().unwrap(); + assert_eq!(rows.value_offsets(), &[0, 1, 1, 2]); + assert_eq!( + rows.values().as_any().downcast_ref::().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::() + .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::().unwrap(); + assert_eq!( + list.values() + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .collect::>(), + 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::() + .unwrap() + .value(0), + 7 + ); + assert_eq!( + rows.column(1) + .as_any() + .downcast_ref::() + .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::().unwrap(); + assert_eq!(rows.value_offsets(), &[0, 1, 1, 2]); + assert_eq!( + rows.values().as_any().downcast_ref::().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 ea9450b34..5a6cff05d 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 { + 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 0ff006639..2fdee4dd3 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 { - 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 { + 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 { + 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 { + 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 { + 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::(&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 801fa5cc7..b036124cd 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 6efce91a0..c4bf56b14 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 a29c92411..411bcb575 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 a650ab204..5b71a91d8 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 d1425d163..b97f051aa 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 3eb95274e..e8a0fea42 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 f50585d6e..44c9a07ac 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 = 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 bd8f2cd58..92d5b7c9f 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 98187099d..83387330f 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 04d74400a..6242a9051 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 57fce0259..ad5966cc4 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 { +/// 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 { normalize_visible_array(array, expected, None) }