jayzhan211 commented on code in PR #26090:
URL: https://github.com/apache/datafusion/pull/26090#discussion_r4429565254
##########
datafusion/physical-plan/src/joins/sort_merge_join/exec.rs:
##########
@@ -140,7 +209,17 @@ impl SortMergeJoinExec {
join_type: JoinType,
sort_options: Vec<SortOptions>,
null_equality: NullEquality,
+ null_aware: bool,
Review Comment:
The upgrade guide says this stays a seven-argument constructor, so I'd
remove `null_aware` here and let `with_null_aware` be the only way to enable
it. The planner call would become:
```diff
- Arc::new(SortMergeJoinExec::try_new(
+ Arc::new(
+ SortMergeJoinExec::try_new(
physical_left,
physical_right,
join_on,
join_filter,
*join_type,
vec![SortOptions::new(false, true)], // ASC
NULLS FIRST
*null_equality,
- true,
- )?)
+ )?
+ .with_null_aware(true)?,
+ )
```
Internal rebuilds (`with_new_children`, `swap_inputs`, projection swap,
proto decode, `enforce_distribution.rs`) can chain
`.with_null_aware(self.null_aware)?` the same way.
##########
datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:
##########
@@ -1092,12 +1156,56 @@ impl BitwiseSortMergeJoinStream {
}
}
+ /// Initializes the inner input's empty/NULL summary for null-aware joins.
+ async fn initialize_null_aware(&mut self) -> Result<()> {
+ if self.subquery_state != SubqueryState::Pending {
+ return Ok(());
+ }
+
+ match self.null_aware {
+ None => return Ok(()),
+ Some(NullAwareMode::LeftAnti) => {}
+ }
+
+ if !self.next_inner_batch().await? {
+ self.subquery_state = SubqueryState::Empty;
+ return Ok(());
+ }
+
+ // Keep the first nonempty inner batch at offset 0 for the merge scan.
+ // The complete right input must be sorted with NULLS FIRST.
+ let first_key_is_null = self.inner_key_arrays[0]
+ .logical_nulls()
+ .is_some_and(|null_keys| null_keys.is_null(0));
+
+ self.subquery_state = if first_key_is_null {
+ SubqueryState::HasNull
+ } else {
+ SubqueryState::NonEmptyNoNull
+ };
+
+ Ok(())
+ }
+
/// Main loop: a classic merge-scan over the two sorted inputs, emitting
/// output batches as they complete.
async fn join(
&mut self,
emitter: &mut TryEmitter<RecordBatch, DataFusionError>,
) -> Result<()> {
+ self.initialize_null_aware().await?;
+ match self.null_aware {
+ None => {}
+ Some(NullAwareMode::LeftAnti)
+ if self.subquery_state == SubqueryState::HasNull =>
+ {
+ // NOT IN cannot be TRUE for any outer row when the inner
+ // input contains a NULL key. (following NOT IN semetics)
Review Comment:
Optional cleanup: typo, "semetics" should be "semantics".
--
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]
