Projection Pushdown in PhysicalPlan #8073

berkaysynnada · 2023-11-07T08:12:10Z

Which issue does this PR close?

Closes #.

Rationale for this change

Pushing down the ProjectionExec is generally beneficial for execution. Hence, whenever feasible and advantageous, we should aim to swap it with its input. This PR introduces a rule for this purpose. While a similar rule exists in the logical planning stage, some cases may emerge for further optimization after some optimizer rules have worked.

What changes are included in this PR?

This PR introduces a PhysicalOptimizerRule ProjectionPushdown implemented at the final optimization step. The rule initially checks if the operation is a ProjectionExec. If it is, the rule attempts to eliminate it if it's redundant. If not, it examines the input of the projection. Each operator has its own conditions for swapping with a projection. If the conditions are satisfied, the plan tree Projection <-- X <-- Y evolves to X <-- Projection <-- Y. Two projections can always be combined into one, and in some scenarios, projections can be removed from the plan if they can be propagated to the source providers.

Are these changes tested?

Yes, unit tests have been added to cover each operator. Additionally, various .slt test changes show successful optimizations. Benchmark results are as follows:

┏━━━━━━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃     main ┃ feature_optimize_projections ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │ 421.93ms │                     421.60ms │     no change │
│ QQuery 2     │  59.86ms │                      59.83ms │     no change │
│ QQuery 3     │ 141.49ms │                     140.42ms │     no change │
│ QQuery 4     │  89.99ms │                      88.31ms │     no change │
│ QQuery 5     │ 173.94ms │                     171.95ms │     no change │
│ QQuery 6     │  84.61ms │                      84.30ms │     no change │
│ QQuery 7     │ 265.81ms │                     251.49ms │ +1.06x faster │
│ QQuery 8     │ 203.04ms │                     200.24ms │     no change │
│ QQuery 9     │ 285.64ms │                     285.41ms │     no change │
│ QQuery 10    │ 224.73ms │                     220.80ms │     no change │
│ QQuery 11    │  57.97ms │                      58.64ms │     no change │
│ QQuery 12    │ 158.46ms │                     158.17ms │     no change │
│ QQuery 13    │ 231.70ms │                     221.08ms │     no change │
│ QQuery 14    │ 119.61ms │                     119.42ms │     no change │
│ QQuery 15    │  82.82ms │                      83.33ms │     no change │
│ QQuery 16    │  51.07ms │                      50.90ms │     no change │
│ QQuery 17    │ 222.45ms │                     204.10ms │ +1.09x faster │
│ QQuery 18    │ 428.66ms │                     436.40ms │     no change │
│ QQuery 19    │ 239.63ms │                     241.73ms │     no change │
│ QQuery 20    │ 137.80ms │                     141.81ms │     no change │
│ QQuery 21    │ 354.14ms │                     362.21ms │     no change │
│ QQuery 22    │  52.01ms │                      53.07ms │     no change │
└──────────────┴──────────┴──────────────────────────────┴───────────────┘

Are there any user-facing changes?

ozankabak

I have collaborated with @berkaysynnada and reviewed this PR carefully over the last week. Almost all the changes are within a new file that implements the rule (projection_pushdown.rs), so it should be an easy review.

alamb · 2023-11-07T13:00:59Z

@crepererum implemented something similar to this in IOx -- can you please review this PR as well @crepererum ? Maybe we can contribute some of IOx's tests cases back upstream?

crepererum · 2023-11-07T16:22:33Z

datafusion/core/src/physical_optimizer/projection_pushdown.rs

+            try_swapping_with_nested_loop_join(projection, nl_join)?
+        } else if let Some(sm_join) = input.downcast_ref::<SortMergeJoinExec>() {
+            try_swapping_with_sort_merge_join(projection, sm_join)?
+        } else if let Some(sym_join) = input.downcast_ref::<SymmetricHashJoinExec>() {


could we make this registry-based so that custom execs could also profit from this pass?

Perhaps we can make a method on ExecutionPlan and then add the relevant methods to each impl ExecutionPlan, similar to what I did in #7936

We like this idea and considered how we can do this, but didn't see an obvious design to follow. Any suggestions on how we can do this? Also, do you think we should get the functionality in first and do the refactor as a follow-on PR, or should we incorporate it in this one?

Maybe we could add a function like this to ExecutionPlan:

trait ExecutionPlan { ... /// Tries to push a projection of the output of this ExecutionPlan /// *into* itself input by rewriting the internal expressions. /// /// For example, /// (TODO EXAMPLE HERE /// /// If the ExecutionPlan does not support pushdown, ,returns Ok(None) (the default) fn push_projection(&self, projection: &[(Arc<dyn PhysicalExpr>, String)]) -> Result<Option<Arc<dyn ExecutionPlan>>>) { Ok(None) } }

So I guess this whole problem is:

How is a set of optimizer passes linked to a set of nodes, while both sets are extendible?

I see the following rough solutions:

A: omniscient optimizer

The optimizer knows all node types. This is what this PR does (and what many other passes do).

This doesn't scale well.

B: omniscient nodes

The nodes know all optimizer passes and implement them themselves. This kinda sounds like what @alamb proposes.

This doesn't scale well.

C: registry-based linking

Developers are aware of the which nodes can be optimized in which way and can fill out gaps in the optimizer-node matrix. The issue is mostly how this registry should be implemented. Rust has a bunch of crates for that that are all not great (due to the issue of the initialization order). Luckily we all know which node types are in a plan (because you could traverse the plan) so we could hook registry initialization in there. Something like:

trait ExecutionPlan { fn register_hook_for_optimizer_pass(....); }

D: abstraction

This is what most other optimizer passes do: they read some abstract property of the node (like "schema", "num children", "output ordering", ...) and infer the correct behavior based on that. I think we could use something like this here as well. Namely if you would know what columns of an input schema are used by the node itself and which are just "pass-through", you could apply projection pushdown automatically.

IIUC what @alamb proposes is almost at the boundary of categories B and D. If we can define what the proposed function does purely in terms of projection behavior (which would still have meaning independent from the pushdown rule), we can consider it to be in category D.

We think that category A is not in-line with Datafusion's philosophy, and I think we all agree on this. However, in many cases, category A-type implementations serve as a good stepping stone as we try to gain a good understanding of how a good category C/D design looks like. So, on our end, we typically employ the strategy of getting a good test suite and a baseline category A implementation done first, and then progressively migrate towards a long-term category C/D solution. This PR is one of such first steps 🙂

I think with a follow-up ticket "make this rule generic" we could accept the solution in this PR, WDYT @alamb ?

I think with a follow-up ticket "make this rule generic" we could accept the solution in this PR, WDYT @alamb ?

I agree -- I filed #8096

alamb

Thanks @berkaysynnada and @ozankabak -- this PR is on my review list but I probably won't get to it until tomorrow

alamb · 2023-11-07T19:01:32Z

datafusion/sqllogictest/test_files/groupby.slt

-physical_plan
-ProjectionExec: expr=[a@0 as a]
--CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c], output_ordering=[a@0 ASC NULLS LAST], has_header=true
+physical_plan CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], has_header=true


this is a nice improvement in plan (it avoids scanning b now)

alamb · 2023-11-07T19:13:33Z

datafusion/core/src/physical_optimizer/projection_pushdown.rs

+            try_swapping_with_nested_loop_join(projection, nl_join)?
+        } else if let Some(sm_join) = input.downcast_ref::<SortMergeJoinExec>() {
+            try_swapping_with_sort_merge_join(projection, sm_join)?
+        } else if let Some(sym_join) = input.downcast_ref::<SymmetricHashJoinExec>() {


Maybe we could add a function like this to ExecutionPlan:

trait ExecutionPlan { ... /// Tries to push a projection of the output of this ExecutionPlan /// *into* itself input by rewriting the internal expressions. /// /// For example, /// (TODO EXAMPLE HERE /// /// If the ExecutionPlan does not support pushdown, ,returns Ok(None) (the default) fn push_projection(&self, projection: &[(Arc<dyn PhysicalExpr>, String)]) -> Result<Option<Arc<dyn ExecutionPlan>>>) { Ok(None) } }

alamb

Thanks @berkaysynnada and @ozankabak -- I took a look at the code and I have some ideas of how to simplify it, but we can do so as a follow on PR.

This is a very nice contribution

alamb · 2023-11-08T16:45:19Z

datafusion/core/src/physical_optimizer/optimizer.rs

@@ -107,6 +108,13 @@ impl PhysicalOptimizer {
            // into an `order by max(x) limit y`. In this case it will copy the limit value down
            // to the aggregation, allowing it to use only y number of accumulators.
            Arc::new(TopKAggregation::new()),
+            // The ProjectionPushdown rule tries to push projections towards
+            // the sources in the execution plan. As a result of this process,


Suggested change

// the sources in the execution plan. As a result of this process,

// the sources in the execution plan, in addition to the projection pushdown

// that happens in the LogicalPlan optimizer. As a result of this process,

alamb · 2023-11-09T13:46:30Z

datafusion/core/src/physical_optimizer/projection_pushdown.rs

+    // If the projection does not narrow the the schema, we should not try to push it down:
+    if projection.expr().len() >= projection.input().schema().fields().len() {
+        return Ok(None);
+    }


This check is repeated for almost every operator -- it might be possible to pull it into remove_unnecessary_projections and remove all the duplication here

alamb · 2023-11-09T14:16:32Z

datafusion/core/src/physical_optimizer/projection_pushdown.rs

+                },
+            ))
+        }
+    } else if let Some(binary) = expr_any.downcast_ref::<BinaryExpr>() {


This pattern is basically recursing through the PhysicalExpr tree manually, and only covers some subset of of the nodes.

I tried rewriting it using TreeNode, which is both less code and covers all PhysicalExpr types, not just a subset, and the tests still pass.

I will make a follow on PR with the proposed simplification

alamb · 2023-11-09T14:27:30Z

I have a follow on PR with some proposed simplifications: #8109

github-actions bot added physical-expr Physical Expressions core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) labels Nov 7, 2023

Projection Pushdown rule and test changes

42aa99f

berkaysynnada mentioned this pull request Nov 7, 2023

Unifying Projections with File Reader Execs #8075

Open

ozankabak approved these changes Nov 7, 2023

View reviewed changes

crepererum reviewed Nov 7, 2023

View reviewed changes

alamb reviewed Nov 7, 2023

View reviewed changes

alamb mentioned this pull request Nov 8, 2023

Allow user defined ExecutionPlans to benefit from pushdown behavior #8096

Open

Merge branch 'apache_main' into feature/pushdown-projection

5f5b6bc

alamb reviewed Nov 9, 2023

View reviewed changes

alamb merged commit 1c17c47 into apache:main Nov 9, 2023

alamb mentioned this pull request Nov 9, 2023

Simplify ProjectionPushdown and make it more general #8109

Merged

berkaysynnada deleted the feature/pushdown-projection branch November 10, 2023 11:40

alamb mentioned this pull request Jan 5, 2024

Regression: Unneeded fields pushed to TableProvider if struct field is part of query #8735

Closed

matthewgapp mentioned this pull request Jan 11, 2024

matt/feat/recursive ctes/config flag matthewgapp/arrow-datafusion#3

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Projection Pushdown in PhysicalPlan #8073

Projection Pushdown in PhysicalPlan #8073

berkaysynnada commented Nov 7, 2023 •

edited

Loading

ozankabak left a comment

alamb commented Nov 7, 2023

crepererum Nov 7, 2023

alamb Nov 7, 2023

ozankabak Nov 7, 2023

alamb Nov 7, 2023

crepererum Nov 8, 2023

ozankabak Nov 8, 2023

crepererum Nov 8, 2023

alamb Nov 8, 2023

alamb left a comment

alamb Nov 7, 2023

alamb Nov 7, 2023

alamb left a comment

alamb Nov 8, 2023

alamb Nov 9, 2023

alamb Nov 9, 2023

alamb Nov 9, 2023

alamb commented Nov 9, 2023

	// the sources in the execution plan. As a result of this process,
	// the sources in the execution plan, in addition to the projection pushdown
	// that happens in the LogicalPlan optimizer. As a result of this process,

Projection Pushdown in PhysicalPlan #8073

Projection Pushdown in PhysicalPlan #8073

Conversation

berkaysynnada commented Nov 7, 2023 • edited Loading

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

ozankabak left a comment

Choose a reason for hiding this comment

alamb commented Nov 7, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

A: omniscient optimizer

B: omniscient nodes

C: registry-based linking

D: abstraction

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented Nov 9, 2023

berkaysynnada commented Nov 7, 2023 •

edited

Loading