dwsmith1983 commented on code in PR #6810:
URL: https://github.com/apache/datafusion-comet/pull/6810#discussion_r4250023866
##########
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)? {
Review Comment:
> Could you run a NOT IN over a sorted streamed side sized near the limit at
`local[1]`, on main and on this branch with `spark.comet.debug.memory=true`, so
we know what this costs?
It costs more than the share alone. At `local[1]` under `fair_unified`, a
NOT IN over a 2M-row sorted partition (1.8M rows survive) needs a 39 MB pool on
main and 165 MB with the sort:
- Below 116 MB the build fails in `HashJoinInput`. `ExternalSorter[0]` and
`ExternalSorterMerge[0]` register first, so the build gets a third of the pool.
- From 116 to 163 MB the build fits, but the sort fails with "Not enough
memory to continue external sort". The join hands it all 1.8M rows as one 28.8
MB batch, and `ExternalSorter` has to reserve twice that inside its own third
with nothing to spill.
So the sort is gone. This version falls back to Spark for a NOT IN whose
streamed side reports an ordering, and the guard's comment points at
apache/datafusion#26226 so it can come out once that fix reaches Comet.
--
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]
