jayzhan211 commented on code in PR #25840:
URL: https://github.com/apache/datafusion/pull/25840#discussion_r4235903494
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs:
##########
@@ -633,18 +595,121 @@ fn build_matched_indices_and_mark_buffered(
)?)
}
-// Creates a record batch from the unmatched indices on the streamed side
-fn create_unmatched_batch(
- streamed_indices: &mut PrimitiveBuilder<UInt32Type>,
- stream_batch: &SortedStreamBatch,
+// The last non-null key of the sorted buffered side, or `None` when it is
empty or every key
+// is NULL. NULLs sort last for `<`/`<=`, so this is not simply the last key.
Logical, as a
+// run-end encoded or `Null`-typed NULL has no physical null buffer.
+fn buffered_extreme(
+ values: &ArrayRef,
+ sort_options: SortOptions,
+) -> Result<Option<ColumnarValue>> {
+ let (start, end) = non_null_range(values.as_ref(), sort_options);
+ if start == end {
+ return Ok(None);
+ }
+ let last = values.slice(end - 1, 1);
+ // `apply_cmp` normalizes `-0.0` only in flat float scalars, not inside a
+ // `ScalarValue::Dictionary`, so normalize the key before taking it.
+ let extreme = normalize_float_zero(&last);
+ Ok(Some(ColumnarValue::Scalar(ScalarValue::try_from_array(
+ &extreme, 0,
+ )?)))
+}
+
+// Which rows of `stream_values` can match at least one buffered row.
+//
+// Every match set is a suffix `[k, non_null_end)` of the sorted buffered
side's non-null keys,
+// so a streamed row matches anything at all iff it matches the last of them,
`buffered_extreme`
+// (`None` when no buffered key is non-null, so nothing can match). A NULL on
either side
+// compares to NULL, which is no match.
+//
+// This must agree exactly with the scan's `JoinKeyComparator`, or it would
change which rows
+// match. One vectorized `apply_cmp` does, nested keys included: it normalizes
`-0.0` to
+// `+0.0` as the comparator does, and orders NULL elements inside a nested key
the same way.
+fn matchable_rows(
+ stream_values: &ArrayRef,
+ buffered_values: &ArrayRef,
+ buffered_extreme: Option<&ColumnarValue>,
+ operator: Operator,
+ sort_options: SortOptions,
+) -> Result<BooleanArray> {
+ let num_rows = stream_values.len();
+ let Some(extreme) = buffered_extreme else {
+ return Ok(BooleanArray::new(BooleanBuffer::new_unset(num_rows), None));
+ };
+
+ // Run-end encoded keys are decided with the scan's own comparator:
arrow's run-end
+ // comparison kernel overflows on an empty batch sliced past its first
run, which the
+ // scan never reaches.
+ if matches!(stream_values.data_type(), DataType::RunEndEncoded(_, _)) {
+ let match_on_equal = matches_on_equal(operator)?;
+ let cmp = JoinKeyComparator::new(
+ &[Arc::clone(stream_values)],
+ &[Arc::clone(buffered_values)],
+ &[sort_options],
+ NullEquality::NullEqualsNothing,
+ )?;
+ let last = non_null_range(buffered_values.as_ref(), sort_options).1 -
1;
+ // Logical NULLs: a run-end encoded NULL has no physical null buffer.
+ let nulls = stream_values.logical_nulls();
+ let matchable = BooleanBuffer::collect_bool(num_rows, |row| {
+ nulls.as_ref().is_none_or(|nulls| nulls.is_valid(row))
+ && is_match(cmp.compare(row, last), match_on_equal)
+ });
+ return Ok(BooleanArray::new(matchable, None));
+ }
+
+ // The predicate is oriented `buffered OP streamed`.
+ let result = apply_cmp(
+ operator,
+ extreme,
+ &ColumnarValue::Array(Arc::clone(stream_values)),
+ )?
+ .into_array(num_rows)?;
+ let result = result.as_boolean();
Review Comment:
`apply_cmp_for_nested` masks only physical nulls (`l.nulls()` /
`r.nulls()`). For `Dictionary<_, List>` those are the keys' nulls, so a valid
key pointing at a NULL list passes as matchable and then hits the scan guard:
`Internal error: PiecewiseMergeJoin: streamed NULL keys must be settled before
the scan.` (Right join, `>`). The previous head sent `is_nested()` types,
dictionary-of-nested included, through the comparator path, which used
`logical_nulls()`.
Repro: add this to `matchable_rows_agrees_with_scan`. It fails on this head
with `dictionary_list_null_values/full >: streamed row 1`:
```rs
(
// A valid key pointing at a NULL list: logically NULL, physically valid.
"dictionary_list_null_values",
Arc::new(DictionaryArray::<Int32Type>::new(
Int32Array::from(vec![0, 1, 2]),
Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
Some(vec![Some(5)]),
None,
Some(vec![Some(1)]),
])),
)),
Arc::new(DictionaryArray::<Int32Type>::new(
Int32Array::from(vec![Some(0), Some(1), Some(2), Some(3), None]),
Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
Some(vec![Some(2)]),
None,
Some(vec![Some(6)]),
Some(vec![Some(0)]),
])),
)),
),
```
Fix (passes the case above and the end-to-end join):
```diff
let result = result.as_boolean();
- Ok(if result.null_count() > 0 {
+ let matchable = if result.null_count() > 0 {
prep_null_mask_filter(result)
} else {
result.clone()
- })
+ };
+ // The nested `apply_cmp` masks only physical NULLs, so a dictionary
key pointing at a
+ // NULL nested value would pass.
+ Ok(match stream_values.logical_nulls() {
+ Some(nulls) => BooleanArray::new(matchable.values() &
nulls.inner(), None),
+ None => matchable,
+ })
```
--
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]
