SubhamSinghal commented on code in PR #25840:
URL: https://github.com/apache/datafusion/pull/25840#discussion_r4236524751
##########
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:
Addressed in 35fa8f25fda5d87d0c7182425d01486eaed75d4a
--
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]
