jayzhan211 commented on code in PR #26118:
URL: https://github.com/apache/datafusion/pull/26118#discussion_r4367068069
##########
datafusion/physical-plan/src/execution_plan.rs:
##########
@@ -233,6 +233,10 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send +
Sync {
/// DataFusion uses this metadata to apply certain optimizations
/// such as automatically repartitioning correctly.
///
+ /// This describes row sequence, not how input expressions map to outputs.
+ /// Custom nodes that support sort pushdown should implement
+ /// [`Self::try_pushdown_sort`]; returning `true` here alone does not opt
in.
Review Comment:
This changes optimizer behavior for downstream `ExecutionPlan`
implementations: nodes that return `true` from `maintains_input_order()` will
now keep sorts above them. Could you add a short note to
`docs/source/library-user-guide/upgrading/56.0.0.md` saying how to restore
pushdown, and add the `api change` label? If you go with the validated fallback
from the other comment, this note isn't needed.
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs:
##########
@@ -661,8 +623,21 @@ fn pushdown_requirement_to_children(
}
} else if let Some(hash_join) = plan.downcast_ref::<HashJoinExec>() {
handle_hash_join(hash_join, parent_required)
+ } else if plan.is::<BufferExec>()
+ || plan.is::<CooperativeExec>()
+ || plan.is::<OutputRequirementExec>()
+ || plan.is::<ScalarSubqueryExec>()
+ {
+ // These operators expose the first child's columns without
transformation.
+ // Scalar subquery children compute side inputs and do not contribute
rows.
+ let mut child_requirements = vec![None; plan.children().len()];
+ child_requirements[0] = Some(parent_required);
+ Ok(Some(child_requirements))
} else {
- handle_custom_pushdown(plan, parent_required, &maintains_input_order)
+ // Preserving row positions does not identify the input expressions
needed
Review Comment:
This also stops pushdown through custom nodes that really are transparent,
and `try_pushdown_sort` can't bring it back when the node sits above a join,
union or window, since only sources and pass-through operators implement it.
You can see it in the updated
`test_custom_node_keeps_sort_with_compatible_input_ordering` snapshot, which
now has two sorts instead of one. Your issue's expected behavior (push only
when the node's own properties prove the ordering) seems to cover both cases. I
tried keeping `handle_custom_pushdown` and adding this check before returning
`Some(result)`:
```rs
let probe_children = plan_children
.iter()
.zip(&result)
.map(|(child, req)| match req {
Some(req) => Arc::new(
SortExec::new(LexOrdering::from(req.first().clone()),
Arc::clone(child))
.with_preserve_partitioning(true),
) as Arc<dyn ExecutionPlan>,
None => Arc::clone(child),
})
.collect();
let probe = Arc::clone(plan).replace_children(
probe_children,
ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
)?;
if !probe
.equivalence_properties()
.ordering_satisfy_requirement(requirement)?
{
return Ok(None);
}
```
With this, your `-a AS a` slt still keeps the sort, the snapshot above goes
back to a single sort, and no other optimizer test or slt changes
(Filter/Repartition still need their explicit exclusion). Would you consider
this instead? If you'd rather keep the conservative default here, a follow-up
is fine too.
--
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]
