refactor(asof_join): enhance match value handling and comparator caching

[jayzhan211]

• Part of feat: support ASOF joins #23738

Rationale for this change

ASOF joins currently convert both match values into ScalarValues for row-level
comparisons. This adds repeated scalar construction and type dispatch to the
ordered scan hot path.

The match expressions are already evaluated into Arrow arrays for each input
batch, so the join can compare array positions directly using an Arrow
comparator constructed once per left/right batch pair.

What changes are included in this PR?

• Compare ASOF match keys directly from their evaluated Arrow arrays.
• Cache the match comparator for each left/right batch pair.
• Cache logical match-key nulls so NULL checks do not require scalar
  materialization.
• Normalize floating-point signed zero once per batch so Arrow comparisons
  preserve SQL semantics where -0.0 and +0.0 compare equal.
• Document why comparator caching, batch-level normalization, shared Arrow
  buffers, and forward-only scanning are important to the hot path.

Are these changes tested?

Yes.

• Added a regression test verifying that floating-point -0.0 and +0.0
  remain equal for ASOF match conditions.
• Existing ASOF tests cover equality groups, NULL handling, batch boundaries,
  comparison directions, shared build memory, and expression validation.

The following checks pass:

• cargo test -p datafusion-physical-plan asof_join --lib
• cargo clippy -p datafusion-physical-plan --lib -- -D warnings
• cargo fmt --all -- --check

Are there any user-facing changes?

No API or query-semantic changes are intended.

This is an internal execution optimization. Floating-point signed-zero and
logical-NULL behavior are explicitly preserved.

[jayzhan211]

@2010YOUY01 @Xuanwo Could you review this when you have a moment? Thanks!

[codecov-commenter]

Codecov Report

❌ Patch coverage is 82.10526% with 17 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.52%. Comparing base (61bf6b9) to head (c69f65a).

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/joins/asof_join.rs	82.10%	11 Missing and 6 partials ⚠️

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #24519   +/-   ##
=======================================
  Coverage   81.52%   81.52%           
=======================================
  Files        1123     1123           
  Lines      405970   406041   +71     
  Branches   405970   406041   +71     
=======================================
+ Hits       330978   331040   +62     
- Misses      55627    55636    +9     
  Partials    19365    19365           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[2010YOUY01]

Thank you! This PR does 1) perf 2) fix float +0.0 and -0.0 normalization, right? I have some thoughts on the zero normalization part:

Is it possible to add an optimizer rule, to rewrite the expression like
match_condition t1.v1 < t2.v1 --> match_condition normalize_zero(t1.v1) < normalize_zero(t2.v1)
, and this way we can move the complexity outside the operators.

Other implementations seem can also benefit: #22835

[jayzhan211]

Thank you! This PR does 1) perf 2) fix float +0.0 and -0.0 normalization, right? I have some thoughts on the zero normalization part:

Is it possible to add an optimizer rule, to rewrite the expression like match_condition t1.v1 < t2.v1 --> match_condition normalize_zero(t1.v1) < normalize_zero(t2.v1) , and this way we can move the complexity outside the operators.

Other implementations seem can also benefit: #22835

I have concerns for moving this out as optimizer rule

1. The match expression is also the sort key. AsOfJoinExec requires both inputs to be sorted on the match expression. If we rewrite it to normalize_zero(v1), the required ordering becomes "sorted by normalize_zero(v1)", and the planner no longer recognizes an input already sorted by v1 as satisfying it — so we'd insert an extra sort on both sides. And that sort would be pointless: -0.0 already sorts right next to +0.0, so normalizing never breaks sortedness in the first place.
2. Correctness shouldn't depend on an optimizer rule running. AsOfJoinExec::try_new is public and used directly by tests and custom planners — those would silently produce wrong results if the rule didn't run. Optimizer rule is optional; correctness belongs in the operator.

[2010YOUY01]

I see, this makes sense to me. One extra challenge is how should we keep the +- 0.0 handling consistent in the entire project, and make this approach documented somewhere, otherwise different implementations will handle it differently #22835

Regarding the optimization part, I suggest we can wait until SQL and benchmark is done for #23738.

Though this PR should be strictly better in performance, but we might be able to find some more aggressive optimization that need to structure the code differently, and it's easier to review when the benchmark is ready.

Ideally we can proceed with a smaller fix-only PR right now, maybe #24375 from @Xuanwo

[Xuanwo]

The perf part looks good to me!

[jayzhan211]

#24375 looks overkill for me.

I see, this makes sense to me. One extra challenge is how should we keep the +- 0.0 handling consistent in the entire project, and make this approach documented somewhere, otherwise different implementations will handle it differently #22835

Another idea that came to mind is to make this a trait method on ExecutionPlan, so whether to apply this conversion can be configurable.

[jayzhan211]

For the performance part, since ASOF is not yet complete, I don’t think a benchmark is a hard requirement. If it looks good to you guys, we can move forward with this first.

Let me split that part into another PR and leave the normalize_neg_zero discussion for later.

[2010YOUY01]

For the performance part, since ASOF is not yet complete, I don’t think a benchmark is a hard requirement. If it looks good to you guys, we can move forward with this first.

Yes, just a suggestion, not a hard requirement. It should be good to go if others can review, I might not be able to get to it timely.

[jayzhan211]

Yes, just a suggestion, not a hard requirement. It should be good to go if others can review, I might not be able to get to it timely.

Thanks for the heads-up! That sounds good to me. I’ll ask others to take a look as well, and no worries if you don’t get a chance to review it in time.

@alamb Could you take a look when you have a chance? This also looks good to @Xuanwo. I think we can merge it without a benchmark, similar to what we did for the physical operator in #23828, and continue iterating on it afterward.

[alamb]

Looks reasonable to me -- thanks @jayzhan211 and @Xuanwo

[jayzhan211]

@kosiew Your comment in #24607 makes a good point — the negative zero case should be resolved together with the comparator. Here's what I'm planning:

• Merge this PR and close perf(asof_join): cache ASOF join match comparator, drop per-row ScalarValue materialization #24607. The approach here is more consistent with the existing methods in DataFusion.
• Start reviewing feat: add ASOF join logical semantics #23829 next.
• Leave feat: support floating-point ASOF equality keys #24375 as is for now, and iterate on it once we have the simple asof join case working.