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/19 19:22:58 UTC
(datafusion-comet) 01/01: feat: opt in to join filter pushdown through deterministic residuals (#6431)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6431-3355602e268819750884aa0ce0dd47002c6ea580
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit b8e11cb361553cd7d5b5d598b098fa966c677f56
Author: Chao Sun <su...@apache.org>
AuthorDate: Mon Oct 19 18:07:47 2026 +0000
feat: opt in to join filter pushdown through deterministic residuals (#6431)
---
docs/source/user-guide/latest/tuning/operators.md | 31 ++++
.../execution/operators/dynamic_filter/early.rs | 2 +
.../operators/dynamic_filter/join/tests.rs | 46 +++++
.../operators/dynamic_filter/parquet_reader.rs | 12 +-
native/core/src/execution/planner.rs | 1 +
native/core/src/execution/planner/join.rs | 187 ++++++++++++++++++++-
native/operators/src/filter.rs | 38 +++++
native/proto/src/proto/operator.proto | 2 +
.../main/scala/org/apache/comet/CometConf.scala | 15 ++
.../org/apache/spark/sql/comet/operators.scala | 4 +
.../org/apache/comet/exec/CometJoinSuite.scala | 83 ++++++++-
11 files changed, 406 insertions(+), 15 deletions(-)
diff --git a/docs/source/user-guide/latest/tuning/operators.md b/docs/source/user-guide/latest/tuning/operators.md
index 94bb76d5ad..c44e408b78 100644
--- a/docs/source/user-guide/latest/tuning/operators.md
+++ b/docs/source/user-guide/latest/tuning/operators.md
@@ -92,6 +92,37 @@ rows, including if later batches become more selective. The `dynamic_filter_earl
metrics belong to the join supplying the filter; intermediate joins and projections
retain their own input/output metrics.
+#### Join runtime filters through residual predicates
+
+Native join runtime filtering is enabled with
+`spark.comet.exec.join.dynamicFilter.enabled`. It can reject probe rows whose keys
+are absent from the completed build side and use eligible Parquet readers to prune
+row groups. By default, an intervening residual filter permits reader propagation
+only for direct-column `IS NOT NULL` checks combined with `AND`.
+
+Set `spark.comet.exec.join.dynamicFilter.allowDeterministicFilterPushdown=true` to
+let the join filter pass through other deterministic residual predicates, such as
+`quantity > 0` or a Spark-generated Bloom check. The original residual and the
+exact join remain in place. For example, when the build contains only item IDs
+17 and 42, a row for item 99 can be rejected before evaluating its quantity.
+Nondeterministic predicates, limits, all standalone projections (including column-only
+projections), and execution boundaries continue to stop reader propagation. Spark can
+insert a standalone projection to drop residual-only columns, so the example requires
+keeping `quantity` in the output alongside `item_id`. Projection within the Parquet
+scan itself remains supported. Parquet schema conversion checks remain active.
+Reader attachment through column-only projections is tracked in
+[#7113](https://github.com/apache/datafusion-comet/issues/7113).
+
+The separate setting lets applications keep the existing runtime-filter behavior
+while choosing whether to allow broader reader pruning. Spark can already skip
+residual evaluation through other pruning paths, but those paths are not present in
+every plan. This setting defaults to false because deterministic expressions can
+throw. Earlier pruning can skip division-by-zero or ANSI cast errors on rows that
+cannot join. Errors on retained rows still propagate. Enable it only when skipping
+such errors on eliminated rows is acceptable. Parquet row filtering during decoding
+also requires `spark.comet.parquet.rowFilterPushdown.enabled`; row-group statistics
+pruning does not require that setting.
+
## 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
index 6f23c54d3c..9b101044bd 100644
--- a/native/core/src/execution/operators/dynamic_filter/early.rs
+++ b/native/core/src/execution/operators/dynamic_filter/early.rs
@@ -101,6 +101,8 @@ fn place(
}
}
} else if let Some(filter) = input.downcast_ref::<CometFilterExec>() {
+ // Runtime-filter pushdown permission controls reader pruning only;
+ // it must not relax the expression barriers for early consumers.
// 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
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 8086ac647e..54ae6737ed 100644
--- a/native/core/src/execution/operators/dynamic_filter/join/tests.rs
+++ b/native/core/src/execution/operators/dynamic_filter/join/tests.rs
@@ -686,6 +686,52 @@ fn filtered_probe(scan: &Arc<DataSourceExec>) -> Arc<CometFilterExec> {
))
}
+#[tokio::test]
+async fn deterministic_residual_opt_in_skips_errors_only_on_pruned_rows() {
+ for (allowed, build_key) in [(false, 3), (true, 3), (true, 0)] {
+ let session = Arc::new(SessionContext::new_with_config(
+ SessionConfig::new().with_target_partitions(1),
+ ));
+ let (_file, scan) = parquet_probe((0..4).collect(), &session, 1);
+ let predicate = Arc::new(BinaryExpr::new(
+ Arc::new(BinaryExpr::new(
+ lit(10_i32),
+ Operator::Divide,
+ Arc::new(Column::new("key", 0)),
+ )),
+ Operator::Gt,
+ lit(0_i32),
+ ));
+ let filter = Arc::new(
+ CometFilterExec::from_datafusion(FilterExec::try_new(predicate, scan).unwrap())
+ .with_runtime_filter_pushdown(allowed),
+ );
+ let build = memory_exec(vec![RecordBatch::try_new(
+ Arc::new(Schema::new(vec![Field::new("key", DataType::Int32, false)])),
+ vec![Arc::new(Int32Array::from(vec![build_key]))],
+ )
+ .unwrap()]);
+ let join = single_key_join_plans(build, filter, PartitionMode::Partitioned);
+ let wrapper =
+ DynamicFilterJoinExec::new(&join, session.copied_config().options().as_ref().clone())
+ .unwrap();
+ assert_eq!(
+ wrapper.build_runtime_join().unwrap().reader_filter_attached,
+ allowed
+ );
+ let result = collect(Arc::new(wrapper), session.task_ctx()).await;
+ if allowed && build_key != 0 {
+ assert_eq!(row_count(&result.unwrap()), 1);
+ } else {
+ assert!(result
+ .unwrap_err()
+ .to_string()
+ .to_lowercase()
+ .contains("divide by zero"));
+ }
+ }
+}
+
fn find_dynamic_filter(expr: &Arc<dyn PhysicalExpr>) -> Option<&DynamicFilterPhysicalExpr> {
if let Some(filter) = expr.downcast_ref::<DynamicFilterPhysicalExpr>() {
return Some(filter);
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 b5dd28d3f3..59009f58fa 100644
--- a/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs
+++ b/native/core/src/execution/operators/dynamic_filter/parquet_reader.rs
@@ -72,9 +72,9 @@ pub(super) fn try_attach_parquet_reader_filter(
}
// 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
- // as a boundary: reader pruning would change which rows reach stateful
- // expressions and can suppress expression errors.
+ // checks because both operations only discard rows. Broader deterministic
+ // residuals require explicit permission from Spark: pruning earlier can skip
+ // their errors on rows that cannot join. Stateful predicates remain barriers.
if let Some(filter) = input.downcast_ref::<CometFilterExec>() {
if filter.has_projection() {
log::debug!(
@@ -82,9 +82,11 @@ pub(super) fn try_attach_parquet_reader_filter(
);
return Ok(None);
}
- if !is_direct_column_null_checks(filter.predicate()) {
+ if !filter.allows_runtime_filter_pushdown()
+ && !is_direct_column_null_checks(filter.predicate())
+ {
log::debug!(
- "Join dynamic filter reader pushdown skipped: probe filter is not direct column IS NOT NULL checks"
+ "Join dynamic filter reader pushdown skipped: probe residual has no pushdown permission and is not direct column IS NOT NULL checks"
);
return Ok(None);
}
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index c664cd758e..bdfdf1142e 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -3791,6 +3791,7 @@ mod tests {
children: vec![child_op],
op_struct: Some(OpStruct::Filter(spark_operator::Filter {
predicate: Some(expr),
+ ..Default::default()
})),
}
}
diff --git a/native/core/src/execution/planner/join.rs b/native/core/src/execution/planner/join.rs
index 3659227deb..0b91f27187 100644
--- a/native/core/src/execution/planner/join.rs
+++ b/native/core/src/execution/planner/join.rs
@@ -46,7 +46,7 @@ use datafusion::{
use datafusion_comet_operators::{CometFilterExec, CometProjectionExec};
use datafusion_comet_proto::{
spark_expression::Expr,
- spark_operator::{BuildSide, JoinType, Operator},
+ spark_operator::{operator::OpStruct, BuildSide, JoinType, Operator},
};
use datafusion_datasource::{file_scan_config::FileScanConfig, source::DataSourceExec};
use itertools::Itertools;
@@ -100,12 +100,15 @@ impl OperatorBuilder for HashJoinBuilder {
// 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
+ let (probe, source) = if join.build_side == BuildSide::BuildLeft as i32 {
+ (&mut join_params.right, &children[1])
} else {
- &mut join_params.left
+ (&mut join_params.left, &children[0])
};
- *probe = PhysicalPlanner::prepare_probe_for_runtime_filters(Arc::clone(probe))?;
+ *probe = PhysicalPlanner::prepare_probe_for_runtime_filters(
+ Arc::clone(probe),
+ Some(source),
+ )?;
}
let mut left = Arc::clone(&join_params.left.native_plan);
@@ -346,10 +349,36 @@ impl PhysicalPlanner {
/// runtime-filter placement can replace. Stop at native/Spark tree boundaries.
fn prepare_probe_for_runtime_filters(
plan: Arc<SparkPlan>,
+ source: Option<&Operator>,
) -> Result<Arc<SparkPlan>, ExecutionError> {
+ // Walk the serialized and Spark trees together. A mismatch must not lend
+ // another filter's permission to this node or any of its descendants.
+ let source = source.filter(|source| {
+ source.plan_id == plan.plan_id
+ && source.children.len() == plan.children.len()
+ && plan.children.len() == plan.native_plan.children().len()
+ && plan
+ .native_plan
+ .children()
+ .iter()
+ .zip(&plan.children)
+ .all(|(native, spark)| Arc::ptr_eq(native, &spark.native_plan))
+ && match &source.op_struct {
+ Some(OpStruct::Filter(_)) => plan.native_plan.is::<FilterExec>(),
+ Some(OpStruct::Projection(_)) => plan.native_plan.is::<ProjectionExec>(),
+ _ => false,
+ }
+ });
let mut native: Arc<dyn ExecutionPlan> =
if let Some(filter) = plan.native_plan.downcast_ref::<FilterExec>() {
- Arc::new(CometFilterExec::from_datafusion(filter.clone()))
+ let allowed = matches!(
+ source.and_then(|source| source.op_struct.as_ref()),
+ Some(OpStruct::Filter(filter)) if filter.allow_runtime_filter_pushdown
+ );
+ Arc::new(
+ CometFilterExec::from_datafusion(filter.clone())
+ .with_runtime_filter_pushdown(allowed),
+ )
} else if let Some(projection) = plan.native_plan.downcast_ref::<ProjectionExec>() {
Arc::new(CometProjectionExec::from_datafusion(projection.clone()))
} else {
@@ -358,7 +387,10 @@ impl PhysicalPlanner {
let mut prepared = plan.as_ref().clone();
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))?;
+ let child = Self::prepare_probe_for_runtime_filters(
+ Arc::clone(child),
+ source.and_then(|source| source.children.first()),
+ )?;
native = native.replace_children(
vec![Arc::clone(&child.native_plan)],
ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
@@ -773,3 +805,144 @@ fn rewrite_physical_expr(
Ok(expr.rewrite(&mut rewriter).data()?)
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use datafusion_comet_proto::{spark_expression, spark_operator};
+
+ fn column() -> Expr {
+ Expr {
+ expr_struct: Some(spark_expression::expr::ExprStruct::Bound(
+ spark_expression::BoundReference {
+ index: 0,
+ datatype: Some(spark_expression::DataType::default()),
+ },
+ )),
+ ..Default::default()
+ }
+ }
+
+ fn scan() -> Operator {
+ Operator {
+ op_struct: Some(OpStruct::Scan(spark_operator::Scan {
+ fields: vec![spark_expression::DataType::default()],
+ ..Default::default()
+ })),
+ ..Default::default()
+ }
+ }
+
+ fn filter(child: Operator, allowed: bool) -> Operator {
+ Operator {
+ plan_id: child.plan_id + 1,
+ children: vec![child],
+ op_struct: Some(OpStruct::Filter(spark_operator::Filter {
+ predicate: Some(column()),
+ allow_runtime_filter_pushdown: allowed,
+ })),
+ ..Default::default()
+ }
+ }
+
+ fn projection(child: Operator) -> Operator {
+ Operator {
+ plan_id: child.plan_id + 1,
+ children: vec![child],
+ op_struct: Some(OpStruct::Projection(spark_operator::Projection {
+ project_list: vec![column()],
+ })),
+ ..Default::default()
+ }
+ }
+
+ fn plan(source: &Operator) -> Arc<SparkPlan> {
+ PhysicalPlanner::default()
+ .create_plan(source, &mut vec![], 1)
+ .unwrap()
+ .2
+ }
+
+ fn allowed(plan: &SparkPlan) -> bool {
+ plan.native_plan
+ .downcast_ref::<CometFilterExec>()
+ .unwrap()
+ .allows_runtime_filter_pushdown()
+ }
+
+ #[test]
+ fn probe_filter_permission_below_projection_follows_its_source() {
+ for permission in [false, true] {
+ let source = projection(filter(scan(), permission));
+ let prepared =
+ PhysicalPlanner::prepare_probe_for_runtime_filters(plan(&source), Some(&source))
+ .unwrap();
+ assert!(prepared.native_plan.is::<CometProjectionExec>());
+ assert_eq!(allowed(&prepared.children[0]), permission);
+ assert!(Arc::ptr_eq(
+ prepared.native_plan.children()[0],
+ &prepared.children[0].native_plan
+ ));
+ }
+ }
+
+ #[test]
+ fn stacked_probe_filters_keep_independent_permissions() {
+ for outer in [false, true] {
+ for inner in [false, true] {
+ let source = filter(filter(scan(), inner), outer);
+ let prepared = PhysicalPlanner::prepare_probe_for_runtime_filters(
+ plan(&source),
+ Some(&source),
+ )
+ .unwrap();
+ assert_eq!(allowed(&prepared), outer);
+ assert_eq!(allowed(&prepared.children[0]), inner);
+ assert!(Arc::ptr_eq(
+ prepared.native_plan.children()[0],
+ &prepared.children[0].native_plan
+ ));
+ }
+ }
+ }
+
+ #[test]
+ fn probe_filter_permissions_default_to_false_on_source_mismatch() {
+ let source = filter(filter(scan(), true), true);
+ let original = plan(&source);
+ let mut wrong_id = source.clone();
+ wrong_id.plan_id += 1;
+ let mut wrong_arity = source.clone();
+ wrong_arity.children.push(scan());
+ let wrong_type = projection(source.children[0].clone());
+ for mismatched in [None, Some(&wrong_id), Some(&wrong_arity), Some(&wrong_type)] {
+ let prepared = PhysicalPlanner::prepare_probe_for_runtime_filters(
+ Arc::clone(&original),
+ mismatched,
+ )
+ .unwrap();
+ assert!(!allowed(&prepared));
+ assert!(!allowed(&prepared.children[0]));
+ }
+
+ // A mismatch below an otherwise matching node cannot borrow its permission.
+ let mut wrong_child = source.clone();
+ wrong_child.children[0] = scan();
+ let prepared = PhysicalPlanner::prepare_probe_for_runtime_filters(
+ Arc::clone(&original),
+ Some(&wrong_child),
+ )
+ .unwrap();
+ assert!(allowed(&prepared));
+ assert!(!allowed(&prepared.children[0]));
+
+ // Native and Spark children can differ at planner-generated boundaries.
+ let mut mismatched = original.as_ref().clone();
+ mismatched.children = vec![plan(&scan())];
+ let prepared =
+ PhysicalPlanner::prepare_probe_for_runtime_filters(Arc::new(mismatched), Some(&source))
+ .unwrap();
+ assert!(!allowed(&prepared));
+ assert!(prepared.native_plan.children()[0].is::<FilterExec>());
+ }
+}
diff --git a/native/operators/src/filter.rs b/native/operators/src/filter.rs
index 56525267ff..6303824378 100644
--- a/native/operators/src/filter.rs
+++ b/native/operators/src/filter.rs
@@ -44,6 +44,7 @@ use datafusion::physical_plan::{
pub struct CometFilterExec {
filter: FilterExec,
metrics: ExecutionPlanMetricsSet,
+ allow_runtime_filter_pushdown: bool,
}
impl CometFilterExec {
@@ -51,9 +52,19 @@ impl CometFilterExec {
Self {
filter,
metrics: ExecutionPlanMetricsSet::new(),
+ allow_runtime_filter_pushdown: false,
}
}
+ pub fn with_runtime_filter_pushdown(mut self, allowed: bool) -> Self {
+ self.allow_runtime_filter_pushdown = allowed;
+ self
+ }
+
+ pub fn allows_runtime_filter_pushdown(&self) -> bool {
+ self.allow_runtime_filter_pushdown
+ }
+
pub fn input(&self) -> &Arc<dyn ExecutionPlan> {
self.filter.input()
}
@@ -90,6 +101,7 @@ impl CometFilterExec {
ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
)?,
metrics: self.metrics.clone(),
+ allow_runtime_filter_pushdown: self.allow_runtime_filter_pushdown,
}))
}
}
@@ -135,6 +147,7 @@ impl ExecutionPlan for CometFilterExec {
Ok(Arc::new(Self {
filter: self.replace_input(children.remove(0), options)?,
metrics: ExecutionPlanMetricsSet::new(),
+ allow_runtime_filter_pushdown: self.allow_runtime_filter_pushdown,
}))
}
@@ -197,6 +210,7 @@ impl ExecutionPlan for CometFilterExec {
Some(Arc::new(Self {
filter,
metrics: ExecutionPlanMetricsSet::new(),
+ allow_runtime_filter_pushdown: self.allow_runtime_filter_pushdown,
}))
}
@@ -206,6 +220,7 @@ impl ExecutionPlan for CometFilterExec {
Some(Arc::new(Self {
filter,
metrics: ExecutionPlanMetricsSet::new(),
+ allow_runtime_filter_pushdown: self.allow_runtime_filter_pushdown,
}))
}
}
@@ -240,6 +255,29 @@ mod tests {
FilterExec::try_new(predicate, input)
}
+ #[test]
+ fn runtime_filter_permission_survives_rebuilding() -> Result<()> {
+ assert!(!CometFilterExec::from_datafusion(filter()?).allows_runtime_filter_pushdown());
+ for allowed in [false, true] {
+ let filter = Arc::new(
+ CometFilterExec::from_datafusion(filter()?).with_runtime_filter_pushdown(allowed),
+ );
+ for rebuilt in [
+ filter.with_execution_input(Arc::clone(filter.input()))?,
+ Arc::clone(&filter).replace_children(
+ vec![Arc::clone(filter.input())],
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )?,
+ filter.with_fetch(Some(1)).unwrap(),
+ ] {
+ let rebuilt = rebuilt.downcast_ref::<CometFilterExec>().unwrap();
+ assert_eq!(rebuilt.allows_runtime_filter_pushdown(), allowed);
+ assert!(Arc::ptr_eq(rebuilt.predicate(), filter.predicate()));
+ }
+ }
+ Ok(())
+ }
+
#[test]
fn expression_visitor_preserves_predicate_root_and_recursion() -> Result<()> {
let filter = CometFilterExec::from_datafusion(filter()?);
diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto
index 48e67ec7c1..afe3183b31 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -873,6 +873,8 @@ message Projection {
message Filter {
spark.spark_expression.Expr predicate = 1;
+ // Explicit opt-in for a deterministic residual. Absent metadata preserves the safe default.
+ bool allow_runtime_filter_pushdown = 2;
}
message Sort {
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index d982dd4cc2..37152f7abf 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -644,6 +644,21 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(false)
+ val COMET_EXEC_JOIN_DYNAMIC_FILTER_ALLOW_DETERMINISTIC_FILTER_PUSHDOWN: ConfigEntry[Boolean] =
+ conf(s"$COMET_EXEC_CONFIG_PREFIX.join.dynamicFilter.allowDeterministicFilterPushdown")
+ .category(CATEGORY_EXEC)
+ .doc(
+ "Allow native join runtime filters to reach Parquet readers through deterministic " +
+ "residual predicates. Requires spark.comet.exec.join.dynamicFilter.enabled. " +
+ "This can skip expression errors on rows eliminated before the residual runs, " +
+ "including division-by-zero and ANSI cast errors. Errors on retained rows and " +
+ "Parquet schema conversion checks are preserved. Nondeterministic predicates, " +
+ "all standalone projections (including column-only projections), and limits remain " +
+ "reader propagation boundaries. When false, " +
+ "only direct-column IS NOT NULL checks combined with AND can be crossed.")
+ .booleanConf
+ .createWithDefault(false)
+
val COMET_EXEC_TOPK_FUSION_ENABLED: ConfigEntry[Boolean] =
conf(s"$COMET_EXEC_CONFIG_PREFIX.topK.fusion.enabled")
.category(CATEGORY_EXEC)
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
index 733f56aade..0e07bb8f07 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
@@ -1501,6 +1501,10 @@ object CometFilterExec extends CometOperatorSerde[FilterExec] {
val filterBuilder = OperatorOuterClass.Filter
.newBuilder()
.setPredicate(cond.get)
+ .setAllowRuntimeFilterPushdown(
+ op.condition.deterministic &&
+ CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ALLOW_DETERMINISTIC_FILTER_PUSHDOWN
+ .get(op.conf))
Some(builder.setFilter(filterBuilder).build())
} else {
None
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 6ae7b1081f..8b8905b0c4 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
@@ -633,6 +633,75 @@ class CometJoinSuite extends CometTestBase {
}
}
+ test("join dynamic filter crosses deterministic residuals only with explicit opt-in") {
+ withTempPath { probePath =>
+ withSQLConf(
+ CometConf.COMET_BATCH_SIZE.key -> "1",
+ CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1") {
+ spark
+ .range(0, 10000, 1, 1)
+ .selectExpr("id AS payload", "CAST(id AS INT) AS probe_key")
+ .write
+ .option("parquet.block.size", "1024")
+ .parquet(probePath.getCanonicalPath)
+ withParquetTable(probePath.getCanonicalPath, "dynamic_residual_probe") {
+ // Key 2506 cannot pass the residual, although it belongs to the join domain.
+ withParquetTable(Seq(Tuple1(2500), Tuple1(2506)), "dynamic_residual_build") {
+ for (buildLeft <- Seq(false, true)) {
+ val from = if (buildLeft) {
+ "dynamic_residual_build b JOIN dynamic_residual_probe p"
+ } else {
+ "dynamic_residual_probe p JOIN dynamic_residual_build b"
+ }
+ val query = "SELECT /*+ BROADCAST(b) */ p.probe_key, p.payload FROM " + from +
+ " ON p.probe_key = b._1 WHERE pmod(p.payload, 7) > 0"
+ var unfilteredBytes = 0L
+ for (allowed <- Seq(false, true)) {
+ withSQLConf(
+ CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ALLOW_DETERMINISTIC_FILTER_PUSHDOWN.key -> allowed.toString) {
+ val (_, plan) = checkSparkAnswerAndOperator(
+ sql(query),
+ Seq(classOf[CometBroadcastHashJoinExec], classOf[CometNativeScanExec]))
+ checkAnswer(sql(query), Seq(Row(2500, 2500L)))
+ val joins = collect(plan) { case join: CometBroadcastHashJoinExec => join }
+ assert(joins.size == 1)
+ val filters = collect(plan) {
+ case filter: CometFilterExec if filter.output.exists(_.name == "probe_key") =>
+ filter
+ }
+ assert(filters.size == 1, s"Expected one native residual filter:\n$plan")
+ assert(filters.head.condition.deterministic)
+ assert(filters.head.nativeOp.getFilter.getAllowRuntimeFilterPushdown == allowed)
+ val scans = collect(plan) {
+ case scan: CometNativeScanExec if scan.output.exists(_.name == "probe_key") =>
+ scan
+ }
+ assert(scans.size == 1)
+ val metrics = scans.head.metrics
+ val bytes = metrics("bytes_scanned").value
+ assert(joins.head.metrics("output_rows").value == 1L)
+ assert(filters.head.metrics("output_rows").value > 0L)
+ if (allowed) {
+ assert(joins.head.metrics("dynamic_filter_join_filters_attached").value > 0L)
+ assert(metrics("row_groups_pruned_statistics").value > 0L)
+ assert(bytes < unfilteredBytes)
+ } else {
+ unfilteredBytes = bytes
+ assert(joins.head.metrics("dynamic_filter_join_filters_attached").value == 0L)
+ assert(joins.head.metrics("dynamic_filter_join_filters_skipped").value > 0L)
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
test("join dynamic filter preserves Parquet schema conversion errors") {
withTempPath { probePath =>
withSQLConf(
@@ -654,14 +723,20 @@ class CometJoinSuite extends CometTestBase {
.parquet(probePath.getCanonicalPath)
.createOrReplaceTempView("dynamic_schema_probe")
withParquetTable(Seq(Tuple1(0)), "dynamic_schema_build") {
- for ((comet, dynamicFilter) <- Seq((false, false), (true, false), (true, true))) {
+ for {
+ (comet, dynamicFilter) <- Seq((false, false), (true, false), (true, true))
+ allowed <- Seq(false, true)
+ // Without a residual, reader filters can attach regardless of the permission flag.
+ residual <- Seq("", " WHERE pmod(p.payload, 2) >= 0")
+ } {
withSQLConf(
CometConf.COMET_ENABLED.key -> comet.toString,
+ CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ALLOW_DETERMINISTIC_FILTER_PUSHDOWN.key -> allowed.toString,
CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.key -> dynamicFilter.toString) {
val df = sql(
"SELECT /*+ BROADCAST(b) */ p.probe_key, p.payload " +
"FROM dynamic_schema_probe p JOIN dynamic_schema_build b " +
- "ON p.probe_key = b._1")
+ "ON p.probe_key = b._1" + residual)
val plan = df.queryExecution.executedPlan
if (comet) {
val joins = collect(plan) { case join: CometBroadcastHashJoinExec => join }
@@ -674,7 +749,8 @@ class CometJoinSuite extends CometTestBase {
}
assert(probes.size == 1, s"Expected one native probe scan:\n$plan")
}
- withClue(s"comet=$comet, dynamicFilter=$dynamicFilter: ") {
+ withClue(s"comet=$comet, dynamicFilter=$dynamicFilter, allowed=$allowed, " +
+ s"residual=$residual: ") {
val error = intercept[SparkException](df.collect())
val chain = causeChain(error)
assert(
@@ -693,6 +769,7 @@ class CometJoinSuite extends CometTestBase {
withTempPath { probePath =>
withSQLConf(
CometConf.COMET_BATCH_SIZE.key -> "16",
+ CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ALLOW_DETERMINISTIC_FILTER_PUSHDOWN.key -> "true",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org