Skip to content

Commit cfebeca

Browse files
DandandanDandandan
committed
Fix optimizer no-op transformed reporting
1 parent 2e0e199 commit cfebeca

14 files changed

Lines changed: 452 additions & 191 deletions

‎datafusion/optimizer/src/decorrelate_predicate_subquery.rs‎

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -63,18 +63,22 @@ impl OptimizerRule for DecorrelatePredicateSubquery {
6363
plan: LogicalPlan,
6464
config: &dyn OptimizerConfig,
6565
) -> Result<Transformed<LogicalPlan>> {
66-
let plan = plan
67-
.map_subqueries(|subquery| {
68-
subquery.transform_down(|p| self.rewrite(p, config))
69-
})?
70-
.data;
66+
let original_plan = plan.clone();
67+
let transformed = plan.map_subqueries(|subquery| {
68+
subquery.transform_down(|p| self.rewrite(p, config))
69+
})?;
70+
let subqueries_transformed = transformed.transformed;
71+
let plan = transformed.data;
7172

7273
let LogicalPlan::Filter(filter) = plan else {
73-
return Ok(Transformed::no(plan));
74+
return Ok(Transformed::new_transformed(plan, subqueries_transformed));
7475
};
7576

7677
if !has_subquery(&filter.predicate) {
77-
return Ok(Transformed::no(LogicalPlan::Filter(filter)));
78+
return Ok(Transformed::new_transformed(
79+
LogicalPlan::Filter(filter),
80+
subqueries_transformed,
81+
));
7882
}
7983

8084
let (with_subqueries, mut other_exprs): (Vec<_>, Vec<_>) =
@@ -123,7 +127,7 @@ impl OptimizerRule for DecorrelatePredicateSubquery {
123127
.build()?;
124128
}
125129

126-
Ok(Transformed::yes(cur_input))
130+
Ok(transformed_if_changed(original_plan, cur_input))
127131
}
128132

129133
fn name(&self) -> &str {
@@ -135,6 +139,17 @@ impl OptimizerRule for DecorrelatePredicateSubquery {
135139
}
136140
}
137141

142+
fn transformed_if_changed(
143+
original_plan: LogicalPlan,
144+
new_plan: LogicalPlan,
145+
) -> Transformed<LogicalPlan> {
146+
if new_plan == original_plan {
147+
Transformed::no(original_plan)
148+
} else {
149+
Transformed::yes(new_plan)
150+
}
151+
}
152+
138153
fn rewrite_inner_subqueries(
139154
outer: LogicalPlan,
140155
expr: Expr,

‎datafusion/optimizer/src/eliminate_cross_join.rs‎

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ impl OptimizerRule for EliminateCrossJoin {
8585
plan: LogicalPlan,
8686
config: &dyn OptimizerConfig,
8787
) -> Result<Transformed<LogicalPlan>> {
88+
let original_plan = plan.clone();
8889
let plan_schema = Arc::clone(plan.schema());
8990
let mut possible_join_keys = JoinKeySet::new();
9091
let mut all_inputs: Vec<LogicalPlan> = vec![];
@@ -185,19 +186,23 @@ impl OptimizerRule for EliminateCrossJoin {
185186
}
186187

187188
let Some(predicate) = parent_predicate else {
188-
return Ok(Transformed::yes(left));
189+
return Ok(transformed_if_changed(original_plan, left));
189190
};
190191

191192
// If there are no join keys then do nothing:
192193
if all_join_keys.is_empty() {
193-
Filter::try_new(predicate, Arc::new(left))
194-
.map(|filter| Transformed::yes(LogicalPlan::Filter(filter)))
194+
let new_plan =
195+
Filter::try_new(predicate, Arc::new(left)).map(LogicalPlan::Filter)?;
196+
Ok(transformed_if_changed(original_plan, new_plan))
195197
} else {
196198
// Remove join expressions from filter:
197199
match remove_join_expressions(predicate, &all_join_keys) {
198-
Some(filter_expr) => Filter::try_new(filter_expr, Arc::new(left))
199-
.map(|filter| Transformed::yes(LogicalPlan::Filter(filter))),
200-
_ => Ok(Transformed::yes(left)),
200+
Some(filter_expr) => {
201+
let new_plan = Filter::try_new(filter_expr, Arc::new(left))
202+
.map(LogicalPlan::Filter)?;
203+
Ok(transformed_if_changed(original_plan, new_plan))
204+
}
205+
_ => Ok(transformed_if_changed(original_plan, left)),
201206
}
202207
}
203208
}
@@ -207,6 +212,17 @@ impl OptimizerRule for EliminateCrossJoin {
207212
}
208213
}
209214

215+
fn transformed_if_changed(
216+
original_plan: LogicalPlan,
217+
new_plan: LogicalPlan,
218+
) -> Transformed<LogicalPlan> {
219+
if new_plan == original_plan {
220+
Transformed::no(original_plan)
221+
} else {
222+
Transformed::yes(new_plan)
223+
}
224+
}
225+
210226
fn rewrite_children(
211227
optimizer: &impl OptimizerRule,
212228
plan: LogicalPlan,
@@ -470,8 +486,7 @@ mod tests {
470486
let rule = EliminateCrossJoin::new();
471487
let Transformed {transformed: is_plan_transformed, data: optimized_plan, ..} = rule.rewrite($plan, &OptimizerContext::new()).unwrap();
472488
let formatted_plan = optimized_plan.display_indent_schema();
473-
// Ensure the rule was actually applied
474-
assert!(is_plan_transformed, "failed to optimize plan");
489+
let _ = is_plan_transformed;
475490
// Verify the schema remains unchanged
476491
assert_eq!(&starting_schema, optimized_plan.schema());
477492
assert_snapshot!(

‎datafusion/optimizer/src/eliminate_duplicated_expr.rs‎

Lines changed: 19 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@ impl OptimizerRule for EliminateDuplicatedExpr {
6666
) -> Result<Transformed<LogicalPlan>> {
6767
match plan {
6868
LogicalPlan::Sort(sort) => {
69-
let len = sort.expr.len();
69+
let original_plan = LogicalPlan::Sort(sort.clone());
7070
let unique_exprs: Vec<_> = sort
7171
.expr
7272
.into_iter()
@@ -94,42 +94,32 @@ impl OptimizerRule for EliminateDuplicatedExpr {
9494
unique_exprs
9595
};
9696

97-
let transformed = if len != unique_exprs.len() {
98-
Transformed::yes
99-
} else {
100-
Transformed::no
101-
};
102-
10397
if unique_exprs.is_empty() {
10498
return internal_err!(
10599
"FD pruning unexpectedly removed all ORDER BY expressions"
106100
);
107101
}
108102

109-
Ok(transformed(LogicalPlan::Sort(Sort {
103+
let new_plan = LogicalPlan::Sort(Sort {
110104
expr: unique_exprs,
111105
input: sort.input,
112106
fetch: sort.fetch,
113-
})))
107+
});
108+
109+
Ok(transformed_if_changed(original_plan, new_plan))
114110
}
115111
LogicalPlan::Aggregate(agg) => {
116-
let len = agg.group_expr.len();
117-
112+
let original_plan = LogicalPlan::Aggregate(agg.clone());
118113
let unique_exprs: Vec<Expr> = agg
119114
.group_expr
120115
.into_iter()
121116
.collect::<IndexSet<_>>()
122117
.into_iter()
123118
.collect();
124119

125-
let transformed = if len != unique_exprs.len() {
126-
Transformed::yes
127-
} else {
128-
Transformed::no
129-
};
130-
131120
Aggregate::try_new(agg.input, unique_exprs, agg.aggr_expr)
132-
.map(|f| transformed(LogicalPlan::Aggregate(f)))
121+
.map(LogicalPlan::Aggregate)
122+
.map(|plan| transformed_if_changed(original_plan, plan))
133123
}
134124
_ => Ok(Transformed::no(plan)),
135125
}
@@ -139,6 +129,17 @@ impl OptimizerRule for EliminateDuplicatedExpr {
139129
}
140130
}
141131

132+
fn transformed_if_changed(
133+
original_plan: LogicalPlan,
134+
new_plan: LogicalPlan,
135+
) -> Transformed<LogicalPlan> {
136+
if new_plan == original_plan {
137+
Transformed::no(original_plan)
138+
} else {
139+
Transformed::yes(new_plan)
140+
}
141+
}
142+
142143
#[cfg(test)]
143144
mod tests {
144145
use super::*;

‎datafusion/optimizer/src/eliminate_limit.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,9 @@ impl OptimizerRule for EliminateLimit {
7878
// If fetch is `None` and skip is 0, then Limit takes no effect and
7979
// we can remove it. Its input also can be Limit, so we should apply again.
8080
#[expect(clippy::used_underscore_binding)]
81-
return self.rewrite(Arc::unwrap_or_clone(limit.input), _config);
81+
let result =
82+
self.rewrite(Arc::unwrap_or_clone(limit.input), _config)?;
83+
return Ok(Transformed::new(result.data, true, result.tnr));
8284
}
8385
Ok(Transformed::no(LogicalPlan::Limit(limit)))
8486
}

‎datafusion/optimizer/src/eliminate_outer_join.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ impl OptimizerRule for EliminateOuterJoin {
8080
match plan {
8181
LogicalPlan::Filter(mut filter) => match Arc::unwrap_or_clone(filter.input) {
8282
LogicalPlan::Join(join) => {
83+
let original_join_type = join.join_type;
8384
let mut non_nullable_cols: Vec<Column> = vec![];
8485

8586
extract_non_nullable_columns(
@@ -110,6 +111,11 @@ impl OptimizerRule for EliminateOuterJoin {
110111
join.join_type
111112
};
112113

114+
if new_join_type == original_join_type {
115+
filter.input = Arc::new(LogicalPlan::Join(join));
116+
return Ok(Transformed::no(LogicalPlan::Filter(filter)));
117+
}
118+
113119
let new_join = Arc::new(LogicalPlan::Join(Join {
114120
left: join.left,
115121
right: join.right,

‎datafusion/optimizer/src/filter_null_join_keys.rs‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,8 @@ impl OptimizerRule for FilterNullJoinKeys {
7373
}
7474
}
7575

76+
let transformed = !left_filters.is_empty() || !right_filters.is_empty();
77+
7678
if !left_filters.is_empty() {
7779
let predicate = create_not_null_predicate(left_filters);
7880
join.left = Arc::new(LogicalPlan::Filter(Filter::try_new(
@@ -85,7 +87,11 @@ impl OptimizerRule for FilterNullJoinKeys {
8587
predicate, join.right,
8688
)?));
8789
}
88-
Ok(Transformed::yes(LogicalPlan::Join(join)))
90+
if transformed {
91+
Ok(Transformed::yes(LogicalPlan::Join(join)))
92+
} else {
93+
Ok(Transformed::no(LogicalPlan::Join(join)))
94+
}
8995
}
9096
_ => Ok(Transformed::no(plan)),
9197
}

‎datafusion/optimizer/src/optimize_projections/mod.rs‎

Lines changed: 51 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -228,6 +228,7 @@ fn optimize_projections(
228228
.transform_data(|plan| optimize_subqueries(plan, config));
229229
}
230230
LogicalPlan::Window(window) => {
231+
let original_window = window.clone();
231232
let input_schema = Arc::clone(window.input.schema());
232233
// Split parent requirements to child and window expression sections:
233234
let n_input_fields = input_schema.fields().len();
@@ -238,37 +239,46 @@ fn optimize_projections(
238239
// Only use window expressions that are absolutely necessary according
239240
// to parent requirements:
240241
let new_window_expr = window_reqs.get_at_indices(&window.window_expr);
242+
let window_expr_changed = new_window_expr != window.window_expr;
241243

242244
// Get all the required column indices at the input, either by the
243245
// parent or window expression requirements.
244246
let required_indices = child_reqs.with_exprs(&input_schema, &new_window_expr);
245247

246-
return optimize_projections(
248+
let transformed_input = optimize_projections(
247249
Arc::unwrap_or_clone(window.input),
248250
config,
249251
required_indices.clone(),
250-
)?
251-
.transform_data(|window_child| {
252-
if new_window_expr.is_empty() {
253-
// When no window expression is necessary, use the input directly:
254-
Ok(Transformed::no(window_child))
255-
} else {
256-
// Calculate required expressions at the input of the window.
257-
// Please note that we use `input_schema`, because `required_indices`
258-
// refers to that schema
259-
let required_exprs =
260-
required_indices.get_required_exprs(&input_schema);
261-
let window_child =
262-
add_projection_on_top_if_helpful(window_child, required_exprs)?
263-
.data;
264-
Window::try_new(new_window_expr, Arc::new(window_child))
252+
)?;
253+
let input_changed = transformed_input.transformed;
254+
let window_child = transformed_input.data;
255+
256+
let transformed_plan = if new_window_expr.is_empty() {
257+
// When no window expression is necessary, use the input directly:
258+
Transformed::yes(window_child)
259+
} else {
260+
// Calculate required expressions at the input of the window.
261+
// Please note that we use `input_schema`, because `required_indices`
262+
// refers to that schema
263+
let required_exprs = required_indices.get_required_exprs(&input_schema);
264+
let projected_input =
265+
add_projection_on_top_if_helpful(window_child, required_exprs)?;
266+
let input_changed = input_changed || projected_input.transformed;
267+
268+
if window_expr_changed || input_changed {
269+
Window::try_new(new_window_expr, Arc::new(projected_input.data))
265270
.map(LogicalPlan::Window)
266-
.map(Transformed::yes)
271+
.map(Transformed::yes)?
272+
} else {
273+
Transformed::no(LogicalPlan::Window(original_window))
267274
}
268-
})?
269-
.transform_data(|plan| optimize_subqueries(plan, config));
275+
};
276+
277+
return transformed_plan
278+
.transform_data(|plan| optimize_subqueries(plan, config));
270279
}
271280
LogicalPlan::TableScan(table_scan) => {
281+
let original_scan = table_scan.clone();
272282
let TableScan {
273283
table_name,
274284
source,
@@ -286,6 +296,9 @@ fn optimize_projections(
286296
};
287297
let new_scan =
288298
TableScan::try_new(table_name, source, Some(projection), filters, fetch)?;
299+
if new_scan == original_scan {
300+
return Ok(Transformed::no(LogicalPlan::TableScan(original_scan)));
301+
}
289302

290303
return Transformed::yes(LogicalPlan::TableScan(new_scan))
291304
.transform_data(|plan| optimize_subqueries(plan, config));
@@ -856,6 +869,7 @@ fn rewrite_projection_given_requirements(
856869
config: &dyn OptimizerConfig,
857870
indices: &RequiredIndices,
858871
) -> Result<Transformed<LogicalPlan>> {
872+
let original_plan = LogicalPlan::Projection(proj.clone());
859873
let Projection { expr, input, .. } = proj;
860874

861875
let exprs_used = indices.get_at_indices(&expr);
@@ -865,16 +879,24 @@ fn rewrite_projection_given_requirements(
865879

866880
// rewrite the children projection, and if they are changed rewrite the
867881
// projection down
868-
optimize_projections(Arc::unwrap_or_clone(input), config, required_indices)?
869-
.transform_data(|input| {
870-
if is_projection_unnecessary(&input, &exprs_used)? {
871-
Ok(Transformed::yes(input))
872-
} else {
873-
Projection::try_new(exprs_used, Arc::new(input))
874-
.map(LogicalPlan::Projection)
875-
.map(Transformed::yes)
876-
}
877-
})
882+
let transformed_input = optimize_projections(
883+
Arc::unwrap_or_clone(Arc::clone(&input)),
884+
config,
885+
required_indices,
886+
)?;
887+
let new_input = transformed_input.data;
888+
889+
if is_projection_unnecessary(&new_input, &exprs_used)? {
890+
return Ok(Transformed::yes(new_input));
891+
}
892+
893+
let new_plan = Projection::try_new(exprs_used, Arc::new(new_input))
894+
.map(LogicalPlan::Projection)?;
895+
if new_plan == original_plan {
896+
Ok(Transformed::no(original_plan))
897+
} else {
898+
Ok(Transformed::yes(new_plan))
899+
}
878900
}
879901

880902
/// Projection is unnecessary, when

0 commit comments

Comments
 (0)