You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "patrickswedish (via GitHub)" <gi...@apache.org> on 2026/10/03 17:46:34 UTC
[PR] fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator [datafusion]
patrickswedish opened a new pull request, #26006:
URL: https://github.com/apache/datafusion/pull/26006
# Which issue does this PR close?
Closes #26003.
# Rationale for this change
In `BitwiseSortMergeJoinStream` (which powers sort-merge semi, anti, and mark joins), when a key group spans across batch boundaries, `keys_match` verifies key equality by calling `compare_join_arrays`.
Prior to this change:
1. `compare_join_arrays` performed a manual downcast across a restricted subset of primitive Arrow types and returned `not_impl_err!("Unsupported data type in sort merge join comparator: {}", dt)` for other data types such as:
- Timezone-aware timestamps (`DataType::Timestamp(_, Some(_))`)
- Dictionary-encoded keys (`DataType::Dictionary(..)`)
- `DataType::Decimal256(..)`
- Time keys (`DataType::Time32(..)`, `DataType::Time64(..)`)
- Duration and Interval keys
2. As a consequence, queries performing semi, anti, or mark sort-merge joins over timezone-aware timestamps or dictionary keys failed at runtime during batch continuation with an "Unsupported data type" error.
3. In addition, `keys_match` previously contained an ad-hoc branch for floating-point arrays to handle IEEE-754 `-0.0` normalization and NaN comparisons via `JoinKeyComparator`, while other types fell back to `compare_join_arrays`.
# What changes are included in this PR?
1. **Route `compare_join_arrays` directly to `JoinKeyComparator`**:
- Replaced the manual match table in `compare_join_arrays` with `JoinKeyComparator::new(left_arrays, right_arrays, sort_options, null_equality)?.compare(left, right)`.
- `JoinKeyComparator` builds Arrow's `make_comparator` closures once per batch pair, providing support for all Arrow data types, preserving IEEE-754 `-0.0` normalization, and respecting `NullEquality` modes consistently across all join streams.
2. **Clean up `keys_match` in `bitwise_stream.rs`**:
- Slices the first row of `right_arrays` and evaluates `JoinKeyComparator::new(left_arrays, &right_keys, sort_options, null_equality)?.is_equal(0, 0)`.
- Eliminates the special-cased float branch in `keys_match` since all types are now handled uniformly.
3. **Tests**:
- Added unit test `test_compare_join_arrays_extended_types` in `joins::utils::tests` validating comparisons across timezone-aware timestamps (`TimestampMicrosecondArray` with `"UTC"`), dictionary arrays (`Dictionary(Int32, Utf8)`), `Decimal256Array`, `Time64NanosecondArray`, and `Time32MillisecondArray`.
# Are these changes tested?
Yes:
- New unit test `test_compare_join_arrays_extended_types` added in `datafusion/physical-plan/src/joins/utils.rs`.
- All 247 existing sort-merge join tests pass (`cargo test --package datafusion-physical-plan --lib joins::sort_merge_join::tests`).
- Formatted with `rustfmt` and verified with `cargo clippy --package datafusion-physical-plan --lib -- -D warnings` (0 warnings).
# Are there any user-facing changes?
No breaking changes. Users can now execute sort-merge semi, anti, and mark joins over timezone-aware timestamps, dictionary keys, `Decimal256`, and time keys without errors.
--
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] fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #26006:
URL: https://github.com/apache/datafusion/pull/26006#issuecomment-5972020363
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/26006?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.15596%` with `14 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 82.64%. Comparing base ([`95a1961`](https://app.codecov.io/gh/apache/datafusion/commit/95a1961cc7ec95bbb7b4015bfb0e2ed814b215c0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`d17ad3f`](https://app.codecov.io/gh/apache/datafusion/commit/d17ad3f11747710c15e98eba79b6927887a4c5dc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/26006?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [datafusion/physical-plan/src/joins/utils.rs](https://app.codecov.io/gh/apache/datafusion/pull/26006?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Fjoins%2Futils.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9qb2lucy91dGlscy5ycw==) | 87.61% | [10 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/26006?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...l-plan/src/joins/sort\_merge\_join/bitwise\_stream.rs](https://app.codecov.io/gh/apache/datafusion/pull/26006?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Fjoins%2Fsort_merge_join%2Fbitwise_stream.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9qb2lucy9zb3J0X21lcmdlX2pvaW4vYml0d2lzZV9zdHJlYW0ucnM=) | 75.00% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/26006?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 #26006 +/- ##
=======================================
Coverage 82.64% 82.64%
=======================================
Files 1147 1147
Lines 445502 445555 +53
Branches 445502 445555 +53
=======================================
+ Hits 368179 368231 +52
+ Misses 54969 54960 -9
- Partials 22354 22364 +10
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/26006?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] fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #26006:
URL: https://github.com/apache/datafusion/pull/26006#issuecomment-6382873251
This PR appears to be waiting on the author to respond to reviewer feedback so marking as draft. Please mark ready to review when it is ready for another look
--
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] fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator [datafusion]
Posted by "patrickswedish (via GitHub)" <gi...@apache.org>.
patrickswedish commented on PR #26006:
URL: https://github.com/apache/datafusion/pull/26006#issuecomment-6385697611
Thank you @kosiew for the thorough architectural review and approval!
I have addressed all the non-blocking suggestions:
1. **Dictionary Cross-Code Ordering**: Updated est_compare_join_arrays_extended_types using DictionaryArray::try_new with independently ordered codes (pple at code 0 in left, code 1 in right; anana at code 1 in left, cherry at code 0 in right) to verify value-based comparison across differing encodings.
2. **Table-Driven Comparator Tests**: Consolidated the extended type test cases into a structured table loop with descriptive failure assertions.
3. **Execution-Level Stream Boundary Regression**: Added join_timestamp_tz_key_batch_boundaries across 96 parameter combinations covering timezone-aware timestamp keys with duplicate groups continuing across batch boundaries for semi, anti, and mark joins.
All tests pass locally. Marked ready for review @alamb.
--
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] fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #26006:
URL: https://github.com/apache/datafusion/pull/26006#discussion_r4370081685
##########
datafusion/physical-plan/src/joins/utils.rs:
##########
@@ -2664,88 +2664,9 @@ pub fn compare_join_arrays(
sort_options: &[SortOptions],
null_equality: NullEquality,
) -> Result<Ordering> {
- let mut res = Ordering::Equal;
- for ((left_array, right_array), sort_options) in
- left_arrays.iter().zip(right_arrays).zip(sort_options)
- {
- macro_rules! compare_value {
- ($T:ty) => {{
- let left_array = left_array.as_any().downcast_ref::<$T>().unwrap();
- let right_array = right_array.as_any().downcast_ref::<$T>().unwrap();
- match (left_array.is_null(left), right_array.is_null(right)) {
- (false, false) => {
- let left_value = &left_array.value(left);
- let right_value = &right_array.value(right);
- res = left_value.partial_cmp(right_value).unwrap();
- if sort_options.descending {
- res = res.reverse();
- }
- }
- (true, false) => {
- res = if sort_options.nulls_first {
- Ordering::Less
- } else {
- Ordering::Greater
- };
- }
- (false, true) => {
- res = if sort_options.nulls_first {
- Ordering::Greater
- } else {
- Ordering::Less
- };
- }
- _ => {
- res = match null_equality {
- NullEquality::NullEqualsNothing => Ordering::Less,
- NullEquality::NullEqualsNull => Ordering::Equal,
- };
- }
- }
- }};
- }
-
- match left_array.data_type() {
- DataType::Null => {}
- DataType::Boolean => compare_value!(BooleanArray),
- DataType::Int8 => compare_value!(Int8Array),
- DataType::Int16 => compare_value!(Int16Array),
- DataType::Int32 => compare_value!(Int32Array),
- DataType::Int64 => compare_value!(Int64Array),
- DataType::UInt8 => compare_value!(UInt8Array),
- DataType::UInt16 => compare_value!(UInt16Array),
- DataType::UInt32 => compare_value!(UInt32Array),
- DataType::UInt64 => compare_value!(UInt64Array),
- DataType::Float32 => compare_value!(Float32Array),
- DataType::Float64 => compare_value!(Float64Array),
- DataType::Binary => compare_value!(BinaryArray),
- DataType::BinaryView => compare_value!(BinaryViewArray),
- DataType::FixedSizeBinary(_) => compare_value!(FixedSizeBinaryArray),
- DataType::LargeBinary => compare_value!(LargeBinaryArray),
- DataType::Utf8 => compare_value!(StringArray),
- DataType::Utf8View => compare_value!(StringViewArray),
- DataType::LargeUtf8 => compare_value!(LargeStringArray),
- DataType::Decimal128(..) => compare_value!(Decimal128Array),
- DataType::Timestamp(time_unit, None) => match time_unit {
- TimeUnit::Second => compare_value!(TimestampSecondArray),
- TimeUnit::Millisecond => compare_value!(TimestampMillisecondArray),
- TimeUnit::Microsecond => compare_value!(TimestampMicrosecondArray),
- TimeUnit::Nanosecond => compare_value!(TimestampNanosecondArray),
- },
- DataType::Date32 => compare_value!(Date32Array),
- DataType::Date64 => compare_value!(Date64Array),
- dt => {
- return not_impl_err!(
- "Unsupported data type in sort merge join comparator: {}",
- dt
- );
- }
- }
- if !res.is_eq() {
- break;
- }
- }
- Ok(res)
+ let cmp =
Review Comment:
nit: nothing in this crate calls `compare_join_arrays` now, but it's still public. Building the comparator over the full arrays makes every call O(len): `normalize_float_zero` scans all the float values and `logical_nulls` scans dictionary keys. Slicing both sides to the one row first keeps it O(1) per call, like before:
```diff
- let cmp =
- JoinKeyComparator::new(left_arrays, right_arrays, sort_options, null_equality)?;
- Ok(cmp.compare(left, right))
+ let left_row: Vec<ArrayRef> = left_arrays.iter().map(|a| a.slice(left, 1)).collect();
+ let right_row: Vec<ArrayRef> =
+ right_arrays.iter().map(|a| a.slice(right, 1)).collect();
+ let cmp = JoinKeyComparator::new(&left_row, &right_row, sort_options, null_equality)?;
+ Ok(cmp.compare(0, 0))
```
--
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] fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb closed pull request #26006: fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator
URL: https://github.com/apache/datafusion/pull/26006
--
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] fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator [datafusion]
Posted by "patrickswedish (via GitHub)" <gi...@apache.org>.
patrickswedish commented on PR #26006:
URL: https://github.com/apache/datafusion/pull/26006#issuecomment-6399803409
Thank you @jayzhan211 for the thorough review, approval, and the sharp catch on `compare_join_arrays`!
You are completely right — constructing `JoinKeyComparator` over the entire array buffers incurred an unnecessary O(N) scan (`normalize_float_zero` and dictionary key inspection) for what should be a constant-time single-row comparison. Slicing both sides to single-row views (`a.slice(..., 1)`) preserves strictly O(1) execution per call without full-array normalization overhead.
Applied your exact suggestion in [`154624c`](https://github.com/apache/datafusion/pull/26006/commits/154624c4129a6bceb5d58eb305c12f8446e041fe). All checks and tests passing cleanly and ready for merge!
--
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] fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #26006:
URL: https://github.com/apache/datafusion/pull/26006#discussion_r4345705719
##########
datafusion/physical-plan/src/joins/utils.rs:
##########
@@ -5293,4 +5215,165 @@ mod tests {
Inexact(10)
);
}
+
+ #[test]
+ fn test_compare_join_arrays_extended_types() -> Result<()> {
+ use arrow::array::{
+ Decimal256Array, Time32MillisecondArray, Time64NanosecondArray,
+ };
+ let sort_opts = vec![SortOptions::default()];
+
+ // 1. Timezone-aware timestamp
+ let ts_left: ArrayRef = Arc::new(
+ TimestampMicrosecondArray::from(vec![1_000_000, 2_000_000])
+ .with_timezone("UTC"),
+ );
+ let ts_right: ArrayRef = Arc::new(
+ TimestampMicrosecondArray::from(vec![1_000_000, 3_000_000])
+ .with_timezone("UTC"),
+ );
+ assert_eq!(
+ compare_join_arrays(
+ &[Arc::clone(&ts_left)],
+ 0,
+ &[Arc::clone(&ts_right)],
+ 0,
+ &sort_opts,
+ NullEquality::NullEqualsNothing
+ )?,
+ Ordering::Equal
+ );
+ assert_eq!(
+ compare_join_arrays(
+ &[ts_left],
+ 1,
+ &[ts_right],
+ 1,
+ &sort_opts,
+ NullEquality::NullEqualsNothing
+ )?,
+ Ordering::Less
+ );
+
+ // 2. Dictionary(Int32, Utf8)
+ let dict_left: ArrayRef = Arc::new(
Review Comment:
Could we also test equal dictionary values with different codes using independently ordered dictionaries built with `DictionaryArray::try_new`? The existing `banana < cherry` check already catches raw-code ordering, but the equal `apple` values both use code 0, so equality across different encodings remains untested. Please keep the existing inequality assertion.
##########
datafusion/physical-plan/src/joins/utils.rs:
##########
@@ -5293,4 +5215,165 @@ mod tests {
Inexact(10)
);
}
+
+ #[test]
+ fn test_compare_join_arrays_extended_types() -> Result<()> {
Review Comment:
As a non-blocking cleanup, consider collecting the five `(name, left, right)` array pairs into a small table and running the shared equal/less assertions in a loop, with the case name in assertion messages. This would remove repeated comparator calls and make additional type coverage easier to maintain.
##########
datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs:
##########
@@ -1216,30 +1216,10 @@ fn keys_match(
null_equality: NullEquality,
) -> Result<bool> {
debug_assert!(left_arrays.iter().all(|a| a.len() == 1));
- if left_arrays
- .iter()
- .any(|array| array.data_type().is_floating())
- {
- // The scalar comparator's partial_cmp panics on NaN. Match the merge
- // scan's floating-point equality, including its signed-zero handling.
- let right_keys = slice_keys(right_arrays, 0);
- return Ok(JoinKeyComparator::new(
- left_arrays,
- &right_keys,
- sort_options,
- null_equality,
- )?
- .is_equal(0, 0));
- }
- let cmp = compare_join_arrays(
- left_arrays,
- 0,
- right_arrays,
- 0,
- sort_options,
- null_equality,
- )?;
- Ok(cmp == Ordering::Equal)
+ let right_keys = slice_keys(right_arrays, 0);
+ let cmp =
+ JoinKeyComparator::new(left_arrays, &right_keys, sort_options, null_equality)?;
Review Comment:
Could we add a non-blocking execution-level regression using the existing `join_float_key_batch_boundaries` harness with timezone-aware timestamp keys? Please cover groups ending and continuing across batch boundaries, asserting semi/anti row IDs and mark bits, since the current tests exercise the helper rather than this stream's continuation path.
--
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] fix: route sort-merge semi/anti/mark key equality through JoinKeyComparator [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #26006:
URL: https://github.com/apache/datafusion/pull/26006#issuecomment-6403505702
Related
- https://github.com/apache/datafusion/pull/26189#issuecomment-6317342804
--
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