comphead commented on code in PR #25840:
URL: https://github.com/apache/datafusion/pull/25840#discussion_r4157589052
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs:
##########
@@ -1185,6 +1227,313 @@ mod tests {
Ok(())
}
+ /// The pre-sort check that settles never-matching streamed rows must
compare keys
+ /// the way the scan's `JoinKeyComparator` does, which normalizes `-0.0`
to `+0.0`.
+ /// Under IEEE 754 total order (arrow's raw kernels, and the sort) `-0.0 <
+0.0`
+ /// holds, so a check built on that would let the `+0.0` row through as a
candidate
+ /// that the scan then rejects; with SQL semantics both rows are unmatched.
+ #[tokio::test]
+ async fn join_right_less_than_signed_zero_prefilter_agrees_with_scan() ->
Result<()> {
Review Comment:
This overlaps the `float64` case of `matchable_rows_agrees_with_scan` and
the `f64` pass of `fuzz_pwmj_matches_nested_loop`. A classic `RIGHT`/`FULL`
case in `piecewise_merge_join_matrix.slt` Part 1 would replace it and be
checked against `NestedLoopJoin` at batch sizes 1, 2, 100 and 8192, instead of
a hard-coded snapshot. The matrix only has `-0.0` for existence joins today.
The same file could take one unsorted streamed batch mixing matching,
non-matching and NULL keys to cover the `positions` remap. I tried both cases
across the matrix locally and they pass on this head.
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs:
##########
@@ -456,126 +494,49 @@ fn resolve_classic_join(
)?;
let mut buffer_idx = batch_process_state.start_buffer_idx;
- let mut stream_idx = batch_process_state.start_stream_idx;
+ let stream_idx = batch_process_state.start_stream_idx;
+ // NULL keys never match and sort first, so the scan starts past the
buffered ones.
+ // Streamed NULL keys never get here: `matchable_rows` settled them before
the sort.
+ debug_assert_eq!(stream_values[0].null_count(), 0);
if !batch_process_state.processed_null_count {
Review Comment:
With streamed NULLs gone from the scan, `processed_null_count` only seeds
`buffer_idx` with the buffered NULL count once per stream batch. Seeding
`start_buffer_idx` right after `batch_process_state.reset()` in
`fetch_stream_batch` removes the flag, its reset and this first-call branch. I
tried it locally and the PWMJ unit tests and the fuzz still pass.
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs:
##########
@@ -1185,6 +1227,313 @@ mod tests {
Ok(())
}
+ /// The pre-sort check that settles never-matching streamed rows must
compare keys
+ /// the way the scan's `JoinKeyComparator` does, which normalizes `-0.0`
to `+0.0`.
+ /// Under IEEE 754 total order (arrow's raw kernels, and the sort) `-0.0 <
+0.0`
+ /// holds, so a check built on that would let the `+0.0` row through as a
candidate
+ /// that the scan then rejects; with SQL semantics both rows are unmatched.
+ #[tokio::test]
+ async fn join_right_less_than_signed_zero_prefilter_agrees_with_scan() ->
Result<()> {
+ let float_exec =
+ |a: &str, b: &str, keys: Vec<f64>| -> Result<Arc<dyn
ExecutionPlan>> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new(a, DataType::Int32, false),
+ Field::new(b, DataType::Float64, false),
+ ]));
+ let ids = (0..keys.len() as i32).collect::<Vec<_>>();
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(arrow::array::Int32Array::from(ids)),
+ Arc::new(arrow::array::Float64Array::from(keys)),
+ ],
+ )?;
+ Ok(TestMemoryExec::try_new_exec(&[vec![batch]], schema, None)?)
+ };
+ let left = float_exec("a1", "b1", vec![-0.0])?;
+ let right = float_exec("a2", "b2", vec![0.0, -1.0])?;
+ let on = (
+ Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("b2", &right.schema())?) as _,
+ );
+
+ let (_, batches, _) =
+ join_collect(left, right, on, Operator::Lt,
JoinType::Right).await?;
+
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+----+----+------+
+ | a1 | b1 | a2 | b2 |
+ +----+----+----+------+
+ | | | 0 | 0.0 |
+ | | | 1 | -1.0 |
+ +----+----+----+------+
+ ");
+ Ok(())
+ }
+
+ /// `matchable_rows` against the scan's own definition of a match -- some
non-NULL
+ /// buffered row `idx` with `is_match(cmp.compare(row, idx))` under the
scan's
+ /// `JoinKeyComparator` -- for every operator and every kind of key it
dispatches on:
+ /// flat keys through `apply_cmp` (incl. signed zeros, NaN, strings,
dictionaries) and
+ /// nested keys with NULL elements through the comparator. Each is also
checked against
+ /// an all-NULL and an empty buffered side.
+ #[test]
+ fn matchable_rows_agrees_with_scan() -> Result<()> {
+ use arrow::array::{
+ BinaryArray, BooleanArray, Decimal128Array, DictionaryArray,
Float64Array,
+ Int32Array, ListArray, StringArray, StringViewArray,
+ TimestampMicrosecondArray,
+ };
+ use arrow::datatypes::Int32Type;
+
+ let strings =
+ |v: &[Option<&str>]| Arc::new(StringArray::from(v.to_vec())) as
ArrayRef;
+ let cases: Vec<(&str, ArrayRef, ArrayRef)> = vec![
+ (
+ "int32",
+ Arc::new(Int32Array::from(vec![
+ Some(5),
+ None,
+ Some(1),
+ Some(9),
+ Some(3),
+ ])),
+ Arc::new(Int32Array::from(vec![
+ Some(0),
+ Some(1),
+ Some(2),
+ Some(3),
+ Some(5),
+ Some(9),
+ Some(10),
+ None,
+ ])),
+ ),
+ (
+ "float64",
+ Arc::new(Float64Array::from(vec![
+ Some(0.0),
+ Some(-0.0),
+ Some(1.5),
+ Some(f64::NAN),
+ None,
+ ])),
+ Arc::new(Float64Array::from(vec![
+ Some(-0.0),
+ Some(0.0),
+ Some(1.5),
+ Some(f64::NAN),
+ Some(-1.0),
+ Some(2.0),
+ Some(f64::INFINITY),
+ Some(f64::NEG_INFINITY),
+ None,
+ ])),
+ ),
+ (
+ "utf8",
+ strings(&[Some("b"), Some(""), Some("d"), None]),
+ strings(&[
+ Some(""),
+ Some("a"),
+ Some("b"),
+ Some("c"),
+ Some("d"),
+ Some("e"),
+ None,
+ ]),
+ ),
+ (
+ "utf8_view",
+ Arc::new(StringViewArray::from(vec![
+ Some("b"),
+ Some(""),
+ Some("d"),
+ None,
+ ])),
+ Arc::new(StringViewArray::from(vec![
+ Some(""),
+ Some("a"),
+ Some("b"),
+ Some("d"),
+ Some("e"),
+ None,
+ ])),
+ ),
+ (
+ "binary",
+ Arc::new(BinaryArray::from(vec![Some(&b"b"[..]), Some(b""),
None])),
+ Arc::new(BinaryArray::from(vec![
+ Some(&b""[..]),
+ Some(b"a"),
+ Some(b"b"),
+ Some(b"c"),
+ None,
+ ])),
+ ),
+ (
+ "dictionary",
+ Arc::new(
+ vec![Some("b"), None, Some("d"), Some("b")]
+ .into_iter()
+ .collect::<DictionaryArray<Int32Type>>(),
+ ),
+ Arc::new(
+ vec![Some("a"), Some("b"), Some("c"), Some("d"),
Some("e"), None]
+ .into_iter()
+ .collect::<DictionaryArray<Int32Type>>(),
+ ),
+ ),
+ (
+ "decimal128",
+ Arc::new(
+ Decimal128Array::from(vec![Some(150), None, Some(-25)])
+ .with_precision_and_scale(10, 2)?,
+ ),
+ Arc::new(
+ Decimal128Array::from(vec![
+ Some(-26),
+ Some(-25),
+ Some(0),
+ Some(150),
+ Some(151),
+ None,
+ ])
+ .with_precision_and_scale(10, 2)?,
+ ),
+ ),
+ (
+ "date32",
+ Arc::new(Date32Array::from(vec![Some(10), None, Some(20)])),
+ Arc::new(Date32Array::from(vec![
+ Some(9),
+ Some(10),
+ Some(15),
+ Some(20),
+ Some(21),
+ None,
+ ])),
+ ),
+ (
+ "timestamp_us",
+ Arc::new(TimestampMicrosecondArray::from(vec![
+ Some(-5),
+ Some(7),
+ None,
+ ])),
+ Arc::new(TimestampMicrosecondArray::from(vec![
+ Some(-6),
+ Some(-5),
+ Some(7),
+ Some(8),
+ None,
+ ])),
+ ),
+ (
+ "boolean",
+ Arc::new(BooleanArray::from(vec![Some(true), None])),
+ Arc::new(BooleanArray::from(vec![Some(false), Some(true),
None])),
+ ),
+ (
+ "list_with_null_elements",
+ Arc::new(ListArray::from_iter_primitive::<Int32Type, _,
_>(vec![
+ Some(vec![Some(5)]),
+ Some(vec![None]),
+ Some(vec![Some(1), Some(2)]),
+ Some(vec![]),
+ None,
+ ])),
+ Arc::new(ListArray::from_iter_primitive::<Int32Type, _,
_>(vec![
+ Some(vec![None]),
+ Some(vec![Some(7)]),
+ Some(vec![Some(1)]),
+ Some(vec![Some(5)]),
+ Some(vec![]),
+ Some(vec![None, Some(1)]),
+ Some(vec![Some(5), None]),
+ None,
+ ])),
+ ),
+ (
+ // No empty list, so the extreme key has a first element to
compare
+ // against the streamed NULL elements.
+ "list_null_elements_vs_extreme",
+ Arc::new(ListArray::from_iter_primitive::<Int32Type, _,
_>(vec![
+ Some(vec![Some(5)]),
+ Some(vec![Some(3)]),
+ None,
+ ])),
+ Arc::new(ListArray::from_iter_primitive::<Int32Type, _,
_>(vec![
+ Some(vec![None]),
+ Some(vec![Some(3), None]),
+ Some(vec![Some(4)]),
+ Some(vec![Some(7)]),
+ Some(vec![Some(1)]),
+ None,
+ ])),
+ ),
+ ];
+
+ for (name, buffered, streamed) in cases {
+ let all_null = new_null_array(buffered.data_type(), 3);
+ let empty = buffered.slice(0, 0);
+ for (side, buffered) in
+ [("full", buffered), ("all_null", all_null), ("empty", empty)]
+ {
+ for operator in
+ [Operator::Lt, Operator::LtEq, Operator::Gt,
Operator::GtEq]
+ {
+ // As `PiecewiseMergeJoinExec::try_new` sorts both sides.
+ let sort_options = match operator {
+ Operator::Lt | Operator::LtEq =>
SortOptions::new(true, true),
+ _ => SortOptions::new(false, true),
+ };
+ let sorted = take(
+ buffered.as_ref(),
+ &sort_to_indices(buffered.as_ref(),
Some(sort_options), None)?,
+ None,
+ )?;
+ let extreme = match sorted.len().checked_sub(1) {
Review Comment:
This copies the extreme computation from `collect_buffered_side`, so the
test would not notice a bug in the production version. A small shared function,
for example `fn buffered_extreme(values: &ArrayRef) ->
Result<Option<ColumnarValue>>`, would remove the duplicate and make the test
exercise the real code.
--
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]
