aectaan commented on code in PR #23288:
URL: https://github.com/apache/datafusion/pull/23288#discussion_r3520555472
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -1042,6 +1055,43 @@ enum RequirementsCompatibility {
NonCompatible,
}
+/// Attempts to push parent ordering requirements through a [`ProjectionExec`].
+///
+/// This is safe when every required sort expression refers to a projected
output
+/// column that is backed by a simple input column. In that case, the
requirement
+/// can be remapped from the projection output schema to the projection input
+/// schema while preserving the original sort options.
+///
+/// For example, a parent requirement on `a@2` over:
+///
+/// ```text
+/// ProjectionExec: expr=[c@2 as c, b@1 as b, a@0 as a]
+/// ```
+///
+/// is remapped to a child requirement on `a@0`.
+///
+/// The implementation is intentionally conservative: computed projection
+/// expressions and non-column sort expressions are not pushed down. Returning
+/// `Ok(None)` leaves sorting above the projection, preserving correctness.
+fn handle_projection_pushdown(
+ projection_exec: &ProjectionExec,
+ parent_required: OrderingRequirements,
+) -> Result<Option<Vec<Option<OrderingRequirements>>>> {
+ let projected_exprs = projection_exec.expr();
+
+ // Only push sorting through pure column projections. Source-dependent
+ // expressions must stay close enough to the scan to be rewritten
+ // by the source and cannot be evaluated by [`ProjectionExec`].
+ if projected_exprs.iter().any(|expr| !expr.expr.is::<Column>()) {
Review Comment:
It's strict enough. Without this check additional projection appears out of
the blue. with such check, they could be pushed to `DataSourceExec`.
For example, this check adds following optimization:
```
- 05)--------ProjectionExec: expr=[__common_expr_3@0 as __common_expr_1,
__common_expr_3@0 AND c2@2 < 4 AND c1@1 > 0 as __common_expr_2, c1@1 as c1,
c2@2 as c2]
- 06)----------ProjectionExec: expr=[c2@1 >= 2 as __common_expr_3, c1@0 as
c1, c2@1 as c2]
- 07)------------SortExec: TopK(fetch=5), expr=[c1@0 ASC NULLS LAST, c2@1
ASC NULLS LAST], preserve_partitioning=[true]
- 08)--------------DataSourceExec: file_groups={2 groups:
[[WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-0.csv,
WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-1.csv],
[WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-2.csv,
WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-3.csv]]},
projection=[c1, c2], file_type=csv, has_header=false
+ 05)--------SortExec: TopK(fetch=5), expr=[c1@2 ASC NULLS LAST, c2@3 ASC
NULLS LAST], preserve_partitioning=[true]
+ 06)----------DataSourceExec: file_groups={2 groups:
[[WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-0.csv,
WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-1.csv],
[WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-2.csv,
WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-3.csv]]},
projection=[c2@1 >= 2 as __common_expr_1, c2@1 >= 2 AND c2@1 < 4 AND c1@0 > 0
as __common_expr_2, c1, c2], file_type=csv, has_header=false
```
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -1042,6 +1055,43 @@ enum RequirementsCompatibility {
NonCompatible,
}
+/// Attempts to push parent ordering requirements through a [`ProjectionExec`].
+///
+/// This is safe when every required sort expression refers to a projected
output
+/// column that is backed by a simple input column. In that case, the
requirement
+/// can be remapped from the projection output schema to the projection input
+/// schema while preserving the original sort options.
+///
+/// For example, a parent requirement on `a@2` over:
+///
+/// ```text
+/// ProjectionExec: expr=[c@2 as c, b@1 as b, a@0 as a]
+/// ```
+///
+/// is remapped to a child requirement on `a@0`.
+///
+/// The implementation is intentionally conservative: computed projection
+/// expressions and non-column sort expressions are not pushed down. Returning
+/// `Ok(None)` leaves sorting above the projection, preserving correctness.
+fn handle_projection_pushdown(
+ projection_exec: &ProjectionExec,
+ parent_required: OrderingRequirements,
+) -> Result<Option<Vec<Option<OrderingRequirements>>>> {
+ let projected_exprs = projection_exec.expr();
+
+ // Only push sorting through pure column projections. Source-dependent
+ // expressions must stay close enough to the scan to be rewritten
+ // by the source and cannot be evaluated by [`ProjectionExec`].
+ if projected_exprs.iter().any(|expr| !expr.expr.is::<Column>()) {
Review Comment:
It's strict enough. Without this check additional projection appears out of
the blue. with such check, they could be pushed to `DataSourceExec`.
For example, this check adds following optimization:
```
- 05)--------ProjectionExec: expr=[__common_expr_3@0 as __common_expr_1,
__common_expr_3@0 AND c2@2 < 4 AND c1@1 > 0 as __common_expr_2, c1@1 as c1,
c2@2 as c2]
- 06)----------ProjectionExec: expr=[c2@1 >= 2 as __common_expr_3, c1@0 as
c1, c2@1 as c2]
- 07)------------SortExec: TopK(fetch=5), expr=[c1@0 ASC NULLS LAST, c2@1
ASC NULLS LAST], preserve_partitioning=[true]
- 08)--------------DataSourceExec: file_groups={2 groups:
[[WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-0.csv,
WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-1.csv],
[WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-2.csv,
WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-3.csv]]},
projection=[c1, c2], file_type=csv, has_header=false
+ 05)--------SortExec: TopK(fetch=5), expr=[c1@2 ASC NULLS LAST, c2@3 ASC
NULLS LAST], preserve_partitioning=[true]
+ 06)----------DataSourceExec: file_groups={2 groups:
[[WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-0.csv,
WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-1.csv],
[WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-2.csv,
WORKSPACE_ROOT/datafusion/core/tests/data/partitioned_csv/partition-3.csv]]},
projection=[c2@1 >= 2 as __common_expr_1, c2@1 >= 2 AND c2@1 < 4 AND c1@0 > 0
as __common_expr_2, c1, c2], file_type=csv, has_header=false
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
