sunchao commented on code in PR #7598: URL: https://github.com/apache/datafusion-comet/pull/7598#discussion_r4555546516
########## native/core/src/execution/operators/dynamic_filter/min_max.rs: ########## @@ -0,0 +1,302 @@ +// 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. + +//! Connect one partial MIN/MAX aggregate's improving bound to its Parquet reader. + +use std::fmt::Formatter; +use std::sync::Arc; + +use arrow::datatypes::DataType; +use datafusion::common::config::ConfigOptions; +use datafusion::common::tree_node::TreeNodeRecursion; +use datafusion::common::{internal_err, Result, Statistics}; +use datafusion::datasource::source::DataSourceExec; +use datafusion::execution::TaskContext; +use datafusion::functions_aggregate::min_max::{Max, Min}; +use datafusion::physical_expr::expressions::{Column, DynamicFilterPhysicalExpr}; +use datafusion::physical_expr::{OrderingRequirements, PhysicalExpr}; +use datafusion::physical_plan::aggregates::{AggregateExec, AggregateMode}; +use datafusion::physical_plan::distribution_requirements::InputDistributionRequirements; +use datafusion::physical_plan::execution_plan::CardinalityEffect; +use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder, MetricsSet}; +use datafusion::physical_plan::statistics::{ChildStats, StatisticsArgs}; +use datafusion::physical_plan::{ + ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, + PlanProperties, ReplaceChildrenOptions, SendableRecordBatchStream, +}; + +use super::parquet_reader::{parquet_file_source, try_attach_parquet_reader_filter}; + +/// Own an unexecuted aggregate template and metric handles. Each execution owns +/// a fresh aggregate, accumulator bounds, and reader predicate; none of that live +/// state is retained by the template or shared with another Spark task. +#[derive(Debug)] +pub(crate) struct MinMaxReaderFilterExec { + template: AggregateExec, + config: ConfigOptions, + metrics: ExecutionPlanMetricsSet, +} + +struct RuntimeAggregate { + aggregate: Arc<dyn ExecutionPlan>, + reader_filter_attached: bool, +} + +impl MinMaxReaderFilterExec { + // Spark planner integration is a separate step of apache/datafusion-comet#5774. + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn try_new( + aggregate: &AggregateExec, + config: &ConfigOptions, + ) -> Result<Option<Self>> { + if !config.optimizer.enable_dynamic_filter_pushdown + || !config.optimizer.enable_aggregate_dynamic_filter_pushdown + || aggregate.mode() != &AggregateMode::Partial + || !aggregate.group_expr().is_true_no_grouping() + || aggregate.limit_options().is_some() + || aggregate.filter_expr().iter().any(Option::is_some) + || aggregate.input().output_partitioning().partition_count() != 1 + || aggregate.input().fetch().is_some() + { + return Ok(None); + } + let [expression] = aggregate.aggr_expr() else { + return Ok(None); + }; + let function = expression.fun().inner(); + if !(function.is::<Min>() || function.is::<Max>()) + || expression.is_distinct() + || !expression.order_bys().is_empty() + { + return Ok(None); + } + let arguments = expression.expressions(); + let [key] = arguments.as_slice() else { + return Ok(None); + }; + if !key.is::<Column>() + || !matches!( + key.data_type(aggregate.input().schema().as_ref())?, + DataType::Int8 | DataType::Int16 | DataType::Int32 | DataType::Int64 + ) + { + return Ok(None); + } + // Keep expressions, residuals, limits, and native/JVM boundaries out of + // this first path. The shared helper still handles reader column mapping + // and per-file schema guards once this direct scan is accepted. + let Some(scan) = aggregate.input().downcast_ref::<DataSourceExec>() else { + return Ok(None); + }; + let Some((_, source)) = parquet_file_source(scan) else { + return Ok(None); + }; + if source.filter().is_some() + || source + .projection() + .is_some_and(|projection| projection.iter().any(|expr| !expr.expr.is::<Column>())) + { + // Pruning before a computed scan projection could change its stateful + // evaluation or suppress errors, even for an unselected output column. + return Ok(None); + } + Ok(Some(Self::new(aggregate, config.clone())?)) Review Comment: [P2] Preserve required-field validation when pruning row groups On `6a0aae9b6b78c6197f74bb86aebd021ba045365a`, write a physically nullable Int32 Parquet key with row groups `[0, 1]` and `[10, NULL]`, then project it as a required key. With the same scan, batch size two, and decoder/page filtering disabled, ordinary partial `MIN` fails with `Column 'key' is declared as non-nullable but contains null values`. This wrapper attaches the live bound, prunes one group, and returns `0`. Declaring the key nullable makes both paths return `0`. The direct-column check still permits the schema adapter to remove a same-type nullability cast, so pruning can skip the decoder validation. Guard attachment against this mismatch before row-group pruning. This is a native component reproduction. ########## native/core/src/execution/operators/dynamic_filter/min_max.rs: ########## @@ -0,0 +1,302 @@ +// 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. + +//! Connect one partial MIN/MAX aggregate's improving bound to its Parquet reader. + +use std::fmt::Formatter; +use std::sync::Arc; + +use arrow::datatypes::DataType; +use datafusion::common::config::ConfigOptions; +use datafusion::common::tree_node::TreeNodeRecursion; +use datafusion::common::{internal_err, Result, Statistics}; +use datafusion::datasource::source::DataSourceExec; +use datafusion::execution::TaskContext; +use datafusion::functions_aggregate::min_max::{Max, Min}; +use datafusion::physical_expr::expressions::{Column, DynamicFilterPhysicalExpr}; +use datafusion::physical_expr::{OrderingRequirements, PhysicalExpr}; +use datafusion::physical_plan::aggregates::{AggregateExec, AggregateMode}; +use datafusion::physical_plan::distribution_requirements::InputDistributionRequirements; +use datafusion::physical_plan::execution_plan::CardinalityEffect; +use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder, MetricsSet}; +use datafusion::physical_plan::statistics::{ChildStats, StatisticsArgs}; +use datafusion::physical_plan::{ + ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, + PlanProperties, ReplaceChildrenOptions, SendableRecordBatchStream, +}; + +use super::parquet_reader::{parquet_file_source, try_attach_parquet_reader_filter}; + +/// Own an unexecuted aggregate template and metric handles. Each execution owns +/// a fresh aggregate, accumulator bounds, and reader predicate; none of that live +/// state is retained by the template or shared with another Spark task. +#[derive(Debug)] +pub(crate) struct MinMaxReaderFilterExec { + template: AggregateExec, + config: ConfigOptions, + metrics: ExecutionPlanMetricsSet, +} + +struct RuntimeAggregate { + aggregate: Arc<dyn ExecutionPlan>, + reader_filter_attached: bool, +} + +impl MinMaxReaderFilterExec { + // Spark planner integration is a separate step of apache/datafusion-comet#5774. + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn try_new( + aggregate: &AggregateExec, + config: &ConfigOptions, + ) -> Result<Option<Self>> { + if !config.optimizer.enable_dynamic_filter_pushdown + || !config.optimizer.enable_aggregate_dynamic_filter_pushdown + || aggregate.mode() != &AggregateMode::Partial + || !aggregate.group_expr().is_true_no_grouping() + || aggregate.limit_options().is_some() + || aggregate.filter_expr().iter().any(Option::is_some) + || aggregate.input().output_partitioning().partition_count() != 1 + || aggregate.input().fetch().is_some() + { + return Ok(None); + } + let [expression] = aggregate.aggr_expr() else { + return Ok(None); + }; + let function = expression.fun().inner(); + if !(function.is::<Min>() || function.is::<Max>()) + || expression.is_distinct() + || !expression.order_bys().is_empty() + { + return Ok(None); + } + let arguments = expression.expressions(); + let [key] = arguments.as_slice() else { + return Ok(None); + }; + if !key.is::<Column>() + || !matches!( + key.data_type(aggregate.input().schema().as_ref())?, + DataType::Int8 | DataType::Int16 | DataType::Int32 | DataType::Int64 + ) + { + return Ok(None); + } + // Keep expressions, residuals, limits, and native/JVM boundaries out of + // this first path. The shared helper still handles reader column mapping + // and per-file schema guards once this direct scan is accepted. + let Some(scan) = aggregate.input().downcast_ref::<DataSourceExec>() else { + return Ok(None); + }; + let Some((_, source)) = parquet_file_source(scan) else { + return Ok(None); + }; + if source.filter().is_some() + || source + .projection() + .is_some_and(|projection| projection.iter().any(|expr| !expr.expr.is::<Column>())) + { + // Pruning before a computed scan projection could change its stateful + // evaluation or suppress errors, even for an unselected output column. + return Ok(None); + } + Ok(Some(Self::new(aggregate, config.clone())?)) + } + + fn new(aggregate: &AggregateExec, config: ConfigOptions) -> Result<Self> { + Ok(Self { + template: Self::fresh_aggregate(aggregate, Arc::clone(aggregate.input()))?, + config, + metrics: ExecutionPlanMetricsSet::new(), + }) + } + + fn fresh_aggregate( + template: &AggregateExec, + input: Arc<dyn ExecutionPlan>, + ) -> Result<AggregateExec> { + // AggregateExec::clone/replace_children keep shared accumulator bounds. + // Replacing just its dynamic expression would leave those bounds stale. + AggregateExec::try_new( + *template.mode(), + template.group_expr().clone(), + template.aggr_expr().to_vec(), + template.filter_expr().to_vec(), + input, + template.input_schema(), + ) + } + + fn build_runtime_aggregate(&self) -> Result<RuntimeAggregate> { + let aggregate: Arc<dyn ExecutionPlan> = Arc::new(Self::fresh_aggregate( + &self.template, + Arc::clone(self.template.input()), + )?); + let predicate = aggregate + .dynamic_expressions_produced() + .into_iter() + .find_map(|expr| { + (expr as Arc<dyn std::any::Any + Send + Sync>) + .downcast::<DynamicFilterPhysicalExpr>() + .ok() + }); + let Some(predicate) = predicate else { + return Ok(RuntimeAggregate { + aggregate, + reader_filter_attached: false, + }); + }; + let reader = + try_attach_parquet_reader_filter(self.template.input(), predicate, &self.config)?; Review Comment: [P2] Bound runtime pruning work when every group improves MIN With one descending Int32 key per row group, batch size one, one file and partition, and no decoder/page filtering, the wrapper and ordinary aggregate both read every row, prune zero groups, and return `1`. The exact debug build takes about 1.11 s versus 0.09 s at 256 groups, 4.24 s versus 0.18 s at 512, and 15.83 s versus 0.38 s at 1,024. The reader reevaluates the live predicate over the unread tail at each boundary, so it repeatedly builds statistics for groups whose bounds can still improve MIN. Bound or adapt that repeated work before enabling this reader path on unfavorable layouts. -- 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]
