perf(asof_join): cache ASOF join match comparator, drop per-row ScalarValue materialization - #24607
perf(asof_join): cache ASOF join match comparator, drop per-row ScalarValue materialization#24607jayzhan211 wants to merge 1 commit into
Conversation
The merge loop built a `ScalarValue` for every match value it looked at: once per left row, and once per right row scanned as a candidate. Each one went through `ScalarValue::try_from_array`, and `is_eligible` then compared them with `ScalarValue::try_cmp`, re-dispatching on type for every single comparison. Compare the arrays directly instead. `InputCursor` caches the logical null buffer at batch load so a row's NULL check is a bitmap lookup, and `AsOfJoinStream` caches an Arrow `DynComparator` keyed on the (right, left) batch-id pair, mirroring how the equality-group comparators are already cached. Type dispatch now happens once per batch pair rather than once per comparison, and no scalars are materialized on the hot path. `is_eligible` drops to `(Operator, Ordering) -> bool`. The comparator always yields natural ascending `right.cmp(left)`, so the four ASOF operators stay the only thing interpreting scan direction. Note: `make_comparator` orders floats by IEEE totalOrder, so `-0.0` no longer compares equal to `+0.0` for float match keys. The previous per-row `normalize_float_zero_scalar` call is gone with `match_value`.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #24607 +/- ##
==========================================
+ Coverage 81.38% 81.39% +0.01%
==========================================
Files 1116 1118 +2
Lines 397960 398711 +751
Branches 397960 398711 +751
==========================================
+ Hits 323880 324535 +655
- Misses 55120 55208 +88
- Partials 18960 18968 +8 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
kosiew
left a comment
kosiew
left a comment
There was a problem hiding this comment.
Thanks for working on this. The comparator caching and removal of per-row ScalarValue materialization look like a solid improvement overall.
I found one behavior change around signed zero for floating-point match keys that I think should be addressed before merging. I also left one non-blocking test suggestion for the new dictionary/null and multi-batch path.
| "ASOF left match array is missing" | ||
| ) | ||
| })?; | ||
| let comparator = make_comparator( |
There was a problem hiding this 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:
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.
| "ASOF left match array is missing" | ||
| ) | ||
| })?; | ||
| let comparator = make_comparator( |
There was a problem hiding this 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.
3 participants
Part of #23738
Performance-related changes split out from #24519. Normalize negative 0 case is not included in this PR.