JunRuiLee commented on code in PR #805: URL: https://github.com/apache/paimon-rust/pull/805#discussion_r3976676927
########## crates/paimon/src/arrow/nested_evolution.rs: ########## @@ -0,0 +1,797 @@ +// 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. + +//! Read-side reconciliation of one decoded column against the read (table) +//! schema, for the cases Arrow's own `cast` cannot express: +//! +//! - a ROW child present in the read schema but **absent from the data file** +//! (an `ALTER TABLE ... ADD COLUMN parent.child` that landed after the file +//! was written) is filled with NULLs; +//! - a ROW child the read schema does not ask for is dropped (nested +//! projection); +//! - children are paired by **field id**, so a renamed nested column still +//! resolves, and a leaf whose type was promoted is cast. +//! +//! Mirrors Java `SchemaEvolutionUtil.createRowCastExecutor` +//! (paimon-core/src/main/java/org/apache/paimon/schema/SchemaEvolutionUtil.java), +//! which builds the same id-based index mapping per ROW level and yields NULL +//! for a target child with no source counterpart. + +use std::sync::Arc; + +use arrow_array::{new_null_array, Array, ArrayRef, ListArray, MapArray, StructArray}; +use arrow_cast::cast; +use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Fields}; + +use crate::arrow::paimon_type_to_arrow; +use crate::spec::{is_variant_extraction_row_type, DataType, MapType, RowType}; + +/// Reconcile `source` (as described by `source_type`, the type the data file +/// 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. +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)?; + // 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 + // different id. Serving the old column there would return the dropped + // field's values instead of NULL. Gate the fast path on the Paimon types + // instead, as Java `SchemaEvolutionUtil.createCastExecutor` does with + // `equalsIgnoreNullable`; the Arrow check stays so a decoded array that does + // not actually match the target still goes through the cast below. + if source_type.equals_ignore_nullable(target_type) && source.data_type() == &target_arrow { + return Ok(source.clone()); + } + + match (target_type, source_type) { + // A variant-extraction ROW is synthetic: its fields are numbered by + // position, not by schema field id, so pairing them by id would mix + // columns up. `prune_data_type` keeps such a row verbatim, so there is + // 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) => {} + (DataType::Row(target_row), DataType::Row(source_row)) => { + return evolve_struct(source, source_row, target_row) + } + (DataType::Array(target_array), DataType::Array(source_array)) => { + return evolve_list( + source, + source_array.element_type(), + target_array.element_type(), + ) + } + (DataType::Map(target_map), DataType::Map(source_map)) => { + return evolve_map(source, source_map, target_map) + } + (DataType::Multiset(target_multiset), DataType::Multiset(source_multiset)) => { + return evolve_multiset( + source, + source_multiset.element_type(), + target_multiset.element_type(), + ) + } + _ => {} + } + + cast(source, &target_arrow).map_err(|e| crate::Error::UnexpectedError { + message: format!( + "failed to cast nested value from {:?} to {:?} during schema evolution", + source.data_type(), + target_arrow + ), + source: Some(Box::new(e)), + }) +} + +/// Rebuild a struct array to `target_row`: pair children by field id, recurse, +/// NULL-fill target children the source does not have, and preserve the +/// source's row-level validity buffer. +fn evolve_struct( + source: &ArrayRef, + source_row: &RowType, + target_row: &RowType, +) -> crate::Result<ArrayRef> { + let source_struct = source + .as_any() + .downcast_ref::<StructArray>() + .ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "expected a struct array for ROW schema evolution, got {:?}", + source.data_type() + ), + source: None, + })?; + + 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())?; + fields.push(Arc::new(ArrowField::new( + target_field.name(), + target_arrow.clone(), + target_field.data_type().is_nullable(), + ))); + + // Pair by id (Java: `createIndexMapping` over the ROW's fields). + let source_field = source_row + .fields() + .iter() + .find(|source_field| source_field.id() == target_field.id()); + + match source_field { + // Not in the data file's schema: the column was added to the ROW + // after this file was written. + None => arrays.push(new_null_array(&target_arrow, source_struct.len())), + Some(source_field) => { + // In the file's schema, so the decoder must have produced it — + // under the source field's own name, which is what the file + // labels the child with. A gap here means the file, its schema + // or the format reader's projection disagree; NULL-filling it + // would pass that off as legitimately absent data. + let column = source_struct + .column_by_name(source_field.name()) + .ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "nested field '{}' (id {}) is declared by the data file's schema \ + but missing from the decoded struct {:?}", + source_field.name(), + source_field.id(), + source_struct.data_type() + ), + source: None, + })?; + arrays.push(evolve_column( + column, + source_field.data_type(), + target_field.data_type(), + )?) + } + } + } + + let evolved = StructArray::try_new(fields.into(), arrays, source_struct.nulls().cloned()) + .map_err(|e| crate::Error::DataInvalid { + message: format!("failed to build schema-evolved struct: {e}"), + source: None, + })?; + Ok(Arc::new(evolved)) +} + +/// Rebuild an ARRAY whose element type evolved, reusing the source offsets and +/// validity so only the element values are reconciled. Mirrors Java +/// `SchemaEvolutionUtil.createArrayCastExecutor`. +fn evolve_list( + source: &ArrayRef, + source_element: &DataType, + target_element: &DataType, +) -> crate::Result<ArrayRef> { + let list = + source + .as_any() + .downcast_ref::<ListArray>() + .ok_or_else(|| crate::Error::DataInvalid { Review Comment: Confirmed, reproduced both layouts. `evolve_list` now normalizes a non-`ListArray` source through Arrow's `cast` to the source element's `List` type first — which range-checks the large offsets — and reconciles the element on the result. Tests cover `LargeList` carrying an added element field, and `FixedSizeList`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
