You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "SubhamSinghal (via GitHub)" <gi...@apache.org> on 2026/08/18 12:57:19 UTC
[PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
SubhamSinghal opened a new pull request, #24457:
URL: https://github.com/apache/datafusion/pull/24457
## Which issue does this PR close?
Part of #17427 (the PiecewiseMergeJoin epic). This is the second half of its existence-join item: #23870 added `LeftSemi`/`LeftAnti`, this adds `RightSemi`/`RightAnti`. Only the Mark joins remain unsupported.
## Rationale for this change
A range-predicate right semi/anti join — `... RIGHT SEMI JOIN r ON l.v > r.v`, or any query that decorrelates to one — currently falls back to `NestedLoopJoinExec`.
## What changes are included in this PR?
New `RightExistencePWMJStream` (`right_existence_join.rs`), a 3-state machine: await the buffered reduction, then per streamed batch evaluate the key, compare it against the reduced key with a vectorized `cmp` kernel, and filter. Plus:
* `exec.rs` — dispatch; `build_buffered_extreme`, which folds every buffered partition on its own `SpawnedTask`; `BufferedExtreme` as the shared state; relaxed ordering, distribution and partitioning declarations for these two join types; removal of the dead `sort_options` flip and the `unimplemented!()` placeholder that the swap design had left behind.
* `physical_planner.rs` — the gate now excludes only the Mark joins.
* `utils.rs` — `is_supported_existence_join` widened; new `is_supported_right_existence_join` (which deliberately excludes `RightMark`, so a Mark join cannot inherit the relaxed input requirements if that gate is ever loosened); no visited-indices bitmap allocated for these join types.
* `extreme_key` made `pub(super)` so both existence streams share one implementation.
## Are these changes tested?
Yes
## Benchmark
Microbenchmark, 20K × 20K rows, single inequality, `enable_piecewise_merge_join` on vs off
| Case | PWMJ | NestedLoopJoin | Speedup |
|---|---|---|---|
| RightSemi, all match | ~0.023 ms | ~77 ms | ~3,300× |
| RightAnti, all match | ~0.024 ms | ~77 ms | ~3,200× |
| RightSemi, no match | ~0.023 ms | ~78 ms | ~3,300× |
| RightAnti, no match | ~0.024 ms | ~78 ms | ~3,300× |
| RightSemi, half match | ~0.036 ms | ~78 ms | ~2,200× |
| RightAnti, half match | ~0.036 ms | ~77 ms | ~2,100× |
## Are there any user-facing changes?
No
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5554176550
Thanks @SubhamSinghal I'm going do another pass today
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead merged PR #24457:
URL: https://github.com/apache/datafusion/pull/24457
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5502374853
Once https://github.com/apache/datafusion/pull/24863 merged we need to rebase and proceed with this PR
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "SubhamSinghal (via GitHub)" <gi...@apache.org>.
SubhamSinghal commented on code in PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#discussion_r3954234322
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs:
##########
@@ -941,6 +1070,147 @@ impl BufferedSideData {
}
}
+/// The entire buffered side of a right existence join, reduced to the one key that decides
+/// every streamed row -- the minimum for `<`/`<=`, the maximum for `>`/`>=`.
+///
+/// Right existence joins never look at a buffered *row*, so unlike [`BufferedSideData`] this
+/// holds no batch: the buffered input is folded away as it streams in and the state is `O(1)`
+/// however large that side is. See `right_existence_join.rs`.
+pub(super) struct BufferedExtreme {
+ /// One-row array. The row is null exactly when no buffered key is non-null (an empty or
+ /// all-NULL buffered side), in which case nothing can ever match.
+ extreme: ArrayRef,
+ _reservation: MemoryReservation,
+}
+
+impl BufferedExtreme {
+ pub(super) fn extreme(&self) -> &ArrayRef {
+ &self.extreme
+ }
+}
+
+/// Reduces one buffered partition to a single extreme key, dropping each batch as it goes.
+/// `None` when the partition produced no batch at all.
+async fn partition_extreme(
+ mut buffered: SendableRecordBatchStream,
+ on_buffered: PhysicalExprRef,
+ metrics: BuildProbeJoinMetrics,
+ reservation: MemoryReservation,
+ descending: bool,
+) -> Result<Option<ArrayRef>> {
+ let mut extreme: Option<ArrayRef> = None;
+
+ while let Some(batch) = buffered.next().await.transpose()? {
+ metrics.build_input_batches.add(1);
+ metrics.build_input_rows.add(batch.num_rows());
+
+ let keys = on_buffered.evaluate(&batch)?.into_array(batch.num_rows())?;
+
+ // Reduced and dropped within this iteration, but as wide as the batch, and every
+ // partition folds one of these at once. Resized rather than grown, so what the pool
+ // sees is the widest key array in flight here and not their sum.
+ reservation.try_resize(keys.get_array_memory_size())?;
+
+ let batch_extreme = extreme_key(&keys, descending)?;
+
+ // Re-reduced as a pair rather than compared, so the running value is ordered exactly
+ // as each single reduction was. `extreme_key` ignores nulls, so a null from an
+ // all-NULL batch never displaces a real key.
+ extreme = Some(match extreme {
+ Some(running) => {
+ let pair = concat(&[running.as_ref(), batch_extreme.as_ref()])?;
+ extreme_key(&pair, descending)?
+ }
+ None => batch_extreme,
+ });
+ }
+
+ Ok(extreme)
+}
+
+/// Folds every buffered partition down to one extreme key. `O(B)` time, and the only state it
+/// retains is that one key -- no buffered batch is concatenated or held. The transient cost is
+/// one key array per folding partition, which each task accounts against the pool for as long
+/// as it holds it.
+///
+/// A min/max combines across partitions, so this side needs no single-partition funnel --
+/// `input_distribution_requirements` asks for `UnspecifiedDistribution` here and the partitions
+/// are folded independently, each in its own spawned task.
+async fn build_buffered_extreme(
+ buffered_streams: Vec<SendableRecordBatchStream>,
+ buffered_schema: SchemaRef,
+ on_buffered: PhysicalExprRef,
+ metrics: BuildProbeJoinMetrics,
+ reservation: MemoryReservation,
+ memory_pool: Arc<dyn MemoryPool>,
+ descending: bool,
+) -> Result<BufferedExtreme> {
+ let tasks: Vec<_> = buffered_streams
+ .into_iter()
+ .enumerate()
+ .map(|(partition, stream)| {
+ let on_buffered = Arc::clone(&on_buffered);
+ let metrics = metrics.clone();
+ // One reservation per task rather than a shared one: `MemoryReservation` is not
+ // shareable, and each task's transient is freed as soon as it finishes.
+ let reservation = MemoryConsumer::new(format!(
+ "PiecewiseMergeJoinBufferedFold[{partition}]"
+ ))
+ .register(&memory_pool);
+ SpawnedTask::spawn(partition_extreme(
+ stream,
+ on_buffered,
+ metrics,
+ reservation,
+ descending,
+ ))
+ })
+ .collect();
+
+ // The tasks run concurrently; this only collects them. Awaiting in order is fine, and a
+ // failure propagates after the rest have been joined, since `SpawnedTask` aborts on drop.
+ let mut extreme: Option<ArrayRef> = None;
+ for task in tasks {
+ let partition_extreme = task.join_unwind().await.map_err(|e| {
+ internal_datafusion_err!("buffered extreme task failed: {e}")
+ })??;
+ if let Some(partition_extreme) = partition_extreme {
+ // Re-reduced as a pair, exactly as the batches within a partition were, so a value
+ // accumulated across partitions is ordered by the same rule.
+ extreme = Some(match extreme {
+ Some(running) => {
+ let pair = concat(&[running.as_ref(), partition_extreme.as_ref()])?;
+ extreme_key(&pair, descending)?
+ }
+ None => partition_extreme,
+ });
Review Comment:
Addressed in [1665517](https://github.com/apache/datafusion/pull/24457/commits/16655173354a996e7c16c3160cc4bbfc8421fc74)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "SubhamSinghal (via GitHub)" <gi...@apache.org>.
SubhamSinghal commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5570782416
Thanks @comphead for review. Resolved issues in 1fedd675467f66485d3a42de0f24259fc26b8187
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5542883050
@coderfender FYI
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#discussion_r3952504659
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs:
##########
@@ -941,6 +1070,147 @@ impl BufferedSideData {
}
}
+/// The entire buffered side of a right existence join, reduced to the one key that decides
+/// every streamed row -- the minimum for `<`/`<=`, the maximum for `>`/`>=`.
+///
+/// Right existence joins never look at a buffered *row*, so unlike [`BufferedSideData`] this
+/// holds no batch: the buffered input is folded away as it streams in and the state is `O(1)`
+/// however large that side is. See `right_existence_join.rs`.
+pub(super) struct BufferedExtreme {
+ /// One-row array. The row is null exactly when no buffered key is non-null (an empty or
+ /// all-NULL buffered side), in which case nothing can ever match.
+ extreme: ArrayRef,
+ _reservation: MemoryReservation,
+}
+
+impl BufferedExtreme {
+ pub(super) fn extreme(&self) -> &ArrayRef {
+ &self.extreme
+ }
+}
+
+/// Reduces one buffered partition to a single extreme key, dropping each batch as it goes.
+/// `None` when the partition produced no batch at all.
+async fn partition_extreme(
+ mut buffered: SendableRecordBatchStream,
+ on_buffered: PhysicalExprRef,
+ metrics: BuildProbeJoinMetrics,
+ reservation: MemoryReservation,
+ descending: bool,
+) -> Result<Option<ArrayRef>> {
+ let mut extreme: Option<ArrayRef> = None;
+
+ while let Some(batch) = buffered.next().await.transpose()? {
+ metrics.build_input_batches.add(1);
+ metrics.build_input_rows.add(batch.num_rows());
+
+ let keys = on_buffered.evaluate(&batch)?.into_array(batch.num_rows())?;
+
+ // Reduced and dropped within this iteration, but as wide as the batch, and every
+ // partition folds one of these at once. Resized rather than grown, so what the pool
+ // sees is the widest key array in flight here and not their sum.
+ reservation.try_resize(keys.get_array_memory_size())?;
+
+ let batch_extreme = extreme_key(&keys, descending)?;
+
+ // Re-reduced as a pair rather than compared, so the running value is ordered exactly
+ // as each single reduction was. `extreme_key` ignores nulls, so a null from an
+ // all-NULL batch never displaces a real key.
+ extreme = Some(match extreme {
+ Some(running) => {
+ let pair = concat(&[running.as_ref(), batch_extreme.as_ref()])?;
+ extreme_key(&pair, descending)?
+ }
+ None => batch_extreme,
+ });
+ }
+
+ Ok(extreme)
+}
+
+/// Folds every buffered partition down to one extreme key. `O(B)` time, and the only state it
+/// retains is that one key -- no buffered batch is concatenated or held. The transient cost is
+/// one key array per folding partition, which each task accounts against the pool for as long
+/// as it holds it.
+///
+/// A min/max combines across partitions, so this side needs no single-partition funnel --
+/// `input_distribution_requirements` asks for `UnspecifiedDistribution` here and the partitions
+/// are folded independently, each in its own spawned task.
+async fn build_buffered_extreme(
+ buffered_streams: Vec<SendableRecordBatchStream>,
+ buffered_schema: SchemaRef,
+ on_buffered: PhysicalExprRef,
+ metrics: BuildProbeJoinMetrics,
+ reservation: MemoryReservation,
+ memory_pool: Arc<dyn MemoryPool>,
+ descending: bool,
+) -> Result<BufferedExtreme> {
+ let tasks: Vec<_> = buffered_streams
+ .into_iter()
+ .enumerate()
+ .map(|(partition, stream)| {
+ let on_buffered = Arc::clone(&on_buffered);
+ let metrics = metrics.clone();
+ // One reservation per task rather than a shared one: `MemoryReservation` is not
+ // shareable, and each task's transient is freed as soon as it finishes.
+ let reservation = MemoryConsumer::new(format!(
+ "PiecewiseMergeJoinBufferedFold[{partition}]"
+ ))
+ .register(&memory_pool);
+ SpawnedTask::spawn(partition_extreme(
+ stream,
+ on_buffered,
+ metrics,
+ reservation,
+ descending,
+ ))
+ })
+ .collect();
+
+ // The tasks run concurrently; this only collects them. Awaiting in order is fine, and a
+ // failure propagates after the rest have been joined, since `SpawnedTask` aborts on drop.
+ let mut extreme: Option<ArrayRef> = None;
+ for task in tasks {
+ let partition_extreme = task.join_unwind().await.map_err(|e| {
+ internal_datafusion_err!("buffered extreme task failed: {e}")
+ })??;
+ if let Some(partition_extreme) = partition_extreme {
+ // Re-reduced as a pair, exactly as the batches within a partition were, so a value
+ // accumulated across partitions is ordered by the same rule.
+ extreme = Some(match extreme {
+ Some(running) => {
+ let pair = concat(&[running.as_ref(), partition_extreme.as_ref()])?;
+ extreme_key(&pair, descending)?
+ }
+ None => partition_extreme,
+ });
Review Comment:
Nit, non-blocking: the pair-reduce is duplicated verbatim between `build_buffered_extreme` and `partition_extreme`. These seven lines are the same as [L1119-L1125](https://github.com/apache/datafusion/blob/bc31504a49de968feabbda864e682079e9f4de49/datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs#L1119-L1125), with `batch_extreme` in place of `partition_extreme`.
Both copies also carry a comment explaining the same invariant, once as "the running value is ordered exactly as each single reduction was" and once as "a value accumulated across partitions is ordered by the same rule". Extracting it would put that invariant in one place:
```rust
/// Re-reduces as a pair rather than comparing, so the running value is ordered exactly as
/// each single reduction was. `extreme_key` ignores nulls, so a null from an all-NULL
/// input never displaces a real key.
fn fold_extreme(
running: Option<ArrayRef>,
next: ArrayRef,
descending: bool,
) -> Result<ArrayRef> {
match running {
Some(running) => {
let pair = concat(&[running.as_ref(), next.as_ref()])?;
extreme_key(&pair, descending)
}
None => Ok(next),
}
}
```
Both call sites then collapse to one line:
```rust
extreme = Some(fold_extreme(extreme, batch_extreme, descending)?);
```
Readability only, no behavior change.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "SubhamSinghal (via GitHub)" <gi...@apache.org>.
SubhamSinghal commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5328616388
benchmark PR: https://github.com/apache/datafusion/pull/24458
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5374414142
@SubhamSinghal sorry for the delay, last time we got some correctness issues for PWJM and Im planning to merge first #24390 to create a generic join test set and we can check this PR against it
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5329849841
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/24457?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `87.98220%` with `81 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 81.26%. Comparing base ([`bb038a6`](https://app.codecov.io/gh/apache/datafusion/commit/bb038a6739091567fe32c196d8c568a98e4f42e2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`5116d05`](https://app.codecov.io/gh/apache/datafusion/commit/5116d05a689275285c46c4e4ed1823fbec0584c4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 4 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/24457?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...joins/piecewise\_merge\_join/right\_existence\_join.rs](https://app.codecov.io/gh/apache/datafusion/pull/24457?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Fjoins%2Fpiecewise_merge_join%2Fright_existence_join.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9qb2lucy9waWVjZXdpc2VfbWVyZ2Vfam9pbi9yaWdodF9leGlzdGVuY2Vfam9pbi5ycw==) | 88.08% | [20 Missing and 36 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/24457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ysical-plan/src/joins/piecewise\_merge\_join/exec.rs](https://app.codecov.io/gh/apache/datafusion/pull/24457?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Fjoins%2Fpiecewise_merge_join%2Fexec.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9qb2lucy9waWVjZXdpc2VfbWVyZ2Vfam9pbi9leGVjLnJz) | 86.97% | [6 Missing and 19 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/24457?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #24457 +/- ##
==========================================
+ Coverage 81.23% 81.26% +0.03%
==========================================
Files 1112 1113 +1
Lines 390635 392183 +1548
Branches 390635 392183 +1548
==========================================
+ Hits 317350 318727 +1377
- Misses 54650 54695 +45
- Partials 18635 18761 +126
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/24457?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "SubhamSinghal (via GitHub)" <gi...@apache.org>.
SubhamSinghal commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5527669880
@comphead Addressed review comment in 597b29ede7a55aca20dbf17d8ef975af71d3e0de commit.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "kumarUjjawal (via GitHub)" <gi...@apache.org>.
kumarUjjawal commented on code in PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#discussion_r3942949526
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/right_existence_join.rs:
##########
@@ -0,0 +1,866 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! PiecewiseMergeJoin stream specialized for right existence joins.
+//!
+//! Instantiated by [`PiecewiseMergeJoinExec`] when the join type is `RightSemi` or
+//! `RightAnti`. `LeftSemi`/`LeftAnti` are served by `ExistencePWMJStream` (see
+//! `existence_join.rs`); the Mark joins are still rejected in
+//! `PiecewiseMergeJoinExec::try_new`.
+//!
+//! # Algorithm
+//!
+//! Left and right existence joins mark opposite sides, and for a single range predicate
+//! that difference is not symmetric — it collapses the work.
+//!
+//! `LeftSemi`/`LeftAnti` ask, for each *buffered* row, whether any streamed row matches, so
+//! the answer depends on the whole streamed side and can only be emitted once it has all
+//! been read. `RightSemi`/`RightAnti` ask the mirror question — for each *streamed* row,
+//! does any buffered row match? — and with only `buffered_key OP streamed_key` to satisfy,
+//! that is decided by a single buffered key:
+//!
+//! ```text
+//! ∃b. b < s ⟺ min(b) < s ∃b. b > s ⟺ max(b) > s
+//! ∃b. b <= s ⟺ min(b) <= s ∃b. b >= s ⟺ max(b) >= s
+//! ```
+//!
+//! The buffered side is therefore reduced to that one key -- `min_batch`/`max_batch`, folded
+//! batch by batch as it streams in -- and each batch is dropped once folded. It is never
+//! concatenated or retained, so the state this join holds is a single row however large that
+//! side is. The reduction is shared, so every streamed partition reads the same key rather
+//! than repeating the pass:
+//!
+//! ```text
+//! operator `<`, buffered keys [NULL, 5, 9, 7] -> min = 5
+//! `b < s` holds for some b iff `5 < s`
+//! ```
+//!
+//! A min/max is `O(B)` from any order, so `required_input_ordering` returns nothing for either
+//! child and no `SortExec` is planned.
+//!
+//! Every streamed row is then decided by comparing it against that one key, which is a
+//! vectorized `cmp` kernel per batch and a filter. Nothing about a streamed row depends on
+//! any other, so:
+//!
+//! * output is produced per batch as it arrives — no watermark, no final pass, and no
+//! election among the streamed partitions,
+//! * all N streamed partitions produce output, rather than one non-empty partition,
+//! * the streamed side is never sorted, not even per batch.
+//!
+//! Rows whose join key is NULL never satisfy a comparison predicate. `min_batch`/`max_batch`
+//! ignore NULLs, so the reduced key is null only when *every* buffered key is (or the buffered
+//! side is empty) — and then no streamed row can match at all, which makes `RightSemi` empty
+//! without reading the streamed side and `RightAnti` a passthrough of it. A NULL streamed key
+//! makes its comparison NULL rather than false, which is "no match" — dropped by `RightSemi`,
+//! kept by `RightAnti`.
+//!
+//! Picking the extreme and comparing against it must agree on ordering. `min_batch`/`max_batch`
+//! and arrow's `lt`/`lt_eq`/`gt`/`gt_eq` kernels both order floats by `total_cmp`
+//! (`arrow-arith`'s `MinAccumulator` compares with `ArrowNativeTypeOp::is_lt`, seeded from
+//! `MAX_TOTAL_ORDER`), which puts `-0.0` strictly below `+0.0` -- but SQL comparisons treat
+//! them as equal, and that is what a real `k < r` predicate actually evaluates to: any
+//! `BinaryExpr` comparison, including the one the `NestedLoopJoinExec` oracle in the
+//! differential fuzz test builds its filter from, normalizes `-0.0` to `+0.0` first (see
+//! `apply_cmp` in `datafusion-physical-expr-common`). Both the extreme and the streamed key
+//! array are normalized with [`normalize_float_zero`] before comparing,
+//! after the reduction: normalizing first would not change which value the reduction picks
+//! (`-0.0` and `+0.0` are numerically equal either way), and normalizing only where the
+//! comparison happens keeps the reduction itself agreeing with the unnormalized `min`/`max`
+//! semantics its docs above describe. `NaN` needs no such fix-up: every kernel involved orders
+//! it as the maximum, so it is treated identically on both sides already.
+//!
+//! # Cost
+//!
+//! Let `B` be the buffered rows and `S` the streamed rows: `O(B)` to reduce the buffered side
+//! plus `O(S)` to filter, and no sort on either side. The state retained across batches is one
+//! row; while folding, each buffered partition also holds the key array of the batch it is
+//! reducing, which it accounts against the memory pool for that long.
+//!
+//! This is why the shared state is [`BufferedExtreme`] and not `BufferedSideData`: every field
+//! of the latter -- the concatenated batch, the key array, the visited-indices bitmap, the
+//! final-pass counter -- would be dead here.
+//!
+//! [`PiecewiseMergeJoinExec`]: super::PiecewiseMergeJoinExec
+
+use std::sync::Arc;
+use std::task::{Poll, ready};
+
+use arrow::array::{Array, ArrayRef, RecordBatch, Scalar};
+use arrow::compute::filter_record_batch;
+use arrow::compute::kernels::boolean::not;
+use arrow::compute::kernels::cmp::{gt, gt_eq, lt, lt_eq};
+use arrow_schema::SchemaRef;
+use datafusion_common::utils::normalize_float_zero;
+use datafusion_common::{Result, internal_err};
+use datafusion_execution::{RecordBatchStream, SendableRecordBatchStream};
+use datafusion_expr::{JoinType, Operator};
+use datafusion_physical_expr::PhysicalExprRef;
+use futures::{Stream, StreamExt};
+
+use crate::handle_state;
+use crate::joins::piecewise_merge_join::exec::BufferedExtreme;
+use crate::joins::utils::{
+ BuildProbeJoinMetrics, OnceFut, StatefulStreamResult, boolean_mask_from_filter,
+};
+use crate::stream::EmptyRecordBatchStream;
+
+pub(super) enum RightExistencePWMJStreamState {
+ /// Await the buffered side's reduction to a single key.
+ WaitBufferedExtreme,
+ /// Fetch streamed batches and emit the rows that do (`RightSemi`) or do not
+ /// (`RightAnti`) have a buffered match.
+ ScanStreamBatches,
+ Completed,
+}
+
+pub(super) struct RightExistencePWMJStream {
+ /// Output schema, which for `RightSemi`/`RightAnti` is the streamed side's schema
+ schema: SchemaRef,
+ /// Physical expression evaluated on the streamed side. The buffered side's
+ /// equivalent is already evaluated when the buffered side is collected.
+ on_streamed: PhysicalExprRef,
+ /// `RightSemi` or `RightAnti`
+ join_type: JoinType,
+ /// Comparison operator
+ operator: Operator,
+ streamed: SendableRecordBatchStream,
+ /// Resolves to the whole buffered side reduced to one key. Shared with the other streamed
+ /// partitions, so that reduction happens exactly once.
+ buffered_extreme_fut: OnceFut<BufferedExtreme>,
+ state: RightExistencePWMJStreamState,
+ /// That key, held as a one-element [`Scalar`] so the comparison against a streamed batch is
+ /// one kernel call. `None` when the buffered side has no non-null key, i.e. nothing can ever
+ /// match. Only populated once `buffered_extreme_fut` has resolved.
+ buffered_extreme: Option<Scalar<ArrayRef>>,
+ join_metrics: BuildProbeJoinMetrics,
+}
+
+impl RightExistencePWMJStream {
+ pub(super) fn try_new(
+ schema: SchemaRef,
+ on_streamed: PhysicalExprRef,
+ join_type: JoinType,
+ operator: Operator,
+ streamed: SendableRecordBatchStream,
+ buffered_extreme_fut: OnceFut<BufferedExtreme>,
+ join_metrics: BuildProbeJoinMetrics,
+ ) -> Self {
+ Self {
+ schema,
+ on_streamed,
+ join_type,
+ operator,
+ streamed,
+ buffered_extreme_fut,
+ state: RightExistencePWMJStreamState::WaitBufferedExtreme,
+ buffered_extreme: None,
+ join_metrics,
+ }
+ }
+
+ fn poll_next_impl(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Option<Result<RecordBatch>>> {
+ loop {
+ return match self.state {
+ RightExistencePWMJStreamState::WaitBufferedExtreme => {
+ handle_state!(ready!(self.collect_buffered_extreme(cx)))
+ }
+ RightExistencePWMJStreamState::ScanStreamBatches => {
+ handle_state!(ready!(self.scan_stream_batch(cx)))
+ }
+ RightExistencePWMJStreamState::Completed => Poll::Ready(None),
+ };
+ }
+ }
+
+ /// Picks up the buffered side's reduced key, which is all this join type needs from it.
+ fn collect_buffered_extreme(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>> {
+ let build_timer = self.join_metrics.build_time.timer();
+ let buffered_extreme = ready!(self.buffered_extreme_fut.get_shared(cx))?;
+ build_timer.done();
+
+ // Null exactly when no buffered key is non-null, and NULLs match nothing. Cloned out by
+ // value -- it is one row -- so the shared state is not kept alive by this stream.
+ //
+ // Normalized because the `lt`/`lt_eq`/`gt`/`gt_eq` kernels called in
+ // `filter_streamed_batch` order `-0.0` strictly below `+0.0`, but SQL comparisons --
+ // including `apply_cmp`, which every other `k < r` predicate in the plan goes through
+ // -- treat them as equal. The streamed key array is normalized the same way, right
+ // before that comparison. `min`/`max_batch`, which reduced this extreme, order `-0.0`
+ // below `+0.0` too, so normalizing after the reduction rather than before leaves the
+ // reduction itself agreeing with the unnormalized ordering its own docs describe.
+ let extreme = normalize_float_zero(buffered_extreme.extreme());
+ self.buffered_extreme = (extreme.null_count() == 0).then(|| Scalar::new(extreme));
+
+ // With no non-null buffered key nothing matches, so `RightSemi` outputs nothing
+ // and does not need to read a single streamed batch. `RightAnti` still has to,
+ // since it outputs all of them.
+ self.state = match (&self.buffered_extreme, self.join_type) {
+ (None, JoinType::RightSemi) => {
+ let streamed_schema = self.streamed.schema();
+ self.streamed = Box::pin(EmptyRecordBatchStream::new(streamed_schema));
+ RightExistencePWMJStreamState::Completed
+ }
+ _ => RightExistencePWMJStreamState::ScanStreamBatches,
+ };
+
+ Poll::Ready(Ok(StatefulStreamResult::Continue))
+ }
+
+ /// Fetches one streamed batch and emits the rows it contributes, if any.
+ fn scan_stream_batch(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>> {
+ match ready!(self.streamed.poll_next_unpin(cx)) {
+ None => self.state = RightExistencePWMJStreamState::Completed,
+ Some(Ok(batch)) => {
+ self.join_metrics.input_batches.add(1);
+ self.join_metrics.input_rows.add(batch.num_rows());
+
+ let output = self.filter_streamed_batch(&batch)?;
+ if output.num_rows() > 0 {
+ return Poll::Ready(Ok(StatefulStreamResult::Ready(Some(output))));
+ }
+ // Nothing survived; take the next batch rather than yielding an empty one.
+ }
+ Some(Err(err)) => return Poll::Ready(Err(err)),
+ }
+
+ Poll::Ready(Ok(StatefulStreamResult::Continue))
+ }
+
+ /// Keeps the streamed rows that have a buffered match (`RightSemi`) or that have none
+ /// (`RightAnti`), by comparing each against the single buffered extreme.
+ fn filter_streamed_batch(&self, batch: &RecordBatch) -> Result<RecordBatch> {
+ let columns = match &self.buffered_extreme {
+ // No non-null buffered key, so no streamed row matches and `RightAnti` keeps
+ // the batch whole. `RightSemi` never gets here: it completed without reading
+ // the streamed side.
+ None => batch.columns().to_vec(),
+ Some(extreme) => {
+ let stream_values = normalize_float_zero(
+ &self
+ .on_streamed
+ .evaluate(batch)?
+ .into_array(batch.num_rows())?,
+ );
+
+ // `extreme` is the buffered key, so it goes on the left of the operator,
+ // matching the `buffered OP streamed` orientation of the predicate.
+ let matched = match self.operator {
Review Comment:
These Arrow comparison kernels reject nested types such as List and Struct. The planner now routes their RightSemi/RightAnti range joins here, whereas the previous NestedLoopJoin path uses `apply_cmp`, which supports nested comparisons.
For example, list keys `[1] < [2]` should produce a matching right row, but this path returns a “Nested comparison” error when PWMJ is enabled.
Could we preserve the nested comparison path or retain the nested-loop fallback for unsupported types? Please add RightSemi/RightAnti regressions comparing results with PWMJ enabled and disabled.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5330227228
Thanks @SubhamSinghal I'll check this today
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "SubhamSinghal (via GitHub)" <gi...@apache.org>.
SubhamSinghal commented on PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#issuecomment-5329854932
@comphead @kumarUjjawal PR for right semi/anti join. Please help in reviewing this PR. Thanks.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(pwmj): support RightSemi/RightAnti existence joins [datafusion]
Posted by "SubhamSinghal (via GitHub)" <gi...@apache.org>.
SubhamSinghal commented on code in PR #24457:
URL: https://github.com/apache/datafusion/pull/24457#discussion_r3943871447
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/right_existence_join.rs:
##########
@@ -0,0 +1,866 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! PiecewiseMergeJoin stream specialized for right existence joins.
+//!
+//! Instantiated by [`PiecewiseMergeJoinExec`] when the join type is `RightSemi` or
+//! `RightAnti`. `LeftSemi`/`LeftAnti` are served by `ExistencePWMJStream` (see
+//! `existence_join.rs`); the Mark joins are still rejected in
+//! `PiecewiseMergeJoinExec::try_new`.
+//!
+//! # Algorithm
+//!
+//! Left and right existence joins mark opposite sides, and for a single range predicate
+//! that difference is not symmetric — it collapses the work.
+//!
+//! `LeftSemi`/`LeftAnti` ask, for each *buffered* row, whether any streamed row matches, so
+//! the answer depends on the whole streamed side and can only be emitted once it has all
+//! been read. `RightSemi`/`RightAnti` ask the mirror question — for each *streamed* row,
+//! does any buffered row match? — and with only `buffered_key OP streamed_key` to satisfy,
+//! that is decided by a single buffered key:
+//!
+//! ```text
+//! ∃b. b < s ⟺ min(b) < s ∃b. b > s ⟺ max(b) > s
+//! ∃b. b <= s ⟺ min(b) <= s ∃b. b >= s ⟺ max(b) >= s
+//! ```
+//!
+//! The buffered side is therefore reduced to that one key -- `min_batch`/`max_batch`, folded
+//! batch by batch as it streams in -- and each batch is dropped once folded. It is never
+//! concatenated or retained, so the state this join holds is a single row however large that
+//! side is. The reduction is shared, so every streamed partition reads the same key rather
+//! than repeating the pass:
+//!
+//! ```text
+//! operator `<`, buffered keys [NULL, 5, 9, 7] -> min = 5
+//! `b < s` holds for some b iff `5 < s`
+//! ```
+//!
+//! A min/max is `O(B)` from any order, so `required_input_ordering` returns nothing for either
+//! child and no `SortExec` is planned.
+//!
+//! Every streamed row is then decided by comparing it against that one key, which is a
+//! vectorized `cmp` kernel per batch and a filter. Nothing about a streamed row depends on
+//! any other, so:
+//!
+//! * output is produced per batch as it arrives — no watermark, no final pass, and no
+//! election among the streamed partitions,
+//! * all N streamed partitions produce output, rather than one non-empty partition,
+//! * the streamed side is never sorted, not even per batch.
+//!
+//! Rows whose join key is NULL never satisfy a comparison predicate. `min_batch`/`max_batch`
+//! ignore NULLs, so the reduced key is null only when *every* buffered key is (or the buffered
+//! side is empty) — and then no streamed row can match at all, which makes `RightSemi` empty
+//! without reading the streamed side and `RightAnti` a passthrough of it. A NULL streamed key
+//! makes its comparison NULL rather than false, which is "no match" — dropped by `RightSemi`,
+//! kept by `RightAnti`.
+//!
+//! Picking the extreme and comparing against it must agree on ordering. `min_batch`/`max_batch`
+//! and arrow's `lt`/`lt_eq`/`gt`/`gt_eq` kernels both order floats by `total_cmp`
+//! (`arrow-arith`'s `MinAccumulator` compares with `ArrowNativeTypeOp::is_lt`, seeded from
+//! `MAX_TOTAL_ORDER`), which puts `-0.0` strictly below `+0.0` -- but SQL comparisons treat
+//! them as equal, and that is what a real `k < r` predicate actually evaluates to: any
+//! `BinaryExpr` comparison, including the one the `NestedLoopJoinExec` oracle in the
+//! differential fuzz test builds its filter from, normalizes `-0.0` to `+0.0` first (see
+//! `apply_cmp` in `datafusion-physical-expr-common`). Both the extreme and the streamed key
+//! array are normalized with [`normalize_float_zero`] before comparing,
+//! after the reduction: normalizing first would not change which value the reduction picks
+//! (`-0.0` and `+0.0` are numerically equal either way), and normalizing only where the
+//! comparison happens keeps the reduction itself agreeing with the unnormalized `min`/`max`
+//! semantics its docs above describe. `NaN` needs no such fix-up: every kernel involved orders
+//! it as the maximum, so it is treated identically on both sides already.
+//!
+//! # Cost
+//!
+//! Let `B` be the buffered rows and `S` the streamed rows: `O(B)` to reduce the buffered side
+//! plus `O(S)` to filter, and no sort on either side. The state retained across batches is one
+//! row; while folding, each buffered partition also holds the key array of the batch it is
+//! reducing, which it accounts against the memory pool for that long.
+//!
+//! This is why the shared state is [`BufferedExtreme`] and not `BufferedSideData`: every field
+//! of the latter -- the concatenated batch, the key array, the visited-indices bitmap, the
+//! final-pass counter -- would be dead here.
+//!
+//! [`PiecewiseMergeJoinExec`]: super::PiecewiseMergeJoinExec
+
+use std::sync::Arc;
+use std::task::{Poll, ready};
+
+use arrow::array::{Array, ArrayRef, RecordBatch, Scalar};
+use arrow::compute::filter_record_batch;
+use arrow::compute::kernels::boolean::not;
+use arrow::compute::kernels::cmp::{gt, gt_eq, lt, lt_eq};
+use arrow_schema::SchemaRef;
+use datafusion_common::utils::normalize_float_zero;
+use datafusion_common::{Result, internal_err};
+use datafusion_execution::{RecordBatchStream, SendableRecordBatchStream};
+use datafusion_expr::{JoinType, Operator};
+use datafusion_physical_expr::PhysicalExprRef;
+use futures::{Stream, StreamExt};
+
+use crate::handle_state;
+use crate::joins::piecewise_merge_join::exec::BufferedExtreme;
+use crate::joins::utils::{
+ BuildProbeJoinMetrics, OnceFut, StatefulStreamResult, boolean_mask_from_filter,
+};
+use crate::stream::EmptyRecordBatchStream;
+
+pub(super) enum RightExistencePWMJStreamState {
+ /// Await the buffered side's reduction to a single key.
+ WaitBufferedExtreme,
+ /// Fetch streamed batches and emit the rows that do (`RightSemi`) or do not
+ /// (`RightAnti`) have a buffered match.
+ ScanStreamBatches,
+ Completed,
+}
+
+pub(super) struct RightExistencePWMJStream {
+ /// Output schema, which for `RightSemi`/`RightAnti` is the streamed side's schema
+ schema: SchemaRef,
+ /// Physical expression evaluated on the streamed side. The buffered side's
+ /// equivalent is already evaluated when the buffered side is collected.
+ on_streamed: PhysicalExprRef,
+ /// `RightSemi` or `RightAnti`
+ join_type: JoinType,
+ /// Comparison operator
+ operator: Operator,
+ streamed: SendableRecordBatchStream,
+ /// Resolves to the whole buffered side reduced to one key. Shared with the other streamed
+ /// partitions, so that reduction happens exactly once.
+ buffered_extreme_fut: OnceFut<BufferedExtreme>,
+ state: RightExistencePWMJStreamState,
+ /// That key, held as a one-element [`Scalar`] so the comparison against a streamed batch is
+ /// one kernel call. `None` when the buffered side has no non-null key, i.e. nothing can ever
+ /// match. Only populated once `buffered_extreme_fut` has resolved.
+ buffered_extreme: Option<Scalar<ArrayRef>>,
+ join_metrics: BuildProbeJoinMetrics,
+}
+
+impl RightExistencePWMJStream {
+ pub(super) fn try_new(
+ schema: SchemaRef,
+ on_streamed: PhysicalExprRef,
+ join_type: JoinType,
+ operator: Operator,
+ streamed: SendableRecordBatchStream,
+ buffered_extreme_fut: OnceFut<BufferedExtreme>,
+ join_metrics: BuildProbeJoinMetrics,
+ ) -> Self {
+ Self {
+ schema,
+ on_streamed,
+ join_type,
+ operator,
+ streamed,
+ buffered_extreme_fut,
+ state: RightExistencePWMJStreamState::WaitBufferedExtreme,
+ buffered_extreme: None,
+ join_metrics,
+ }
+ }
+
+ fn poll_next_impl(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Option<Result<RecordBatch>>> {
+ loop {
+ return match self.state {
+ RightExistencePWMJStreamState::WaitBufferedExtreme => {
+ handle_state!(ready!(self.collect_buffered_extreme(cx)))
+ }
+ RightExistencePWMJStreamState::ScanStreamBatches => {
+ handle_state!(ready!(self.scan_stream_batch(cx)))
+ }
+ RightExistencePWMJStreamState::Completed => Poll::Ready(None),
+ };
+ }
+ }
+
+ /// Picks up the buffered side's reduced key, which is all this join type needs from it.
+ fn collect_buffered_extreme(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>> {
+ let build_timer = self.join_metrics.build_time.timer();
+ let buffered_extreme = ready!(self.buffered_extreme_fut.get_shared(cx))?;
+ build_timer.done();
+
+ // Null exactly when no buffered key is non-null, and NULLs match nothing. Cloned out by
+ // value -- it is one row -- so the shared state is not kept alive by this stream.
+ //
+ // Normalized because the `lt`/`lt_eq`/`gt`/`gt_eq` kernels called in
+ // `filter_streamed_batch` order `-0.0` strictly below `+0.0`, but SQL comparisons --
+ // including `apply_cmp`, which every other `k < r` predicate in the plan goes through
+ // -- treat them as equal. The streamed key array is normalized the same way, right
+ // before that comparison. `min`/`max_batch`, which reduced this extreme, order `-0.0`
+ // below `+0.0` too, so normalizing after the reduction rather than before leaves the
+ // reduction itself agreeing with the unnormalized ordering its own docs describe.
+ let extreme = normalize_float_zero(buffered_extreme.extreme());
+ self.buffered_extreme = (extreme.null_count() == 0).then(|| Scalar::new(extreme));
+
+ // With no non-null buffered key nothing matches, so `RightSemi` outputs nothing
+ // and does not need to read a single streamed batch. `RightAnti` still has to,
+ // since it outputs all of them.
+ self.state = match (&self.buffered_extreme, self.join_type) {
+ (None, JoinType::RightSemi) => {
+ let streamed_schema = self.streamed.schema();
+ self.streamed = Box::pin(EmptyRecordBatchStream::new(streamed_schema));
+ RightExistencePWMJStreamState::Completed
+ }
+ _ => RightExistencePWMJStreamState::ScanStreamBatches,
+ };
+
+ Poll::Ready(Ok(StatefulStreamResult::Continue))
+ }
+
+ /// Fetches one streamed batch and emits the rows it contributes, if any.
+ fn scan_stream_batch(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>> {
+ match ready!(self.streamed.poll_next_unpin(cx)) {
+ None => self.state = RightExistencePWMJStreamState::Completed,
+ Some(Ok(batch)) => {
+ self.join_metrics.input_batches.add(1);
+ self.join_metrics.input_rows.add(batch.num_rows());
+
+ let output = self.filter_streamed_batch(&batch)?;
+ if output.num_rows() > 0 {
+ return Poll::Ready(Ok(StatefulStreamResult::Ready(Some(output))));
+ }
+ // Nothing survived; take the next batch rather than yielding an empty one.
+ }
+ Some(Err(err)) => return Poll::Ready(Err(err)),
+ }
+
+ Poll::Ready(Ok(StatefulStreamResult::Continue))
+ }
+
+ /// Keeps the streamed rows that have a buffered match (`RightSemi`) or that have none
+ /// (`RightAnti`), by comparing each against the single buffered extreme.
+ fn filter_streamed_batch(&self, batch: &RecordBatch) -> Result<RecordBatch> {
+ let columns = match &self.buffered_extreme {
+ // No non-null buffered key, so no streamed row matches and `RightAnti` keeps
+ // the batch whole. `RightSemi` never gets here: it completed without reading
+ // the streamed side.
+ None => batch.columns().to_vec(),
+ Some(extreme) => {
+ let stream_values = normalize_float_zero(
+ &self
+ .on_streamed
+ .evaluate(batch)?
+ .into_array(batch.num_rows())?,
+ );
+
+ // `extreme` is the buffered key, so it goes on the left of the operator,
+ // matching the `buffered OP streamed` orientation of the predicate.
+ let matched = match self.operator {
Review Comment:
Thanks @kumarUjjawal , addressed in 85a002b945fddc8f0eccedc2fab7f64ee041e13d
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org