gstamatakis95 commented on code in PR #25554:
URL: https://github.com/apache/datafusion/pull/25554#discussion_r4370919021
##########
datafusion/physical-plan/src/joins/utils.rs:
##########
@@ -1984,14 +1990,67 @@ pub(crate) fn symmetric_join_output_partitioning(
JoinType::Inner | JoinType::Right => {
adjust_right_output_partitioning(right_partitioning,
left_columns_len)?
}
- JoinType::Full => {
- // We could also use left partition count as they are necessarily
equal.
-
Partitioning::UnknownPartitioning(right_partitioning.partition_count())
- }
+ JoinType::Full => full_join_output_partitioning(left, right, on,
eq_properties)?,
};
Ok(result)
}
+fn full_join_output_partitioning(
+ left: &Arc<dyn ExecutionPlan>,
+ right: &Arc<dyn ExecutionPlan>,
+ on: JoinOnRef,
+ eq_properties: &mut EquivalenceProperties,
+) -> Result<Partitioning> {
+ let left_columns_len = left.schema().fields.len();
+ let left_partitioning = left.output_partitioning();
+ let unknown =
+
Partitioning::UnknownPartitioning(right.output_partitioning().partition_count());
+ if on.is_empty() || left_partitioning.partition_count() < 2 {
+ return Ok(unknown);
+ }
+ let (left_keys, right_keys) = on
+ .iter()
+ .map(|(l, r)| (Arc::clone(l), Arc::clone(r)))
+ .unzip();
+ let requirements = InputDistributionRequirements::co_partitioned(vec![
+ Distribution::KeyPartitioned(left_keys),
+ Distribution::KeyPartitioned(right_keys),
+ ]);
+ let children = [left.as_ref(), right.as_ref()];
+ if !requirements
+ .unsatisfied_co_partitioned_children("Full join", &children)?
+ .is_empty()
+ {
+ return Ok(unknown);
+ }
+ // Either side of a row may be null, so each key becomes `coalesce(left,
right)`.
+ let mut coalesced = Vec::with_capacity(on.len());
+ for (l, r) in on {
+ let r = add_offset_to_expr(Arc::clone(r), left_columns_len as _)?;
+ let left_first = coalesce_keys(Arc::clone(l), Arc::clone(&r))?;
Review Comment:
Good catch, thanks! It now only kicks in when both join keys are plain
columns, otherwise it falls back to UnknownPartitioning like before. I added a
unit test
--
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]
