1use std::sync::Arc;
41
42use crate::logical_plan::plan::RangePartitioning;
43use crate::{
44 Aggregate, Analyze, CreateMemoryTable, CreateView, DdlStatement, Distinct,
45 DistinctOn, DmlStatement, Execute, Explain, Expr, Extension, Filter, Join, Limit,
46 LogicalPlan, Partitioning, Prepare, Projection, RecursiveQuery, Repartition, Sort,
47 Statement, Subquery, SubqueryAlias, TableScan, Union, Unnest, UserDefinedLogicalNode,
48 Values, Window, WriteOp, builder::unnest_with_options, dml::CopyTo,
49};
50use datafusion_common::tree_node::TreeNodeRefContainer;
51
52use crate::expr::{Exists, InSubquery, SetComparison};
53use datafusion_common::tree_node::{
54 Transformed, TreeNode, TreeNodeContainer, TreeNodeIterator, TreeNodeRecursion,
55 TreeNodeRewriter, TreeNodeVisitor,
56};
57use datafusion_common::{Result, internal_err};
58
59impl TreeNode for LogicalPlan {
60 fn apply_children<'n, F: FnMut(&'n Self) -> Result<TreeNodeRecursion>>(
61 &'n self,
62 f: F,
63 ) -> Result<TreeNodeRecursion> {
64 self.inputs().apply_ref_elements(f)
65 }
66
67 fn map_children<F: FnMut(Self) -> Result<Transformed<Self>>>(
76 self,
77 f: F,
78 ) -> Result<Transformed<Self>> {
79 Ok(match self {
80 LogicalPlan::Projection(Projection {
81 expr,
82 input,
83 schema,
84 }) => input.map_elements(f)?.update_data(|input| {
85 LogicalPlan::Projection(Projection {
86 expr,
87 input,
88 schema,
89 })
90 }),
91 LogicalPlan::Filter(Filter { predicate, input }) => input
92 .map_elements(f)?
93 .update_data(|input| LogicalPlan::Filter(Filter { predicate, input })),
94 LogicalPlan::Repartition(Repartition {
95 input,
96 partitioning_scheme,
97 }) => input.map_elements(f)?.update_data(|input| {
98 LogicalPlan::Repartition(Repartition {
99 input,
100 partitioning_scheme,
101 })
102 }),
103 LogicalPlan::Window(Window {
104 input,
105 window_expr,
106 schema,
107 }) => input.map_elements(f)?.update_data(|input| {
108 LogicalPlan::Window(Window {
109 input,
110 window_expr,
111 schema,
112 })
113 }),
114 LogicalPlan::Aggregate(Aggregate {
115 input,
116 group_expr,
117 aggr_expr,
118 schema,
119 }) => input.map_elements(f)?.update_data(|input| {
120 LogicalPlan::Aggregate(Aggregate {
121 input,
122 group_expr,
123 aggr_expr,
124 schema,
125 })
126 }),
127 LogicalPlan::Sort(Sort { expr, input, fetch }) => input
128 .map_elements(f)?
129 .update_data(|input| LogicalPlan::Sort(Sort { expr, input, fetch })),
130 LogicalPlan::Join(Join {
131 left,
132 right,
133 on,
134 filter,
135 join_type,
136 join_constraint,
137 schema,
138 null_equality,
139 null_aware,
140 }) => (left, right).map_elements(f)?.update_data(|(left, right)| {
141 LogicalPlan::Join(Join {
142 left,
143 right,
144 on,
145 filter,
146 join_type,
147 join_constraint,
148 schema,
149 null_equality,
150 null_aware,
151 })
152 }),
153 LogicalPlan::Limit(Limit { skip, fetch, input }) => input
154 .map_elements(f)?
155 .update_data(|input| LogicalPlan::Limit(Limit { skip, fetch, input })),
156 LogicalPlan::Subquery(Subquery {
157 subquery,
158 outer_ref_columns,
159 spans,
160 }) => subquery.map_elements(f)?.update_data(|subquery| {
161 LogicalPlan::Subquery(Subquery {
162 subquery,
163 outer_ref_columns,
164 spans,
165 })
166 }),
167 LogicalPlan::SubqueryAlias(SubqueryAlias {
168 input,
169 alias,
170 schema,
171 }) => input.map_elements(f)?.update_data(|input| {
172 LogicalPlan::SubqueryAlias(SubqueryAlias {
173 input,
174 alias,
175 schema,
176 })
177 }),
178 LogicalPlan::Extension(extension) => rewrite_extension_inputs(extension, f)?
179 .update_data(LogicalPlan::Extension),
180 LogicalPlan::Union(Union { inputs, schema }) => inputs
181 .map_elements(f)?
182 .update_data(|inputs| LogicalPlan::Union(Union { inputs, schema })),
183 LogicalPlan::Distinct(distinct) => match distinct {
184 Distinct::All(input) => input.map_elements(f)?.update_data(Distinct::All),
185 Distinct::On(DistinctOn {
186 on_expr,
187 select_expr,
188 sort_expr,
189 input,
190 schema,
191 }) => input.map_elements(f)?.update_data(|input| {
192 Distinct::On(DistinctOn {
193 on_expr,
194 select_expr,
195 sort_expr,
196 input,
197 schema,
198 })
199 }),
200 }
201 .update_data(LogicalPlan::Distinct),
202 LogicalPlan::Explain(Explain {
203 verbose,
204 explain_format: format,
205 plan,
206 stringified_plans,
207 schema,
208 logical_optimization_succeeded,
209 show_statistics,
210 }) => plan.map_elements(f)?.update_data(|plan| {
211 LogicalPlan::Explain(Explain {
212 verbose,
213 explain_format: format,
214 plan,
215 stringified_plans,
216 schema,
217 logical_optimization_succeeded,
218 show_statistics,
219 })
220 }),
221 LogicalPlan::Analyze(Analyze {
222 verbose,
223 format,
224 input,
225 schema,
226 analyze_level,
227 analyze_categories,
228 }) => input.map_elements(f)?.update_data(|input| {
229 LogicalPlan::Analyze(Analyze {
230 verbose,
231 format,
232 input,
233 schema,
234 analyze_level,
235 analyze_categories,
236 })
237 }),
238 LogicalPlan::Dml(DmlStatement {
239 table_name,
240 target,
241 op,
242 input,
243 output_schema,
244 }) => input.map_elements(f)?.update_data(|input| {
245 LogicalPlan::Dml(DmlStatement {
246 table_name,
247 target,
248 op,
249 input,
250 output_schema,
251 })
252 }),
253 LogicalPlan::Copy(CopyTo {
254 input,
255 output_url,
256 partition_by,
257 file_type,
258 options,
259 output_schema,
260 }) => input.map_elements(f)?.update_data(|input| {
261 LogicalPlan::Copy(CopyTo {
262 input,
263 output_url,
264 partition_by,
265 file_type,
266 options,
267 output_schema,
268 })
269 }),
270 LogicalPlan::Ddl(ddl) => {
271 match ddl {
272 DdlStatement::CreateMemoryTable(CreateMemoryTable {
273 name,
274 constraints,
275 input,
276 if_not_exists,
277 or_replace,
278 column_defaults,
279 temporary,
280 }) => input.map_elements(f)?.update_data(|input| {
281 DdlStatement::CreateMemoryTable(CreateMemoryTable {
282 name,
283 constraints,
284 input,
285 if_not_exists,
286 or_replace,
287 column_defaults,
288 temporary,
289 })
290 }),
291 DdlStatement::CreateView(CreateView {
292 name,
293 input,
294 or_replace,
295 definition,
296 temporary,
297 }) => input.map_elements(f)?.update_data(|input| {
298 DdlStatement::CreateView(CreateView {
299 name,
300 input,
301 or_replace,
302 definition,
303 temporary,
304 })
305 }),
306 DdlStatement::CreateExternalTable(_)
308 | DdlStatement::CreateCatalogSchema(_)
309 | DdlStatement::CreateCatalog(_)
310 | DdlStatement::CreateIndex(_)
311 | DdlStatement::DropTable(_)
312 | DdlStatement::DropView(_)
313 | DdlStatement::DropCatalogSchema(_)
314 | DdlStatement::CreateFunction(_)
315 | DdlStatement::DropFunction(_) => Transformed::no(ddl),
316 }
317 .update_data(LogicalPlan::Ddl)
318 }
319 LogicalPlan::Unnest(Unnest {
320 input,
321 exec_columns: input_columns,
322 list_type_columns,
323 struct_type_columns,
324 dependency_indices,
325 schema,
326 options,
327 }) => input.map_elements(f)?.update_data(|input| {
328 LogicalPlan::Unnest(Unnest {
329 input,
330 exec_columns: input_columns,
331 list_type_columns,
332 struct_type_columns,
333 dependency_indices,
334 schema,
335 options,
336 })
337 }),
338 LogicalPlan::RecursiveQuery(RecursiveQuery {
339 name,
340 static_term,
341 recursive_term,
342 is_distinct,
343 schema,
344 }) => (static_term, recursive_term).map_elements(f)?.update_data(
345 |(static_term, recursive_term)| {
346 LogicalPlan::RecursiveQuery(RecursiveQuery {
350 name,
351 static_term,
352 recursive_term,
353 is_distinct,
354 schema,
355 })
356 },
357 ),
358 LogicalPlan::Statement(stmt) => match stmt {
359 Statement::Prepare(p) => p
360 .input
361 .map_elements(f)?
362 .update_data(|input| Statement::Prepare(Prepare { input, ..p })),
363 _ => Transformed::no(stmt),
364 }
365 .update_data(LogicalPlan::Statement),
366 LogicalPlan::TableScan { .. }
368 | LogicalPlan::EmptyRelation { .. }
369 | LogicalPlan::Values { .. }
370 | LogicalPlan::DescribeTable(_) => Transformed::no(self),
371 })
372 }
373}
374
375fn rewrite_extension_inputs<F: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>>(
381 extension: Extension,
382 f: F,
383) -> Result<Transformed<Extension>> {
384 let Extension { node } = extension;
385
386 node.inputs()
387 .into_iter()
388 .cloned()
389 .map_until_stop_and_collect(f)?
390 .map_data(|new_inputs| {
391 let exprs = node.expressions();
392 Ok(Extension {
393 node: node.with_exprs_and_inputs(exprs, new_inputs)?,
394 })
395 })
396}
397
398macro_rules! handle_transform_recursion {
401 ($F_DOWN:expr, $F_CHILD:expr, $F_UP:expr) => {{
402 $F_DOWN?
403 .transform_children(|n| {
404 n.map_subqueries($F_CHILD)?
405 .transform_sibling(|n| n.map_children($F_CHILD))
406 })?
407 .transform_parent($F_UP)
408 }};
409}
410
411impl LogicalPlan {
412 pub fn apply_expressions<F: FnMut(&Expr) -> Result<TreeNodeRecursion>>(
419 &self,
420 mut f: F,
421 ) -> Result<TreeNodeRecursion> {
422 match self {
423 LogicalPlan::Projection(Projection { expr, .. }) => expr.apply_elements(f),
424 LogicalPlan::Values(Values { values, .. }) => values.apply_elements(f),
425 LogicalPlan::Filter(Filter { predicate, .. }) => f(predicate),
426 LogicalPlan::Repartition(Repartition {
427 partitioning_scheme,
428 ..
429 }) => match partitioning_scheme {
430 Partitioning::Hash(expr, _) | Partitioning::DistributeBy(expr) => {
431 expr.apply_elements(f)
432 }
433 Partitioning::Range(range) => range.ordering().to_vec().apply_elements(f),
434 Partitioning::RoundRobinBatch(_) => Ok(TreeNodeRecursion::Continue),
435 },
436 LogicalPlan::Window(Window { window_expr, .. }) => {
437 window_expr.apply_elements(f)
438 }
439 LogicalPlan::Aggregate(Aggregate {
440 group_expr,
441 aggr_expr,
442 ..
443 }) => (group_expr, aggr_expr).apply_ref_elements(f),
444 LogicalPlan::Join(Join { on, filter, .. }) => {
448 (on, filter).apply_ref_elements(f)
449 }
450 LogicalPlan::Sort(Sort { expr, .. }) => expr.apply_elements(f),
451 LogicalPlan::Extension(extension) => {
452 extension.node.expressions().apply_elements(f)
455 }
456 LogicalPlan::TableScan(TableScan { filters, .. }) => {
457 filters.apply_elements(f)
458 }
459 LogicalPlan::Unnest(unnest) => {
460 let exprs = unnest
461 .exec_columns
462 .iter()
463 .cloned()
464 .map(Expr::Column)
465 .collect::<Vec<_>>();
466 exprs.apply_elements(f)
467 }
468 LogicalPlan::Distinct(Distinct::On(DistinctOn {
469 on_expr,
470 select_expr,
471 sort_expr,
472 ..
473 })) => (on_expr, select_expr, sort_expr).apply_ref_elements(f),
474 LogicalPlan::Limit(Limit { skip, fetch, .. }) => {
475 (skip, fetch).apply_ref_elements(f)
476 }
477 LogicalPlan::Statement(stmt) => match stmt {
478 Statement::Execute(Execute { parameters, .. }) => {
479 parameters.apply_elements(f)
480 }
481 _ => Ok(TreeNodeRecursion::Continue),
482 },
483 LogicalPlan::Dml(DmlStatement {
484 op: WriteOp::MergeInto(merge_op),
485 ..
486 }) => merge_op.exprs().apply_ref_elements(f),
487 LogicalPlan::EmptyRelation(_)
489 | LogicalPlan::RecursiveQuery(_)
490 | LogicalPlan::Subquery(_)
491 | LogicalPlan::SubqueryAlias(_)
492 | LogicalPlan::Analyze(_)
493 | LogicalPlan::Explain(_)
494 | LogicalPlan::Union(_)
495 | LogicalPlan::Distinct(Distinct::All(_))
496 | LogicalPlan::Dml(_)
497 | LogicalPlan::Ddl(_)
498 | LogicalPlan::Copy(_)
499 | LogicalPlan::DescribeTable(_) => Ok(TreeNodeRecursion::Continue),
500 }
501 }
502
503 pub fn map_expressions<F: FnMut(Expr) -> Result<Transformed<Expr>>>(
511 self,
512 mut f: F,
513 ) -> Result<Transformed<Self>> {
514 Ok(match self {
515 LogicalPlan::Projection(Projection {
516 expr,
517 input,
518 schema,
519 }) => expr.map_elements(f)?.update_data(|expr| {
520 LogicalPlan::Projection(Projection {
521 expr,
522 input,
523 schema,
524 })
525 }),
526 LogicalPlan::Values(Values { schema, values }) => values
527 .map_elements(f)?
528 .update_data(|values| LogicalPlan::Values(Values { schema, values })),
529 LogicalPlan::Filter(Filter { predicate, input }) => f(predicate)?
530 .update_data(|predicate| {
531 LogicalPlan::Filter(Filter { predicate, input })
532 }),
533 LogicalPlan::Repartition(Repartition {
534 input,
535 partitioning_scheme,
536 }) => match partitioning_scheme {
537 Partitioning::Hash(expr, usize) => expr
538 .map_elements(f)?
539 .update_data(|expr| Partitioning::Hash(expr, usize)),
540 Partitioning::DistributeBy(expr) => expr
541 .map_elements(f)?
542 .update_data(Partitioning::DistributeBy),
543 Partitioning::Range(range) => {
544 let split_points = range.split_points().to_vec();
545 range
546 .ordering()
547 .to_vec()
548 .map_elements(f)?
549 .map_data(|ordering| {
550 Ok(Partitioning::Range(RangePartitioning::try_new(
551 ordering,
552 split_points,
553 )?))
554 })?
555 }
556 Partitioning::RoundRobinBatch(_) => Transformed::no(partitioning_scheme),
557 }
558 .update_data(|partitioning_scheme| {
559 LogicalPlan::Repartition(Repartition {
560 input,
561 partitioning_scheme,
562 })
563 }),
564 LogicalPlan::Window(Window {
565 input,
566 window_expr,
567 schema,
568 }) => window_expr.map_elements(f)?.update_data(|window_expr| {
569 LogicalPlan::Window(Window {
570 input,
571 window_expr,
572 schema,
573 })
574 }),
575 LogicalPlan::Aggregate(Aggregate {
576 input,
577 group_expr,
578 aggr_expr,
579 schema,
580 }) => (group_expr, aggr_expr).map_elements(f)?.update_data(
581 |(group_expr, aggr_expr)| {
582 LogicalPlan::Aggregate(Aggregate {
583 input,
584 group_expr,
585 aggr_expr,
586 schema,
587 })
588 },
589 ),
590
591 LogicalPlan::Join(Join {
595 left,
596 right,
597 on,
598 filter,
599 join_type,
600 join_constraint,
601 schema,
602 null_equality,
603 null_aware,
604 }) => (on, filter).map_elements(f)?.update_data(|(on, filter)| {
605 LogicalPlan::Join(Join {
606 left,
607 right,
608 on,
609 filter,
610 join_type,
611 join_constraint,
612 schema,
613 null_equality,
614 null_aware,
615 })
616 }),
617 LogicalPlan::Sort(Sort { expr, input, fetch }) => expr
618 .map_elements(f)?
619 .update_data(|expr| LogicalPlan::Sort(Sort { expr, input, fetch })),
620 LogicalPlan::Extension(Extension { node }) => {
621 let raw_exprs = node.expressions();
622 if raw_exprs.is_empty() {
623 Transformed::no(LogicalPlan::Extension(Extension { node }))
626 } else {
627 let exprs = raw_exprs.map_elements(f)?;
635 let plan = LogicalPlan::Extension(Extension {
636 node: UserDefinedLogicalNode::with_exprs_and_inputs(
637 node.as_ref(),
638 exprs.data,
639 node.inputs().into_iter().cloned().collect::<Vec<_>>(),
640 )?,
641 });
642 Transformed::new(plan, exprs.transformed, exprs.tnr)
643 }
644 }
645 LogicalPlan::TableScan(TableScan {
646 table_name,
647 source,
648 projection,
649 projected_schema,
650 filters,
651 fetch,
652 statistics_requests,
653 }) => filters.map_elements(f)?.update_data(|filters| {
654 LogicalPlan::TableScan(TableScan {
655 table_name,
656 source,
657 projection,
658 projected_schema,
659 filters,
660 fetch,
661 statistics_requests,
662 })
663 }),
664 LogicalPlan::Distinct(Distinct::On(DistinctOn {
665 on_expr,
666 select_expr,
667 sort_expr,
668 input,
669 schema,
670 })) => (on_expr, select_expr, sort_expr)
671 .map_elements(f)?
672 .update_data(|(on_expr, select_expr, sort_expr)| {
673 LogicalPlan::Distinct(Distinct::On(DistinctOn {
674 on_expr,
675 select_expr,
676 sort_expr,
677 input,
678 schema,
679 }))
680 }),
681 LogicalPlan::Limit(Limit { skip, fetch, input }) => {
682 (skip, fetch).map_elements(f)?.update_data(|(skip, fetch)| {
683 LogicalPlan::Limit(Limit { skip, fetch, input })
684 })
685 }
686 LogicalPlan::Statement(stmt) => match stmt {
687 Statement::Execute(e) => {
688 e.parameters.map_elements(f)?.update_data(|parameters| {
689 Statement::Execute(Execute { parameters, ..e })
690 })
691 }
692 _ => Transformed::no(stmt),
693 }
694 .update_data(LogicalPlan::Statement),
695 LogicalPlan::Unnest(Unnest {
696 input,
697 exec_columns,
698 options,
699 ..
700 }) => {
701 let exprs: Vec<Expr> =
702 exec_columns.into_iter().map(Expr::Column).collect();
703 exprs.map_elements(f)?.map_data(|mapped_exprs| {
704 let new_columns = mapped_exprs
705 .into_iter()
706 .map(|e| match e {
707 Expr::Column(c) => Ok(c),
708 other => internal_err!(
709 "Expected Expr::Column for Unnest exec_columns, got {other:?}"
710 ),
711 })
712 .collect::<Result<Vec<_>>>()?;
713 unnest_with_options(
720 Arc::unwrap_or_clone(input),
721 new_columns,
722 options,
723 )
724 })?
725 }
726 LogicalPlan::Dml(DmlStatement {
727 table_name,
728 target,
729 op: WriteOp::MergeInto(merge_op),
730 input,
731 output_schema,
732 }) => {
733 let owned_exprs: Vec<Expr> =
734 merge_op.exprs().into_iter().cloned().collect();
735 owned_exprs.map_elements(f)?.transform_data(|new_exprs| {
736 Ok(Transformed::no(LogicalPlan::Dml(DmlStatement {
737 table_name,
738 target,
739 op: WriteOp::MergeInto(Box::new(
740 merge_op.with_new_exprs(new_exprs)?,
741 )),
742 input,
743 output_schema,
744 })))
745 })?
746 }
747 LogicalPlan::EmptyRelation(_)
749 | LogicalPlan::RecursiveQuery(_)
750 | LogicalPlan::Subquery(_)
751 | LogicalPlan::SubqueryAlias(_)
752 | LogicalPlan::Analyze(_)
753 | LogicalPlan::Explain(_)
754 | LogicalPlan::Union(_)
755 | LogicalPlan::Distinct(Distinct::All(_))
756 | LogicalPlan::Dml(_)
757 | LogicalPlan::Ddl(_)
758 | LogicalPlan::Copy(_)
759 | LogicalPlan::DescribeTable(_) => Transformed::no(self),
760 })
761 }
762
763 #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
766 pub fn visit_with_subqueries<V: for<'n> TreeNodeVisitor<'n, Node = Self>>(
767 &self,
768 visitor: &mut V,
769 ) -> Result<TreeNodeRecursion> {
770 visitor
771 .f_down(self)?
772 .visit_children(|| {
773 self.apply_subqueries(|c| c.visit_with_subqueries(visitor))?
774 .visit_sibling(|| {
775 self.apply_children(|c| c.visit_with_subqueries(visitor))
776 })
777 })?
778 .visit_parent(|| visitor.f_up(self))
779 }
780
781 #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
785 pub fn rewrite_with_subqueries<R: TreeNodeRewriter<Node = Self>>(
786 self,
787 rewriter: &mut R,
788 ) -> Result<Transformed<Self>> {
789 handle_transform_recursion!(
790 rewriter.f_down(self),
791 |c| c.rewrite_with_subqueries(rewriter),
792 |n| rewriter.f_up(n)
793 )
794 }
795
796 pub fn apply_with_subqueries<F: FnMut(&Self) -> Result<TreeNodeRecursion>>(
800 &self,
801 mut f: F,
802 ) -> Result<TreeNodeRecursion> {
803 #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
804 fn apply_with_subqueries_impl<
805 F: FnMut(&LogicalPlan) -> Result<TreeNodeRecursion>,
806 >(
807 node: &LogicalPlan,
808 f: &mut F,
809 ) -> Result<TreeNodeRecursion> {
810 f(node)?.visit_children(|| {
811 node.apply_subqueries(|c| apply_with_subqueries_impl(c, f))?
812 .visit_sibling(|| {
813 node.apply_children(|c| apply_with_subqueries_impl(c, f))
814 })
815 })
816 }
817
818 apply_with_subqueries_impl(self, &mut f)
819 }
820
821 pub fn transform_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
825 self,
826 f: F,
827 ) -> Result<Transformed<Self>> {
828 self.transform_up_with_subqueries(f)
829 }
830
831 pub fn transform_down_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
835 self,
836 mut f: F,
837 ) -> Result<Transformed<Self>> {
838 #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
839 fn transform_down_with_subqueries_impl<
840 F: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
841 >(
842 node: LogicalPlan,
843 f: &mut F,
844 ) -> Result<Transformed<LogicalPlan>> {
845 f(node)?.transform_children(|n| {
846 n.map_subqueries(|c| transform_down_with_subqueries_impl(c, f))?
847 .transform_sibling(|n| {
848 n.map_children(|c| transform_down_with_subqueries_impl(c, f))
849 })
850 })
851 }
852
853 transform_down_with_subqueries_impl(self, &mut f)
854 }
855
856 pub fn transform_up_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
860 self,
861 mut f: F,
862 ) -> Result<Transformed<Self>> {
863 #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
864 fn transform_up_with_subqueries_impl<
865 F: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
866 >(
867 node: LogicalPlan,
868 f: &mut F,
869 ) -> Result<Transformed<LogicalPlan>> {
870 node.map_subqueries(|c| transform_up_with_subqueries_impl(c, f))?
871 .transform_sibling(|n| {
872 n.map_children(|c| transform_up_with_subqueries_impl(c, f))
873 })?
874 .transform_parent(f)
875 }
876
877 transform_up_with_subqueries_impl(self, &mut f)
878 }
879
880 pub fn transform_down_up_with_subqueries<
884 FD: FnMut(Self) -> Result<Transformed<Self>>,
885 FU: FnMut(Self) -> Result<Transformed<Self>>,
886 >(
887 self,
888 mut f_down: FD,
889 mut f_up: FU,
890 ) -> Result<Transformed<Self>> {
891 #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
892 fn transform_down_up_with_subqueries_impl<
893 FD: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
894 FU: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
895 >(
896 node: LogicalPlan,
897 f_down: &mut FD,
898 f_up: &mut FU,
899 ) -> Result<Transformed<LogicalPlan>> {
900 handle_transform_recursion!(
901 f_down(node),
902 |c| transform_down_up_with_subqueries_impl(c, f_down, f_up),
903 f_up
904 )
905 }
906
907 transform_down_up_with_subqueries_impl(self, &mut f_down, &mut f_up)
908 }
909
910 pub fn apply_subqueries<F: FnMut(&Self) -> Result<TreeNodeRecursion>>(
914 &self,
915 mut f: F,
916 ) -> Result<TreeNodeRecursion> {
917 self.apply_expressions(|expr| {
918 expr.apply(|expr| match expr {
919 Expr::Exists(Exists { subquery, .. })
920 | Expr::InSubquery(InSubquery { subquery, .. })
921 | Expr::SetComparison(SetComparison { subquery, .. })
922 | Expr::ScalarSubquery(subquery) => {
923 f(&LogicalPlan::Subquery(subquery.clone()))
925 }
926 _ => Ok(TreeNodeRecursion::Continue),
927 })
928 })
929 }
930
931 fn has_subquery_expressions(&self) -> bool {
934 let mut found = false;
935 let _ = self.apply_expressions(|expr| {
936 if found {
937 return Ok(TreeNodeRecursion::Stop);
938 }
939 expr.apply(|e| {
940 if matches!(
941 e,
942 Expr::Exists(_)
943 | Expr::InSubquery(_)
944 | Expr::SetComparison(_)
945 | Expr::ScalarSubquery(_)
946 ) {
947 found = true;
948 Ok(TreeNodeRecursion::Stop)
949 } else {
950 Ok(TreeNodeRecursion::Continue)
951 }
952 })
953 });
954 found
955 }
956
957 pub fn map_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
962 self,
963 mut f: F,
964 ) -> Result<Transformed<Self>> {
965 if !self.has_subquery_expressions() {
970 return Ok(Transformed::no(self));
971 }
972
973 self.map_expressions(|expr| {
974 expr.transform_down(|expr| match expr {
975 Expr::Exists(Exists { subquery, negated }) => {
976 f(LogicalPlan::Subquery(subquery))?.map_data(|s| match s {
977 LogicalPlan::Subquery(subquery) => {
978 Ok(Expr::Exists(Exists { subquery, negated }))
979 }
980 _ => internal_err!("Transformation should return Subquery"),
981 })
982 }
983 Expr::InSubquery(InSubquery {
984 expr,
985 subquery,
986 negated,
987 }) => f(LogicalPlan::Subquery(subquery))?.map_data(|s| match s {
988 LogicalPlan::Subquery(subquery) => Ok(Expr::InSubquery(InSubquery {
989 expr,
990 subquery,
991 negated,
992 })),
993 _ => internal_err!("Transformation should return Subquery"),
994 }),
995 Expr::SetComparison(SetComparison {
996 expr,
997 subquery,
998 op,
999 quantifier,
1000 }) => f(LogicalPlan::Subquery(subquery))?.map_data(|s| match s {
1001 LogicalPlan::Subquery(subquery) => {
1002 Ok(Expr::SetComparison(SetComparison {
1003 expr,
1004 subquery,
1005 op,
1006 quantifier,
1007 }))
1008 }
1009 _ => internal_err!("Transformation should return Subquery"),
1010 }),
1011 Expr::ScalarSubquery(subquery) => f(LogicalPlan::Subquery(subquery))?
1012 .map_data(|s| match s {
1013 LogicalPlan::Subquery(subquery) => {
1014 Ok(Expr::ScalarSubquery(subquery))
1015 }
1016 _ => internal_err!("Transformation should return Subquery"),
1017 }),
1018 _ => Ok(Transformed::no(expr)),
1019 })
1020 })
1021 }
1022
1023 pub fn map_uncorrelated_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
1026 self,
1027 mut f: F,
1028 ) -> Result<Transformed<Self>> {
1029 self.map_subqueries(|subquery_plan| match &subquery_plan {
1030 LogicalPlan::Subquery(sq) if sq.outer_ref_columns.is_empty() => {
1031 f(subquery_plan)
1032 }
1033 _ => Ok(Transformed::no(subquery_plan)),
1034 })
1035 }
1036}