From a405ac1af85a3553a1f80bfc630f92ba1dc837f7 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 01:16:11 +0800 Subject: [PATCH] feat(read): complete nested projection pushdown for Parquet --- crates/paimon/src/arrow/format/avro.rs | 35 +- crates/paimon/src/arrow/format/mod.rs | 82 +- crates/paimon/src/arrow/format/parquet.rs | 106 ++- crates/paimon/src/arrow/format/shredding.rs | 13 + crates/paimon/src/arrow/format/text.rs | 2 +- crates/paimon/src/arrow/format/vortex.rs | 152 ++++ .../paimon/src/arrow/format/vortex/nested.rs | 278 ++++++ crates/paimon/src/arrow/shredding/map.rs | 31 +- crates/paimon/src/table/data_file_reader.rs | 223 +++-- .../nested_projection_tests.rs | 835 ++++++++++++++++++ .../GenerateLegacyLists.java | 61 ++ .../testdata/nested_projection/array.parquet | Bin 0 -> 550 bytes .../nested_projection/generate_orc.py | 45 + .../testdata/nested_projection/nested.orc | Bin 0 -> 4018 bytes .../nested_projection/repeated.parquet | Bin 0 -> 520 bytes .../nested_projection/standard.parquet | Bin 0 -> 578 bytes .../testdata/nested_projection/tuple.parquet | Bin 0 -> 568 bytes 17 files changed, 1706 insertions(+), 157 deletions(-) create mode 100644 crates/paimon/src/arrow/format/vortex/nested.rs create mode 100644 crates/paimon/src/table/data_file_reader/nested_projection_tests.rs create mode 100644 crates/paimon/testdata/nested_projection/GenerateLegacyLists.java create mode 100644 crates/paimon/testdata/nested_projection/array.parquet create mode 100644 crates/paimon/testdata/nested_projection/generate_orc.py create mode 100644 crates/paimon/testdata/nested_projection/nested.orc create mode 100644 crates/paimon/testdata/nested_projection/repeated.parquet create mode 100644 crates/paimon/testdata/nested_projection/standard.parquet create mode 100644 crates/paimon/testdata/nested_projection/tuple.parquet diff --git a/crates/paimon/src/arrow/format/avro.rs b/crates/paimon/src/arrow/format/avro.rs index 37f01c513..2b593559c 100644 --- a/crates/paimon/src/arrow/format/avro.rs +++ b/crates/paimon/src/arrow/format/avro.rs @@ -119,14 +119,15 @@ impl FormatFileReader for AvroFormatReader { // Value access helpers — work directly with apache_avro::types::Value // --------------------------------------------------------------------------- -/// Find the position of a named field from the first record. +/// Find the position of a named field from a populated record. /// All records in an Avro file share the same schema, so the index is valid /// for every record. fn field_index(records: &[Value], name: &str) -> Option { - match records.first() { - Some(Value::Record(fields)) => fields.iter().position(|(n, _)| n == name), + // Null parent ROWs are represented by empty records while building children. + records.iter().find_map(|record| match record { + Value::Record(fields) => fields.iter().position(|(n, _)| n == name), _ => None, - } + }) } /// Look up a field by cached index, unwrapping unions. @@ -829,6 +830,32 @@ mod tests { // unwrap_value // ----------------------------------------------------------------------- + #[test] + fn first_null_nested_row_does_not_hide_later_children() { + let typ = DataType::Row(RowType::new(vec![DataField::new( + 1, + "keep".into(), + DataType::Int(IntType::new()), + )])); + let records = vec![ + Value::Record(vec![("profile".into(), Value::Null)]), + Value::Record(vec![( + "profile".into(), + Value::Record(vec![("keep".into(), Value::Int(7))]), + )]), + ]; + let array = build_column(&records, "profile", &typ, records.len()).unwrap(); + let rows = array.as_any().downcast_ref::().unwrap(); + assert!(rows.is_null(0)); + let child = rows + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert!(child.is_valid(1)); + assert_eq!(child.value(1), 7); + } + #[test] fn test_unwrap_value_null() { assert!(unwrap_value(&av_null()).is_none()); diff --git a/crates/paimon/src/arrow/format/mod.rs b/crates/paimon/src/arrow/format/mod.rs index 3de32c7e9..c85c0578a 100644 --- a/crates/paimon/src/arrow/format/mod.rs +++ b/crates/paimon/src/arrow/format/mod.rs @@ -69,15 +69,24 @@ pub(crate) struct FilePredicates { /// - Row range selection #[async_trait] pub(crate) trait FormatFileReader: Send + Sync { + /// Whether the decoder can prune fields inside ROW, ARRAY and MAP types. + /// Other formats read complete top-level fields and let the caller project. + fn supports_nested_projection(&self) -> bool { + false + } + /// Choose the fields the decoder must actually read. Most columnar formats /// can read the projection, while positional formats need the complete /// physical data schema to decode each row. fn select_read_fields( &self, - _data_schema_fields: &[DataField], + data_schema_fields: &[DataField], projected_fields: &[DataField], ) -> Vec { - projected_fields.to_vec() + if self.supports_nested_projection() { + return projected_fields.to_vec(); + } + top_level_read_fields(data_schema_fields, projected_fields) } /// Read a single data file, returning a stream of RecordBatches containing @@ -367,6 +376,29 @@ pub(crate) fn create_format_reader( .map(|configured| configured.reader) } +/// Expand nested projections for readers that only select whole root fields. +/// Keep selected-key instructions so the outer reader can assemble their ROW. +fn top_level_read_fields( + data_schema_fields: &[DataField], + projected_fields: &[DataField], +) -> Vec { + projected_fields + .iter() + .map(|projected| { + let Some(source) = data_schema_fields.iter().find(|f| f.id() == projected.id()) else { + return projected.clone(); + }; + let mut source = source.clone(); + if projected.description().is_some_and(|description| { + description.starts_with(crate::spec::MAP_SELECTED_KEYS_METADATA_KEY) + }) { + source = source.with_description(projected.description().map(str::to_string)); + } + source + }) + .collect() +} + pub(crate) struct ConfiguredFormatReader { pub reader: Box, pub read_fields: Vec, @@ -773,6 +805,52 @@ mod tests { } } + #[test] + fn nested_projection_is_expanded_for_formats_without_support() { + use crate::spec::{ArrayType, RowType}; + let child = |id, name: &str| DataField::new(id, name.into(), DataType::Int(IntType::new())); + let field = |children| { + DataField::new( + 0, + "items".into(), + DataType::Array(ArrayType::new(DataType::Row(RowType::new(children)))), + ) + }; + let physical = vec![field(vec![child(1, "first"), child(2, "second")])]; + let projected = vec![field(vec![child(2, "second")])]; + for path in [ + "data.parquet", + "data.orc", + "data.avro", + "data.json", + "data.row", + ] { + let configured = create_format_reader_with_budget( + path, + false, + FormatReadFields { + data_schema: &physical, + projected: &projected, + }, + &HashMap::new(), + None, + blob::DEFAULT_BLOB_READ_PARALLELISM, + MosaicPrefetchOptions::default(), + ) + .unwrap(); + assert_eq!( + configured.read_fields, + if path.ends_with("parquet") { + &projected + } else { + &physical + } + .clone(), + "{path}" + ); + } + } + #[tokio::test] async fn direct_writer_rejects_unsupported_map_shredding_before_creating_file() { use crate::spec::{MapType, VarCharType}; diff --git a/crates/paimon/src/arrow/format/parquet.rs b/crates/paimon/src/arrow/format/parquet.rs index c678b5296..0ade1fbf6 100644 --- a/crates/paimon/src/arrow/format/parquet.rs +++ b/crates/paimon/src/arrow/format/parquet.rs @@ -882,6 +882,10 @@ fn dictionary_variant_metadata_schema( #[async_trait] impl FormatFileReader for ParquetFormatReader { + fn supports_nested_projection(&self) -> bool { + true + } + async fn read_batch_stream( &self, reader: Box, @@ -1378,6 +1382,15 @@ fn read_type_projection_mask( predicates: &[Predicate], null_residuals: &std::collections::HashSet>, ) -> ProjectionMask { + fn is_list(physical: &parquet::schema::types::Type) -> bool { + let info = physical.get_basic_info(); + !physical.is_primitive() + && match info.logical_type_ref() { + Some(logical) => logical == &parquet::basic::LogicalType::List, + None => info.converted_type() == parquet::basic::ConvertedType::LIST, + } + } + fn requested_paths( field: &DataField, physical: &parquet::schema::types::Type, @@ -1393,18 +1406,54 @@ fn read_type_projection_mask( return; } let data_type = field.data_type(); + if let DataType::Array(array) = data_type { + let element = + DataField::new(field.id(), "element".into(), array.element_type().clone()); + if is_list(physical) && physical.get_fields().len() == 1 { + // Follow arrow-rs's LIST conversion rules, including legacy + // two-level lists of ROWs named `array` or `_tuple`. + let repeated = &physical.get_fields()[0]; + path.push(repeated.name().to_string()); + if !repeated.is_primitive() + && repeated.get_fields().len() == 1 + && (is_list(repeated) + || repeated.get_fields()[0].get_basic_info().repetition() + == parquet::basic::Repetition::REPEATED + || (repeated.name() != "array" + && repeated.name() != format!("{}_tuple", physical.name()))) + { + let child = &repeated.get_fields()[0]; + path.push(child.name().to_string()); + requested_paths(&element, child, path, output, null_residuals); + path.pop(); + } else { + requested_paths(&element, repeated, path, output, null_residuals); + } + path.pop(); + return; + } + if physical.get_basic_info().has_repetition() + && physical.get_basic_info().repetition() == parquet::basic::Repetition::REPEATED + { + // Unannotated repeated fields are also decoded as lists. + requested_paths(&element, physical, path, output, null_residuals); + return; + } + } if let DataType::Map(map) = data_type { if !physical.is_primitive() && physical.get_fields().len() == 1 { let entries = &physical.get_fields()[0]; - if !entries.is_primitive() - && entries.get_fields().len() == 2 - && entries.get_fields()[0].name() == "key" - && entries.get_fields()[1].name() == "value" - { + if !entries.is_primitive() && entries.get_fields().len() == 2 { path.push(entries.name().to_string()); let key = &entries.get_fields()[0]; path.push(key.name().to_string()); - super::variant_projection::all_leaves(key, path, output); + requested_paths( + &DataField::new(field.id(), key.name().into(), map.key_type().clone()), + key, + path, + output, + null_residuals, + ); path.pop(); let value = &entries.get_fields()[1]; path.push(value.name().to_string()); @@ -1476,18 +1525,16 @@ fn read_type_projection_mask( .any(|(column, _)| *column == field.name()) { super::variant_projection::all_leaves(root, &mut path, &mut selected); - } else if let Some(children) = - map_plan.and_then(|plan| plan.projected_physical_children(field.name())) + } else if let Some(physical_field) = + map_plan.and_then(|plan| plan.projected_physical_field(field.name())) { - for child in root - .get_fields() - .iter() - .filter(|child| children.contains(child.name())) - { - path.push(child.name().to_string()); - super::variant_projection::all_leaves(child, &mut path, &mut selected); - path.pop(); - } + requested_paths( + &physical_field, + root, + &mut path, + &mut selected, + null_residuals, + ); } else { requested_paths(field, root, &mut path, &mut selected, null_residuals); } @@ -3409,6 +3456,31 @@ mod tests { } } + #[test] + fn array_read_type_masks_standard_and_legacy_list_leaves() { + use crate::spec::RowType; + let requested = vec![DataField::new( + 0, + "items".into(), + DataType::Array(ArrayType::new(DataType::Row(RowType::new(vec![ + DataField::new(2, "wanted".into(), DataType::Int(IntType::new())), + ])))), + )]; + for layout in [ + "optional group items (LIST) { repeated group list { optional group element { optional int32 unused; optional int32 wanted; } } }", + "optional group items (LIST) { repeated group array { optional int32 unused; optional int32 wanted; } }", + "optional group items (LIST) { repeated group items_tuple { optional int32 unused; optional int32 wanted; } }", + "repeated group items { optional int32 unused; optional int32 wanted; }", + ] { + let descriptor = SchemaDescriptor::new(Arc::new( + parse_message_type(&format!("message schema {{ {layout} }}")).unwrap())); + let mask = super::read_type_projection_mask( + &descriptor, &requested, None, &[], &Default::default()); + assert!(!mask.leaf_included(0), "{layout}"); + assert!(mask.leaf_included(1), "{layout}"); + } + } + #[test] fn selected_map_rows_keep_all_map_leaves_without_a_shredding_plan() { use crate::spec::RowType; diff --git a/crates/paimon/src/arrow/format/shredding.rs b/crates/paimon/src/arrow/format/shredding.rs index 5a341a8e9..981e2eccc 100644 --- a/crates/paimon/src/arrow/format/shredding.rs +++ b/crates/paimon/src/arrow/format/shredding.rs @@ -148,6 +148,19 @@ impl ShreddingFormatReader { #[async_trait] impl FormatFileReader for ShreddingFormatReader { + fn supports_nested_projection(&self) -> bool { + self.inner.supports_nested_projection() + } + + fn select_read_fields( + &self, + data_schema_fields: &[DataField], + projected_fields: &[DataField], + ) -> Vec { + self.inner + .select_read_fields(data_schema_fields, projected_fields) + } + async fn read_batch_stream( &self, reader: Box, diff --git a/crates/paimon/src/arrow/format/text.rs b/crates/paimon/src/arrow/format/text.rs index 61f1ce0f2..c2e431e54 100644 --- a/crates/paimon/src/arrow/format/text.rs +++ b/crates/paimon/src/arrow/format/text.rs @@ -633,7 +633,7 @@ impl FormatFileReader for TextFormatReader { // still has to decode their original physical column positions. match self.kind { TextKind::Csv | TextKind::Text => data_schema_fields.to_vec(), - TextKind::Json => projected_fields.to_vec(), + TextKind::Json => super::top_level_read_fields(data_schema_fields, projected_fields), } } diff --git a/crates/paimon/src/arrow/format/vortex.rs b/crates/paimon/src/arrow/format/vortex.rs index 2075b3ab4..92fb7bf95 100644 --- a/crates/paimon/src/arrow/format/vortex.rs +++ b/crates/paimon/src/arrow/format/vortex.rs @@ -42,6 +42,8 @@ use vortex::scan::selection::Selection; use vortex::session::VortexSession; use vortex::VortexSessionDefault; +mod nested; + async fn acquire_vortex_io_permit() -> crate::Result> { static SEMAPHORE: OnceLock = OnceLock::new(); SEMAPHORE @@ -174,6 +176,49 @@ fn read_vortex_batches( return Ok(vec![batch]); } + if scan_schema + .fields() + .iter() + .any(|field| matches!(field.data_type(), ArrowDataType::Struct(_))) + { + // Vortex 0.75 can apply a child ROW's validity to its parent while + // projecting recursive struct layouts. Read complete ROWs and their + // independent validity before applying selection and outer projection. + let names = scan_schema + .fields() + .iter() + .map(|field| field.name().clone()) + .collect::>(); + let mut array = nested::read_array(runtime, &session, &vortex_file, &names)?; + if let Some(ranges) = row_selection.as_ref() { + let Selection::IncludeRoaring(rows) = + row_ranges_to_selection(ranges, vortex_file.row_count()) + else { + unreachable!() + }; + let indices = arrow_array::UInt64Array::from_iter_values(rows.iter()); + let indices = + ArrayRef::from_arrow(&indices, false).map_err(|error| Error::DataInvalid { + message: format!("Failed to build Vortex row selection: {error}"), + source: None, + })?; + // Take before Arrow export so sparse reads do not expand every + // unselected ROW into Arrow buffers. + array = array.take(indices).map_err(|error| Error::DataInvalid { + message: format!("Failed to select Vortex rows: {error}"), + source: None, + })?; + } + let batch = vortex_array_to_record_batch(&session, array, &scan_schema)?; + return Ok(vec![filter_and_project_batch( + batch, + &target_schema, + &read_fields, + &scan_fields, + predicates.as_ref(), + )?]); + } + // Build projection expression for requested fields. let projected_names: Vec<&str> = scan_fields.iter().map(|f| f.name()).collect(); @@ -614,6 +659,113 @@ mod tests { ])) } + #[tokio::test] + async fn nullable_single_child_row_preserves_parent_validity() { + use crate::spec::{IntType, RowType}; + use arrow_array::{Array, StructArray}; + use arrow_buffer::NullBuffer; + use futures::TryStreamExt; + let fields = vec![DataField::new( + 0, + "profile".into(), + DataType::Row(RowType::new(vec![DataField::new( + 1, + "detail".into(), + DataType::Row(RowType::new(vec![DataField::new( + 2, + "keep".into(), + DataType::Int(IntType::new()), + )])), + )])), + )]; + let schema = crate::arrow::build_target_arrow_schema(&fields).unwrap(); + let ArrowDataType::Struct(profile_fields) = schema.field(0).data_type() else { + unreachable!() + }; + let ArrowDataType::Struct(detail_fields) = profile_fields[0].data_type() else { + unreachable!() + }; + let detail = StructArray::new( + detail_fields.clone(), + vec![Arc::new(Int32Array::from(vec![Some(7), None, None]))], + Some(NullBuffer::from(vec![true, false, false])), + ); + let profile = StructArray::new( + profile_fields.clone(), + vec![Arc::new(detail)], + Some(NullBuffer::from(vec![true, true, false])), + ); + let original = RecordBatch::try_new(schema.clone(), vec![Arc::new(profile)]).unwrap(); + let memory = vortex_array_to_record_batch( + &VortexSession::default(), + ArrayRef::from_arrow(original.clone(), false).unwrap(), + &schema, + ) + .unwrap(); + assert!(memory.column(0).is_valid(1), "in-memory conversion"); + for repeats in [1, 3000] { + let input = + arrow_select::concat::concat_batches(&schema, (0..repeats).map(|_| &original)) + .unwrap(); + let io = FileIOBuilder::new("memory").build().unwrap(); + let path = format!("memory:/nullable_single_child_{repeats}.vortex"); + let output = io.new_output(&path).unwrap(); + let mut writer = VortexFormatWriter::new(&output, schema.clone()) + .await + .unwrap(); + writer.write(&input).await.unwrap(); + let size = Box::new(writer).close().await.unwrap().file_size; + let reader = VortexFormatReader; + let batches = reader + .read_batch_stream( + Box::new(io.new_input(&path).unwrap().reader().await.unwrap()), + size, + &fields, + None, + None, + None, + ) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + let actual = arrow_select::concat::concat_batches(&schema, &batches).unwrap(); + assert_eq!(actual, input, "{repeats} repeats"); + if repeats == 3000 { + let ranges = vec![ + RowRange::new(-2, 1), + RowRange::new(1, 1), + RowRange::new(8191, 8192), + RowRange::new(8999, 10000), + ]; + let batches = reader + .read_batch_stream( + Box::new(io.new_input(&path).unwrap().reader().await.unwrap()), + size, + &fields, + None, + None, + Some(ranges), + ) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + let expected = arrow_select::take::take_record_batch( + &input, + &arrow_array::UInt64Array::from(vec![0, 1, 8191, 8192, 8999]), + ) + .unwrap(); + assert_eq!( + arrow_select::concat::concat_batches(&schema, &batches).unwrap(), + expected + ); + } + } + } + fn test_batch(schema: &Arc, ids: Vec, values: Vec) -> RecordBatch { RecordBatch::try_new( schema.clone(), diff --git a/crates/paimon/src/arrow/format/vortex/nested.rs b/crates/paimon/src/arrow/format/vortex/nested.rs new file mode 100644 index 000000000..b19dfd2a0 --- /dev/null +++ b/crates/paimon/src/arrow/format/vortex/nested.rs @@ -0,0 +1,278 @@ +// 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 complete ROWs without Vortex 0.75's recursive StructReader projection. +//! Each ROW has independent validity; a null child must not null its parent. + +use super::CurrentThreadRuntime; +use crate::{Error, Result}; +use std::sync::Arc; +use vortex::array::arrays::struct_::StructArrayExt; +use vortex::array::arrays::{ChunkedArray, DictArray, StructArray}; +use vortex::array::dtype::{DType, FieldNames, StructFields}; +use vortex::array::validity::Validity; +use vortex::array::{ArrayRef, IntoArray, VortexSessionExecute}; +use vortex::file::VortexFile; +use vortex::io::runtime::BlockingRuntime; +use vortex::layout::layouts::{chunked::Chunked, dict::Dict, struct_::Struct}; +use vortex::layout::scan::{scan_builder::ScanBuilder, split_by::SplitBy}; +use vortex::layout::segments::SegmentSource; +use vortex::layout::{LayoutChildType, LayoutRef}; +use vortex::session::VortexSession; + +fn invalid(error: impl std::fmt::Display) -> Error { + Error::DataInvalid { + message: format!("Failed to read complete Vortex ROW: {error}"), + source: None, + } +} + +pub(super) fn read_array( + runtime: &CurrentThreadRuntime, + session: &VortexSession, + file: &VortexFile, + names: &[String], +) -> Result { + read_layout( + runtime, + session, + file.segment_source(), + file.footer().layout(), + Some(names), + ) +} + +fn projected_dtype(dtype: &DType, names: Option<&[String]>) -> Result { + let Some(names) = names else { + return Ok(dtype.clone()); + }; + let DType::Struct(fields, nullable) = dtype else { + return Err(invalid("expected a root ROW")); + }; + let types = names + .iter() + .map(|name| { + fields + .field(name.as_str()) + .ok_or_else(|| invalid(format!("missing root field {name}"))) + }) + .collect::>>()?; + Ok(DType::Struct( + StructFields::new( + FieldNames::from_iter(names.iter().map(String::as_str)), + types, + ), + *nullable, + )) +} + +fn read_layout( + runtime: &CurrentThreadRuntime, + session: &VortexSession, + source: Arc, + layout: &LayoutRef, + roots: Option<&[String]>, +) -> Result { + let dtype = projected_dtype(layout.dtype(), roots)?; + if layout.is::() { + let DType::Struct(fields, _) = &dtype else { + return Err(invalid("invalid struct layout")); + }; + let DType::Struct(original, _) = layout.dtype() else { + return Err(invalid("invalid struct layout")); + }; + let nullable = layout.dtype().is_nullable(); + let validity = if nullable { + Validity::Array(read_layout( + runtime, + session, + source.clone(), + &layout.child(0).map_err(invalid)?, + None, + )?) + } else { + Validity::NonNullable + }; + let mut columns = Vec::with_capacity(fields.nfields()); + for name in fields.names().iter() { + let index = original + .find(name) + .ok_or_else(|| invalid(format!("missing root field {name}")))?; + let child = layout + .child(index + usize::from(nullable)) + .map_err(invalid)?; + columns.push(read_layout(runtime, session, source.clone(), &child, None)?); + } + return StructArray::try_new( + fields.names().clone(), + columns, + usize::try_from(layout.row_count()).map_err(invalid)?, + validity, + ) + .map(IntoArray::into_array) + .map_err(invalid); + } + if layout.is::() { + let mut chunks = Vec::with_capacity(layout.nchildren()); + for index in 0..layout.nchildren() { + chunks.push(read_layout( + runtime, + session, + source.clone(), + &layout.child(index).map_err(invalid)?, + roots, + )?); + } + return ChunkedArray::try_new(chunks, dtype) + .map(IntoArray::into_array) + .map_err(invalid); + } + if layout.is::() { + // Values use dictionary coordinates, codes use row coordinates. + let values = read_layout( + runtime, + session, + source.clone(), + &layout.child(0).map_err(invalid)?, + roots, + )?; + let codes = read_layout( + runtime, + session, + source, + &layout.child(1).map_err(invalid)?, + None, + )?; + return DictArray::try_new(codes, values) + .map(IntoArray::into_array) + .map_err(invalid); + } + // Zone maps wrap data without changing its schema or row coordinates. + for (index, kind) in layout.child_types().enumerate() { + if matches!(kind, LayoutChildType::Transparent(_)) { + let child = layout.child(index).map_err(invalid)?; + if child.dtype() == layout.dtype() && child.row_count() == layout.row_count() { + return read_layout(runtime, session, source, &child, roots); + } + } + } + let reader = layout + .new_reader("".into(), source, session, &Default::default()) + .map_err(invalid)?; + let tasks = ScanBuilder::new(session.clone(), reader) + .with_concurrency(1) + .with_split_by(SplitBy::RowCount( + usize::try_from(layout.row_count()).map_err(invalid)?.max(1), + )) + .build() + .map_err(invalid)?; + let mut chunks = Vec::new(); + for task in tasks { + if let Some(array) = runtime.block_on(task).map_err(invalid)? { + chunks.push(array); + } + } + let array = ChunkedArray::try_new(chunks, layout.dtype().clone()) + .map(IntoArray::into_array) + .map_err(invalid)?; + let Some(names) = roots else { + return Ok(array); + }; + // A flat root cannot skip decoding fields, but selecting from the decoded + // array preserves validity and does not invoke layout projection. + let mut ctx = session.create_execution_ctx(); + let rows = array.execute::(&mut ctx).map_err(invalid)?; + let columns = names + .iter() + .map(|name| { + rows.unmasked_field_by_name(name.as_str()) + .cloned() + .map_err(invalid) + }) + .collect::>>()?; + StructArray::try_new( + FieldNames::from_iter(names.iter().map(String::as_str)), + columns, + rows.len(), + rows.validity().map_err(invalid)?, + ) + .map(IntoArray::into_array) + .map_err(invalid) +} + +#[cfg(test)] +mod tests { + use super::*; + use vortex::array::dtype::arrow::FromArrowType; + use vortex::io::session::RuntimeSessionExt; + use vortex::layout::layouts::chunked::ChunkedLayout; + use vortex::layout::segments::{SegmentFuture, SegmentId}; + use vortex::layout::IntoLayout; + use vortex::layout::LayoutChildren; + use vortex::VortexSessionDefault; + + #[derive(Clone)] + struct Empty; + impl SegmentSource for Empty { + fn request(&self, _: SegmentId) -> SegmentFuture { + panic!("empty layout must not read segments") + } + } + impl LayoutChildren for Empty { + fn to_arc(&self) -> Arc { + Arc::new(self.clone()) + } + fn child(&self, _: usize, _: &DType) -> vortex::error::VortexResult { + panic!("empty layout has no children") + } + fn child_row_count(&self, _: usize) -> u64 { + panic!("empty layout has no children") + } + fn nchildren(&self) -> usize { + 0 + } + } + + #[test] + fn empty_chunked_layout_preserves_projected_row_schema() { + use arrow_schema::{DataType, Field, Schema}; + let schema = Arc::new(Schema::new(vec![ + Field::new("unused", DataType::Int32, true), + Field::new( + "profile", + DataType::Struct(vec![Field::new("keep", DataType::Int32, true)].into()), + true, + ), + ])); + let layout = + ChunkedLayout::new(0, DType::from_arrow(schema.clone()), Arc::new(Empty)).into_layout(); + let runtime = CurrentThreadRuntime::new(); + let session = VortexSession::default().with_handle(runtime.handle()); + let array = read_layout( + &runtime, + &session, + Arc::new(Empty), + &layout, + Some(&["profile".into()]), + ) + .unwrap(); + let schema = Arc::new(schema.project(&[1]).unwrap()); + let batch = super::super::vortex_array_to_record_batch(&session, array, &schema).unwrap(); + assert_eq!(batch.num_rows(), 0); + assert_eq!(batch.schema(), schema); + } +} diff --git a/crates/paimon/src/arrow/shredding/map.rs b/crates/paimon/src/arrow/shredding/map.rs index 7a2af66a4..7bacb3e24 100644 --- a/crates/paimon/src/arrow/shredding/map.rs +++ b/crates/paimon/src/arrow/shredding/map.rs @@ -1115,9 +1115,7 @@ fn map_array_to_physical( /// shared-shredding physical ROW values. pub(crate) struct MapShreddingReadPlan { logical_fields: Vec, - /// Physical fields decoded from the file; part of the mirrored - /// [`ShreddingReadPlan`] API surface. - #[allow(dead_code)] + /// Physical read types, including projections inside each MAP value. physical_fields: Vec, logical_schema: arrow_schema::SchemaRef, contexts: HashMap, @@ -1209,16 +1207,17 @@ impl MapShreddingReadPlan { })) } - /// Return the physical struct children needed for one logical field. - /// `None` means the whole root field must be decoded. - pub(crate) fn projected_physical_children(&self, field_name: &str) -> Option> { + /// Map logical value projections onto every direct and overflow source. + pub(crate) fn projected_physical_field(&self, field_name: &str) -> Option { let field_index = self .logical_fields .iter() .position(|field| field.name() == field_name)?; let ctx = self.contexts.get(&field_index)?; - ctx.selected_names.as_ref()?; - + let physical = &self.physical_fields[field_index]; + if ctx.selected_names.is_none() { + return Some(physical.clone()); + } let mut children = HashSet::from([FIELD_MAPPING_NAME.to_string()]); children.extend( ctx.selected_columns @@ -1228,7 +1227,21 @@ impl MapShreddingReadPlan { if ctx.include_overflow { children.insert(OVERFLOW_NAME.to_string()); } - Some(children) + let DataType::Row(row) = physical.data_type() else { + return None; + }; + Some(DataField::new( + physical.id(), + physical.name().into(), + DataType::Row(RowType::with_nullable( + physical.data_type().is_nullable(), + row.fields() + .iter() + .filter(|child| children.contains(child.name())) + .cloned() + .collect(), + )), + )) } } diff --git a/crates/paimon/src/table/data_file_reader.rs b/crates/paimon/src/table/data_file_reader.rs index 9d5483217..7b775da3a 100644 --- a/crates/paimon/src/table/data_file_reader.rs +++ b/crates/paimon/src/table/data_file_reader.rs @@ -698,23 +698,8 @@ impl DataFileReader { let target_schema = crate::arrow::build_read_arrow_schema(&read_type)?; // What the reader is asked for. - let projected_read_fields: Vec = if data_fields.is_some() - || file_meta.write_cols.is_some() - || read_type.iter().any(contains_selected_map) - { - read_data_fields_for_format( - &data_schema_fields, - &read_type, - self.nested_field_enabled, - file_meta.file_name.ends_with(".parquet"), - )? - } else { - read_type - .iter() - .filter(|field| field.name() != ROW_ID_FIELD_NAME) - .cloned() - .collect() - }; + let projected_read_fields = + read_data_fields(&data_schema_fields, &read_type, self.nested_field_enabled)?; let path_to_read = split.data_file_path(&file_meta); let configured_reader = create_format_reader_with_budget( &path_to_read, @@ -771,17 +756,9 @@ impl DataFileReader { // The decoded batch is described by `format_read_fields`, so map // `read_type` onto *that* list: its entries carry the types the columns // actually come back as, which is what reconciling them needs. - let (index_mapping, source_fields) = if data_fields.is_some() - || file_meta.write_cols.is_some() - || read_type.iter().any(contains_selected_map) - { - ( - create_index_mapping(&read_type, &format_read_fields), - Some(format_read_fields.clone()), - ) - } else { - (None, None) - }; + let source_fields = projected_source_fields(&format_read_fields, &projected_read_fields); + let index_mapping = create_index_mapping(&read_type, &source_fields); + let source_fields = Some(source_fields); let input_file = file_io.new_input(&read_path)?; let open_start = read_timing.as_ref().map(|_| Instant::now()); @@ -1031,23 +1008,8 @@ impl DataFileReader { let file_fields = data_fields.clone().unwrap_or_else(|| table_fields.clone()); let data_schema_fields = self.physical_data_fields(&file_meta, &file_fields, None)?; // What the reader is asked for. - let projected_read_fields: Vec = if data_fields.is_some() - || file_meta.write_cols.is_some() - || read_type.iter().any(contains_selected_map) - { - read_data_fields_for_format( - &data_schema_fields, - &read_type, - self.nested_field_enabled, - file_meta.file_name.ends_with(".parquet"), - )? - } else { - read_type - .iter() - .filter(|field| field.name() != ROW_ID_FIELD_NAME) - .cloned() - .collect() - }; + let projected_read_fields = + read_data_fields(&data_schema_fields, &read_type, self.nested_field_enabled)?; let path_to_read = split.data_file_path(&file_meta); let configured_reader = create_format_reader_with_budget( &path_to_read, @@ -1066,17 +1028,9 @@ impl DataFileReader { // The decoded batch is described by `format_read_fields`, so map // `read_type` onto *that* list: its entries carry the types the columns // actually come back as, which is what reconciling them needs. - let (index_mapping, source_fields) = if data_fields.is_some() - || file_meta.write_cols.is_some() - || read_type.iter().any(contains_selected_map) - { - ( - create_index_mapping(&read_type, &format_read_fields), - Some(format_read_fields.clone()), - ) - } else { - (None, None) - }; + let source_fields = projected_source_fields(&format_read_fields, &projected_read_fields); + let index_mapping = create_index_mapping(&read_type, &source_fields); + let source_fields = Some(source_fields); // Remap predicates from table-level to file-level indices. let file_predicates = { @@ -1297,11 +1251,6 @@ fn project_file_batch( }) } -fn contains_selected_map(field: &DataField) -> bool { - crate::spec::is_map_selected_keys_field(field) - || matches!(field.data_type(), DataType::Row(row) if row.fields().iter().any(contains_selected_map)) -} - fn row_tracking_sequence_column( physical: Option<&Arc>, max_sequence_number: i64, @@ -1327,25 +1276,55 @@ fn row_tracking_sequence_column( )))) } -#[cfg(test)] -fn read_data_fields( - all_data_fields: &[DataField], - expected_fields: &[DataField], - nested_field_enabled: bool, -) -> crate::Result> { - read_data_fields_for_format( - all_data_fields, - expected_fields, - nested_field_enabled, - false, - ) +/// Fallback formats decode complete roots. Keep the decoder's actual types, +/// but do not resurrect descendants omitted by Java-compatible schema pruning. +fn projected_source_fields(decoded: &[DataField], projected: &[DataField]) -> Vec { + decoded + .iter() + .filter_map(|source| { + let projected = projected.iter().find(|field| field.id() == source.id())?; + Some(data_field_with_type( + source, + projected_source_type(source.data_type(), projected.data_type()), + )) + }) + .collect() } -fn read_data_fields_for_format( +fn projected_source_type(decoded: &DataType, projected: &DataType) -> DataType { + // Synthetic ROW children describe extraction instructions, not physical + // field IDs. Reconciliation must see the decoder's VARIANT/MAP context. + if is_variant_extraction_row_type(projected) { + return decoded.clone(); + } + match (decoded, projected) { + (DataType::Row(source), DataType::Row(target)) => { + DataType::Row(crate::spec::RowType::with_nullable( + decoded.is_nullable(), + projected_source_fields(source.fields(), target.fields()), + )) + } + (DataType::Array(source), DataType::Array(target)) => { + DataType::Array(crate::spec::ArrayType::with_nullable( + decoded.is_nullable(), + projected_source_type(source.element_type(), target.element_type()), + )) + } + (DataType::Map(source), DataType::Map(target)) => { + DataType::Map(crate::spec::MapType::with_nullable( + decoded.is_nullable(), + projected_source_type(source.key_type(), target.key_type()), + projected_source_type(source.value_type(), target.value_type()), + )) + } + _ => decoded.clone(), + } +} + +fn read_data_fields( all_data_fields: &[DataField], expected_fields: &[DataField], nested_field_enabled: bool, - prune_map_values: bool, ) -> crate::Result> { let mut read_fields = Vec::new(); for data_field in all_data_fields { @@ -1353,11 +1332,10 @@ fn read_data_fields_for_format( .iter() .find(|field| field.id() == data_field.id()) { - if let Some(pruned_type) = prune_data_type_for_format( + if let Some(pruned_type) = prune_data_type( expected.data_type(), data_field.data_type(), nested_field_enabled, - prune_map_values, )? { let mut field = data_field_with_type(data_field, pruned_type); if crate::spec::is_map_selected_keys_field(expected) { @@ -1393,39 +1371,49 @@ fn is_physical_system_field(field: &DataField) -> bool { ) } -#[cfg(test)] fn prune_data_type( read_type: &DataType, data_type: &DataType, nested_field_enabled: bool, -) -> crate::Result> { - prune_data_type_for_format(read_type, data_type, nested_field_enabled, false) -} - -fn prune_data_type_for_format( - read_type: &DataType, - data_type: &DataType, - nested_field_enabled: bool, - prune_map_values: bool, ) -> crate::Result> { match read_type { - DataType::Map(read_map) if prune_map_values => { + DataType::Array(read_array) => { + let DataType::Array(data_array) = data_type else { + return Ok(Some(data_type.clone())); + }; + Ok(prune_data_type( + read_array.element_type(), + data_array.element_type(), + nested_field_enabled, + )? + .map(|element| { + DataType::Array(crate::spec::ArrayType::with_nullable( + data_type.is_nullable(), + element, + )) + })) + } + DataType::Map(read_map) => { let DataType::Map(data_map) = data_type else { return Ok(Some(data_type.clone())); }; - // Keep file field IDs and ordering; reconciliation handles renames. - let value = prune_data_type_for_format( + let key = prune_data_type( + read_map.key_type(), + data_map.key_type(), + nested_field_enabled, + )?; + let value = prune_data_type( read_map.value_type(), data_map.value_type(), - true, - true, - )? - .unwrap_or_else(|| data_map.value_type().clone()); - Ok(Some(DataType::Map(crate::spec::MapType::with_nullable( - data_type.is_nullable(), - data_map.key_type().clone(), - value, - )))) + nested_field_enabled, + )?; + Ok(key.zip(value).map(|(key, value)| { + DataType::Map(crate::spec::MapType::with_nullable( + data_type.is_nullable(), + key, + value, + )) + })) } DataType::Row(read_row) if is_variant_extraction_row_type(read_type) => { Ok(Some(DataType::Row(read_row.clone()))) @@ -1434,12 +1422,8 @@ fn prune_data_type_for_format( let DataType::Row(data_row) = data_type else { return Ok(Some(data_type.clone())); }; - let mut fields = read_data_fields_for_format( - data_row.fields(), - read_row.fields(), - nested_field_enabled, - prune_map_values, - )?; + let mut fields = + read_data_fields(data_row.fields(), read_row.fields(), nested_field_enabled)?; if fields.is_empty() { if !nested_field_enabled { return Ok(None); @@ -1451,17 +1435,10 @@ fn prune_data_type_for_format( } } Ok(Some(DataType::Row(crate::spec::RowType::with_nullable( - read_type.is_nullable(), + data_type.is_nullable(), fields, )))) } - // ARRAY and MAP are deliberately NOT descended, even though Java's - // `pruneDataType` does: the pruned type is also what the Vortex reader is - // asked for, and it projects top-level columns only, then reinterprets a - // decoded nested struct against the requested Arrow type positionally, so - // a container pruned into read order would silently relabel its children. - // Nested containers are reconciled after decoding instead, by - // `arrow::nested_evolution::evolve_column`, which pairs by field id. _ => Ok(Some(data_type.clone())), } } @@ -1469,6 +1446,7 @@ fn prune_data_type_for_format( fn data_field_with_type(field: &DataField, data_type: DataType) -> DataField { DataField::new(field.id(), field.name().to_string(), data_type) .with_description(field.description().map(ToString::to_string)) + .with_default_value(field.default_value().map(ToString::to_string)) } fn data_schema_fields_for_file( @@ -4627,6 +4605,9 @@ mod vector_parquet_tests { } } +#[cfg(test)] +mod nested_projection_tests; + #[cfg(test)] mod prune_container_tests { use super::*; @@ -4645,28 +4626,25 @@ mod prune_container_tests { } #[test] - fn keeps_an_array_element_type_verbatim() { - // Not descended on purpose — see the comment in `prune_data_type`: the - // Vortex reader would relabel a reordered container positionally. The - // element is reconciled after decoding instead. + fn prunes_an_array_element_by_field_id() { let data = DataType::Array(ArrayType::new(row(vec![ f(2, "name", str_t()), f(3, "lang", str_t()), ]))); let read = DataType::Array(ArrayType::new(row(vec![f(3, "lang", str_t())]))); - assert_eq!(prune_data_type(&read, &data, false).unwrap().unwrap(), data); + assert_eq!(prune_data_type(&read, &data, false).unwrap().unwrap(), read); } #[test] - fn keeps_a_map_value_type_verbatim() { + fn prunes_a_map_value_by_field_id() { let data = DataType::Map(MapType::new( str_t(), row(vec![f(5, "v", str_t()), f(6, "unit", str_t())]), )); let read = DataType::Map(MapType::new(str_t(), row(vec![f(5, "v", str_t())]))); - assert_eq!(prune_data_type(&read, &data, false).unwrap().unwrap(), data); + assert_eq!(prune_data_type(&read, &data, false).unwrap().unwrap(), read); } #[test] @@ -4688,11 +4666,8 @@ mod prune_container_tests { row(vec![f(5, "old_name", str_t()), f(7, "last", str_t())]), )); assert_eq!( - prune_data_type_for_format(&read, &data, false, true) - .unwrap() - .unwrap(), + prune_data_type(&read, &data, false).unwrap().unwrap(), expected ); - assert_eq!(prune_data_type(&read, &data, false).unwrap().unwrap(), data); } } diff --git a/crates/paimon/src/table/data_file_reader/nested_projection_tests.rs b/crates/paimon/src/table/data_file_reader/nested_projection_tests.rs new file mode 100644 index 000000000..06a726905 --- /dev/null +++ b/crates/paimon/src/table/data_file_reader/nested_projection_tests.rs @@ -0,0 +1,835 @@ +// 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 super::*; +use crate::arrow::format::{ + create_format_reader_with_budget, create_format_writer, FormatReadFields, MosaicPrefetchOptions, +}; +use crate::arrow::{build_read_arrow_schema, build_target_arrow_schema}; +use crate::io::FileIOBuilder; +use crate::spec::stats::BinaryTableStats; +use crate::spec::{ArrayType, BinaryRow, DataFileMeta, IntType, MapType, RowType, VarCharType}; +use crate::table::source::DataSplitBuilder; +use arrow_array::UInt32Array; +use arrow_select::concat::concat_batches; +use bytes::Bytes; +use futures::TryStreamExt; +use std::io::Cursor; +use std::sync::atomic::{AtomicU64, Ordering}; + +fn f(id: i32, name: &str, typ: DataType) -> DataField { + DataField::new(id, name.into(), typ) +} +fn int() -> DataType { + DataType::Int(IntType::new()) +} +fn string() -> DataType { + DataType::VarChar(VarCharType::string_type()) +} +fn row(fields: Vec) -> DataType { + DataType::Row(RowType::new(fields)) +} +fn array(element: DataType) -> DataType { + DataType::Array(ArrayType::new(element)) +} +fn map(value: DataType) -> DataType { + DataType::Map(MapType::new(string(), value)) +} +fn leaves(base: i32, projected: bool) -> DataType { + if projected { + row(vec![ + f( + base + 2, + "renamed_tail", + DataType::BigInt(crate::spec::BigIntType::new()), + ), + f(base + 1, "keep", int()), + ]) + } else { + row(vec![ + f(base + 1, "keep", int()), + f(base + 2, "tail", int()), + f(base + 3, "unused", string()), + ]) + } +} +fn fields(projected: bool) -> Vec { + vec![ + f( + 10, + "profile", + row(vec![f(11, "detail", leaves(11, projected))]), + ), + f(20, "items", array(leaves(20, projected))), + f(30, "groups", array(array(leaves(30, projected)))), + f(40, "attrs", map(leaves(40, projected))), + f( + 50, + "complex", + row(vec![f(51, "lookup", map(array(leaves(51, projected))))]), + ), + f(60, "array_maps", array(map(leaves(60, projected)))), + ] +} +fn input() -> RecordBatch { + let leaf = serde_json::json!({"keep": 7, "tail": 9, "unused": "discard".repeat(2048)}); + let rows = [ + serde_json::json!({"profile":{"detail":leaf}, "items":[leaf,null,leaf], "groups":[[leaf,null],[],null], "attrs":{"a":leaf,"b":null}, "complex":{"lookup":{"x":[leaf,null],"y":[]}},"array_maps":[{"k":leaf},null,{}]}), + serde_json::json!({"profile":null,"items":null,"groups":null,"attrs":null,"complex":null,"array_maps":null}), + serde_json::json!({"profile":{"detail":null},"items":[],"groups":[],"attrs":{},"complex":{"lookup":{}},"array_maps":[]}), + serde_json::json!({"profile":{"detail":{"keep":null,"tail":3}},"items":[{"keep":null,"tail":4}],"groups":[[null]],"attrs":{"z":{"keep":null,"tail":5}},"complex":{"lookup":null},"array_maps":[{"z":null}]}), + serde_json::json!({"profile":{"detail":{"unused":"present"}},"items":[{"unused":"present"}],"groups":[[{"unused":"present"}]],"attrs":{"k":{"unused":"present"}},"complex":{"lookup":{"k":[{"unused":"present"}]}},"array_maps":[{"k":{"unused":"present"}}]}), + ]; + let json = rows + .iter() + .map(|row| row.to_string()) + .collect::>() + .join("\n"); + arrow_json::ReaderBuilder::new(build_target_arrow_schema(&fields(false)).unwrap()) + .build(Cursor::new(json)) + .unwrap() + .next() + .unwrap() + .unwrap() +} +fn expected(input: &RecordBatch, source: &[DataField], target: &[DataField]) -> RecordBatch { + let columns = target + .iter() + .map(|target| { + let index = source + .iter() + .position(|source| source.id() == target.id()) + .unwrap(); + crate::arrow::nested_evolution::evolve_field( + input.column(index), + &source[index], + target, + ) + .unwrap() + }) + .collect(); + RecordBatch::try_new(build_read_arrow_schema(target).unwrap(), columns).unwrap() +} + +// Avro's native MAP has unordered entries. Compare mappings recursively while +// retaining ARRAY order, ROW nullness and empty-versus-null containers. +fn json_rows(batch: &RecordBatch) -> serde_json::Value { + let mut writer = arrow_json::ArrayWriter::new(Vec::new()); + writer.write_batches(&[batch]).unwrap(); + writer.finish().unwrap(); + serde_json::from_slice(&writer.into_inner()).unwrap() +} +fn meta(name: &str, size: u64, rows: usize) -> DataFileMeta { + DataFileMeta { + file_name: name.into(), + file_size: size as i64, + row_count: rows as i64, + min_key: vec![], + max_key: vec![], + key_stats: BinaryTableStats::empty(), + value_stats: BinaryTableStats::empty(), + min_sequence_number: 0, + max_sequence_number: 0, + schema_id: 1, + level: 0, + extra_files: vec![], + creation_time: None, + delete_row_count: None, + embedded_index: None, + file_source: None, + value_stats_cols: None, + external_path: None, + first_row_id: None, + write_cols: None, + column_max_sequence_numbers: None, + } +} + +#[tokio::test] +async fn nested_projection_roundtrips_parquet_and_fallback_formats() { + let source = fields(false); + let input = input(); + let target = fields(true).into_iter().rev().collect::>(); + for extension in ["parquet", "orc", "avro", "row", "json"] + .into_iter() + .chain(cfg!(feature = "vortex").then_some("vortex")) + { + let source = if extension == "vortex" { + source[..3].to_vec() + } else { + source.clone() + }; + let input = if extension == "vortex" { + input.project(&[0, 1, 2]).unwrap() + } else { + input.clone() + }; + let target = target + .iter() + .filter(|field| source.iter().any(|source| source.id() == field.id())) + .cloned() + .collect::>(); + let expected = expected(&input, &source, &target); + assert_eq!( + json_rows(&expected)[0]["profile"]["detail"], + serde_json::json!({"renamed_tail":9,"keep":7}) + ); + let io = FileIOBuilder::new("memory").build().unwrap(); + let table_path = format!("memory:/nested_projection_{extension}"); + let bucket = format!("{table_path}/bucket-0"); + let name = format!("part.{extension}"); + let output = io.new_output(&format!("{bucket}/{name}")).unwrap(); + let size = if extension == "orc" { + // orc-rust's writer cannot encode nested types. This fixture was + // written by Apache ORC through PyArrow; tests need no Python. + let bytes = include_bytes!("../../../testdata/nested_projection/nested.orc"); + output.write(Bytes::from_static(bytes)).await.unwrap(); + bytes.len() as u64 + } else { + let mut writer = create_format_writer( + &output, + input.schema(), + "none", + 1, + None, + Some(&source), + None, + ) + .await + .unwrap(); + writer.write(&input).await.unwrap(); + writer.close().await.unwrap().file_size + }; + let file = meta(&name, size, input.num_rows()); + let split = DataSplitBuilder::new() + .with_snapshot(1) + .with_partition(BinaryRow::new(0)) + .with_bucket(0) + .with_bucket_path(bucket) + .with_total_buckets(1) + .with_data_files(vec![file.clone()]) + .build() + .unwrap(); + let reader = DataFileReader::new( + io.clone(), + SchemaManager::new(io, table_path), + 1, + source.clone(), + |fields| Ok(fields.to_vec()), + target.clone(), + vec![], + ); + for schema in [None, Some(source.clone())] { + let batches = reader + .read_single_file_stream(&split, file.clone(), schema.clone(), None, None) + .unwrap() + .try_collect::>() + .await + .unwrap_or_else(|err| panic!("{extension}: {err}")); + let actual = concat_batches(&expected.schema(), &batches).unwrap(); + assert_eq!(actual.schema(), expected.schema()); + assert_eq!( + json_rows(&actual), + json_rows(&expected), + "{extension}, explicit schema: {}", + schema.is_some() + ); + if extension == "json" { + continue; + } // line-oriented formats have no positional reads + let batches = reader + .read_single_file_stream_local(&split, file.clone(), schema, None, vec![0, 2]) + .unwrap() + .try_collect::>() + .await + .unwrap_or_else(|err| panic!("{extension}: {err}")); + let selected = + arrow_select::take::take_record_batch(&expected, &UInt32Array::from(vec![0, 2])) + .unwrap(); + assert_eq!( + json_rows(&concat_batches(&selected.schema(), &batches).unwrap()), + json_rows(&selected), + "local {extension}" + ); + } + let missing = row(vec![f(100, "added", int())]); + let requested = vec![ + f(20, "items", array(missing.clone())), + f(40, "attrs", map(missing)), + ] + .into_iter() + .filter(|field| source.iter().any(|source| source.id() == field.id())) + .collect::>(); + for nested_enabled in [false, true] { + let reader = reader + .clone() + .with_read_type(requested.clone()) + .with_nested_field_enabled(nested_enabled); + let batches = reader + .read_single_file_stream(&split, file.clone(), Some(source.clone()), None, None) + .unwrap() + .try_collect::>() + .await + .unwrap(); + let mut wanted = self::expected(&input, &source, &requested); + if !nested_enabled { + wanted = RecordBatch::try_new( + wanted.schema(), + wanted + .schema() + .fields() + .iter() + .map(|field| { + arrow_array::new_null_array(field.data_type(), input.num_rows()) + }) + .collect(), + ) + .unwrap(); + } + let actual = concat_batches(&wanted.schema(), &batches).unwrap(); + assert_eq!( + json_rows(&actual), + json_rows(&wanted), + "{extension}: all-missing, nested={nested_enabled}" + ); + } + } +} + +#[derive(Clone)] +struct CountingRead { + data: Bytes, + bytes: Arc, +} +#[async_trait::async_trait] +impl FileRead for CountingRead { + async fn read(&self, range: std::ops::Range) -> crate::Result { + self.bytes + .fetch_add(range.end - range.start, Ordering::Relaxed); + Ok(self.data.slice(range.start as usize..range.end as usize)) + } +} + +#[tokio::test] +async fn nested_projection_reduces_parquet_data_io_in_every_container() { + let source = fields(false); + let original = input(); + let input = concat_batches(&original.schema(), (0..64).map(|_| &original)).unwrap(); + let io = FileIOBuilder::new("memory").build().unwrap(); + let output = io + .new_output("memory:/nested_projection_io.parquet") + .unwrap(); + let options = HashMap::from([("parquet.enable.dictionary".into(), "false".into())]); + let mut writer = create_format_writer( + &output, + input.schema(), + "none", + 1, + None, + Some(&source), + Some(&options), + ) + .await + .unwrap(); + writer.write(&input).await.unwrap(); + let size = writer.close().await.unwrap().file_size; + let data = io + .new_input(output.location()) + .unwrap() + .read() + .await + .unwrap(); + for (full, projected) in source.iter().zip(fields(true)) { + let mut read_bytes = vec![]; + for requested in [full, &projected] { + let physical = + read_data_fields(&source, std::slice::from_ref(requested), false).unwrap(); + let configured = create_format_reader_with_budget( + output.location(), + false, + FormatReadFields { + data_schema: &source, + projected: &physical, + }, + &HashMap::new(), + None, + 1, + MosaicPrefetchOptions::default(), + ) + .unwrap(); + let count = Arc::new(AtomicU64::new(0)); + let stream = configured + .reader + .read_batch_stream( + Box::new(CountingRead { + data: data.clone(), + bytes: count.clone(), + }), + size, + &configured.read_fields, + None, + None, + None, + ) + .await + .unwrap(); + count.store(0, Ordering::Relaxed); + let batches = stream.try_collect::>().await.unwrap(); + let actual = concat_batches(&batches[0].schema(), &batches).unwrap(); + let wanted = expected(&input, &source, std::slice::from_ref(requested)); + let normalized = expected( + &actual, + &configured.read_fields, + std::slice::from_ref(requested), + ); + assert_eq!(normalized, wanted, "{}", requested.name()); + read_bytes.push(count.load(Ordering::Relaxed)); + } + assert!( + read_bytes[1] * 2 < read_bytes[0], + "{}: projected {} bytes, full {} bytes", + full.name(), + read_bytes[1], + read_bytes[0] + ); + } +} + +#[test] +fn missing_container_leaves_follow_java_nested_field_mode() { + let source = row(vec![f(1, "old", int())]); + let requested = row(vec![f(2, "added", int())]); + for (source, requested) in [ + (array(source.clone()), array(requested.clone())), + (map(source.clone()), map(requested.clone())), + ( + DataType::Map(MapType::new(source.clone(), int())), + DataType::Map(MapType::new(requested.clone(), int())), + ), + ] { + // FormatReaderMapping returns null for the whole type when a ROW has + // no surviving children outside nested-field mode. In nested-field + // mode it keeps a physical anchor, recursively through containers. + assert_eq!(prune_data_type(&requested, &source, false).unwrap(), None); + assert_eq!( + prune_data_type(&requested, &source, true).unwrap(), + Some(source) + ); + } +} + +#[tokio::test] +async fn shared_map_value_nested_projection_prunes_direct_and_overflow_sources() { + let source = vec![fields(false)[3].clone()]; + let target = vec![fields(true)[3].clone()]; + let original = input(); + let input = RecordBatch::try_new( + build_target_arrow_schema(&source).unwrap(), + vec![original.column(3).clone()], + ) + .unwrap(); + let io = FileIOBuilder::new("memory").build().unwrap(); + let output = io + .new_output("memory:/shared_map_value_nested_projection.parquet") + .unwrap(); + let options = HashMap::from([ + ( + "fields.attrs.map.storage-layout".into(), + "shared-shredding".into(), + ), + ( + "fields.attrs.map.shared-shredding.max-columns".into(), + "1".into(), + ), + ]); + let mut writer = create_format_writer( + &output, + input.schema(), + "none", + 1, + None, + Some(&source), + Some(&options), + ) + .await + .unwrap(); + writer.write(&input).await.unwrap(); + let size = writer.close().await.unwrap().file_size; + let physical = read_data_fields(&source, &target, false).unwrap(); + let configured = create_format_reader_with_budget( + output.location(), + false, + FormatReadFields { + data_schema: &source, + projected: &physical, + }, + &HashMap::new(), + None, + 1, + MosaicPrefetchOptions::default(), + ) + .unwrap(); + let batches = configured + .reader + .read_batch_stream( + Box::new( + io.new_input(output.location()) + .unwrap() + .reader() + .await + .unwrap(), + ), + size, + &configured.read_fields, + None, + None, + None, + ) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + let decoded = concat_batches(&batches[0].schema(), &batches).unwrap(); + assert_eq!( + expected(&decoded, &physical, &target), + expected(&input, &source, &target) + ); + let values = decoded + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values(); + assert_eq!( + values + .as_any() + .downcast_ref::() + .unwrap() + .num_columns(), + 2 + ); +} + +#[tokio::test] +async fn map_row_key_projection_preserves_entries_even_when_keys_become_equal() { + use arrow_array::{Int32Array, MapArray, StringArray, StructArray}; + use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer}; + use arrow_schema::DataType as ArrowType; + let map_field = |children| { + f( + 0, + "m", + DataType::Map(MapType::new( + DataType::Row(RowType::with_nullable(false, children)), + int(), + )), + ) + }; + let source = vec![map_field(vec![ + f(1, "keep", int()), + f(2, "unused", string()), + ])]; + let target = vec![map_field(vec![f(1, "keep", int())])]; + let schema = build_target_arrow_schema(&source).unwrap(); + let ArrowType::Map(entries_field, _) = schema.field(0).data_type() else { + unreachable!() + }; + let ArrowType::Struct(entry_fields) = entries_field.data_type() else { + unreachable!() + }; + let ArrowType::Struct(key_fields) = entry_fields[0].data_type() else { + unreachable!() + }; + let keys = StructArray::new( + key_fields.clone(), + vec![ + Arc::new(Int32Array::from(vec![1, 1, 2])), + Arc::new(StringArray::from(vec!["a", "b", "c"])), + ], + None, + ); + let entries = StructArray::new( + entry_fields.clone(), + vec![Arc::new(keys), Arc::new(Int32Array::from(vec![10, 20, 30]))], + None, + ); + let maps = MapArray::new( + entries_field.clone(), + OffsetBuffer::new(ScalarBuffer::from(vec![0, 2, 2, 3])), + entries, + Some(NullBuffer::from(vec![true, false, true])), + false, + ); + let input = RecordBatch::try_new(schema, vec![Arc::new(maps)]).unwrap(); + let io = FileIOBuilder::new("memory").build().unwrap(); + let output = io + .new_output("memory:/nested_map_row_keys.parquet") + .unwrap(); + let mut writer = create_format_writer( + &output, + input.schema(), + "none", + 1, + None, + Some(&source), + None, + ) + .await + .unwrap(); + writer.write(&input).await.unwrap(); + let size = writer.close().await.unwrap().file_size; + let physical = read_data_fields(&source, &target, false).unwrap(); + let configured = create_format_reader_with_budget( + output.location(), + false, + FormatReadFields { + data_schema: &source, + projected: &physical, + }, + &HashMap::new(), + None, + 1, + MosaicPrefetchOptions::default(), + ) + .unwrap(); + let batches = configured + .reader + .read_batch_stream( + Box::new( + io.new_input(output.location()) + .unwrap() + .reader() + .await + .unwrap(), + ), + size, + &configured.read_fields, + None, + None, + None, + ) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + let decoded = concat_batches(&batches[0].schema(), &batches).unwrap(); + let maps = decoded + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + maps.keys() + .as_any() + .downcast_ref::() + .unwrap() + .num_columns(), + 1 + ); + assert_eq!( + expected(&decoded, &physical, &target), + expected(&input, &source, &target) + ); +} + +#[tokio::test] +async fn java_list_layouts_decode_the_projected_row_without_shape_changes() { + use arrow_array::{Array, Int32Array, ListArray, StructArray}; + let target = vec![f(0, "items", array(row(vec![f(2, "wanted", int())])))]; + for (name, bytes, repeated) in [ + ( + "standard", + include_bytes!("../../../testdata/nested_projection/standard.parquet").as_slice(), + false, + ), + ( + "array", + include_bytes!("../../../testdata/nested_projection/array.parquet").as_slice(), + false, + ), + ( + "tuple", + include_bytes!("../../../testdata/nested_projection/tuple.parquet").as_slice(), + false, + ), + ( + "repeated", + include_bytes!("../../../testdata/nested_projection/repeated.parquet").as_slice(), + true, + ), + ] { + let configured = create_format_reader_with_budget( + "data.parquet", + false, + FormatReadFields { + data_schema: &target, + projected: &target, + }, + &HashMap::new(), + None, + 1, + MosaicPrefetchOptions::default(), + ) + .unwrap(); + let batches = configured + .reader + .read_batch_stream( + Box::new(CountingRead { + data: Bytes::from_static(bytes), + bytes: Arc::new(AtomicU64::new(0)), + }), + bytes.len() as u64, + &target, + None, + None, + None, + ) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + let decoded = concat_batches(&batches[0].schema(), &batches).unwrap(); + let list = decoded + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(list.value_offsets(), &[0, 2, 2, 2], "{name}"); + assert!(list.is_valid(1), "{name}: empty list"); + assert_eq!(list.is_null(2), !repeated, "{name}: null list"); + let elements = list + .values() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(elements.column_names(), vec!["wanted"], "{name}"); + assert_eq!( + elements + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values(), + &[7, 8], + "{name}" + ); + } +} + +#[tokio::test] +async fn fallback_does_not_resurrect_omitted_children_inside_retained_rows() { + let nested = |base, projected| { + let leaf = |old, new| { + row(vec![if projected { + f(new, "added", int()) + } else { + f(old, "old", int()) + }]) + }; + row(vec![ + f(base + 1, "keep", int()), + f(base + 2, "inner", leaf(base + 3, base + 4)), + f(base + 5, "items", array(leaf(base + 6, base + 7))), + f(base + 8, "attrs", map(leaf(base + 9, base + 10))), + ]) + }; + let schema_fields = |projected| { + vec![ + f(0, "p", nested(10, projected)), + f(100, "arr", array(nested(110, projected))), + f(200, "m", map(nested(210, projected))), + ] + }; + let source = schema_fields(false); + let requested = schema_fields(true); + let leaf = + serde_json::json!({"keep":9,"inner":{"old":7},"items":[{"old":7}],"attrs":{"a":{"old":7}}}); + let json = format!( + "{}\n{{}}", + serde_json::json!({"p":leaf,"arr":[leaf,null],"m":{"x":leaf}}) + ); + let input = arrow_json::ReaderBuilder::new(build_target_arrow_schema(&source).unwrap()) + .build(Cursor::new(json)) + .unwrap() + .next() + .unwrap() + .unwrap(); + for extension in ["parquet", "avro", "row", "json"] { + let io = FileIOBuilder::new("memory").build().unwrap(); + let table = format!("memory:/deep_missing_{extension}"); + let bucket = format!("{table}/bucket-0"); + let name = format!("part.{extension}"); + let output = io.new_output(&format!("{bucket}/{name}")).unwrap(); + let mut writer = create_format_writer( + &output, + input.schema(), + "none", + 1, + None, + Some(&source), + None, + ) + .await + .unwrap(); + writer.write(&input).await.unwrap(); + let file = meta( + &name, + writer.close().await.unwrap().file_size, + input.num_rows(), + ); + let split = DataSplitBuilder::new() + .with_snapshot(1) + .with_partition(BinaryRow::new(0)) + .with_bucket(0) + .with_bucket_path(bucket) + .with_total_buckets(1) + .with_data_files(vec![file.clone()]) + .build() + .unwrap(); + let reader = DataFileReader::new( + io.clone(), + SchemaManager::new(io, table), + 1, + source.clone(), + |fields| Ok(fields.to_vec()), + requested.clone(), + vec![], + ); + for enabled in [false, true] { + let batches = reader + .clone() + .with_nested_field_enabled(enabled) + .read_single_file_stream(&split, file.clone(), Some(source.clone()), None, None) + .unwrap() + .try_collect::>() + .await + .unwrap(); + let actual = + concat_batches(&build_read_arrow_schema(&requested).unwrap(), &batches).unwrap(); + let leaf = if enabled { + serde_json::json!({"keep":9,"inner":{},"items":[{}],"attrs":{"a":{}}}) + } else { + serde_json::json!({"keep":9}) + }; + let wanted = serde_json::json!([{"p":leaf,"arr":[leaf,null],"m":{"x":leaf}},{}]); + assert_eq!(json_rows(&actual), wanted, "{extension}, nested={enabled}"); + } + } +} diff --git a/crates/paimon/testdata/nested_projection/GenerateLegacyLists.java b/crates/paimon/testdata/nested_projection/GenerateLegacyLists.java new file mode 100644 index 000000000..803a3a4ac --- /dev/null +++ b/crates/paimon/testdata/nested_projection/GenerateLegacyLists.java @@ -0,0 +1,61 @@ +// 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. + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; + +// Run with parquet-hadoop/example and Hadoop common/mapreduce on the classpath. +// Generates fixtures using Parquet's Java writer, including legacy LIST layouts. +public class GenerateLegacyLists { + public static void main(String[] args) throws Exception { + String[] names = {"standard", "array", "tuple", "repeated"}; + String[] layouts = { + "optional group items (LIST) { repeated group list { optional group element { optional int32 unused; optional int32 wanted; } } }", + "optional group items (LIST) { repeated group array { optional int32 unused; optional int32 wanted; } }", + "optional group items (LIST) { repeated group items_tuple { optional int32 unused; optional int32 wanted; } }", + "repeated group items { optional int32 unused; optional int32 wanted; }" + }; + for (int i = 0; i < names.length; i++) { + MessageType schema = MessageTypeParser.parseMessageType("message schema {" + layouts[i] + "}"); + SimpleGroupFactory factory = new SimpleGroupFactory(schema); + try (ParquetWriter writer = ExampleParquetWriter.builder(new Path(args[0] + "/" + names[i] + ".parquet")) + .withConf(new Configuration()).withType(schema) + .withCompressionCodec(CompressionCodecName.UNCOMPRESSED).build()) { + Group record = factory.newGroup(); + Group root = i == 3 ? record : record.addGroup("items"); + for (int value = 7; value <= 8; value++) { + Group element = i == 0 ? root.addGroup("list").addGroup("element") + : root.addGroup(i == 1 ? "array" : i == 2 ? "items_tuple" : "items"); + element.add("unused", 100 + value); + element.add("wanted", value); + } + writer.write(record); + Group empty = factory.newGroup(); + if (i != 3) empty.addGroup("items"); + writer.write(empty); + writer.write(factory.newGroup()); + } + } + } +} diff --git a/crates/paimon/testdata/nested_projection/array.parquet b/crates/paimon/testdata/nested_projection/array.parquet new file mode 100644 index 0000000000000000000000000000000000000000..51e58491bff11649a4af65a88f0368ee3e59675c GIT binary patch literal 550 zcmZ{iF;Buk6vrP&Il7W?mo{l43j_@|T%b}C7B^!Yh!cYY2X~1?3fh9`=I-k3?C9tR za4_+E_yyG6cN|d%O?!R)zy9B^?_I9He=HcV9lT!e-=7Nsb5T-80SEx(N!~oS;0(hQ z*T#GM^7)bf&s#*Fi_0$&-*ZeQhIKLVxwAu2NHsB^5@S%OfFDC~y8=vY# zG6E6t3X_FSGDa9Z>Y+7qyn;+qdCeTeREbBIvt_2$My%v)fr3*%@EbKZKwg!CA6ts8 z+*B~&E+6ng&F$G{ZRgynuUoeYTf+@NH~_yRtjwQsEQH$dZ?OIQH+YC(Gp^^D*$wzd z{RTf6@S6ei#~i0M7em?XidR~tIDQ%?bET)@OeWeZ>MNO?Pc?Xl=Q6!m>cwu7RxWj# zP3E(T2t_LtmHKEonW~B!wME<#&8~_@omN|qqi(CI;!bm{q!xQZx0|t2Bf0H20sy?> KCI`^R6Y>QU9$z8= literal 0 HcmV?d00001 diff --git a/crates/paimon/testdata/nested_projection/generate_orc.py b/crates/paimon/testdata/nested_projection/generate_orc.py new file mode 100644 index 000000000..93b86a733 --- /dev/null +++ b/crates/paimon/testdata/nested_projection/generate_orc.py @@ -0,0 +1,45 @@ +# 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. + +# Fixture generation only; Rust tests consume the saved file without Python. +from pathlib import Path +import pyarrow as pa +import pyarrow.orc as orc + +leaf = pa.struct([('keep', pa.int32()), ('tail', pa.int32()), ('unused', pa.string())]) +schema = pa.schema([ + ('profile', pa.struct([('detail', leaf)])), + ('items', pa.list_(leaf)), + ('groups', pa.list_(pa.list_(leaf))), + ('attrs', pa.map_(pa.string(), leaf)), + ('complex', pa.struct([('lookup', pa.map_(pa.string(), pa.list_(leaf)))])), + ('array_maps', pa.list_(pa.map_(pa.string(), leaf))), +]) +x = {'keep': 7, 'tail': 9, 'unused': 'discard'} +rows = [ + {'profile': {'detail': x}, 'items': [x, None, x], 'groups': [[x, None], [], None], + 'attrs': [('a', x), ('b', None)], 'complex': {'lookup': [('x', [x, None]), ('y', [])]}, + 'array_maps': [[('k', x)], None, []]}, + {'profile': None, 'items': None, 'groups': None, 'attrs': None, 'complex': None, 'array_maps': None}, + {'profile': {'detail': None}, 'items': [], 'groups': [], 'attrs': [], 'complex': {'lookup': []}, 'array_maps': []}, + {'profile': {'detail': {'keep': None, 'tail': 3}}, 'items': [{'keep': None, 'tail': 4}], 'groups': [[None]], + 'attrs': [('z', {'keep': None, 'tail': 5})], 'complex': {'lookup': None}, 'array_maps': [[('z', None)]]}, + {'profile': {'detail': {'unused': 'present'}}, 'items': [{'unused': 'present'}], 'groups': [[{'unused': 'present'}]], + 'attrs': [('k', {'unused': 'present'})], 'complex': {'lookup': [('k', [{'unused': 'present'}])]}, + 'array_maps': [[('k', {'unused': 'present'})]]}, +] +orc.write_table(pa.Table.from_pylist(rows, schema=schema), Path(__file__).with_name('nested.orc'), compression='uncompressed') diff --git a/crates/paimon/testdata/nested_projection/nested.orc b/crates/paimon/testdata/nested_projection/nested.orc new file mode 100644 index 0000000000000000000000000000000000000000..c9d9546e5ea931832da2d020a7a45bb6927dbb17 GIT binary patch literal 4018 zcmcgvO;0075bf^ya5EmvFkongFWZ<1MNx!u66t(%FaCF>z(&!$= zg-*BGYd$?wD|u&%cj!wNIm#Qv)k#=!bpRcA$kidWq_`gq#dSu;{V>mI$F)NB)k3+0 zbVS~DOc}H~rgY+=bpBNtblh@wFG|v_$$&RveyOB=i*A2c0!l|0?RIx1_Y7Dyp?KyN z3FQoOb98R4m|rNLaw`Rbo)9UW19m8s^*H^HrPEK4m39A4@=;k|nyjxk{2yhRzD{4I zS9|PT`evV=j`Zao{}%59-iGw`uJa3-z1ib`r0@1c9Lw}A>%TOkY=7Hzo~JJ|)~kIM zYu6E^&pB@#UbgJ`!N~aO+g;o5ICxNFR1K1UTL&#~^mw;y=BiPRZ$B=Wdp9d~OF2H^f>k9fM$V5dZ z&tc0;CI_xUI6@x`o1y2?0HG^5S0sRK_b>KSirokfG zG?~4vYJ}3BFcLX_uE=u@kUq^_sZW>TEbU?8c3En!;gMi` z1^OrdB<}v6?eo1vyR;AX!IA#H_MwmIL51|`JcyYc%lC>To#?08WSYe9zWR$-ewABZ z?Nb@XKKF~Wi|YvuqV2DbPnwB$eAYbeCH^e5VfXX;QEV-e3?ZzY%We0kx{#HOw)@lV?A)_Di~CfZVR zBbq7Kvc0F3)wMf^Hf$9;WVLo9q^aCSee)TmxomHN;G9i!7{<9Uy$;+%ikeHPp5yQ~+&jB6n6r+U4N0 zgd*hH*3Na?dT;xnwq@0e+Ljg5w5|I^dhlZZbAIT3@-hA7&r>uvUtPI1!NRQ{-EIH7 MKON`Miq%>V!Z literal 0 HcmV?d00001 diff --git a/crates/paimon/testdata/nested_projection/repeated.parquet b/crates/paimon/testdata/nested_projection/repeated.parquet new file mode 100644 index 0000000000000000000000000000000000000000..57b4e3d1788af55169afcd6c51056393acf8b604 GIT binary patch literal 520 zcmZ`%F;9a)82x%hO5B?CE|4&=lvGm*7h2mUE^fv+G)^6X-X%5ytx&B0Lsusk2Zv6^ z|KR^{@<;exA!Qo0H0MH~&_1=eTED7$K z`|`NG*Z$+SkZ0hw+Qc$7rNpum6N_neTamD>c(iKU%%C%-+BOTTFZ^?HrWf-onMRbr zP&4-yYnc{QLs6HQdYP4l6gAjH*%|@UE!*n5hK=g34py(y)LaV8U^6yzjgxAqYC@*< zb|Y2@N5G(H3&0)-?Xb57^}r4PT2?21TZU+<0-2ghr!0TaZ_8(8`JpV?w!w7Qz|v8< z&Gsh8fVr10g-pCGN~PJ7cTu`ZB$(%`D8DJ?`Y6qvTbUP&W#({?2Of8Nak)r@BjN!c r2i%{CVLS>3az2~{z8H`Exrih`=5pYVg@~gAt8W9~g17-N#XIl?P(fC6 literal 0 HcmV?d00001 diff --git a/crates/paimon/testdata/nested_projection/standard.parquet b/crates/paimon/testdata/nested_projection/standard.parquet new file mode 100644 index 0000000000000000000000000000000000000000..9d06d8d17be585e502e97c6f0b9f719e892aac35 GIT binary patch literal 578 zcmaKq!AiqG5Qe8~+~!*7E(zq2OGAYk64Kfhdhu4oL-AC4YT8{en51ozTJPd3c<~K9 zdK5u?8+`{4&L$PTD9L6rv;TbA`N`z;!e_ubc)fhwZ7il^Mi~Vl0B}Gw^%MYC7-L*} z?dR+58~syUN1=|(Y!K7XR4azEo0weNmz4<1O2?|Q#dS92s%-H&o2fV4iN`cwUdkk7 z1bW0cj0&0Lj4*mWLP1YMa+DX}(N(Zqscp#Xi}EX@Lo+re$^<-9mVa zf;p~`CFUEjZ)CuAKIKzKA5{ZX({Qa_j#FiLTyzD$0l@DGYx3tBbHSPa%dgH-@@IVJ z%#p90Ff?V0;{Rm7`HzbKuK4_^#;K!YXez^u#x6r=GB;TX8M|qiNTV)q!ekvwFwWLt zw%N+!ILYj5ndQ+cwSCtQT;Falwoxo>vFQ4P!1sotx9A65x$F%CPYn9rQiRe!@nzQ= U2(bu{%#HfKq|C>Kq4JAKnPZ)aY1bZ{aVuw8uK9zI@54l@x_Mga%_6iHFPx8VXq zh&$)rzJ7fc|KgUBXX3Ui#I_7wiD6w#Y;NRfO~SP1u`KOyla0AfJA9IT;h)n(x}2V? z$Y%sRVwGlzik6HpdOSp`XINY7%vq9NhKd;|Rwous=jl?(9IX6#f`C)ov0Dw(LDf11 zyPs)o*5`E$xXH(S+%Sh(&^ literal 0 HcmV?d00001