feat: Adds IntervalJoinExec for point-in-interval range joins

[SubhamSinghal]

Which issue does this PR close?

• Partially closes #8393
• Related to #318

Rationale for this change

Point-in-interval range joins (e.time >= w.start AND e.time < w.end) currently fall through to NestedLoopJoin which
is O(N×M). This adds a specialized IntervalJoinExec operator that uses a sort-merge approach — O(N log N + M log M +
scan) — with sequential memory access and no intermediate JoinFilter batch construction overhead.
Common use cases: temporal joins (event → time window), financial tier lookups (amount → price range), SCD joins.

What changes are included in this PR?

• IntervalJoinExec (physical-plan/src/joins/interval_join.rs): sort-merge range join operator for inner joins
  with point-in-interval pattern. Collects + sorts build side by low bound, sorts each probe batch internally, uses a
  monotonic boundary pointer for amortized O(1) per-probe-row matching.
• Pattern detection (core/src/physical_planner.rs): detects probe_col {>= | >} build_low AND probe_col {< | <=} build_high in the non-equi join path. Handles flipped operands and reordered conditions. Falls through to NLJ if
  pattern doesn't match.
• Config flag (common/src/config.rs): datafusion.optimizer.enable_interval_join (default true).
• SLT tests (interval_join.slt): basic matching, edge cases (empty sides, NULLs, overlapping intervals, boundary
  inclusivity), EXPLAIN plan verification, timestamp + interval arithmetic.
• Fuzz tests (join_fuzz.rs): compares IntervalJoinExec output against NLJ for correctness with random data.

Are these changes tested?

Yes:

• 273 lines of SQL logic tests covering basic joins, edge cases, NULL handling, overlapping intervals, boundary
  operators, EXPLAIN output, and timestamp types
• Fuzz tests comparing IntervalJoin results against NLJ on random data
• Unit tests within the operator

Are there any user-facing changes?

No

[2010YOUY01]

• IntervalJoinExec (physical-plan/src/joins/interval_join.rs): sort-merge range join operator for inner joins
  with point-in-interval pattern. Collects + sorts build side by low bound, sorts each probe batch internally, uses a
  monotonic boundary pointer for amortized O(1) per-probe-row matching.
• Pattern detection (core/src/physical_planner.rs): detects probe_col {>= | >} build_low AND probe_col {< | <=} build_high in the non-equi join path. Handles flipped operands and reordered conditions. Falls through to NLJ if
  pattern doesn't match.

This is a cool idea. I got a question: for this predicate e.time >= w.start AND e.time < w.end, let's say e.time >= w.start can be handled efficiently with monotonic pointer (amortized O(1)), do we have some mechanism to handle the residual filter e.time < w.end also efficiently? Or we can only do the first stage filter efficiently, the residual filter still has to be evaluated row-by-row.

The existing Piecewise Merge join has already implemented the above idea, though the residual filter has't been implemented yet.

[SubhamSinghal]

e.time >= w.start (lower bound): The build side is sorted by start_time ascending, and a monotonic boundary pointer advances forward as we process sorted probe rows. For probe value v, all build rows [0..=boundary] satisfy start_time <= v. This is amortized O(1) per probe row since the pointer only moves forward.

e.time < w.end (upper bound): The boundary pointer already eliminates all intervals that haven't started yet, so the end_time check only runs on the remaining smaller window. Expired intervals are still visited but not emitted.

If needed, we can also sort by end_time as a secondary key to enable skipping expired intervals entirely — happy to explore this as a follow-up. I didn't include sorting by end_time thinking this wouldn't add much value since window is already getting smaller by start_time pointer.

[2010YOUY01]

e.time >= w.start (lower bound): The build side is sorted by start_time ascending, and a monotonic boundary pointer advances forward as we process sorted probe rows. For probe value v, all build rows [0..=boundary] satisfy start_time <= v. This is amortized O(1) per probe row since the pointer only moves forward.

e.time < w.end (upper bound): The boundary pointer already eliminates all intervals that haven't started yet, so the end_time check only runs on the remaining smaller window. Expired intervals are still visited but not emitted.

This idea has been already implemented in

datafusion/datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs

Line 68 in 7fa6e21

/// `PiecewiseMergeJoinExec` is a join execution plan that only evaluates single range filter and show much

, though it's not working right now, it will work if the predicate only includes e.time >= w.start, but it requires some follow-up work for remaining filters handling (like e.time < w.end)

If needed, we can also sort by end_time as a secondary key to enable skipping expired intervals entirely — happy to explore this as a follow-up. I didn't include sorting by end_time thinking this wouldn't add much value since window is already getting smaller by start_time pointer.

This is a good idea to explore in the future! I think PiecewiseMergeJoin is a more general solution for only one in-equality join condition, if there are more in-equality conditions, there might be other indexing scheme to accelerate those specific workloads.

For coordination, I’m hoping to refactor PWMJ to build on top of the JoinAccelerator API: #21851 (comment). I’ll share a more detailed write-up soon.

[SubhamSinghal]

@2010YOUY01 Thanks for the feedback. I will refactor code once JoinAccelerator API is ready.

[github-actions]

Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.