Support IEJoin for inner joins with two range predicates

[xudong963]

Is your feature request related to a problem or challenge?

Interval-overlap joins and similar workloads commonly contain two cross-input inequality predicates, for example:

left.start_at <= right.end_at
AND right.start_at <= left.end_at

DataFusion can currently exploit a single range predicate with PiecewiseMergeJoinExec. With two range predicates, an unkeyed join may use NestedLoopJoinExec, while a join with equality keys may use HashJoinExec and evaluate both inequalities as residual filters. These plans can examine or materialize a large candidate set even when the final result is small.

This is a concrete implementation follow-up to the broader range-join performance issue #8393.

Describe the solution you'd like

Add an experimental IEJoinExec physical operator for bounded inner joins driven by exactly two cross-input <, <=, >, or >= predicates.

The initial implementation should:

• Support unkeyed joins and optional equality keys.
• Normalize range predicates to left-input expression, operator, and right-input expression.
• Use two sorted orders, an inverse permutation, and a sparse bitmap to enumerate matches.
• Recheck complete equality keys after hash grouping so hash collisions cannot affect correctness.
• Preserve any additional join predicates as residual filters.
• Exclude volatile expressions from IEJoin driver selection.
• Emit bounded output batches, bound work performed by each stream poll, account for memory-pool usage, and return controlled resource-exhaustion errors.
• Be selected by the physical planner only when an experimental configuration option is enabled; the option should default to false.

The initial scope excludes outer, semi, anti, and null-aware joins, as well as spilling.

Describe alternatives you've considered

• Continue using NestedLoopJoinExec for unkeyed joins. This is general but may evaluate a quadratic candidate space.
• Use HashJoinExec for equality keys and apply inequalities as residual filters. This helps keyed workloads but can still produce a large candidate set within each key.
• Extend PiecewiseMergeJoinExec. It is designed around a single range predicate and does not use both inequalities to prune candidates.
• Wait for a more general range-join framework. IEJoin is a focused first step for the common two-inequality pattern and can remain opt-in while its scope evolves.

Additional context

Draft implementation and initial benchmark results are available in #23741.

The implementation spans the physical operator, keyed and residual-filter support, planner integration, configuration and documentation, end-to-end tests, and benchmarks. To keep each change independently reviewable, it will be delivered as a stack of smaller PRs; the proposed stack is documented in a follow-up comment.

[xudong963]

Given the initial implementation is huge, I'd like to finish it by stacked PRs.

Proposed stacked PR plan

Dependency order:

main
└─ PR 1: planner refactor
   └─ PR 2: unkeyed IEJoinExec
      └─ PR 3: equality keys and residual filters
         └─ PR 4: planner and configuration integration
            └─ PR 5: benchmarks

PR 1 — refactor(core): extract range-join planning helpers

A behavior-preserving planner refactor:

• Extract construction of a physical JoinFilter from a logical join expression.
• Extract cross-input expression classification and inequality normalization, including operand swapping.
• Reuse the helpers in the existing Piecewise Merge Join planning path.
• Add focused tests showing that existing PWMJ and fallback plans do not change.
• Do not add IEJoin symbols, configuration, or behavior.

This removes unrelated refactoring from the eventual IEJoin planner diff.

PR 2 — feat(physical-plan): add unkeyed IEJoinExec

Introduce the smallest complete and directly usable execution operator:

• Add IEJoinCondition and IEJoinExec.
• Support bounded inner joins with exactly two range conditions and no equality keys.
• Implement input materialization, memory reservation, the two sorted orders, inverse permutation, sparse bitmap scan, bounded output batches, and bounded work per stream poll.
• Reject unsupported operators, volatile conditions, type mismatches, and unbounded inputs with clear errors.
• Export the operator for programmatic construction, but do not select it from the SQL planner.
• Cover all operator pairs, strict/non-strict boundaries, NULL, NaN, signed zero, empty inputs, bitmap word boundaries, cross-batch cursor continuation, memory limits, and dictionary-buffer accounting.

The algorithm, stream, and execution-plan adapter should stay together in this PR because none is independently useful or testable as production functionality.

PR 3 — feat(physical-plan): support keyed and residual IEJoin predicates

Extend the physical operator without changing planner behavior:

• Add optional equality keys and co-partitioned input requirements.
• Group rows by equality-key hash and recheck complete keys before output.
• Support NullEquality.
• Support an additional residual JoinFilter.
• Forward the force_hash_collisions feature needed for collision testing.
• Add keyed differential, multiple-key, residual-filter, null-equality, and forced-hash-collision tests.

The public construction API should remain compatible between PR 2 and PR 3, either through an options/builder shape established in PR 2 or by retaining the unkeyed constructor as a convenience API.

PR 4 — feat(core): plan IEJoin for two range predicates

Add the user-facing opt-in:

• Select IEJoin only for eligible inner, non-null-aware joins with at least two non-volatile cross-input range predicates.
• Use two normalized predicates as IEJoin drivers and preserve every other conjunct as a residual filter.
• Support both keyed and unkeyed planning.
• Add datafusion.optimizer.enable_ie_join, defaulting to false.
• Update generated configuration documentation and information-schema expectations.
• Add physical-planner unit tests and SQLLogicTests for keyed, unkeyed, reversed-expression, residual-filter, disabled-option, and fallback cases.
• Keep the configuration/API-change labeling and release-note impact isolated to this PR.

PR 5 — bench(physical-plan): add IEJoin benchmarks

Add performance evidence without production-code changes:

• Compare unkeyed IEJoin with NestedLoopJoinExec.
• Compare keyed IEJoin with HashJoinExec plus residual filtering.
• Include input construction, materialization, sorting, and complete output collection.
• Validate result cardinality before measuring.
• Document the benchmark command and representative results in the PR description.

Each PR should compile and pass its focused tests independently. Formatting and clippy run on every layer; the full workspace test suite and documentation formatting checks run on the top of the stack.

#23741 can remain a Draft umbrella while the stack is prepared. Its description should link these PRs and show their dependency/merge order; review should happen on the individual layers rather than on the cumulative Draft diff.

[2010YOUY01]

Thank you for driving this feature!

I'm also curious how many people find this algorithm easy to reason about. I tried to understand it before (with the paper + duckdb blog), but after spending several hours on it, I still didn't fully understand it, and don't have the confidence to implement it in datafusion. In particular, some of the bitmask construction/manipulation steps were difficult for me to follow. This is probably the only operator I've worked with that I still find hard to fully understand.

If the algorithm is inherently this tricky, I think there may be a real long-term maintenance cost. Piecewise merge join might be fast enough and also way easier to understand.

That said, it's also possible that IEJoin is easy if explained from the right angle, I'm wondering do you have a clear explanation, and I'm interested in seeing it.