feat(read): complete nested projection pushdown for Parquet

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());

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};

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;

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>,

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),
         }
     }