viirya commented on code in PR #6433: URL: https://github.com/apache/datafusion-comet/pull/6433#discussion_r4210479974
########## native/core/src/execution/operators/dynamic_filter/early.rs: ########## @@ -0,0 +1,135 @@ +// 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. + +//! Place a live ancestor filter before an intermediate join's probe work. +//! +//! This is decoded-batch filtering only. It leaves scan/schema conversion and +//! arbitrary expressions in place, and never propagates into an intermediate +//! build side. The downstream join remains the authority for matching rows. + +use std::any::Any; +use std::sync::Arc; + +use datafusion::common::{internal_err, Result}; +use datafusion::physical_expr::expressions::{Column, DynamicFilterPhysicalExpr}; +use datafusion::physical_expr::PhysicalExpr; +use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet; +use datafusion::physical_plan::ExecutionPlan; +use datafusion_comet_operators::CometFilterExec; + +use super::parquet_reader::is_direct_column_null_checks; +use super::{DynamicFilterExec, DynamicFilterJoinExec}; +use crate::execution::operators::CometProjectionExec; + +pub(super) fn place_early_filter( + input: &Arc<dyn ExecutionPlan>, + predicate: Arc<DynamicFilterPhysicalExpr>, + metrics: &ExecutionPlanMetricsSet, +) -> Result<Arc<dyn ExecutionPlan>> { + Ok(place(input, predicate, metrics, false)?.unwrap_or_else(|| Arc::clone(input))) +} + +fn remap( + predicate: Arc<DynamicFilterPhysicalExpr>, + column: Arc<dyn PhysicalExpr>, +) -> Result<Arc<DynamicFilterPhysicalExpr>> { + // Derived expressions share producer updates. Taking current() here would + // capture the initial TRUE placeholder instead of the completed build domain. + let mapped: Arc<dyn Any + Send + Sync> = predicate.with_new_children(vec![column])?; + mapped.downcast::<DynamicFilterPhysicalExpr>().map_err(|_| { + datafusion::common::DataFusionError::Internal( + "Dynamic filter remapping changed type".into(), + ) + }) +} + +fn place( + input: &Arc<dyn ExecutionPlan>, + predicate: Arc<DynamicFilterPhysicalExpr>, + metrics: &ExecutionPlanMetricsSet, + crossed_join: bool, +) -> Result<Option<Arc<dyn ExecutionPlan>>> { + let children = predicate.children(); + let [key] = children.as_slice() else { + return internal_err!("Early join filtering requires one key"); + }; + let Some(key) = key.downcast_ref::<Column>() else { + return internal_err!("Early join filtering requires a column key"); + }; + if input.fetch().is_none() { + if let Some(join) = input.downcast_ref::<DynamicFilterJoinExec>() { + let template = join.template(); + // This wrapper is already limited to ordinary inner, single-key joins. + // Do not skip fallible join residuals, or guess an embedded projection. + if template.filter().is_none() && !template.contains_projection() { + let build_columns = template.left().schema().fields().len(); + if let Some(index) = key.index().checked_sub(build_columns) { + if let Some(field) = template.right().schema().fields().get(index) { + let mapped = remap( + Arc::clone(&predicate), + Arc::new(Column::new(field.name(), index)), + )?; + if let Some(probe) = place(template.right(), mapped, metrics, true)? { + return Ok(Some(join.with_execution_probe(probe)?)); + } + } + } + } + } else if let Some(projection) = input.downcast_ref::<CometProjectionExec>() { + let exprs = projection.projection().expr(); + // Even an unrelated computed expression may fail or be stateful. + if exprs.iter().all(|expr| expr.expr.is::<Column>()) { + if let Some(expr) = exprs.get(key.index()) { + let mapped = remap(Arc::clone(&predicate), Arc::clone(&expr.expr))?; + if let Some(child) = place(projection.input(), mapped, metrics, crossed_join)? { + return Ok(Some(projection.with_execution_input(child)?)); + } + } + } + } else if let Some(filter) = input.downcast_ref::<CometFilterExec>() { Review Comment: Thanks, this fixes it. I checked that `nullable_unselective_ancestor_bypasses_above_inferred_null_checks` would fail on the previous placement (all rows evaluated, nothing bypassed) and passes now. ########## native/core/src/execution/operators/dynamic_filter/mod.rs: ########## @@ -172,8 +194,18 @@ impl ExecutionPlan for DynamicFilterExec { // add its input/output counts or elapsed time to the join's existing metrics. let eval_time = MetricBuilder::new(&self.metrics) .subset_time(format!("{}_eval_time", self.metric_prefix), partition); + // Early filtering duplicates the final consumer. Stop that extra work after + // two nonempty evaluated batches remove nothing. The downstream join still + // verifies every row, so later selectivity changes only lose an optimization. + // Keep this decision per stream; an inactive TRUE placeholder is not a sample. + let adaptive = self.adaptive; Review Comment: The documented rationale works for me. Keeping the policy simple until there is workload evidence is reasonable, and the final consumer keeps it correct either way. ########## spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala: ########## @@ -431,6 +431,60 @@ class CometJoinSuite extends CometTestBase { } } + test("join dynamic filter rejects rows before an intermediate join") { Review Comment: Thanks for the q7/q19/q42 runs and the SQL file matrix. Those results were on a base that already includes #6396, which also helps answer the question in comphead's review. -- 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]
