kosiew commented on code in PR #24607:
URL: https://github.com/apache/datafusion/pull/24607#discussion_r3878963290
##########
datafusion/physical-plan/src/joins/asof_join.rs:
##########
@@ -982,6 +999,50 @@ impl AsOfJoinStream {
Ok(comparator.compare(candidate.row, self.left.row) != Ordering::Equal)
}
+ /// Compares the current right match value with the current left match
value.
+ ///
+ /// This always returns natural ascending order (`right.cmp(left)`),
regardless
+ /// of scan direction. [`is_eligible`] interprets that order for the four
ASOF
+ /// operators, keeping direction-specific logic out of the comparator
cache.
+ fn compare_input_matches(&mut self) -> Result<Ordering> {
+ let _timer = self.metrics.baseline.elapsed_compute().timer();
+ let right_batch_id = self.right.key_batch_id;
+ let left_batch_id = self.left.key_batch_id;
+ if self
+ .input_match_comparator
+ .as_ref()
+ .is_none_or(|(right, left, _)| {
+ *right != right_batch_id || *left != left_batch_id
+ })
+ {
+ let right = self.right.match_array.as_ref().ok_or_else(|| {
+ datafusion_common::internal_datafusion_err!(
+ "ASOF right match array is missing"
+ )
+ })?;
+ let left = self.left.match_array.as_ref().ok_or_else(|| {
+ datafusion_common::internal_datafusion_err!(
+ "ASOF left match array is missing"
+ )
+ })?;
+ let comparator = make_comparator(
Review Comment:
I think this changes ASOF match semantics for floating-point match keys
because the previous signed-zero normalization is no longer applied.
Float match expressions are still accepted here. The validation only rejects
floating-point types for equality keys, while match expressions are checked for
left/right type agreement.
Previously, `match_value()` called `normalize_float_zero_scalar()`. That was
important because `ScalarValue::partial_cmp` already uses `total_cmp` for
Float16/32/64. Without the normalization, `-0.0` and `+0.0` are ordered rather
than treated as equal. Arrow's `make_comparator` also uses total ordering,
where `-0.0 < +0.0`.
I reproduced this with a Float64 match key, no equality keys, `GtEq`, left
`ts = -0.0`, and right `ts = +0.0`:
| before | after |
| --- | --- |
| matched, `price = 99` | no match, `price = NULL` |
This also conflicts with the documented signed-zero invariant in this
operator.
I think the clean fix is to normalize once per batch when the match array is
evaluated, for example:
```rust
let match_array = normalize_float_zero(
&self.match_expr.evaluate(&batch)?.into_array(batch.num_rows())?,
);
```
That keeps the normalization cost at one scan per batch instead of per row,
and mapping `-0.0` to `+0.0` should preserve the merge ordering invariants.
Could you also add a regression test for the signed-zero case?
A couple of notes on scope: NaN behavior appears unchanged because both
paths use total ordering. Also, `normalize_float_zero` only handles top-level
Float16/32/64, so dictionary-encoded float values would retain the existing
limitation.
##########
datafusion/physical-plan/src/joins/asof_join.rs:
##########
@@ -982,6 +999,50 @@ impl AsOfJoinStream {
Ok(comparator.compare(candidate.row, self.left.row) != Ordering::Equal)
}
+ /// Compares the current right match value with the current left match
value.
+ ///
+ /// This always returns natural ascending order (`right.cmp(left)`),
regardless
+ /// of scan direction. [`is_eligible`] interprets that order for the four
ASOF
+ /// operators, keeping direction-specific logic out of the comparator
cache.
+ fn compare_input_matches(&mut self) -> Result<Ordering> {
+ let _timer = self.metrics.baseline.elapsed_compute().timer();
+ let right_batch_id = self.right.key_batch_id;
+ let left_batch_id = self.left.key_batch_id;
+ if self
+ .input_match_comparator
+ .as_ref()
+ .is_none_or(|(right, left, _)| {
+ *right != right_batch_id || *left != left_batch_id
+ })
+ {
+ let right = self.right.match_array.as_ref().ok_or_else(|| {
+ datafusion_common::internal_datafusion_err!(
+ "ASOF right match array is missing"
+ )
+ })?;
+ let left = self.left.match_array.as_ref().ok_or_else(|| {
+ datafusion_common::internal_datafusion_err!(
+ "ASOF left match array is missing"
+ )
+ })?;
+ let comparator = make_comparator(
Review Comment:
One non-blocking suggestion: it would be useful to add coverage for nullable
dictionary-encoded match keys across multiple input batches.
This is a genuinely new path now. Dictionary values go through
`compare_dict` in `make_comparator` instead of being decoded into
`ScalarValue`, and null handling now comes from `logical_nulls()` instead of
`ScalarValue::is_null()`.
The multi-batch case is especially useful because `input_match_comparator`
is cached by `(right.key_batch_id, left.key_batch_id)`. Existing tests do not
appear to exercise enough batch transitions to cover cache invalidation.
The cache key itself looks sound to me since both batch counters are
monotonic per cursor, but a regression test would make that behavior much
easier to protect.
--
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]
