You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/11 12:43:07 UTC
(datafusion-comet) branch main updated: feat: filter decoded probe batches before intermediate joins (#6433)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 19b0dd264c feat: filter decoded probe batches before intermediate joins (#6433)
19b0dd264c is described below
commit 19b0dd264cd39c2d34a7bffa98694d44f48c8163
Author: Chao Sun <su...@apache.org>
AuthorDate: Sun Oct 11 11:35:27 2026 +0000
feat: filter decoded probe batches before intermediate joins (#6433)
* feat: filter decoded probe batches before intermediate joins
* fix: address early join runtime filter review feedback
---
docs/source/user-guide/latest/metrics.md | 23 +-
docs/source/user-guide/latest/tuning/operators.md | 25 ++
.../execution/operators/dynamic_filter/early.rs | 142 ++++++++
.../src/execution/operators/dynamic_filter/join.rs | 36 +-
.../operators/dynamic_filter/join/tests.rs | 394 +++++++++++++++++++++
.../src/execution/operators/dynamic_filter/mod.rs | 62 +++-
.../operators/dynamic_filter/parquet_reader.rs | 8 +
.../execution/operators/dynamic_filter/tests.rs | 120 +++++++
native/core/src/execution/planner.rs | 42 ++-
native/operators/src/lib.rs | 2 +
native/operators/src/projection.rs | 295 +++++++++++++++
.../apache/spark/sql/comet/CometMetricNode.scala | 10 +-
.../sql-tests/join/dynamic_filter_chain.sql | 72 ++++
.../org/apache/comet/exec/CometJoinSuite.scala | 104 +++++-
14 files changed, 1295 insertions(+), 40 deletions(-)
diff --git a/docs/source/user-guide/latest/metrics.md b/docs/source/user-guide/latest/metrics.md
index 56491281db..75d9d1c282 100644
--- a/docs/source/user-guide/latest/metrics.md
+++ b/docs/source/user-guide/latest/metrics.md
@@ -93,14 +93,21 @@ With `spark.comet.exec.join.dynamicFilter.enabled=true`, native broadcast and sh
report these additional metric keys. See [Join Runtime Filters](tuning/operators.md#join-runtime-filters) for
eligibility and reader restrictions.
-| Metric | Description |
-| -------------------------------------- | ----------------------------------------------------------------- |
-| `dynamic_filter_join_rows_evaluated` | Probe rows evaluated by the runtime filter. |
-| `dynamic_filter_join_rows_pruned` | Probe rows rejected by that filter before the hash probe. |
-| `dynamic_filter_join_rows_bypassed` | Probe rows passed through while the runtime filter is inactive. |
-| `dynamic_filter_join_eval_time` | Time evaluating the runtime filter. |
-| `dynamic_filter_join_filters_attached` | Executions that attach their runtime filter to a native reader. |
-| `dynamic_filter_join_filters_skipped` | Executions whose probe input is ineligible for reader attachment. |
+| Metric | Description |
+| -------------------------------------- | ------------------------------------------------------------------------------------------------------------- |
+| `dynamic_filter_join_rows_evaluated` | Probe rows evaluated by the runtime filter. |
+| `dynamic_filter_join_rows_pruned` | Probe rows rejected by that filter before the hash probe. |
+| `dynamic_filter_join_rows_bypassed` | Probe rows passed through while the runtime filter is inactive. |
+| `dynamic_filter_join_eval_time` | Time evaluating the runtime filter. |
+| `dynamic_filter_join_filters_attached` | Executions that attach their runtime filter to a native reader. |
+| `dynamic_filter_join_filters_skipped` | Executions whose probe input is ineligible for reader attachment. |
+| `dynamic_filter_early_rows_evaluated` | Rows evaluated before an intermediate join on this join's probe path. |
+| `dynamic_filter_early_rows_pruned` | Rows rejected there before the intermediate join does its probe work. |
+| `dynamic_filter_early_rows_bypassed` | Rows passed through while early filtering is inactive or has stopped after two nonempty all-matching batches. |
+| `dynamic_filter_early_eval_time` | Time evaluating that early filter. |
+
+See [early filtering in join chains](tuning/operators.md#early-filtering-in-join-chains)
+for placement and adaptive bypass behavior.
The row counters measure residual filtering of decoded probe batches. They exclude rows skipped
by the reader. An attached filter does not guarantee that any row groups are pruned: compare the
diff --git a/docs/source/user-guide/latest/tuning/operators.md b/docs/source/user-guide/latest/tuning/operators.md
index cb538fb63d..94bb76d5ad 100644
--- a/docs/source/user-guide/latest/tuning/operators.md
+++ b/docs/source/user-guide/latest/tuning/operators.md
@@ -67,6 +67,31 @@ boundaries. A shuffled hash join can still filter probe batches after shuffle, b
its filter back to an earlier scan stage. Compare the [runtime-filter and scan metrics](../metrics.md#hash-joins)
with the setting disabled to distinguish reduced hash-probe work from reader I/O savings.
+#### Early filtering in join chains
+
+With `spark.comet.exec.join.dynamicFilter.enabled=true`, an eligible join can also
+use its completed build keys to filter decoded rows before an intermediate inner
+join. For example, when a final join accepts only customer 42, it can remove other
+customers before an earlier join looks up their orders. This reduces intermediate
+join work while the final join still checks every match and preserves duplicates.
+
+Placement follows direct probe-side columns through ordinary inner joins already
+eligible for runtime filtering (a single signed-integer key),
+column-only projections, and direct column null checks above the first intermediate join.
+After crossing a join, the early consumer stays above inferred null checks so null
+keys do not prevent adaptive bypass. It stops at computed
+expressions, other predicates, limits, unsupported joins, and execution boundaries.
+These early filters operate on decoded batches; they do not introduce additional
+Parquet pruning or skip schema conversions. Each intermediate join retains its
+existing reader-filter checks.
+
+Early filtering stops evaluating after two consecutive nonempty evaluated batches
+remove no rows, avoiding repeated membership checks on unselective inputs. That
+decision is local to one execution. The final join remains responsible for matching
+rows, including if later batches become more selective. The `dynamic_filter_early_*`
+metrics belong to the join supplying the filter; intermediate joins and projections
+retain their own input/output metrics.
+
## Adaptive Partial Aggregation
Set `spark.comet.exec.aggregate.skipPartial.enabled=true` to let Comet bypass partial hash
diff --git a/native/core/src/execution/operators/dynamic_filter/early.rs b/native/core/src/execution/operators/dynamic_filter/early.rs
new file mode 100644
index 0000000000..6f23c54d3c
--- /dev/null
+++ b/native/core/src/execution/operators/dynamic_filter/early.rs
@@ -0,0 +1,142 @@
+// 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 datafusion_comet_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>() {
+ // Once below a join, leave the terminal consumer above Spark's
+ // inferred null checks. Rejected nulls otherwise make an unselective
+ // ancestor look selective and prevent adaptive bypass. Early consumers
+ // do not prune readers, so descending farther offers no I/O benefit.
+ if !crossed_join
+ && !filter.has_projection()
+ && is_direct_column_null_checks(filter.predicate())
+ {
+ if let Some(child) = place(
+ filter.input(),
+ Arc::clone(&predicate),
+ metrics,
+ crossed_join,
+ )? {
+ return Ok(Some(filter.with_execution_input(child)?));
+ }
+ }
+ } else if let Some(filter) = input.downcast_ref::<DynamicFilterExec>() {
+ if let Some(child) =
+ place(&filter.input, Arc::clone(&predicate), metrics, crossed_join)?
+ {
+ return Ok(Some(filter.with_execution_input(child)));
+ }
+ }
+ }
+ // A terminal consumer is useful only after crossing a join. Above other
+ // boundaries the existing final consumer already performs the same work.
+ Ok(crossed_join.then(|| {
+ Arc::new(
+ DynamicFilterExec::new(
+ Arc::clone(input),
+ predicate,
+ metrics.clone(),
+ "dynamic_filter_early",
+ )
+ .adaptive(),
+ ) as Arc<dyn ExecutionPlan>
+ }))
+}
diff --git a/native/core/src/execution/operators/dynamic_filter/join.rs b/native/core/src/execution/operators/dynamic_filter/join.rs
index ba2e275960..c2688781b1 100644
--- a/native/core/src/execution/operators/dynamic_filter/join.rs
+++ b/native/core/src/execution/operators/dynamic_filter/join.rs
@@ -45,6 +45,7 @@ use datafusion::physical_plan::{
};
use futures::StreamExt;
+use super::early::place_early_filter;
use super::parquet_reader::try_attach_parquet_reader_filter;
use super::DynamicFilterExec;
@@ -84,19 +85,38 @@ impl DynamicFilterJoinExec {
})
}
+ pub(super) fn template(&self) -> &HashJoinExec {
+ &self.template
+ }
+
+ pub(super) fn with_execution_probe(
+ &self,
+ probe: Arc<dyn ExecutionPlan>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ Ok(Arc::new(Self {
+ template: self
+ .template
+ .builder()
+ .reset_state()
+ .with_new_children(vec![Arc::clone(self.template.left()), probe])?
+ .build()?,
+ config: self.config.clone(),
+ metrics: self.metrics.clone(),
+ }))
+ }
+
fn build_runtime_join(&self) -> Result<RuntimeDynamicFilterJoin> {
let predicate = Arc::new(DynamicFilterPhysicalExpr::new(
vec![Arc::clone(&self.template.on()[0].1)],
lit(true),
));
- let reader = try_attach_parquet_reader_filter(
- self.template.right(),
- Arc::clone(&predicate),
- &self.config,
- )?;
+ let probe =
+ place_early_filter(self.template.right(), Arc::clone(&predicate), &self.metrics)?;
+ let reader =
+ try_attach_parquet_reader_filter(&probe, Arc::clone(&predicate), &self.config)?;
let reader_filter_attached = reader.is_some();
let consumer = Arc::new(DynamicFilterExec::new(
- reader.unwrap_or_else(|| Arc::clone(self.template.right())),
+ reader.unwrap_or(probe),
Arc::clone(&predicate),
self.metrics.clone(),
"dynamic_filter_join",
@@ -192,6 +212,10 @@ impl ExecutionPlan for DynamicFilterJoinExec {
self.template.maintains_input_order()
}
+ fn fetch(&self) -> Option<usize> {
+ self.template.fetch()
+ }
+
fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
self.template.children()
}
diff --git a/native/core/src/execution/operators/dynamic_filter/join/tests.rs b/native/core/src/execution/operators/dynamic_filter/join/tests.rs
index 14242b4e4f..fd2248a053 100644
--- a/native/core/src/execution/operators/dynamic_filter/join/tests.rs
+++ b/native/core/src/execution/operators/dynamic_filter/join/tests.rs
@@ -1564,3 +1564,397 @@ fn wrapper_preserves_join_statistics_and_distribution() {
);
}
}
+
+fn chain_join(
+ build: Arc<dyn ExecutionPlan>,
+ probe: Arc<dyn ExecutionPlan>,
+ build_key: usize,
+ probe_key: usize,
+) -> HashJoinExec {
+ let left = Arc::new(Column::new(
+ build.schema().field(build_key).name(),
+ build_key,
+ ));
+ let right = Arc::new(Column::new(
+ probe.schema().field(probe_key).name(),
+ probe_key,
+ ));
+ HashJoinExec::try_new(
+ build,
+ probe,
+ vec![(left, right)],
+ None,
+ &JoinType::Inner,
+ None,
+ PartitionMode::Partitioned,
+ NullEquality::NullEqualsNothing,
+ false,
+ )
+ .unwrap()
+}
+
+#[tokio::test]
+async fn ancestor_filter_prunes_before_intermediate_join_and_keeps_metric_owners() {
+ use datafusion_comet_operators::CometProjectionExec;
+ let ctx = SessionContext::new();
+ let fact = input(
+ vec![Some(0), Some(1), Some(2), Some(2), Some(3), None],
+ &DataType::Int32,
+ 0,
+ );
+ let dimension = input(
+ vec![Some(0), Some(1), Some(2), Some(2), Some(3)],
+ &DataType::Int32,
+ 0,
+ );
+ let ancestor_build = input(vec![Some(2), Some(2)], &DataType::Int32, 0);
+ let inner = chain_join(dimension, fact, 0, 0);
+ let inner = Arc::new(
+ DynamicFilterJoinExec::try_new(&inner, ctx.copied_config().options())
+ .unwrap()
+ .unwrap(),
+ );
+ let projection = Arc::new(CometProjectionExec::from_datafusion(
+ ProjectionExec::try_new(
+ vec![
+ (
+ Arc::new(Column::new("payload", 3)) as Arc<dyn PhysicalExpr>,
+ "payload".into(),
+ ),
+ (
+ Arc::new(Column::new("key", 2)) as Arc<dyn PhysicalExpr>,
+ "key".into(),
+ ),
+ ],
+ Arc::clone(&inner) as Arc<dyn ExecutionPlan>,
+ )
+ .unwrap(),
+ ));
+ let outer = chain_join(
+ ancestor_build,
+ Arc::clone(&projection) as Arc<dyn ExecutionPlan>,
+ 0,
+ 1,
+ );
+ let expected = collect(
+ Arc::new(outer.builder().reset_state().build().unwrap()),
+ ctx.task_ctx(),
+ )
+ .await
+ .unwrap();
+ assert_eq!(row_count(&expected), 8);
+ // Start fresh metric owners after executing the ordinary reference plan.
+ let before_inner = inner.metrics().unwrap().output_rows().unwrap_or(0);
+ let before_projection = projection.metrics().unwrap().output_rows().unwrap_or(0);
+ let outer = Arc::new(
+ DynamicFilterJoinExec::try_new(&outer, ctx.copied_config().options())
+ .unwrap()
+ .unwrap(),
+ ) as Arc<dyn ExecutionPlan>;
+ for execution in 1..=2 {
+ let actual = collect(Arc::clone(&outer), ctx.task_ctx()).await.unwrap();
+ assert_eq!(
+ batches_to_sort_string(&actual),
+ batches_to_sort_string(&expected)
+ );
+ assert_eq!(
+ metric(&outer, "dynamic_filter_early_rows_evaluated"),
+ 6 * execution
+ );
+ assert_eq!(
+ metric(&outer, "dynamic_filter_early_rows_pruned"),
+ 4 * execution
+ );
+ assert_eq!(
+ inner.metrics().unwrap().output_rows().unwrap(),
+ before_inner + 4 * execution
+ );
+ assert_eq!(
+ projection.metrics().unwrap().output_rows().unwrap(),
+ before_projection + 4 * execution
+ );
+ }
+}
+
+#[tokio::test]
+async fn early_filter_preserves_probe_limit_selection() {
+ use datafusion::physical_plan::limit::LocalLimitExec;
+ let ctx = SessionContext::new();
+ let fact = Arc::new(LocalLimitExec::new(
+ input(vec![Some(0), Some(1), Some(2)], &DataType::Int32, 0),
+ 2,
+ ));
+ let dimension = input(vec![Some(0), Some(1), Some(2)], &DataType::Int32, 0);
+ let inner = chain_join(dimension, fact, 0, 0);
+ let inner = Arc::new(
+ DynamicFilterJoinExec::try_new(&inner, ctx.copied_config().options())
+ .unwrap()
+ .unwrap(),
+ );
+ let outer = chain_join(input(vec![Some(2)], &DataType::Int32, 0), inner, 0, 2);
+ let expected = collect(
+ Arc::new(outer.builder().reset_state().build().unwrap()),
+ ctx.task_ctx(),
+ )
+ .await
+ .unwrap();
+ let outer = Arc::new(
+ DynamicFilterJoinExec::try_new(&outer, ctx.copied_config().options())
+ .unwrap()
+ .unwrap(),
+ ) as Arc<dyn ExecutionPlan>;
+ let actual = collect(Arc::clone(&outer), ctx.task_ctx()).await.unwrap();
+ assert_eq!(row_count(&expected), 0);
+ assert_eq!(row_count(&actual), 0);
+ // The early consumer sees exactly the two already-selected rows.
+ assert_eq!(metric(&outer, "dynamic_filter_early_rows_evaluated"), 2);
+ assert_eq!(metric(&outer, "dynamic_filter_early_rows_pruned"), 2);
+}
+
+#[tokio::test]
+async fn early_filter_does_not_hide_computed_projection_errors() {
+ use datafusion_comet_operators::CometProjectionExec;
+ let ctx = SessionContext::new();
+ let inner = chain_join(
+ input(vec![Some(0), Some(2)], &DataType::Int32, 0),
+ input(vec![Some(0), Some(2)], &DataType::Int32, 0),
+ 0,
+ 0,
+ );
+ let inner = Arc::new(
+ DynamicFilterJoinExec::try_new(&inner, ctx.copied_config().options())
+ .unwrap()
+ .unwrap(),
+ );
+ let projection = Arc::new(CometProjectionExec::from_datafusion(
+ ProjectionExec::try_new(
+ vec![
+ (
+ Arc::new(BinaryExpr::new(
+ lit(10i32),
+ Operator::Divide,
+ Arc::new(Column::new("key", 2)),
+ )) as Arc<dyn PhysicalExpr>,
+ "quotient".into(),
+ ),
+ (
+ Arc::new(Column::new("key", 2)) as Arc<dyn PhysicalExpr>,
+ "key".into(),
+ ),
+ ],
+ inner,
+ )
+ .unwrap(),
+ ));
+ let outer = chain_join(input(vec![Some(2)], &DataType::Int32, 0), projection, 0, 1);
+ for enabled in [false, true] {
+ let plan = PhysicalPlanner::apply_join_dynamic_filter(
+ Arc::new(outer.builder().reset_state().build().unwrap()),
+ enabled,
+ ctx.copied_config().options(),
+ )
+ .unwrap();
+ let error = collect(plan, ctx.task_ctx()).await.unwrap_err();
+ assert!(error.to_string().to_lowercase().contains("zero"), "{error}");
+ }
+}
+
+#[test]
+fn early_filter_stops_at_build_columns_and_join_residuals() {
+ use super::super::early::place_early_filter;
+ use datafusion::physical_plan::joins::utils::JoinFilter;
+ let config = ConfigOptions::default();
+ let metrics = ExecutionPlanMetricsSet::new();
+ let inner = plain_join();
+ let wrapped = Arc::new(
+ DynamicFilterJoinExec::try_new(&inner, &config)
+ .unwrap()
+ .unwrap(),
+ ) as Arc<dyn ExecutionPlan>;
+ // Column 1 belongs to this intermediate join's build; do not filter its probe.
+ let predicate = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::new(Column::new("key", 1))],
+ lit(true),
+ ));
+ assert!(Arc::ptr_eq(
+ &wrapped,
+ &place_early_filter(&wrapped, predicate, &metrics).unwrap()
+ ));
+ let filtered = inner
+ .builder()
+ .with_filter(Some(JoinFilter::new(
+ lit(true),
+ vec![],
+ Arc::new(Schema::empty()),
+ )))
+ .build()
+ .unwrap();
+ let wrapped = Arc::new(
+ DynamicFilterJoinExec::try_new(&filtered, &config)
+ .unwrap()
+ .unwrap(),
+ ) as Arc<dyn ExecutionPlan>;
+ let predicate = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::new(Column::new("key", 2))],
+ lit(true),
+ ));
+ assert!(Arc::ptr_eq(
+ &wrapped,
+ &place_early_filter(&wrapped, predicate, &metrics).unwrap()
+ ));
+}
+
+#[tokio::test]
+async fn early_consumers_release_ancestor_domains_when_cancelled() {
+ for after_output in [false, true] {
+ let (session, pool) = limited_session(8 * 1024 * 1024);
+ let inner = chain_join(
+ input((0..4096).map(Some).collect(), &DataType::Int32, 0),
+ input((0..4096).map(Some).collect(), &DataType::Int32, 0),
+ 0,
+ 0,
+ );
+ let inner = Arc::new(DynamicFilterJoinExec::new(&inner, ConfigOptions::default()).unwrap());
+ let outer = chain_join(
+ input(vec![Some(3)], &DataType::Int32, 0),
+ Arc::clone(&inner) as Arc<dyn ExecutionPlan>,
+ 0,
+ 2,
+ );
+ let outer = DynamicFilterJoinExec::new(&outer, ConfigOptions::default()).unwrap();
+ let runtime = outer.build_runtime_join().unwrap();
+ let predicate = Arc::downgrade(&produced_join_filter(&runtime.join));
+ let final_consumer = runtime
+ .join
+ .right()
+ .downcast_ref::<DynamicFilterExec>()
+ .unwrap();
+ let intermediate = final_consumer
+ .input
+ .downcast_ref::<DynamicFilterJoinExec>()
+ .unwrap();
+ let early_consumer = intermediate
+ .template
+ .right()
+ .downcast_ref::<DynamicFilterExec>()
+ .unwrap();
+ let early_predicate = Arc::downgrade(&early_consumer.predicate);
+ let mut stream = outer
+ .execute_runtime_join(runtime.join, 0, session.task_ctx())
+ .unwrap();
+ assert!(predicate.upgrade().is_some());
+ if after_output {
+ loop {
+ if stream.next().await.unwrap().unwrap().num_rows() > 0 {
+ break;
+ }
+ }
+ assert!(pool.reserved() > 0);
+ assert!(
+ outer
+ .metrics()
+ .unwrap()
+ .sum_by_name("dynamic_filter_early_rows_pruned")
+ .unwrap()
+ .as_usize()
+ > 0
+ );
+ }
+ drop(stream);
+ assert_eq!(pool.reserved(), 0);
+ assert!(predicate.upgrade().is_none());
+ assert!(early_predicate.upgrade().is_none());
+ assert!(inner.template.dynamic_expressions_produced().is_empty());
+ assert!(outer.template.dynamic_expressions_produced().is_empty());
+ }
+}
+
+#[tokio::test]
+async fn nullable_unselective_ancestor_bypasses_above_inferred_null_checks() {
+ let ctx = SessionContext::new();
+ // Every decoded batch has a null FK, as in nullable fact-table keys. Null
+ // rejection must belong to Spark's inferred filter, not the adaptation sample.
+ let fact = input(
+ vec![
+ Some(0),
+ None,
+ Some(1),
+ None,
+ Some(2),
+ None,
+ Some(3),
+ None,
+ Some(4),
+ None,
+ ],
+ &DataType::Int32,
+ 0,
+ );
+ let checked = Arc::new(CometFilterExec::from_datafusion(
+ FilterExec::try_new(
+ Arc::new(IsNotNullExpr::new(Arc::new(Column::new("key", 0)))),
+ fact,
+ )
+ .unwrap()
+ .with_batch_size(1)
+ .unwrap(),
+ ));
+ let inner = chain_join(
+ input((0..5).map(Some).collect(), &DataType::Int32, 0),
+ checked,
+ 0,
+ 0,
+ );
+ let inner = Arc::new(
+ DynamicFilterJoinExec::try_new(&inner, ctx.copied_config().options())
+ .unwrap()
+ .unwrap(),
+ );
+ let outer = chain_join(
+ input((0..5).map(Some).collect(), &DataType::Int32, 0),
+ inner,
+ 0,
+ 2,
+ );
+ let expected = collect(
+ Arc::new(outer.builder().reset_state().build().unwrap()),
+ ctx.task_ctx(),
+ )
+ .await
+ .unwrap();
+ let outer = Arc::new(
+ DynamicFilterJoinExec::try_new(&outer, ctx.copied_config().options())
+ .unwrap()
+ .unwrap(),
+ ) as Arc<dyn ExecutionPlan>;
+ let actual = collect(Arc::clone(&outer), ctx.task_ctx()).await.unwrap();
+ assert_eq!(
+ batches_to_sort_string(&actual),
+ batches_to_sort_string(&expected)
+ );
+ assert_eq!(row_count(&actual), 5);
+ assert_eq!(metric(&outer, "dynamic_filter_early_rows_evaluated"), 2);
+ assert_eq!(metric(&outer, "dynamic_filter_early_rows_pruned"), 0);
+ assert_eq!(metric(&outer, "dynamic_filter_early_rows_bypassed"), 3);
+}
+
+#[test]
+fn early_filter_stops_at_intermediate_join_fetch() {
+ use super::super::early::place_early_filter;
+ let config = ConfigOptions::default();
+ let limited = plain_join().with_fetch(Some(1)).unwrap();
+ let wrapped = Arc::new(
+ DynamicFilterJoinExec::try_new(limited.downcast_ref::<HashJoinExec>().unwrap(), &config)
+ .unwrap()
+ .unwrap(),
+ ) as Arc<dyn ExecutionPlan>;
+ assert_eq!(wrapped.fetch(), Some(1));
+ let predicate = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::new(Column::new("key", 2))],
+ lit(true),
+ ));
+ assert!(Arc::ptr_eq(
+ &wrapped,
+ &place_early_filter(&wrapped, predicate, &ExecutionPlanMetricsSet::new()).unwrap()
+ ));
+}
diff --git a/native/core/src/execution/operators/dynamic_filter/mod.rs b/native/core/src/execution/operators/dynamic_filter/mod.rs
index 3dc33ce15d..ab6d329c66 100644
--- a/native/core/src/execution/operators/dynamic_filter/mod.rs
+++ b/native/core/src/execution/operators/dynamic_filter/mod.rs
@@ -17,6 +17,7 @@
//! Runtime-filter wiring and shared filtering of decoded batches.
+mod early;
mod join;
mod parquet_reader;
mod topk;
@@ -45,12 +46,13 @@ use datafusion::physical_plan::{
use futures::StreamExt;
/// A task-local consumer of a live runtime predicate.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub(crate) struct DynamicFilterExec {
input: Arc<dyn ExecutionPlan>,
predicate: Arc<DynamicFilterPhysicalExpr>,
metrics: ExecutionPlanMetricsSet,
metric_prefix: &'static str,
+ adaptive: bool,
}
impl DynamicFilterExec {
@@ -65,8 +67,21 @@ impl DynamicFilterExec {
predicate,
metrics,
metric_prefix,
+ adaptive: false,
}
}
+
+ fn adaptive(mut self) -> Self {
+ self.adaptive = true;
+ self
+ }
+
+ fn with_execution_input(&self, input: Arc<dyn ExecutionPlan>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(Self {
+ input,
+ ..self.clone()
+ })
+ }
}
impl DisplayAs for DynamicFilterExec {
@@ -122,12 +137,11 @@ impl ExecutionPlan for DynamicFilterExec {
if children.len() != 1 {
return internal_err!("CometDynamicFilterExec requires one child");
}
- Ok(Arc::new(Self::new(
- children.remove(0),
- Arc::clone(&self.predicate),
- ExecutionPlanMetricsSet::new(),
- self.metric_prefix,
- )))
+ Ok(Arc::new(Self {
+ input: children.remove(0),
+ metrics: ExecutionPlanMetricsSet::new(),
+ ..self.as_ref().clone()
+ }))
}
fn reset_state(self: Arc<Self>) -> Result<Arc<dyn ExecutionPlan>> {
@@ -138,12 +152,11 @@ impl ExecutionPlan for DynamicFilterExec {
self.predicate.children().into_iter().cloned().collect(),
lit(true),
));
- Ok(Arc::new(Self::new(
- Arc::clone(&self.input),
+ Ok(Arc::new(Self {
predicate,
- ExecutionPlanMetricsSet::new(),
- self.metric_prefix,
- )))
+ metrics: ExecutionPlanMetricsSet::new(),
+ ..self.as_ref().clone()
+ }))
}
fn execute(
@@ -172,8 +185,23 @@ 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.
+ // Exact zero avoids a workload-specific break-even ratio: even a small
+ // reduction may save substantial work in an intermediate join. Permanent
+ // bypass bounds duplicate evaluation; clustered inputs may lose later
+ // pruning, but the final consumer preserves correctness. Resampling and
+ // ratio thresholds would need workload evidence before adding more policy.
+ // Keep this decision per stream; an inactive TRUE placeholder is not a sample.
+ let adaptive = self.adaptive;
+ let mut unselective_batches = 0;
let stream = input.map(move |batch| {
let batch = batch?;
+ if adaptive && unselective_batches >= 2 {
+ bypassed.add(batch.num_rows());
+ return Ok(batch);
+ }
let _timer = eval_time.timer();
// AND may prefilter its input before evaluating hash membership. A
// zero-copy key projection keeps payload columns out of that temporary
@@ -187,12 +215,22 @@ impl ExecutionPlan for DynamicFilterExec {
Ok(batch)
}
ColumnarValue::Scalar(ScalarValue::Boolean(Some(false) | None)) => {
+ if adaptive {
+ unselective_batches = 0;
+ }
evaluated.add(batch.num_rows());
pruned.add(batch.num_rows());
Ok(batch.slice(0, 0))
}
ColumnarValue::Array(mask) => {
let filtered = filter_record_batch(&batch, as_boolean_array(&mask)?)?;
+ if adaptive && batch.num_rows() > 0 {
+ unselective_batches = if filtered.num_rows() == batch.num_rows() {
+ unselective_batches + 1
+ } else {
+ 0
+ };
+ }
evaluated.add(batch.num_rows());
pruned.add(batch.num_rows() - filtered.num_rows());
Ok(filtered)
diff --git a/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs b/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs
index f40fbfc6cb..b5dd28d3f3 100644
--- a/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs
+++ b/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs
@@ -62,6 +62,14 @@ pub(super) fn try_attach_parquet_reader_filter(
log::debug!("Join dynamic filter reader pushdown skipped: probe has a fetch limit");
return Ok(None);
}
+ // An ancestor join may already filter decoded batches below this join. Keep
+ // that consumer while attaching this join's own predicate to its reader.
+ if let Some(filter) = input.downcast_ref::<super::DynamicFilterExec>() {
+ return Ok(
+ try_attach_parquet_reader_filter(&filter.input, predicate, config)?
+ .map(|reader| filter.with_execution_input(reader)),
+ );
+ }
// Spark inserts IS NOT NULL residuals above equijoin inputs, including AND
// chains of inferred null checks. A reader predicate can cross those direct
// checks because both operations only discard rows. Keep every other filter
diff --git a/native/core/src/execution/operators/dynamic_filter/tests.rs b/native/core/src/execution/operators/dynamic_filter/tests.rs
index e610f8b4a6..76e2fb963c 100644
--- a/native/core/src/execution/operators/dynamic_filter/tests.rs
+++ b/native/core/src/execution/operators/dynamic_filter/tests.rs
@@ -123,3 +123,123 @@ async fn placeholder_updates_and_errors_are_not_hidden() {
let error = collect(wrapper, task).await.unwrap_err();
assert!(error.to_string().contains("must evaluate to a Boolean"));
}
+
+#[tokio::test]
+async fn early_filter_bypasses_only_after_nonempty_evaluated_batches_and_resets_per_stream() {
+ use datafusion::physical_expr::expressions::IsNotNullExpr;
+ let ctx = SessionContext::new();
+ let plan = input((0..10).map(Some).collect(), &DataType::Int32, 0);
+ let key = Arc::new(Column::new("key", 0)) as Arc<dyn PhysicalExpr>;
+ let predicate = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::clone(&key)],
+ lit(true),
+ ));
+ let metrics = ExecutionPlanMetricsSet::new();
+ let filter = Arc::new(
+ DynamicFilterExec::new(
+ plan,
+ Arc::clone(&predicate),
+ metrics,
+ "dynamic_filter_early",
+ )
+ .adaptive(),
+ ) as Arc<dyn ExecutionPlan>;
+ // An unpopulated producer does not use up the adaptation samples.
+ let mut stream = filter.execute(0, ctx.task_ctx()).unwrap();
+ assert_eq!(stream.next().await.unwrap().unwrap().num_rows(), 2);
+ assert_eq!(stream.next().await.unwrap().unwrap().num_rows(), 2);
+ predicate
+ .update(Arc::new(IsNotNullExpr::new(Arc::clone(&key))))
+ .unwrap();
+ while let Some(batch) = stream.next().await {
+ assert_eq!(batch.unwrap().num_rows(), 2);
+ }
+ assert_eq!(metric(&filter, "dynamic_filter_early_rows_evaluated"), 4);
+ assert_eq!(metric(&filter, "dynamic_filter_early_rows_bypassed"), 6);
+ // Adaptation belongs to a stream, and the downstream join remains responsible
+ // for matching rows if a later producer update would become more selective.
+ predicate
+ .update(Arc::new(BinaryExpr::new(key, Operator::Eq, lit(7i32))))
+ .unwrap();
+ let rows = collect(Arc::clone(&filter), ctx.task_ctx()).await.unwrap();
+ assert_eq!(rows.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ assert_eq!(metric(&filter, "dynamic_filter_early_rows_evaluated"), 14);
+ assert_eq!(metric(&filter, "dynamic_filter_early_rows_pruned"), 9);
+}
+
+#[tokio::test]
+async fn empty_batches_do_not_disable_early_filtering() {
+ use datafusion::physical_expr::expressions::IsNotNullExpr;
+ let ctx = SessionContext::new();
+ let schema = Arc::new(Schema::new(vec![Field::new("key", DataType::Int32, true)]));
+ let first = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![Some(1)]))],
+ )
+ .unwrap();
+ let last = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![None::<i32>]))],
+ )
+ .unwrap();
+ let plan = memory_exec(vec![first, RecordBatch::new_empty(schema), last]);
+ let key = Arc::new(Column::new("key", 0)) as Arc<dyn PhysicalExpr>;
+ let predicate = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::clone(&key)],
+ Arc::new(IsNotNullExpr::new(key)),
+ ));
+ let filter = Arc::new(
+ DynamicFilterExec::new(
+ plan,
+ predicate,
+ ExecutionPlanMetricsSet::new(),
+ "dynamic_filter_early",
+ )
+ .adaptive(),
+ ) as Arc<dyn ExecutionPlan>;
+ let output = collect(Arc::clone(&filter), ctx.task_ctx()).await.unwrap();
+ assert_eq!(output.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ assert_eq!(metric(&filter, "dynamic_filter_early_rows_pruned"), 1);
+}
+
+#[test]
+fn adaptive_mode_survives_rebuilding() {
+ let source = input(vec![Some(1)], &DataType::Int32, 0);
+ let predicate = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::new(Column::new("key", 0))],
+ lit(true),
+ ));
+ let original = Arc::new(
+ DynamicFilterExec::new(
+ Arc::clone(&source),
+ Arc::clone(&predicate),
+ ExecutionPlanMetricsSet::new(),
+ "dynamic_filter_early",
+ )
+ .adaptive(),
+ );
+ let execution = original.with_execution_input(Arc::clone(&source));
+ let replaced = Arc::clone(&original)
+ .replace_children(
+ vec![source],
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )
+ .unwrap();
+ let reset = original.reset_state().unwrap();
+ for plan in [&execution, &replaced, &reset] {
+ let filter = plan.downcast_ref::<DynamicFilterExec>().unwrap();
+ assert!(filter.adaptive);
+ assert_eq!(filter.metric_prefix, "dynamic_filter_early");
+ }
+ assert!(Arc::ptr_eq(
+ &execution
+ .downcast_ref::<DynamicFilterExec>()
+ .unwrap()
+ .predicate,
+ &predicate
+ ));
+ assert!(!Arc::ptr_eq(
+ &reset.downcast_ref::<DynamicFilterExec>().unwrap().predicate,
+ &predicate
+ ));
+}
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 96802ca805..cc9157f846 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -84,8 +84,8 @@ use datafusion::{
prelude::SessionContext,
};
use datafusion_comet_operators::{
- range_exec, CometFilterExec, ExpandExec, ExplodeExec, PartitionedRankLimitExec, SampleExec,
- WindowFnKind,
+ range_exec, CometFilterExec, CometProjectionExec, ExpandExec, ExplodeExec,
+ PartitionedRankLimitExec, SampleExec, WindowFnKind,
};
use datafusion_comet_spark_expr::{
create_comet_physical_fun, create_comet_physical_fun_with_eval_mode, BinaryOutputStyle,
@@ -2616,15 +2616,15 @@ impl PhysicalPlanner {
partition_count,
)?;
- // Reader attachment replaces the probe filter's child per execution.
- // Keep its metrics owned by the same Spark filter node.
+ // Reader attachment and early placement recursively replace probe-side
+ // filters and projections. Keep metrics owned by their Spark nodes.
if join.dynamic_filter_enabled && !join.null_aware_anti_join {
let probe = if join.build_side == BuildSide::BuildLeft as i32 {
&mut join_params.right
} else {
&mut join_params.left
};
- *probe = Self::prepare_probe_filter_for_runtime_reader(Arc::clone(probe));
+ *probe = Self::prepare_probe_for_runtime_filters(Arc::clone(probe))?;
}
let left = Arc::clone(&join_params.left.native_plan);
@@ -2909,14 +2909,32 @@ impl PhysicalPlanner {
}
}
- /// Keep the Spark filter's metric identity when its reader is replaced for an execution.
- fn prepare_probe_filter_for_runtime_reader(plan: Arc<SparkPlan>) -> Arc<SparkPlan> {
- let Some(filter) = plan.native_plan.downcast_ref::<FilterExec>() else {
- return plan;
- };
+ /// Keep Spark metric identities for the small set of nodes whose children
+ /// runtime-filter placement can replace. Stop at native/Spark tree boundaries.
+ fn prepare_probe_for_runtime_filters(
+ plan: Arc<SparkPlan>,
+ ) -> Result<Arc<SparkPlan>, ExecutionError> {
+ let mut native: Arc<dyn ExecutionPlan> =
+ if let Some(filter) = plan.native_plan.downcast_ref::<FilterExec>() {
+ Arc::new(CometFilterExec::from_datafusion(filter.clone()))
+ } else if let Some(projection) = plan.native_plan.downcast_ref::<ProjectionExec>() {
+ Arc::new(CometProjectionExec::from_datafusion(projection.clone()))
+ } else {
+ return Ok(plan);
+ };
let mut prepared = plan.as_ref().clone();
- prepared.native_plan = Arc::new(CometFilterExec::from_datafusion(filter.clone()));
- Arc::new(prepared)
+ if let ([child], [only]) = (plan.children.as_slice(), native.children().as_slice()) {
+ if Arc::ptr_eq(only, &child.native_plan) {
+ let child = Self::prepare_probe_for_runtime_filters(Arc::clone(child))?;
+ native = native.replace_children(
+ vec![Arc::clone(&child.native_plan)],
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )?;
+ prepared.children = vec![child];
+ }
+ }
+ prepared.native_plan = native;
+ Ok(Arc::new(prepared))
}
/// Attach after choosing the final build side, including the projection emitted
diff --git a/native/operators/src/lib.rs b/native/operators/src/lib.rs
index cfc92a0eb6..b5f762025a 100644
--- a/native/operators/src/lib.rs
+++ b/native/operators/src/lib.rs
@@ -28,6 +28,7 @@
mod expand;
mod explode;
mod filter;
+mod projection;
mod range;
mod rank_limit;
mod sample;
@@ -35,6 +36,7 @@ mod sample;
pub use expand::ExpandExec;
pub use explode::ExplodeExec;
pub use filter::CometFilterExec;
+pub use projection::CometProjectionExec;
pub use range::range_exec;
pub use rank_limit::{PartitionedRankLimitExec, WindowFnKind};
pub use sample::SampleExec;
diff --git a/native/operators/src/projection.rs b/native/operators/src/projection.rs
new file mode 100644
index 0000000000..b35afcae31
--- /dev/null
+++ b/native/operators/src/projection.rs
@@ -0,0 +1,295 @@
+// 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.
+
+//! A DataFusion projection whose metrics remain owned by its Spark plan node.
+//!
+//! Comet normally keeps a one-to-one Spark/native plan tree so native metric
+//! handles map back to the corresponding Spark operator. Some execution-local
+//! rewrites need to replace a projection's child. DataFusion gives that replacement
+//! a new private metric set, so this adapter owns the stable metric set and
+//! registers the handles from the projection that actually executes.
+
+use std::fmt::Formatter;
+use std::sync::Arc;
+
+use datafusion::common::tree_node::TreeNodeRecursion;
+use datafusion::common::{internal_err, Result, Statistics};
+use datafusion::execution::TaskContext;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::execution_plan::{
+ CardinalityEffect, ChildrenPropertiesMode, ReplaceChildrenOptions,
+};
+use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
+use datafusion::physical_plan::projection::ProjectionExec;
+use datafusion::physical_plan::statistics::{ChildStats, StatisticsArgs};
+use datafusion::physical_plan::{
+ DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, SendableRecordBatchStream,
+};
+
+#[derive(Debug)]
+pub struct CometProjectionExec {
+ projection: ProjectionExec,
+ metrics: ExecutionPlanMetricsSet,
+}
+
+impl CometProjectionExec {
+ pub fn from_datafusion(projection: ProjectionExec) -> Self {
+ Self {
+ projection,
+ metrics: ExecutionPlanMetricsSet::new(),
+ }
+ }
+
+ pub fn input(&self) -> &Arc<dyn ExecutionPlan> {
+ self.projection.input()
+ }
+
+ pub fn projection(&self) -> &ProjectionExec {
+ &self.projection
+ }
+
+ fn replace_input(
+ &self,
+ input: Arc<dyn ExecutionPlan>,
+ options: ReplaceChildrenOptions,
+ ) -> Result<ProjectionExec> {
+ let replaced = Arc::new(self.projection.clone()).replace_children(vec![input], options)?;
+ let Some(projection) = replaced.downcast_ref::<ProjectionExec>() else {
+ return internal_err!("ProjectionExec child replacement changed its plan type");
+ };
+ Ok(projection.clone())
+ }
+
+ /// Replace the child for one execution while keeping the metric identity
+ /// owned by the permanent Spark projection node.
+ pub fn with_execution_input(
+ &self,
+ input: Arc<dyn ExecutionPlan>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ Ok(Arc::new(Self {
+ projection: self.replace_input(
+ input,
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )?,
+ metrics: self.metrics.clone(),
+ }))
+ }
+}
+
+impl DisplayAs for CometProjectionExec {
+ fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
+ self.projection.fmt_as(t, f)
+ }
+}
+
+impl ExecutionPlan for CometProjectionExec {
+ fn name(&self) -> &str {
+ self.projection.name()
+ }
+
+ fn properties(&self) -> &Arc<PlanProperties> {
+ self.projection.properties()
+ }
+
+ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+ vec![self.projection.input()]
+ }
+
+ fn apply_expressions(
+ &self,
+ f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+ ) -> Result<TreeNodeRecursion> {
+ self.projection.apply_expressions(f)
+ }
+
+ fn benefits_from_input_partitioning(&self) -> Vec<bool> {
+ self.projection.benefits_from_input_partitioning()
+ }
+
+ fn maintains_input_order(&self) -> Vec<bool> {
+ self.projection.maintains_input_order()
+ }
+
+ fn replace_children(
+ self: Arc<Self>,
+ mut children: Vec<Arc<dyn ExecutionPlan>>,
+ options: ReplaceChildrenOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ if children.len() != 1 {
+ return internal_err!("CometProjectionExec requires one child");
+ }
+ Ok(Arc::new(Self {
+ projection: self.replace_input(children.remove(0), options)?,
+ metrics: ExecutionPlanMetricsSet::new(),
+ }))
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ self.replace_children(
+ children,
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )
+ }
+
+ fn execute(
+ &self,
+ partition: usize,
+ context: Arc<TaskContext>,
+ ) -> Result<SendableRecordBatchStream> {
+ // Every execution gets fresh DataFusion projection metrics. Register their
+ // handles on the stable set before returning the stream, including when
+ // opening the child fails.
+ let projection = self.replace_input(
+ Arc::clone(self.projection.input()),
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )?;
+ let result = projection.execute(partition, context);
+ for metric in projection.metrics().unwrap_or_default().iter() {
+ self.metrics.register(Arc::clone(metric));
+ }
+ result
+ }
+
+ fn metrics(&self) -> Option<MetricsSet> {
+ Some(self.metrics.clone_inner())
+ }
+
+ fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+ self.projection.child_stats_requests(partition)
+ }
+
+ fn statistics_from_inputs(
+ &self,
+ input_stats: &[Arc<Statistics>],
+ args: &StatisticsArgs,
+ ) -> Result<Arc<Statistics>> {
+ self.projection.statistics_from_inputs(input_stats, args)
+ }
+
+ fn cardinality_effect(&self) -> CardinalityEffect {
+ self.projection.cardinality_effect()
+ }
+
+ fn fetch(&self) -> Option<usize> {
+ self.projection.fetch()
+ }
+
+ fn with_fetch(&self, fetch: Option<usize>) -> Option<Arc<dyn ExecutionPlan>> {
+ let projection = self.projection.with_fetch(fetch)?;
+ let projection = projection.downcast_ref::<ProjectionExec>()?.clone();
+ Some(Arc::new(Self {
+ projection,
+ metrics: ExecutionPlanMetricsSet::new(),
+ }))
+ }
+
+ fn with_preserve_order(&self, preserve_order: bool) -> Option<Arc<dyn ExecutionPlan>> {
+ let projection = self.projection.with_preserve_order(preserve_order)?;
+ let projection = projection.downcast_ref::<ProjectionExec>()?.clone();
+ Some(Arc::new(Self {
+ projection,
+ metrics: ExecutionPlanMetricsSet::new(),
+ }))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use arrow::array::Int32Array;
+ use arrow::datatypes::{DataType, Field, Schema};
+ use arrow::record_batch::RecordBatch;
+ use datafusion::common::stats::Precision;
+ use datafusion::datasource::memory::MemorySourceConfig;
+ use datafusion::logical_expr::Operator;
+ use datafusion::physical_expr::expressions::{binary, col, lit};
+ use datafusion::physical_plan::statistics::StatisticsContext;
+
+ fn projection() -> Result<ProjectionExec> {
+ let schema = Arc::new(Schema::new(vec![Field::new("key", DataType::Int32, false)]));
+ let partitions = [1..11, 11..41]
+ .into_iter()
+ .map(|values| {
+ RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from_iter_values(values))],
+ )
+ .map(|batch| vec![batch])
+ })
+ .collect::<std::result::Result<Vec<_>, _>>()?;
+ let input = MemorySourceConfig::try_new_exec(&partitions, Arc::clone(&schema), None)?;
+ ProjectionExec::try_new(
+ vec![
+ (
+ binary(col("key", &schema)?, Operator::Plus, lit(5i32), &schema)?,
+ "offset".into(),
+ ),
+ (col("key", &schema)?, "key".into()),
+ ],
+ input,
+ )
+ }
+
+ #[test]
+ fn expression_visitor_preserves_roots_and_recursion() -> Result<()> {
+ let wrapper = CometProjectionExec::from_datafusion(projection()?);
+ assert_eq!(wrapper.projection().expr()[0].expr.children().len(), 2);
+ for recursion in [TreeNodeRecursion::Continue, TreeNodeRecursion::Stop] {
+ let mut visits = 0;
+ let result = wrapper.apply_expressions(&mut |expression| {
+ assert!(Arc::ptr_eq(
+ expression,
+ &wrapper.projection().expr()[visits].expr
+ ));
+ visits += 1;
+ Ok(recursion)
+ })?;
+ assert_eq!(
+ visits,
+ if recursion == TreeNodeRecursion::Stop {
+ 1
+ } else {
+ 2
+ }
+ );
+ assert_eq!(result, recursion);
+ }
+ Ok(())
+ }
+
+ #[test]
+ fn statistics_match_datafusion_for_each_partition() -> Result<()> {
+ let projection = projection()?;
+ let wrapper = CometProjectionExec::from_datafusion(projection.clone());
+ let context = StatisticsContext::new();
+ let mut row_counts = Vec::new();
+ for partition in [None, Some(0), Some(1)] {
+ let args = StatisticsArgs::new().with_partition(partition);
+ let expected = context.compute(&projection, &args)?;
+ let actual = context.compute(&wrapper, &args)?;
+ assert_eq!(actual, expected);
+ assert_ne!(actual.num_rows, Precision::Absent);
+ row_counts.push(actual.num_rows);
+ }
+ assert_ne!(row_counts[0], row_counts[1]);
+ assert_ne!(row_counts[1], row_counts[2]);
+ Ok(())
+ }
+}
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala
index 7d86de43c4..e038eb78d8 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala
@@ -526,7 +526,15 @@ object CometMetricNode {
"dynamic_filter_join_filters_attached" ->
SQLMetrics.createMetric(sc, "Join runtime filters attached to native readers"),
"dynamic_filter_join_filters_skipped" ->
- SQLMetrics.createMetric(sc, "Join runtime filters not eligible for native readers"))
+ SQLMetrics.createMetric(sc, "Join runtime filters not eligible for native readers"),
+ "dynamic_filter_early_rows_evaluated" ->
+ SQLMetrics.createMetric(sc, "Rows evaluated before an intermediate join"),
+ "dynamic_filter_early_rows_pruned" ->
+ SQLMetrics.createMetric(sc, "Rows rejected before an intermediate join"),
+ "dynamic_filter_early_rows_bypassed" ->
+ SQLMetrics.createMetric(sc, "Rows bypassing inactive or unselective early filtering"),
+ "dynamic_filter_early_eval_time" ->
+ SQLMetrics.createNanoTimingMetric(sc, "Time evaluating the early join runtime filter"))
}
/**
diff --git a/spark/src/test/resources/sql-tests/join/dynamic_filter_chain.sql b/spark/src/test/resources/sql-tests/join/dynamic_filter_chain.sql
new file mode 100644
index 0000000000..5237e225bf
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/join/dynamic_filter_chain.sql
@@ -0,0 +1,72 @@
+-- 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.
+
+-- Compare complete join results with Spark with runtime filtering enabled and disabled.
+-- Keep the join order fixed so both broadcast build sides and stacked ancestor consumers run.
+-- Config: spark.sql.adaptive.enabled=false
+-- Config: spark.sql.autoBroadcastJoinThreshold=-1
+-- Config: spark.sql.cbo.joinReorder.enabled=false
+-- Config: spark.comet.batchSize=16
+-- ConfigMatrix: spark.comet.exec.join.dynamicFilter.enabled=false,true
+
+statement
+CREATE TABLE chain_fact(k INT, payload INT) USING parquet
+
+statement
+INSERT INTO chain_fact VALUES (1, 10), (1, 11), (2, 20), (3, 30), (4, 40), (NULL, 50)
+
+statement
+CREATE TABLE chain_dimension(k INT, detail INT) USING parquet
+
+statement
+INSERT INTO chain_dimension VALUES (1, 100), (1, 101), (2, 200), (3, 300), (NULL, 500)
+
+statement
+CREATE TABLE chain_selection(k INT, selected INT) USING parquet
+
+statement
+INSERT INTO chain_selection VALUES (1, 1000), (1, 1001), (2, 2000), (NULL, 5000)
+
+statement
+CREATE TABLE chain_final(k INT, selected INT) USING parquet
+
+statement
+INSERT INTO chain_final VALUES (1, 10000), (1, 10001), (3, 30000), (NULL, 50000)
+
+-- Duplicate keys multiply matches, while NULL keys never match each other.
+query
+SELECT /*+ BROADCAST(d), BROADCAST(s) */ f.*, d.detail, s.selected
+FROM chain_fact f JOIN chain_dimension d ON f.k = d.k
+JOIN chain_selection s ON f.k = s.k
+
+query
+SELECT /*+ BROADCAST(d), BROADCAST(s) */ f.*, d.detail, s.selected
+FROM chain_dimension d JOIN chain_fact f ON f.k = d.k
+JOIN chain_selection s ON f.k = s.k
+
+-- Two ancestors install early consumers below the innermost join.
+query
+SELECT /*+ BROADCAST(d), BROADCAST(s), BROADCAST(t) */ f.*, d.detail, s.selected, t.selected
+FROM chain_fact f JOIN chain_dimension d ON f.k = d.k
+JOIN chain_selection s ON f.k = s.k
+JOIN chain_final t ON f.k = t.k
+
+query
+SELECT /*+ BROADCAST(d), BROADCAST(s), BROADCAST(t) */ f.*, d.detail, s.selected, t.selected
+FROM chain_dimension d JOIN chain_fact f ON f.k = d.k
+JOIN chain_selection s ON f.k = s.k
+JOIN chain_final t ON f.k = t.k
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
index 369c251763..36e7c5c4f4 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
@@ -33,7 +33,7 @@ import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, Dynam
import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.logical.Join
-import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometFilterExec, CometHashJoinExec, CometNativeScanExec, CometSortExec, CometSortMergeJoinExec, CometUnionExec, CometWindowExec}
+import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometFilterExec, CometHashJoinExec, CometNativeScanExec, CometProjectExec, CometSortExec, CometSortMergeJoinExec, CometUnionExec, CometWindowExec}
import org.apache.spark.sql.execution.{ColumnarToRowTransition, InputAdapter, LocalTableScanExec, SortExec, SparkPlan, WholeStageCodegenExec}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec, QueryStageExec}
import org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException
@@ -435,6 +435,108 @@ class CometJoinSuite extends CometTestBase {
}
}
+ test("join dynamic filter rejects rows before an intermediate join") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.JOIN_REORDER_ENABLED.key -> "false",
+ SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1",
+ CometConf.COMET_BATCH_SIZE.key -> "128") {
+ withParquetTable((0 until 1000).map(i => (i, i.toLong)), "early_fact") {
+ withParquetTable((0 until 2000).map(i => (i % 1000, i)), "early_dimension") {
+ withParquetTable(Seq((42, 1), (42, 2)), "early_selection") {
+ for (buildLeft <- Seq(false, true); enabled <- Seq(false, true)) {
+ val from = if (buildLeft) {
+ "early_dimension d JOIN early_fact f"
+ } else {
+ "early_fact f JOIN early_dimension d"
+ }
+ val query = "SELECT /*+ BROADCAST(d), BROADCAST(s) */ " +
+ "f._1 AS selected_key, f._2 AS payload, d._2 AS detail, s._2 AS selection " +
+ s"FROM $from ON f._1 = d._1 " +
+ "JOIN early_selection s ON f._1 = s._1"
+ withSQLConf(
+ CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.key -> enabled.toString) {
+ val (_, plan) = checkSparkAnswerAndOperator(
+ sql(query),
+ Seq(classOf[CometBroadcastHashJoinExec]))
+ val joins = collect(plan) { case join: CometBroadcastHashJoinExec => join }
+ assert(joins.size == 2, s"Expected two native broadcast joins:\n$plan")
+ val outputs = joins.map(_.metrics("output_rows").value).sorted
+ assert(outputs == (if (enabled) Seq(2L, 4L) else Seq(4L, 2000L)))
+ if (enabled) {
+ assert(
+ joins.map(_.metrics("dynamic_filter_early_rows_pruned").value).sum == 999L)
+ assert(
+ joins
+ .map(_.metrics("dynamic_filter_early_rows_evaluated").value)
+ .sum == 1000L)
+ // The ancestor consumer must not hide the scan from the intermediate join's
+ // own reader filter. collect visits the outer join before the intermediate one.
+ assert(joins.last.metrics("dynamic_filter_join_filters_attached").value > 0L)
+ val projections = collect(plan) { case project: CometProjectExec => project }
+ assert(projections.nonEmpty)
+ assert(projections.forall(_.metrics("output_rows").value > 0L))
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("join dynamic filters stack across three joins and preserve reader attachment") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.JOIN_REORDER_ENABLED.key -> "false",
+ SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1",
+ CometConf.COMET_BATCH_SIZE.key -> "128") {
+ withParquetTable((0 until 1000).map(i => (i, i.toLong)), "stacked_fact") {
+ withParquetTable((0 until 2000).map(i => (i % 1000, i)), "stacked_dimension") {
+ withParquetTable(Seq((42, 1), (43, 2)), "stacked_selection") {
+ withParquetTable(Seq((42, 1), (42, 2)), "stacked_final") {
+ for (buildLeft <- Seq(false, true); enabled <- Seq(false, true)) {
+ val from = if (buildLeft) {
+ "stacked_dimension d JOIN stacked_fact f"
+ } else {
+ "stacked_fact f JOIN stacked_dimension d"
+ }
+ val query = "SELECT /*+ BROADCAST(d), BROADCAST(s), BROADCAST(t) */ " +
+ "f._1 AS selected_key, f._2 AS payload, d._2 AS detail, " +
+ "s._2 AS selection, t._2 AS final_selection " +
+ s"FROM $from ON f._1 = d._1 " +
+ "JOIN stacked_selection s ON f._1 = s._1 " +
+ "JOIN stacked_final t ON f._1 = t._1"
+ withSQLConf(
+ CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.key -> enabled.toString) {
+ val (_, plan) = checkSparkAnswerAndOperator(
+ sql(query),
+ Seq(classOf[CometBroadcastHashJoinExec]))
+ val joins = collect(plan) { case join: CometBroadcastHashJoinExec => join }
+ assert(joins.size == 3, s"Expected three native broadcast joins:\n$plan")
+ val outputs = joins.map(_.metrics("output_rows").value).sorted
+ assert(outputs == (if (enabled) Seq(2L, 2L, 4L) else Seq(4L, 4L, 2000L)))
+ if (enabled) {
+ // Each ancestor installs a consumer beneath the first join, so both must
+ // evaluate rows even when one has already filtered most of the batch.
+ assert(
+ joins.count(_.metrics("dynamic_filter_early_rows_evaluated").value > 0L)
+ == 2)
+ assert(
+ joins.map(_.metrics("dynamic_filter_early_rows_pruned").value).sum == 999L)
+ assert(joins.last.metrics("dynamic_filter_join_filters_attached").value > 0L)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
test("join dynamic filter prunes a projected Parquet reader through null-check conjunctions") {
withTempPath { probePath =>
withSQLConf(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org