Skip to content

Commit b9fa393

Browse files
xudong963xudong963
committed
feat: rewrite cross joins and folded expressions against a constant row
- Fold the constant side's expressions at planning time, so values over a statistics-folded aggregate, such as `min(ts) + 1`, are recognized. Volatile expressions and ones that fail to evaluate are still rejected. - Rewrite `CrossJoinExec` against a constant row too. It drops input ordering and may buffer the other input as its build side. - Show in an EXPLAIN that a left join against a constant row stays a join.
1 parent 0981f5b commit b9fa393

4 files changed

Lines changed: 123 additions & 34 deletions

File tree

‎datafusion/physical-optimizer/src/eliminate_constant_join.rs‎

Lines changed: 68 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -19,25 +19,26 @@
1919
2020
use std::sync::Arc;
2121

22+
use arrow::datatypes::SchemaRef;
2223
use datafusion_common::Result;
2324
use datafusion_common::config::ConfigOptions;
2425
use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
2526
use datafusion_common::{JoinSide, JoinType, internal_datafusion_err};
26-
use datafusion_physical_expr::PhysicalExpr;
2727
use datafusion_physical_expr::expressions::{Column, LambdaExpr, Literal};
28+
use datafusion_physical_expr::{PhysicalExpr, PhysicalExprSimplifier};
2829
use datafusion_physical_plan::filter::FilterExec;
29-
use datafusion_physical_plan::joins::NestedLoopJoinExec;
30-
use datafusion_physical_plan::joins::utils::ColumnIndex;
30+
use datafusion_physical_plan::joins::utils::{ColumnIndex, JoinFilter};
31+
use datafusion_physical_plan::joins::{CrossJoinExec, NestedLoopJoinExec};
3132
use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
3233
use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
3334
use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
3435

3536
use crate::PhysicalOptimizerRule;
3637

37-
/// Rewrites an inner [`NestedLoopJoinExec`] whose input provably produces one row
38-
/// of literals into a [`FilterExec`] over the other input, with the literals
39-
/// substituted into the join filter, under a [`ProjectionExec`] that restores the
40-
/// join's output.
38+
/// Rewrites an inner [`NestedLoopJoinExec`] or a [`CrossJoinExec`] whose input
39+
/// provably produces one row of literals into a [`FilterExec`] over the other
40+
/// input, with the literals substituted into the join filter, under a
41+
/// [`ProjectionExec`] that restores the join's output.
4142
///
4243
/// For example, [`AggregateStatistics`] folds the scalar subquery in
4344
///
@@ -70,10 +71,15 @@ impl PhysicalOptimizerRule for EliminateConstantJoin {
7071
_config: &ConfigOptions,
7172
) -> Result<Arc<dyn ExecutionPlan>> {
7273
plan.transform_up(|plan| {
73-
let Some(join) = plan.downcast_ref::<NestedLoopJoinExec>() else {
74-
return Ok(Transformed::no(plan));
74+
let rewritten = if let Some(join) = plan.downcast_ref::<NestedLoopJoinExec>()
75+
{
76+
eliminate_nested_loop_join(join)?
77+
} else if let Some(join) = plan.downcast_ref::<CrossJoinExec>() {
78+
eliminate(join.left(), join.right(), None, None, &join.schema())?
79+
} else {
80+
None
7581
};
76-
Ok(match eliminate(join)? {
82+
Ok(match rewritten {
7783
Some(rewritten) => Transformed::yes(rewritten),
7884
None => Transformed::no(plan),
7985
})
@@ -90,8 +96,11 @@ impl PhysicalOptimizerRule for EliminateConstantJoin {
9096
}
9197
}
9298

93-
/// Rewrites `join` over its non-constant input, if the other input is a constant row.
94-
fn eliminate(join: &NestedLoopJoinExec) -> Result<Option<Arc<dyn ExecutionPlan>>> {
99+
/// Rewrites `join` if it is an inner join against a constant row and its filter
100+
/// has no lambdas.
101+
fn eliminate_nested_loop_join(
102+
join: &NestedLoopJoinExec,
103+
) -> Result<Option<Arc<dyn ExecutionPlan>>> {
95104
if join.join_type() != &JoinType::Inner {
96105
return Ok(None);
97106
}
@@ -104,17 +113,35 @@ fn eliminate(join: &NestedLoopJoinExec) -> Result<Option<Arc<dyn ExecutionPlan>>
104113
{
105114
return Ok(None);
106115
}
116+
eliminate(
117+
join.left(),
118+
join.right(),
119+
join.filter(),
120+
join.projection().as_deref(),
121+
&join.schema(),
122+
)
123+
}
124+
125+
/// Rewrites an inner join of `left_input` and `right_input` over its non-constant
126+
/// input, if the other input is a constant row.
127+
fn eliminate(
128+
left_input: &Arc<dyn ExecutionPlan>,
129+
right_input: &Arc<dyn ExecutionPlan>,
130+
filter: Option<&JoinFilter>,
131+
projection: Option<&[usize]>,
132+
schema: &SchemaRef,
133+
) -> Result<Option<Arc<dyn ExecutionPlan>>> {
107134
// Each input's columns as expressions over the non-constant input.
108-
let (left, right, input) = if let Some(row) = constant_row(join.left()) {
109-
(row, columns(join.right()), join.right())
110-
} else if let Some(row) = constant_row(join.right()) {
111-
(columns(join.left()), row, join.left())
135+
let (left, right, input) = if let Some(row) = constant_row(left_input) {
136+
(row, columns(right_input), right_input)
137+
} else if let Some(row) = constant_row(right_input) {
138+
(columns(left_input), row, left_input)
112139
} else {
113140
return Ok(None);
114141
};
115142

116143
let mut input = Arc::clone(input);
117-
if let Some(filter) = join.filter() {
144+
if let Some(filter) = filter {
118145
let predicate = Arc::clone(filter.expression())
119146
.transform_up(|expr| {
120147
let Some(column) = expr.downcast_ref::<Column>() else {
@@ -141,14 +168,13 @@ fn eliminate(join: &NestedLoopJoinExec) -> Result<Option<Arc<dyn ExecutionPlan>>
141168
}
142169

143170
let mut outputs = left.into_iter().chain(right).collect::<Vec<_>>();
144-
if let Some(projection) = join.projection() {
171+
if let Some(projection) = projection {
145172
outputs = projection
146173
.iter()
147174
.map(|&index| outputs.get(index).cloned())
148175
.collect::<Option<_>>()
149176
.ok_or_else(|| internal_datafusion_err!("Join projection out of range"))?;
150177
}
151-
let schema = join.schema();
152178
let exprs = outputs
153179
.into_iter()
154180
.zip(schema.fields())
@@ -158,13 +184,13 @@ fn eliminate(join: &NestedLoopJoinExec) -> Result<Option<Arc<dyn ExecutionPlan>>
158184
});
159185
// Keep the join's metadata, which may come from the constant input.
160186
Ok(Some(Arc::new(
161-
ProjectionExec::try_new_with_schema_metadata(exprs, input, &schema)?,
187+
ProjectionExec::try_new_with_schema_metadata(exprs, input, schema)?,
162188
)))
163189
}
164190

165191
/// Returns the literal values of the one row `plan` produces, if it is a stack of
166-
/// projections over a single-partition [`PlaceholderRowExec`] that project only
167-
/// literals and columns of the projection below.
192+
/// projections over a single-partition [`PlaceholderRowExec`] whose expressions
193+
/// evaluate to literals at planning time.
168194
fn constant_row(plan: &Arc<dyn ExecutionPlan>) -> Option<Vec<Arc<dyn PhysicalExpr>>> {
169195
let projection = plan.downcast_ref::<ProjectionExec>()?;
170196
let input = projection.input();
@@ -178,15 +204,30 @@ fn constant_row(plan: &Arc<dyn ExecutionPlan>) -> Option<Vec<Arc<dyn PhysicalExp
178204
} else {
179205
constant_row(input)?
180206
};
207+
// The logical simplifier already folds constant expressions, but those over a
208+
// value `AggregateStatistics` folded, such as `min(ts) + 1`, are only constant
209+
// now. Volatile expressions and ones that fail to evaluate are not folded, and
210+
// are rejected: the filter would evaluate them for every row instead of once.
211+
let schema = input.schema();
212+
let simplifier = PhysicalExprSimplifier::new(&schema);
181213
projection
182214
.expr()
183215
.iter()
184216
.map(|ProjectionExpr { expr, .. }| {
185-
if expr.is::<Literal>() {
186-
return Some(Arc::clone(expr));
187-
}
188-
let column = expr.downcast_ref::<Column>()?;
189-
input_row.get(column.index()).cloned()
217+
let expr = Arc::clone(expr)
218+
.transform_up(|expr| {
219+
let value = expr
220+
.downcast_ref::<Column>()
221+
.and_then(|column| input_row.get(column.index()));
222+
Ok(match value {
223+
Some(value) => Transformed::yes(Arc::clone(value)),
224+
None => Transformed::no(expr),
225+
})
226+
})
227+
.data()
228+
.and_then(|expr| simplifier.simplify(expr))
229+
.ok()?;
230+
expr.is::<Literal>().then_some(expr)
190231
})
191232
.collect()
192233
}

‎datafusion/sqllogictest/test_files/cte.slt‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1155,10 +1155,8 @@ logical_plan
11551155
04)----Projection: Int64(1)
11561156
05)------EmptyRelation: rows=1
11571157
physical_plan
1158-
01)CrossJoinExec
1158+
01)ProjectionExec: expr=[a@0 as a, 1 as Int64(1)]
11591159
02)--DataSourceExec: partitions=1, partition_sizes=[0]
1160-
03)--ProjectionExec: expr=[1 as Int64(1)]
1161-
04)----PlaceholderRowExec
11621160

11631161
statement count 0
11641162
drop table j1;

‎datafusion/sqllogictest/test_files/eliminate_constant_join.slt‎

Lines changed: 53 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,46 @@ a 1
7575
b 1
7676
c 1
7777

78+
# Expressions over a value folded from statistics are evaluated at planning time
79+
query TT
80+
EXPLAIN SELECT e.acct
81+
FROM events e CROSS JOIN (SELECT min(ts) + 1 AS cutoff FROM live) l
82+
WHERE e.t < l.cutoff;
83+
----
84+
physical_plan
85+
01)FilterExec: 26 > t@1, projection=[acct@0]
86+
02)--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=26 > t@1, pruning_predicate=t_null_count@1 != row_count@2 AND t_min@0 < 26, required_guarantees=[]
87+
88+
query T rowsort
89+
SELECT e.acct
90+
FROM events e CROSS JOIN (SELECT min(ts) + 1 AS cutoff FROM live) l
91+
WHERE e.t < l.cutoff;
92+
----
93+
a
94+
b
95+
c
96+
97+
# A cross join with a constant row keeps the other input's ordering
98+
query TT
99+
EXPLAIN SELECT e.acct, l.first_ts, count(*) AS n
100+
FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
101+
GROUP BY e.acct, l.first_ts;
102+
----
103+
physical_plan
104+
01)ProjectionExec: expr=[acct@0 as acct, first_ts@1 as first_ts, count(Int64(1))@2 as n]
105+
02)--AggregateExec: mode=Single, gby=[acct@0 as acct, first_ts@1 as first_ts], aggr=[count(Int64(1))], group_clustering_mode=Full
106+
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/eliminate_constant_join/events.parquet]]}, projection=[acct, 25 as first_ts], output_ordering=[acct@0 ASC NULLS LAST], file_type=parquet
107+
108+
query TII
109+
SELECT e.acct, l.first_ts, count(*) AS n
110+
FROM events e CROSS JOIN (SELECT min(ts) AS first_ts FROM live) l
111+
GROUP BY e.acct, l.first_ts
112+
ORDER BY e.acct;
113+
----
114+
a 25 2
115+
b 25 2
116+
c 25 1
117+
78118
# The constant row's columns stay in the output
79119
query TT
80120
EXPLAIN SELECT e.acct, e.t, c.cutoff
@@ -93,7 +133,19 @@ a 10 25
93133
b 20 25
94134
c 5 25
95135

96-
# A left join keeps the constant row when nothing matches
136+
# A left join keeps the constant row when nothing matches, so it stays a join
137+
query TT
138+
EXPLAIN SELECT c.cutoff, e.acct
139+
FROM (SELECT 25 AS cutoff) c LEFT JOIN events e ON e.t > c.cutoff + 100;
140+
----
141+
physical_plan
142+
01)ProjectionExec: expr=[cutoff@0 as cutoff, acct@1 as acct]
143+
02)--NestedLoopJoinExec: join_type=Left, filter=t@0 > join_proj_push_down_1@1, projection=[cutoff@0, acct@2, t@3]
144+
03)----ProjectionExec: expr=[cutoff@0 as cutoff, cutoff@0 + 100 as join_proj_push_down_1]
145+
04)------ProjectionExec: expr=[25 as cutoff]
146+
05)--------PlaceholderRowExec
147+
06)----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
148+
97149
query IT
98150
SELECT c.cutoff, e.acct
99151
FROM (SELECT 25 AS cutoff) c LEFT JOIN events e ON e.t > c.cutoff + 100;

‎datafusion/sqllogictest/test_files/joins.slt‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4843,10 +4843,8 @@ logical_plan
48434843
04)----Projection: Int64(1)
48444844
05)------EmptyRelation: rows=1
48454845
physical_plan
4846-
01)CrossJoinExec
4846+
01)ProjectionExec: expr=[j1_string@0 as j1_string, j1_id@1 as j1_id, 1 as Int64(1)]
48474847
02)--DataSourceExec: partitions=1, partition_sizes=[0]
4848-
03)--ProjectionExec: expr=[1 as Int64(1)]
4849-
04)----PlaceholderRowExec
48504848

48514849
statement count 0
48524850
drop table j1;

0 commit comments

Comments
 (0)