This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new fc732b08 feat(read): complete nested projection pushdown for Parquet
(#1248)
fc732b08 is described below
commit fc732b08961c020ecf3c6654d5d2b0b222394bb7
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 29 08:44:19 2026 +0800
feat(read): complete nested projection pushdown for Parquet (#1248)
---
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 ++++
crates/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 +++---
.../data_file_reader/nested_projection_tests.rs | 835 +++++++++++++++++++++
.../nested_projection/GenerateLegacyLists.java | 61 ++
.../testdata/nested_projection/array.parquet | Bin 0 -> 550 bytes
.../testdata/nested_projection/generate_orc.py | 45 ++
.../paimon/testdata/nested_projection/nested.orc | Bin 0 -> 4018 bytes
.../testdata/nested_projection/repeated.parquet | Bin 0 -> 520 bytes
.../testdata/nested_projection/standard.parquet | Bin 0 -> 578 bytes
.../testdata/nested_projection/tuple.parquet | Bin 0 -> 568 bytes
17 files changed, 1706 insertions(+), 157 deletions(-)
diff --git a/crates/paimon/src/arrow/format/avro.rs
b/crates/paimon/src/arrow/format/avro.rs
index 37f01c51..2b593559 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<usize> {
- 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::<StructArray>().unwrap();
+ assert!(rows.is_null(0));
+ let child = rows
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .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 3de32c7e..c85c0578 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<DataField> {
- 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<DataField> {
+ 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<dyn FormatFileReader>,
pub read_fields: Vec<DataField>,
@@ -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 c678b529..0ade1fbf 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<dyn FileRead>,
@@ -1378,6 +1382,15 @@ fn read_type_projection_mask(
predicates: &[Predicate],
null_residuals: &std::collections::HashSet<Vec<String>>,
) -> 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 `<name>_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 5a341a8e..981e2ecc 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<DataField> {
+ self.inner
+ .select_read_fields(data_schema_fields, projected_fields)
+ }
+
async fn read_batch_stream(
&self,
reader: Box<dyn FileRead>,
diff --git a/crates/paimon/src/arrow/format/text.rs
b/crates/paimon/src/arrow/format/text.rs
index 61f1ce0f..c2e431e5 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 2075b3ab..92fb7bf9 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<tokio::sync::SemaphorePermit<'static>> {
static SEMAPHORE: OnceLock<tokio::sync::Semaphore> = 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::<Vec<_>>();
+ 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::<Vec<_>>()
+ .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::<Vec<_>>()
+ .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<ArrowSchema>, ids: Vec<i32>, values: Vec<i32>)
-> 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 00000000..b19dfd2a
--- /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<ArrayRef> {
+ read_layout(
+ runtime,
+ session,
+ file.segment_source(),
+ file.footer().layout(),
+ Some(names),
+ )
+}
+
+fn projected_dtype(dtype: &DType, names: Option<&[String]>) -> Result<DType> {
+ 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::<Result<Vec<_>>>()?;
+ 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<dyn SegmentSource>,
+ layout: &LayoutRef,
+ roots: Option<&[String]>,
+) -> Result<ArrayRef> {
+ let dtype = projected_dtype(layout.dtype(), roots)?;
+ if layout.is::<Struct>() {
+ 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::<Chunked>() {
+ 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::<Dict>() {
+ // 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::<StructArray>(&mut ctx).map_err(invalid)?;
+ let columns = names
+ .iter()
+ .map(|name| {
+ rows.unmasked_field_by_name(name.as_str())
+ .cloned()
+ .map_err(invalid)
+ })
+ .collect::<Result<Vec<_>>>()?;
+ 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<dyn LayoutChildren> {
+ Arc::new(self.clone())
+ }
+ fn child(&self, _: usize, _: &DType) ->
vortex::error::VortexResult<LayoutRef> {
+ 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 7a2af66a..7bacb3e2 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<DataField>,
- /// 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<DataField>,
logical_schema: arrow_schema::SchemaRef,
contexts: HashMap<usize, MapReadContext>,
@@ -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<HashSet<String>> {
+ /// Map logical value projections onto every direct and overflow source.
+ pub(crate) fn projected_physical_field(&self, field_name: &str) ->
Option<DataField> {
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 9d548321..7b775da3 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<DataField> = 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<DataField> = 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<dyn Array>>,
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<Vec<DataField>> {
- 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<DataField> {
+ 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<Vec<DataField>> {
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<Option<DataType>> {
- 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<Option<DataType>> {
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 00000000..06a72690
--- /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<DataField>) -> 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<DataField> {
+ 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::<Vec<_>>()
+ .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::<Vec<_>>();
+ 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::<Vec<_>>();
+ 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::<Vec<_>>()
+ .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::<Vec<_>>()
+ .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::<Vec<_>>();
+ 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::<Vec<_>>()
+ .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<AtomicU64>,
+}
+#[async_trait::async_trait]
+impl FileRead for CountingRead {
+ async fn read(&self, range: std::ops::Range<u64>) -> crate::Result<Bytes> {
+ 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::<Vec<_>>().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::<Vec<_>>()
+ .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::<arrow_array::MapArray>()
+ .unwrap()
+ .values();
+ assert_eq!(
+ values
+ .as_any()
+ .downcast_ref::<arrow_array::StructArray>()
+ .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::<Vec<_>>()
+ .await
+ .unwrap();
+ let decoded = concat_batches(&batches[0].schema(), &batches).unwrap();
+ let maps = decoded
+ .column(0)
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .unwrap();
+ assert_eq!(
+ maps.keys()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .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::<Vec<_>>()
+ .await
+ .unwrap();
+ let decoded = concat_batches(&batches[0].schema(), &batches).unwrap();
+ let list = decoded
+ .column(0)
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .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::<StructArray>()
+ .unwrap();
+ assert_eq!(elements.column_names(), vec!["wanted"], "{name}");
+ assert_eq!(
+ elements
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .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::<Vec<_>>()
+ .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 00000000..803a3a4a
--- /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<Group> 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 00000000..51e58491
Binary files /dev/null and
b/crates/paimon/testdata/nested_projection/array.parquet differ
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 00000000..93b86a73
--- /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 00000000..c9d9546e
Binary files /dev/null and
b/crates/paimon/testdata/nested_projection/nested.orc differ
diff --git a/crates/paimon/testdata/nested_projection/repeated.parquet
b/crates/paimon/testdata/nested_projection/repeated.parquet
new file mode 100644
index 00000000..57b4e3d1
Binary files /dev/null and
b/crates/paimon/testdata/nested_projection/repeated.parquet differ
diff --git a/crates/paimon/testdata/nested_projection/standard.parquet
b/crates/paimon/testdata/nested_projection/standard.parquet
new file mode 100644
index 00000000..9d06d8d1
Binary files /dev/null and
b/crates/paimon/testdata/nested_projection/standard.parquet differ
diff --git a/crates/paimon/testdata/nested_projection/tuple.parquet
b/crates/paimon/testdata/nested_projection/tuple.parquet
new file mode 100644
index 00000000..0bc59143
Binary files /dev/null and
b/crates/paimon/testdata/nested_projection/tuple.parquet differ
