dwsmith1983 commented on code in PR #6810:
URL: https://github.com/apache/datafusion-comet/pull/6810#discussion_r4250053117
##########
native/core/src/execution/planner.rs:
##########
@@ -2688,18 +2693,38 @@ impl PhysicalPlanner {
// a projection was added to the hash join
additional_native_plans.push(Arc::clone(swapped_hash_join.children()[0]));
}
+ (swapped_hash_join, additional_native_plans)
+ };
- Ok((
- scans,
- shuffle_scans,
- Arc::new(SparkPlan::new_with_additional(
- spark_plan.plan_id,
- swapped_hash_join,
- vec![join_params.left, join_params.right],
- additional_native_plans,
- )),
- ))
- }
+ // Spark reports the streamed side's ordering and may have
removed a sort above
+ // the join on that basis. DataFusion keeps unmatched probe
rows in probe order
+ // only when it sees the probe input sorted, and does not keep
a null-aware anti
+ // join's build order, so sort the output otherwise.
+ let join_schema = join_root.schema();
+ let output_ordering = join
+ .output_ordering
+ .iter()
+ .map(|expr| self.create_sort_expr(expr,
Arc::clone(&join_schema)))
+ .collect::<Result<Vec<_>, _>>()?;
+ let native_plan =
+ match Self::sort_unless_ordered(Arc::clone(&join_root),
output_ordering)? {
+ Some(sort) => {
+ additional_native_plans.push(join_root);
Review Comment:
> Could the hash join execs add the spill metrics when
`getOutputOrderingCount > 0`, the way they add `joinDynamicFilterMetrics`?
With the sort gone there is no second plan under the join, so the node's
metrics are the same as on main.
--
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]
