You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "xudong963 (via GitHub)" <gi...@apache.org> on 2026/10/16 05:30:18 UTC
[PR] feat: rewrite inner joins against a single constant row into a filter [datafusion]
xudong963 opened a new pull request, #26404:
URL: https://github.com/apache/datafusion/pull/26404
## Which issue does this PR close?
- Closes #26349.
## Rationale for this change
`AggregateStatistics` can answer a scalar subquery such as `(SELECT min(ts) FROM live)` from statistics and replace it with a row of literals. A join against that row still runs as a `NestedLoopJoinExec`.
That join only filters its other input, but `NestedLoopJoinExec` preserves neither input's ordering. Operators above it therefore lose the input's ordering. For example, an aggregation grouped by the input's sort key can no longer use `ordering_mode=Sorted` and must hold every group until the end. The literal bound is also not pushed into the scan for pruning.
## What changes are included in this PR?
A new physical optimizer rule, `EliminateConstantJoin`, runs right after `AggregateStatistics`. It applies when one input of an inner `NestedLoopJoinExec` is provably a single row of literals: projections over a single-partition `PlaceholderRowExec`. The rule substitutes the literals into the join filter. It then replaces the join with a `FilterExec` over the other input, under a `ProjectionExec` that restores the join's output schema and metadata.
At this point in the optimizer, a folded `min(ts) AS first_ts` is still an alias projection over the literal projection, so the rule follows stacked projections.
The rule leaves the join unchanged when:
- the join is not an inner join;
- the placeholder has more than one partition, because it emits one row per partition;
- a value is not a literal; for example, `random()` would then be evaluated per row instead of once;
- the join filter contains a lambda, because lambda variables are numbered after the filter's own columns, not the other input's.
## What is the testing strategy for this PR?
- `eliminate_constant_join.slt` has plans and results for two rewrites: a scalar folded from Parquet statistics, and a literal subquery. In the first, the aggregation becomes `ordering_mode=Sorted` and the bound is pushed into the Parquet scan. The file also checks that a left join and a `random()` constant keep the join.
- `physical_optimizer/eliminate_constant_join.rs` covers cases SQL cannot reach:
- a multi-partition placeholder;
- a lambda join filter, planned without optimizer rules because projection pushdown cannot plan it yet (#26399);
- schema metadata that comes from the constant input.
- `explain.slt`: `EXPLAIN VERBOSE` lists the new rule.
## Are there any user-facing changes?
Queries that join against a row folded from statistics, or against a literal subquery, now plan a filter and a projection instead of a nested loop join. `EXPLAIN VERBOSE` and `PhysicalOptimizer::new()` include the new rule. There are no API changes.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: rewrite inner joins against a single constant row into a filter [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 merged PR #26404:
URL: https://github.com/apache/datafusion/pull/26404
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: rewrite inner joins against a single constant row into a filter [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #26404:
URL: https://github.com/apache/datafusion/pull/26404#issuecomment-6258428724
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/26404?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `82.56881%` with `19 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 82.89%. Comparing base ([`fe612e2`](https://app.codecov.io/gh/apache/datafusion/commit/fe612e2da212016905d600eb0dd689910c0899b7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`0981f5b`](https://app.codecov.io/gh/apache/datafusion/commit/0981f5bb2fe6b4ad4825438a5aedba53a6e614e7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/26404?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../physical-optimizer/src/eliminate\_constant\_join.rs](https://app.codecov.io/gh/apache/datafusion/pull/26404?src=pr&el=tree&filepath=datafusion%2Fphysical-optimizer%2Fsrc%2Feliminate_constant_join.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1vcHRpbWl6ZXIvc3JjL2VsaW1pbmF0ZV9jb25zdGFudF9qb2luLnJz) | 82.40% | [11 Missing and 8 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/26404?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #26404 +/- ##
==========================================
- Coverage 82.89% 82.89% -0.01%
==========================================
Files 1154 1155 +1
Lines 458022 458131 +109
Branches 458022 458131 +109
==========================================
+ Hits 379683 379767 +84
- Misses 55303 55315 +12
- Partials 23036 23049 +13
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/26404?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: rewrite inner joins against a single constant row into a filter [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #26404:
URL: https://github.com/apache/datafusion/pull/26404#discussion_r4319959656
##########
datafusion/physical-optimizer/src/eliminate_constant_join.rs:
##########
@@ -0,0 +1,202 @@
+// 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.
+
+//! Replace inner joins against a single row of literals with a filter.
+
+use std::sync::Arc;
+
+use datafusion_common::Result;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::{JoinSide, JoinType, internal_datafusion_err};
+use datafusion_physical_expr::PhysicalExpr;
+use datafusion_physical_expr::expressions::{Column, LambdaExpr, Literal};
+use datafusion_physical_plan::filter::FilterExec;
+use datafusion_physical_plan::joins::NestedLoopJoinExec;
+use datafusion_physical_plan::joins::utils::ColumnIndex;
+use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
+use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+
+use crate::PhysicalOptimizerRule;
+
+/// Rewrites an inner [`NestedLoopJoinExec`] whose input provably produces one row
+/// of literals into a [`FilterExec`] over the other input, with the literals
+/// substituted into the join filter, under a [`ProjectionExec`] that restores the
+/// join's output.
+///
+/// For example, [`AggregateStatistics`] folds the scalar subquery in
+///
+/// ```sql
+/// SELECT e.* FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
+/// WHERE e.ts < l.first_ts
+/// ```
+///
+/// into literals over a [`PlaceholderRowExec`], leaving a join that only filters
+/// `events`. A nested-loop join preserves neither input's order, while the filter
+/// and projection preserve the other input's ordering and partitioning. This lets
+/// later operators, such as an aggregation grouped by a sort key of `events`,
+/// use that ordering.
+///
+/// [`AggregateStatistics`]: crate::aggregate_statistics::AggregateStatistics
+#[derive(Debug, Default)]
+pub struct EliminateConstantJoin {}
+
+impl EliminateConstantJoin {
+ #[expect(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for EliminateConstantJoin {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ _config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ plan.transform_up(|plan| {
+ let Some(join) = plan.downcast_ref::<NestedLoopJoinExec>() else {
+ return Ok(Transformed::no(plan));
+ };
+ Ok(match eliminate(join)? {
+ Some(rewritten) => Transformed::yes(rewritten),
+ None => Transformed::no(plan),
+ })
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "EliminateConstantJoin"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Rewrites `join` over its non-constant input, if the other input is a constant row.
+fn eliminate(join: &NestedLoopJoinExec) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ if join.join_type() != &JoinType::Inner {
+ return Ok(None);
+ }
+ if let Some(filter) = join.filter()
+ // Lambda variables are numbered after the filter's own columns, not the
+ // other input's.
+ && filter
+ .expression()
+ .exists(|expr| Ok(expr.is::<LambdaExpr>()))?
+ {
+ return Ok(None);
+ }
+ // Each input's columns as expressions over the non-constant input.
+ let (left, right, input) = if let Some(row) = constant_row(join.left()) {
+ (row, columns(join.right()), join.right())
+ } else if let Some(row) = constant_row(join.right()) {
+ (columns(join.left()), row, join.left())
+ } else {
+ return Ok(None);
+ };
+
+ let mut input = Arc::clone(input);
+ if let Some(filter) = join.filter() {
+ let predicate = Arc::clone(filter.expression())
+ .transform_up(|expr| {
+ let Some(column) = expr.downcast_ref::<Column>() else {
+ return Ok(Transformed::no(expr));
+ };
+ // Filter columns index the filter's own schema, which
+ // `column_indices` maps back to the join inputs.
+ let ColumnIndex { index, side } =
+ filter.column_indices().get(column.index()).ok_or_else(|| {
+ internal_datafusion_err!("Unmapped join filter column")
+ })?;
+ let value = match side {
+ JoinSide::Left => left.get(*index),
+ JoinSide::Right => right.get(*index),
+ JoinSide::None => None,
+ }
+ .ok_or_else(|| {
+ internal_datafusion_err!("Join filter column out of range")
+ })?;
+ Ok(Transformed::yes(Arc::clone(value)))
+ })
+ .data()?;
+ input = Arc::new(FilterExec::try_new(predicate, input)?);
+ }
+
+ let mut outputs = left.into_iter().chain(right).collect::<Vec<_>>();
+ if let Some(projection) = join.projection() {
+ outputs = projection
+ .iter()
+ .map(|&index| outputs.get(index).cloned())
+ .collect::<Option<_>>()
+ .ok_or_else(|| internal_datafusion_err!("Join projection out of range"))?;
+ }
+ let schema = join.schema();
+ let exprs = outputs
+ .into_iter()
+ .zip(schema.fields())
+ .map(|(expr, field)| ProjectionExpr {
+ expr,
+ alias: field.name().clone(),
+ });
+ // Keep the join's metadata, which may come from the constant input.
+ Ok(Some(Arc::new(
+ ProjectionExec::try_new_with_schema_metadata(exprs, input, &schema)?,
+ )))
+}
+
+/// Returns the literal values of the one row `plan` produces, if it is a stack of
+/// projections over a single-partition [`PlaceholderRowExec`] that project only
+/// literals and columns of the projection below.
+fn constant_row(plan: &Arc<dyn ExecutionPlan>) -> Option<Vec<Arc<dyn PhysicalExpr>>> {
+ let projection = plan.downcast_ref::<ProjectionExec>()?;
+ let input = projection.input();
+ let input_row = if input.is::<PlaceholderRowExec>() {
+ // A placeholder emits one row per partition, and its columns are not
+ // literals.
+ if input.output_partitioning().partition_count() != 1 {
+ return None;
+ }
+ Vec::new()
+ } else {
+ constant_row(input)?
+ };
+ projection
+ .expr()
+ .iter()
+ .map(|ProjectionExpr { expr, .. }| {
+ if expr.is::<Literal>() {
Review Comment:
Checked this: constants written in SQL are folded by the logical simplifier, but values folded by `AggregateStatistics` only become literals during physical planning, so `(SELECT min(ts) + 1 ...)` reached the rule as `25 + 1` and was skipped. The rule now substitutes the row below and folds with `PhysicalExprSimplifier`, accepting only a literal result; volatile expressions and ones that fail to evaluate are still rejected. Added a comment and an slt case in b9fa393cd.
##########
datafusion/sqllogictest/test_files/eliminate_constant_join.slt:
##########
@@ -0,0 +1,126 @@
+# 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.
+
+# Inner joins against a single row of literals become a filter over the other
+# input, which keeps that input's ordering (EliminateConstantJoin).
+
+statement ok
+set datafusion.execution.target_partitions = 1;
+
+statement ok
+set datafusion.explain.physical_plan_only = true;
+
+# Events sorted by account
+query I
+COPY (SELECT * FROM (VALUES ('a', 10), ('a', 30), ('b', 20), ('b', NULL), ('c', 5)) AS v(acct, t))
+TO 'test_files/scratch/eliminate_constant_join/events.parquet'
+STORED AS PARQUET;
+----
+5
+
+statement ok
+CREATE EXTERNAL TABLE events
+STORED AS PARQUET
+WITH ORDER (acct ASC NULLS LAST)
+LOCATION 'test_files/scratch/eliminate_constant_join/events.parquet';
+
+query I
+COPY (SELECT * FROM (VALUES (25), (40)) AS v(ts))
+TO 'test_files/scratch/eliminate_constant_join/live.parquet'
+STORED AS PARQUET;
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE live
+STORED AS PARQUET
+LOCATION 'test_files/scratch/eliminate_constant_join/live.parquet';
+
+# min(ts) is answered from Parquet statistics, so the join only filters the
+# events, and the aggregation keeps their account order
+query TT
+EXPLAIN SELECT e.acct, count(*) AS n
+FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
+WHERE e.t < l.first_ts
+GROUP BY e.acct;
+----
+physical_plan
+01)ProjectionExec: expr=[acct@0 as acct, count(Int64(1))@1 as n]
+02)--AggregateExec: mode=Single, gby=[acct@0 as acct], aggr=[count(Int64(1))], group_clustering_mode=Full
+03)----FilterExec: 25 > t@1, projection=[acct@0]
+04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/eliminate_constant_join/events.parquet]]}, projection=[acct, t], output_ordering=[acct@0 ASC NULLS LAST], file_type=parquet, predicate=25 > t@1, pruning_predicate=t_null_count@1 != row_count@2 AND t_min@0 < 25, required_guarantees=[]
+
+query TI
+SELECT e.acct, count(*) AS n
+FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
+WHERE e.t < l.first_ts
+GROUP BY e.acct
+ORDER BY e.acct;
+----
+a 1
+b 1
+c 1
+
+# The constant row's columns stay in the output
+query TT
+EXPLAIN SELECT e.acct, e.t, c.cutoff
+FROM (SELECT 25 AS cutoff) c JOIN events e ON e.t < c.cutoff;
+----
+physical_plan
+01)ProjectionExec: expr=[acct@0 as acct, t@1 as t, 25 as cutoff]
+02)--FilterExec: t@1 < 25
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/eliminate_constant_join/events.parquet]]}, projection=[acct, t], output_ordering=[acct@0 ASC NULLS LAST], file_type=parquet, predicate=t@1 < 25, pruning_predicate=t_null_count@1 != row_count@2 AND t_min@0 < 25, required_guarantees=[]
+
+query TII rowsort
+SELECT e.acct, e.t, c.cutoff
+FROM (SELECT 25 AS cutoff) c JOIN events e ON e.t < c.cutoff;
+----
+a 10 25
+b 20 25
+c 5 25
+
+# A left join keeps the constant row when nothing matches
Review Comment:
Added an `EXPLAIN` showing the left join stays a `NestedLoopJoinExec` (b9fa393cd).
##########
datafusion/physical-optimizer/src/eliminate_constant_join.rs:
##########
@@ -0,0 +1,202 @@
+// 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.
+
+//! Replace inner joins against a single row of literals with a filter.
+
+use std::sync::Arc;
+
+use datafusion_common::Result;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::{JoinSide, JoinType, internal_datafusion_err};
+use datafusion_physical_expr::PhysicalExpr;
+use datafusion_physical_expr::expressions::{Column, LambdaExpr, Literal};
+use datafusion_physical_plan::filter::FilterExec;
+use datafusion_physical_plan::joins::NestedLoopJoinExec;
+use datafusion_physical_plan::joins::utils::ColumnIndex;
+use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
+use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+
+use crate::PhysicalOptimizerRule;
+
+/// Rewrites an inner [`NestedLoopJoinExec`] whose input provably produces one row
+/// of literals into a [`FilterExec`] over the other input, with the literals
+/// substituted into the join filter, under a [`ProjectionExec`] that restores the
+/// join's output.
+///
+/// For example, [`AggregateStatistics`] folds the scalar subquery in
+///
+/// ```sql
+/// SELECT e.* FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
+/// WHERE e.ts < l.first_ts
+/// ```
+///
+/// into literals over a [`PlaceholderRowExec`], leaving a join that only filters
+/// `events`. A nested-loop join preserves neither input's order, while the filter
+/// and projection preserve the other input's ordering and partitioning. This lets
+/// later operators, such as an aggregation grouped by a sort key of `events`,
+/// use that ordering.
+///
+/// [`AggregateStatistics`]: crate::aggregate_statistics::AggregateStatistics
+#[derive(Debug, Default)]
+pub struct EliminateConstantJoin {}
+
+impl EliminateConstantJoin {
+ #[expect(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for EliminateConstantJoin {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ _config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ plan.transform_up(|plan| {
+ let Some(join) = plan.downcast_ref::<NestedLoopJoinExec>() else {
+ return Ok(Transformed::no(plan));
+ };
+ Ok(match eliminate(join)? {
+ Some(rewritten) => Transformed::yes(rewritten),
+ None => Transformed::no(plan),
+ })
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "EliminateConstantJoin"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Rewrites `join` over its non-constant input, if the other input is a constant row.
+fn eliminate(join: &NestedLoopJoinExec) -> Result<Option<Arc<dyn ExecutionPlan>>> {
Review Comment:
Added in b9fa393cd. `CrossJoinExec` drops input ordering and, when the other input has no statistics, buffers it as the build side, so the rewrite helps there as well. This changes the `j1, LATERAL (SELECT 1)` plans in `cte.slt` and `joins.slt`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: rewrite inner joins against a single constant row into a filter [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on code in PR #26404:
URL: https://github.com/apache/datafusion/pull/26404#discussion_r4312300818
##########
datafusion/sqllogictest/test_files/eliminate_constant_join.slt:
##########
@@ -0,0 +1,126 @@
+# 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.
+
+# Inner joins against a single row of literals become a filter over the other
+# input, which keeps that input's ordering (EliminateConstantJoin).
+
+statement ok
+set datafusion.execution.target_partitions = 1;
+
+statement ok
+set datafusion.explain.physical_plan_only = true;
+
+# Events sorted by account
+query I
+COPY (SELECT * FROM (VALUES ('a', 10), ('a', 30), ('b', 20), ('b', NULL), ('c', 5)) AS v(acct, t))
+TO 'test_files/scratch/eliminate_constant_join/events.parquet'
+STORED AS PARQUET;
+----
+5
+
+statement ok
+CREATE EXTERNAL TABLE events
+STORED AS PARQUET
+WITH ORDER (acct ASC NULLS LAST)
+LOCATION 'test_files/scratch/eliminate_constant_join/events.parquet';
+
+query I
+COPY (SELECT * FROM (VALUES (25), (40)) AS v(ts))
+TO 'test_files/scratch/eliminate_constant_join/live.parquet'
+STORED AS PARQUET;
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE live
+STORED AS PARQUET
+LOCATION 'test_files/scratch/eliminate_constant_join/live.parquet';
+
+# min(ts) is answered from Parquet statistics, so the join only filters the
+# events, and the aggregation keeps their account order
+query TT
+EXPLAIN SELECT e.acct, count(*) AS n
+FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
+WHERE e.t < l.first_ts
+GROUP BY e.acct;
+----
+physical_plan
+01)ProjectionExec: expr=[acct@0 as acct, count(Int64(1))@1 as n]
+02)--AggregateExec: mode=Single, gby=[acct@0 as acct], aggr=[count(Int64(1))], group_clustering_mode=Full
+03)----FilterExec: 25 > t@1, projection=[acct@0]
+04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/eliminate_constant_join/events.parquet]]}, projection=[acct, t], output_ordering=[acct@0 ASC NULLS LAST], file_type=parquet, predicate=25 > t@1, pruning_predicate=t_null_count@1 != row_count@2 AND t_min@0 < 25, required_guarantees=[]
+
+query TI
+SELECT e.acct, count(*) AS n
+FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
+WHERE e.t < l.first_ts
+GROUP BY e.acct
+ORDER BY e.acct;
+----
+a 1
+b 1
+c 1
+
+# The constant row's columns stay in the output
+query TT
+EXPLAIN SELECT e.acct, e.t, c.cutoff
+FROM (SELECT 25 AS cutoff) c JOIN events e ON e.t < c.cutoff;
+----
+physical_plan
+01)ProjectionExec: expr=[acct@0 as acct, t@1 as t, 25 as cutoff]
+02)--FilterExec: t@1 < 25
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/eliminate_constant_join/events.parquet]]}, projection=[acct, t], output_ordering=[acct@0 ASC NULLS LAST], file_type=parquet, predicate=t@1 < 25, pruning_predicate=t_null_count@1 != row_count@2 AND t_min@0 < 25, required_guarantees=[]
+
+query TII rowsort
+SELECT e.acct, e.t, c.cutoff
+FROM (SELECT 25 AS cutoff) c JOIN events e ON e.t < c.cutoff;
+----
+a 10 25
+b 20 25
+c 5 25
+
+# A left join keeps the constant row when nothing matches
Review Comment:
Nit: I think an `EXPLAIN` here would help to show the bail-out of the rule, beyond the correct result returned
##########
datafusion/physical-optimizer/src/eliminate_constant_join.rs:
##########
@@ -0,0 +1,202 @@
+// 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.
+
+//! Replace inner joins against a single row of literals with a filter.
+
+use std::sync::Arc;
+
+use datafusion_common::Result;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::{JoinSide, JoinType, internal_datafusion_err};
+use datafusion_physical_expr::PhysicalExpr;
+use datafusion_physical_expr::expressions::{Column, LambdaExpr, Literal};
+use datafusion_physical_plan::filter::FilterExec;
+use datafusion_physical_plan::joins::NestedLoopJoinExec;
+use datafusion_physical_plan::joins::utils::ColumnIndex;
+use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
+use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+
+use crate::PhysicalOptimizerRule;
+
+/// Rewrites an inner [`NestedLoopJoinExec`] whose input provably produces one row
+/// of literals into a [`FilterExec`] over the other input, with the literals
+/// substituted into the join filter, under a [`ProjectionExec`] that restores the
+/// join's output.
+///
+/// For example, [`AggregateStatistics`] folds the scalar subquery in
+///
+/// ```sql
+/// SELECT e.* FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
+/// WHERE e.ts < l.first_ts
+/// ```
+///
+/// into literals over a [`PlaceholderRowExec`], leaving a join that only filters
+/// `events`. A nested-loop join preserves neither input's order, while the filter
+/// and projection preserve the other input's ordering and partitioning. This lets
+/// later operators, such as an aggregation grouped by a sort key of `events`,
+/// use that ordering.
+///
+/// [`AggregateStatistics`]: crate::aggregate_statistics::AggregateStatistics
+#[derive(Debug, Default)]
+pub struct EliminateConstantJoin {}
+
+impl EliminateConstantJoin {
+ #[expect(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for EliminateConstantJoin {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ _config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ plan.transform_up(|plan| {
+ let Some(join) = plan.downcast_ref::<NestedLoopJoinExec>() else {
+ return Ok(Transformed::no(plan));
+ };
+ Ok(match eliminate(join)? {
+ Some(rewritten) => Transformed::yes(rewritten),
+ None => Transformed::no(plan),
+ })
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "EliminateConstantJoin"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Rewrites `join` over its non-constant input, if the other input is a constant row.
+fn eliminate(join: &NestedLoopJoinExec) -> Result<Option<Arc<dyn ExecutionPlan>>> {
Review Comment:
This optimization would work also for `CrossJoinExec`, I think we could either add support for it (it doesn't seem much work from a cursory look), or document the gap and possibly file a follow-up issue. wdyt?
##########
datafusion/physical-optimizer/src/eliminate_constant_join.rs:
##########
@@ -0,0 +1,202 @@
+// 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.
+
+//! Replace inner joins against a single row of literals with a filter.
+
+use std::sync::Arc;
+
+use datafusion_common::Result;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::{JoinSide, JoinType, internal_datafusion_err};
+use datafusion_physical_expr::PhysicalExpr;
+use datafusion_physical_expr::expressions::{Column, LambdaExpr, Literal};
+use datafusion_physical_plan::filter::FilterExec;
+use datafusion_physical_plan::joins::NestedLoopJoinExec;
+use datafusion_physical_plan::joins::utils::ColumnIndex;
+use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
+use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
+use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+
+use crate::PhysicalOptimizerRule;
+
+/// Rewrites an inner [`NestedLoopJoinExec`] whose input provably produces one row
+/// of literals into a [`FilterExec`] over the other input, with the literals
+/// substituted into the join filter, under a [`ProjectionExec`] that restores the
+/// join's output.
+///
+/// For example, [`AggregateStatistics`] folds the scalar subquery in
+///
+/// ```sql
+/// SELECT e.* FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
+/// WHERE e.ts < l.first_ts
+/// ```
+///
+/// into literals over a [`PlaceholderRowExec`], leaving a join that only filters
+/// `events`. A nested-loop join preserves neither input's order, while the filter
+/// and projection preserve the other input's ordering and partitioning. This lets
+/// later operators, such as an aggregation grouped by a sort key of `events`,
+/// use that ordering.
+///
+/// [`AggregateStatistics`]: crate::aggregate_statistics::AggregateStatistics
+#[derive(Debug, Default)]
+pub struct EliminateConstantJoin {}
+
+impl EliminateConstantJoin {
+ #[expect(missing_docs)]
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl PhysicalOptimizerRule for EliminateConstantJoin {
+ fn optimize(
+ &self,
+ plan: Arc<dyn ExecutionPlan>,
+ _config: &ConfigOptions,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ plan.transform_up(|plan| {
+ let Some(join) = plan.downcast_ref::<NestedLoopJoinExec>() else {
+ return Ok(Transformed::no(plan));
+ };
+ Ok(match eliminate(join)? {
+ Some(rewritten) => Transformed::yes(rewritten),
+ None => Transformed::no(plan),
+ })
+ })
+ .data()
+ }
+
+ fn name(&self) -> &str {
+ "EliminateConstantJoin"
+ }
+
+ fn schema_check(&self) -> bool {
+ true
+ }
+}
+
+/// Rewrites `join` over its non-constant input, if the other input is a constant row.
+fn eliminate(join: &NestedLoopJoinExec) -> Result<Option<Arc<dyn ExecutionPlan>>> {
+ if join.join_type() != &JoinType::Inner {
+ return Ok(None);
+ }
+ if let Some(filter) = join.filter()
+ // Lambda variables are numbered after the filter's own columns, not the
+ // other input's.
+ && filter
+ .expression()
+ .exists(|expr| Ok(expr.is::<LambdaExpr>()))?
+ {
+ return Ok(None);
+ }
+ // Each input's columns as expressions over the non-constant input.
+ let (left, right, input) = if let Some(row) = constant_row(join.left()) {
+ (row, columns(join.right()), join.right())
+ } else if let Some(row) = constant_row(join.right()) {
+ (columns(join.left()), row, join.left())
+ } else {
+ return Ok(None);
+ };
+
+ let mut input = Arc::clone(input);
+ if let Some(filter) = join.filter() {
+ let predicate = Arc::clone(filter.expression())
+ .transform_up(|expr| {
+ let Some(column) = expr.downcast_ref::<Column>() else {
+ return Ok(Transformed::no(expr));
+ };
+ // Filter columns index the filter's own schema, which
+ // `column_indices` maps back to the join inputs.
+ let ColumnIndex { index, side } =
+ filter.column_indices().get(column.index()).ok_or_else(|| {
+ internal_datafusion_err!("Unmapped join filter column")
+ })?;
+ let value = match side {
+ JoinSide::Left => left.get(*index),
+ JoinSide::Right => right.get(*index),
+ JoinSide::None => None,
+ }
+ .ok_or_else(|| {
+ internal_datafusion_err!("Join filter column out of range")
+ })?;
+ Ok(Transformed::yes(Arc::clone(value)))
+ })
+ .data()?;
+ input = Arc::new(FilterExec::try_new(predicate, input)?);
+ }
+
+ let mut outputs = left.into_iter().chain(right).collect::<Vec<_>>();
+ if let Some(projection) = join.projection() {
+ outputs = projection
+ .iter()
+ .map(|&index| outputs.get(index).cloned())
+ .collect::<Option<_>>()
+ .ok_or_else(|| internal_datafusion_err!("Join projection out of range"))?;
+ }
+ let schema = join.schema();
+ let exprs = outputs
+ .into_iter()
+ .zip(schema.fields())
+ .map(|(expr, field)| ProjectionExpr {
+ expr,
+ alias: field.name().clone(),
+ });
+ // Keep the join's metadata, which may come from the constant input.
+ Ok(Some(Arc::new(
+ ProjectionExec::try_new_with_schema_metadata(exprs, input, &schema)?,
+ )))
+}
+
+/// Returns the literal values of the one row `plan` produces, if it is a stack of
+/// projections over a single-partition [`PlaceholderRowExec`] that project only
+/// literals and columns of the projection below.
+fn constant_row(plan: &Arc<dyn ExecutionPlan>) -> Option<Vec<Arc<dyn PhysicalExpr>>> {
+ let projection = plan.downcast_ref::<ProjectionExec>()?;
+ let input = projection.input();
+ let input_row = if input.is::<PlaceholderRowExec>() {
+ // A placeholder emits one row per partition, and its columns are not
+ // literals.
+ if input.output_partitioning().partition_count() != 1 {
+ return None;
+ }
+ Vec::new()
+ } else {
+ constant_row(input)?
+ };
+ projection
+ .expr()
+ .iter()
+ .map(|ProjectionExpr { expr, .. }| {
+ if expr.is::<Literal>() {
Review Comment:
Nit: `ConstEvaluator` seems to be always constant-folding constant expressions (no column references, no volatile functions), we could possibly add a comment around that, as at first sight it looked too restrictive to me, and I had to play with a few SLT tests to verify.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org