zhuqi-lucas commented on code in PR #23288:
URL: https://github.com/apache/datafusion/pull/23288#discussion_r3520349150
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -469,15 +482,15 @@ fn pushdown_requirement_to_children(
}
} else if let Some(aggregate_exec) = plan.downcast_ref::<AggregateExec>() {
handle_aggregate_pushdown(aggregate_exec, parent_required)
+ } else if let Some(projection_exec) =
plan.downcast_ref::<ProjectionExec>() {
+ // A `ProjectionExec` with a fetch is handled in the fetch branch above
Review Comment:
The comments here is not easy to understanding, we may polish it or move to
other places.
A `ProjectionExec` with a fetch is handled in the fetch branch above
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -356,15 +356,28 @@ fn pushdown_requirement_to_children(
return Ok(None);
};
match determine_children_requirement(&parent_required, &child_req,
child_plan) {
- RequirementsCompatibility::Satisfy =>
Ok(Some(vec![Some(child_req)])),
+ RequirementsCompatibility::Satisfy => {
Review Comment:
The Window branch `Satisfy` fix looks orthogonal — could be split into a
separate PR for cleaner history.
##########
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:
The upfront `projected_exprs.iter().all(is::<Column>())` bail is stricter
than needed — the per-column check inside the loop already handles it?
##########
datafusion/core/tests/physical_optimizer/enforce_sorting.rs:
##########
@@ -2974,3 +2974,45 @@ async fn
test_parallelize_sorts_remaps_index_through_reordering_projection() ->
Ok(())
}
+
+#[tokio::test]
+async fn test_push_sort_through_reordered_projection_to_union() -> Result<()> {
Review Comment:
Test coverage is currently thin (one Union scenario); a few negative /alias
/ fetch-interaction cases would help.
--
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]
