feat: Refactor NLJ into an extensible framework for specialized joins#21983
feat: Refactor NLJ into an extensible framework for specialized joins#219832010YOUY01 wants to merge 1 commit intoapache:mainfrom
Conversation
github-actions
Bot
added
core
|
|
||
| /// Add one build-side input batch to the buffer, and optionally build a | ||
| /// runtime index. | ||
| fn add_build_batch(&mut self, batch: RecordBatch) -> Result<()>; |
There was a problem hiding this comment.
To extend dynamic filter: a common pattern for it is calculating bounds from the build side, and build an expression to filter the probe side.
The bound calculation step can be included inside this API naturally, and we only have to extend an API for returning the dynamic filter, in the follow-up work.
| /// The default implementation is the nested-loop fallback: it returns the | ||
| /// probe batch unchanged and emits one buffered build-side row at a time. | ||
| /// The caller joins that row against the current probe batch. | ||
| fn init_prober( |
There was a problem hiding this comment.
This function creates an iterator on candidates for ALL_BUILD_BATCHES x CURRENT_PROBE_BATCH
An easier alternative can be ALL_BUILD_BATCHES x CURRENT_PROBE_ROW for common indexed joins, however I think processing a probe batch each step is more flexible. For example, probe side may need certain batched pre-processing on the probe batch. So I prefer this design for now.
| /// This materializes the candidate pairs, evaluates any remaining join | ||
| /// filter, updates match bitmaps for joins that need unmatched rows, and | ||
| /// produces the final joined batch. | ||
| fn process_probe_candidates( |
There was a problem hiding this comment.
Here is a major place that is needed to update when the first indexed join implementation is added.
Now the BuildRow variant is a specialized implementation that is the most efficient for NLJ cartesian join fallback, for indexed join it needs a more general representation like (build_index, probe_index).
|
NLJ microbenchmark with Difference should be local noise -- the execution flow has almost no change. |
1 participant
Which issue does this PR close?
Rationale for this change
See issue for details.
The high-level idea is to make it easier to add specialized joins that can be accelerated with runtime indexes or dynamic filters.
With this refactor, future accelerated joins only need to implement the
JoinAcceleratortrait to define how to build and probe a runtime join index for target join conditions. The join state machine, outer join state tracking, and residual join predicate handling can then be handled reliably by common driver code in NLJ.What changes are included in this PR?
JoinAcceleratorAPI for runtime index, the dynamic filter part is planned to be added in a follow-up PRJoinAcceleratorAre these changes tested?
Existing tests
Are there any user-facing changes?
No