sunchao commented on code in PR #7598: URL: https://github.com/apache/datafusion-comet/pull/7598#discussion_r4532041455
########## native/core/src/execution/operators/dynamic_filter/min_max.rs: ########## @@ -0,0 +1,888 @@ +// 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::{lit, 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::stream::RecordBatchStreamAdapter; +use datafusion::physical_plan::{ + ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, + PlanProperties, ReplaceChildrenOptions, SendableRecordBatchStream, +}; +use futures::StreamExt; + +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: AggregateExec, + reader_filter_attached: bool, +} + +impl MinMaxReaderFilterExec { + 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 { Review Comment: [P2] Reject computed expressions inside the scan projection before attaching this predicate. Checking that the aggregate argument is a `Column` does not prove it is a direct file column: `DataSourceExec` can project a computed expression and expose its result as a column to the aggregate. I reproduced this with a volatile integer UDF embedded in the Parquet scan projection and decoder filter pushdown enabled. Ordinary partial MIN returns 0 after two UDF calls; this wrapper is accepted, remaps the dynamic predicate through the projection, evaluates the UDF again during decoder filtering, and returns 10 after three calls. With decoder filtering disabled, both return 0. Please require the selected scan projection to map directly to the file column, or otherwise prove that re-evaluating it cannot change the result. This is an execution-level component reproduction using the locked DataFusion/Parquet versions; this module is not yet wired into Spark, so it does not assert an existing S park query regression. -- 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]
