xudong963 commented on code in PR #23396: URL: https://github.com/apache/datafusion/pull/23396#discussion_r3549067403
########## datafusion/datasource-parquet/src/projection_read_plan.rs: ########## @@ -0,0 +1,481 @@ +// 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. + +//! Resolution of expressions against a Parquet file's schema into a +//! [`ParquetReadPlan`]: the leaf-level [`ProjectionMask`] to install on the +//! decoder plus the Arrow schema the decoder will emit under that mask. +//! +//! This is shared by the opener's projection handling (via +//! [`build_projection_read_plan`]) and row-filter construction (via +//! [`crate::row_filter`]), which both need to translate column and struct +//! field references into Parquet leaf indices. + +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Arc; + +use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use parquet::arrow::ProjectionMask; +use parquet::schema::types::SchemaDescriptor; + +use datafusion_common::tree_node::TreeNode; +use datafusion_physical_expr::PhysicalExpr; +use datafusion_physical_expr::expressions::Column; +use datafusion_physical_expr::utils::collect_columns; + +use crate::row_filter::PushdownChecker; Review Comment: `projection_read_plan` imports `PushdownChecker` from `row_filter`, while`row_filter` imports `ParquetReadPlan` and helpers from `projection_read_plan`. How about moving `PushdownChecker` /`PushdownColumns` into `projection_read_plan.rs` or a neutral module like `column_access.rs`, then have `row_filter.rs` call into that ########## datafusion/datasource-parquet/src/projection_read_plan.rs: ########## @@ -0,0 +1,481 @@ +// 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. + +//! Resolution of expressions against a Parquet file's schema into a +//! [`ParquetReadPlan`]: the leaf-level [`ProjectionMask`] to install on the +//! decoder plus the Arrow schema the decoder will emit under that mask. +//! +//! This is shared by the opener's projection handling (via +//! [`build_projection_read_plan`]) and row-filter construction (via +//! [`crate::row_filter`]), which both need to translate column and struct +//! field references into Parquet leaf indices. + +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Arc; + +use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use parquet::arrow::ProjectionMask; +use parquet::schema::types::SchemaDescriptor; + +use datafusion_common::tree_node::TreeNode; +use datafusion_physical_expr::PhysicalExpr; +use datafusion_physical_expr::expressions::Column; +use datafusion_physical_expr::utils::collect_columns; + +use crate::row_filter::PushdownChecker; + +/// The result of resolving which Parquet leaf columns and Arrow schema fields +/// are needed to evaluate an expression against a Parquet file +/// +/// This is the shared output of the column resolution pipeline used by both +/// the row filter to build `ArrowPredicate`s and the opener to build `ProjectionMask`s +#[derive(Debug, Clone)] +pub(crate) struct ParquetReadPlan { + /// Projection mask built from leaf column indices in the Parquet schema. + /// Using a `ProjectionMask` directly (rather than raw indices) prevents + /// bugs from accidentally mixing up root vs leaf indices. + pub projection_mask: ProjectionMask, + /// The projected Arrow schema containing only the columns/fields required + /// Struct types are pruned to include only the accessed sub-fields + pub projected_schema: SchemaRef, +} + +/// Records a struct field access via `get_field(struct_col, 'field1', 'field2', ...)`. +/// +/// This allows the row filter to project only the specific Parquet leaf columns +/// needed by the filter, rather than all leaves of the struct. +#[derive(Debug, Clone)] +pub(crate) struct StructFieldAccess { + /// Arrow root column index of the struct in the file schema. + pub(crate) root_index: usize, + /// Field names forming the path into the struct. + /// e.g., `["value"]` for `s['value']`, `["outer", "inner"]` for `s['outer']['inner']`. + pub(crate) field_path: Vec<String>, +} + +/// Builds a unified [`ParquetReadPlan`] for a set of projection expressions +/// +/// Unlike [`crate::row_filter::build_parquet_read_plan`] (which is used for +/// filter pushdown and returns `None` when an expression references +/// unsupported nested types or missing columns), this function always +/// succeeds. It collects every column that *can* be resolved in the file and +/// produces a leaf-level projection mask. Columns missing from the file are +/// silently skipped since the projection layer handles those by inserting +/// nulls. +pub(crate) fn build_projection_read_plan( + exprs: impl IntoIterator<Item = Arc<dyn PhysicalExpr>>, + file_schema: &Schema, + schema_descr: &SchemaDescriptor, +) -> ParquetReadPlan { + // fast path: if every expression is a plain Column reference, skip all + // struct analysis and use root-level projection directly + let exprs = exprs.into_iter().collect::<Vec<_>>(); + let all_plain_columns = exprs.iter().all(|e| e.downcast_ref::<Column>().is_some()); + + if all_plain_columns { + let mut root_indices: Vec<usize> = exprs + .iter() + .map(|e| e.downcast_ref::<Column>().unwrap().index()) + .collect(); + root_indices.sort_unstable(); + root_indices.dedup(); + + let projection_mask = + ProjectionMask::roots(schema_descr, root_indices.iter().copied()); + let projected_schema = Arc::new( + file_schema + .project(&root_indices) + .expect("valid column indices"), + ); + + return ParquetReadPlan { + projection_mask, + projected_schema, + }; + } + + // secondary fast path: if the schema has no struct columns, we can skip + // PushdownChecker traversal and use root-level projection + let has_struct_columns = file_schema + .fields() + .iter() + .any(|f| matches!(f.data_type(), DataType::Struct(_))); + + if !has_struct_columns { + let mut root_indices = exprs + .into_iter() + .flat_map(|e| collect_columns(&e).into_iter().map(|col| col.index())) + .collect::<Vec<_>>(); + + root_indices.sort_unstable(); + root_indices.dedup(); + + let projection_mask = + ProjectionMask::roots(schema_descr, root_indices.iter().copied()); + + let projected_schema = Arc::new( + file_schema + .project(&root_indices) + .expect("valid column indices"), + ); + + return ParquetReadPlan { + projection_mask, + projected_schema, + }; + } + + let mut all_root_indices = Vec::new(); + let mut all_struct_accesses = Vec::new(); + + for expr in exprs { + let mut checker = PushdownChecker::new(file_schema, true); + let _ = expr.visit(&mut checker); + let columns = checker.into_sorted_columns(); + + all_root_indices.extend_from_slice(&columns.required_columns); + all_struct_accesses.extend(columns.struct_field_accesses); + } + + all_root_indices.sort_unstable(); + all_root_indices.dedup(); + + // when no struct field accesses were found, fall back to root-level projection + // to match the performance of the simple path + if all_struct_accesses.is_empty() { + let projection_mask = + ProjectionMask::roots(schema_descr, all_root_indices.iter().copied()); + let projected_schema = Arc::new( + file_schema + .project(&all_root_indices) + .expect("valid column indices"), + ); + + return ParquetReadPlan { + projection_mask, + projected_schema, + }; + } + + let leaf_indices = { + let mut out = + leaf_indices_for_roots(all_root_indices.iter().copied(), schema_descr); + let struct_leaf_indices = + resolve_struct_field_leaves(&all_struct_accesses, file_schema, schema_descr); + + out.extend_from_slice(&struct_leaf_indices); + out.sort_unstable(); + out.dedup(); + + out + }; + + let projection_mask = + ProjectionMask::leaves(schema_descr, leaf_indices.iter().copied()); + + let projected_schema = + build_filter_schema(file_schema, &all_root_indices, &all_struct_accesses); + + ParquetReadPlan { + projection_mask, + projected_schema, + } +} + +pub(crate) fn leaf_indices_for_roots<I>( + root_indices: I, + schema_descr: &SchemaDescriptor, +) -> Vec<usize> +where + I: IntoIterator<Item = usize>, +{ + // Always map root (Arrow) indices to Parquet leaf indices via the schema + // descriptor. Arrow root indices only equal Parquet leaf indices when the + // schema has no group columns (Struct, Map, etc.); when group columns + // exist, their children become separate leaves and shift all subsequent + // leaf indices. + // Struct columns are unsupported. + let root_set: BTreeSet<_> = root_indices.into_iter().collect(); + + (0..schema_descr.num_columns()) + .filter(|leaf_idx| { + root_set.contains(&schema_descr.get_column_root_idx(*leaf_idx)) + }) + .collect() +} + +/// Resolves struct field access to specific Parquet leaf column indices +/// +/// For every `StructFieldAccess`, finds the leaf columns in the Parquet schema +/// whose path matches the struct root name + field path. This avoids reading all +/// leaves of a struct when only specific fields are needed +pub(crate) fn resolve_struct_field_leaves( + accesses: &[StructFieldAccess], + file_schema: &Schema, + schema_descr: &SchemaDescriptor, +) -> Vec<usize> { + let mut leaf_indices = Vec::new(); + + for access in accesses { + let root_name = file_schema.field(access.root_index).name(); + let prefix = std::iter::once(root_name.as_str()) + .chain(access.field_path.iter().map(|p| p.as_str())) + .collect::<Vec<_>>(); + + for leaf_idx in 0..schema_descr.num_columns() { + let col = schema_descr.column(leaf_idx); + let col_path = col.path().parts(); + + // A leaf matches if its path starts with our prefix. + // e.g., prefix=["s", "value"] matches leaf path ["s", "value"] + // prefix=["s", "outer"] matches ["s", "outer", "inner"] + + // a leaf matches if its path starts with our prefix + // for example: prefix=["s", "value"] matches leaf path ["s", "value"] + // prefix=["s", "outer"] matches ["s", "outer", "inner"] Review Comment: duplicated comments -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
