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