jayzhan211 commented on code in PR #25840:
URL: https://github.com/apache/datafusion/pull/25840#discussion_r4166675528
##########
datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs:
##########
@@ -631,18 +587,111 @@ 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 key of the sorted buffered side, or `None` when it is empty or
every key is NULL:
+// NULLs sort first, so the last key is NULL only when every buffered key is.
+fn buffered_extreme(values: &ArrayRef) -> Result<Option<ColumnarValue>> {
+ Ok(match values.len().checked_sub(1) {
+ Some(last) if values.is_valid(last) => Some(ColumnarValue::Scalar(
+ ScalarValue::try_from_array(values, last)?,
Review Comment:
For `Dictionary(_, Float*)` keys the pre-filter disagrees with the scan on
`-0.0`. `apply_cmp` normalizes the streamed array (dictionary values included),
but `normalize_float_zero_scalar` skips `ScalarValue::Dictionary`, so the
extreme keeps `-0.0` under total order while `JoinKeyComparator` treats it as
`0.0`. This gives an internal error when the filter lets a row through, and
wrong rows when it drops a real match. The base commit returns the right
results.
Repro:
```sql
set datafusion.optimizer.enable_piecewise_merge_join = true;
CREATE TABLE dl AS SELECT * FROM (VALUES (1, arrow_cast(-0.0,
'Dictionary(Int32, Float64)')), (2, arrow_cast(0.5, 'Dictionary(Int32,
Float64)'))) t(lid, lv);
CREATE TABLE dr AS SELECT * FROM (VALUES (1, arrow_cast(0.0,
'Dictionary(Int32, Float64)')), (2, arrow_cast(0.25, 'Dictionary(Int32,
Float64)'))) t(rid, rv);
SELECT * FROM dl JOIN dr ON dl.lv < dr.rv;
-- Internal error: PiecewiseMergeJoin: streamed row 1 reached the scan
without a match
CREATE TABLE dl2 AS SELECT * FROM (VALUES (1, arrow_cast(-1.0,
'Dictionary(Int32, Float64)')), (2, arrow_cast(-0.0, 'Dictionary(Int32,
Float64)'))) t(lid, lv);
CREATE TABLE dr2 AS SELECT * FROM (VALUES (1, arrow_cast(0.0,
'Dictionary(Int32, Float64)'))) t(rid, rv);
SELECT * FROM dl2 RIGHT JOIN dr2 ON dl2.lv >= dr2.rv;
-- got `NULL NULL 1 0`, expected `2 0 1 0` (NLJ and the base commit agree)
```
Fix (normalize the extreme before taking it as a scalar):
```diff
+use datafusion_common::utils::normalize_float_zero;
```
```diff
fn buffered_extreme(values: &ArrayRef) -> Result<Option<ColumnarValue>> {
Ok(match values.len().checked_sub(1) {
- Some(last) if values.is_valid(last) => Some(ColumnarValue::Scalar(
- ScalarValue::try_from_array(values, last)?,
- )),
+ // `apply_cmp` normalizes `-0.0` only in flat float scalars, not
inside a
+ // `ScalarValue::Dictionary`, so normalize the key before taking it.
+ Some(last) if values.is_valid(last) => {
+ let extreme = normalize_float_zero(&values.slice(last, 1));
+ Some(ColumnarValue::Scalar(ScalarValue::try_from_array(
+ &extreme, 0,
+ )?))
+ }
_ => None,
})
}
```
Cases for `matchable_rows_agrees_with_scan`. They fail on the current head
(`dictionary_float64_neg_zero_min/full <: streamed row 0`) and pass with the
fix:
```rs
(
// `-0.0` as the buffered extreme inside a dictionary: the smallest
// key for `<`/`<=`, the largest for `>`/`>=`.
"dictionary_float64_neg_zero_min",
Arc::new(DictionaryArray::<Int32Type>::new(
Int32Array::from(vec![0, 1]),
Arc::new(Float64Array::from(vec![-0.0, 0.5])),
)),
Arc::new(DictionaryArray::<Int32Type>::new(
Int32Array::from(vec![0, 1, 2]),
Arc::new(Float64Array::from(vec![0.0, -0.0, 0.25])),
)),
),
(
"dictionary_float64_neg_zero_max",
Arc::new(DictionaryArray::<Int32Type>::new(
Int32Array::from(vec![0, 1]),
Arc::new(Float64Array::from(vec![-1.0, -0.0])),
)),
Arc::new(DictionaryArray::<Int32Type>::new(
Int32Array::from(vec![0, 1, 2]),
Arc::new(Float64Array::from(vec![0.0, -0.0, -0.5])),
)),
),
```
The root cause, `normalize_float_zero_scalar` not recursing into
`ScalarValue::Dictionary`, is already on main: `SELECT arrow_cast(0.0,
'Dictionary(Int32, Float64)') > arrow_cast(-0.0, 'Dictionary(Int32, Float64)')`
returns `true`. Worth filing on its own; fixing it there would also cover this.
--
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]
