gitmodimo opened a new pull request, #51094:
URL: https://github.com/apache/arrow/pull/51094
### Rationale for this change
This is a second atteempt at solving #46421 first one in #46140 caused
performance degradation. It was caused by output batch merging multiple input
batches and producing huge output batch. This caused unmaterialized table to
grow and reallocate(+copy) multiple times. I decided to do deeper rewrite and
fix more problems at once by redesigning data flow in the node.
### What changes are included in this PR?
- **Execution and materialization model:** Replace the dedicated
processing thread with a coordinator and independently scheduled processing for
each RHS input. Per-input sequencing makes results independent of physical
batch arrival order, fixing #36651. The coordinator activates one LHS batch at
a time, while each RHS input finds matches and materializes its payload columns
independently, implementing the parallel execution proposed in #34135. Once
every RHS input completes, the coordinator reuses the original LHS arrays and
scalars, resolving the unnecessary copying described in #41873, appends the RHS
columns, and emits one output batch preserving the LHS boundary and index. This
also avoids the cross-boundary output growth amplified by the earlier
`asof_join_pause` attempt. The same implementation supports threaded, serial,
and `ARROW_ENABLE_THREADING=OFF` execution.
- **Backpressure:** Bound the LHS and each RHS input queue using
batch-count watermarks, applying pause and resume upstream as queues cross
their watermarks. A downstream pause allows the active LHS batch to finish but
prevents another from starting, fixing #46421.
- **Key matching:** Use hashes for lookup followed by exact `by`-key
comparison, preventing collisions from producing incorrect joins and fixing
#32894 without switching to 128-bit hashes. Support additional flat key types
and scalar hashing. Benchmarks show no regression. Dictionary-encoded key
columns remain unsupported.
- **Non-key fields:** Reuse LHS payloads directly and materialize RHS
payloads using generic Arrow builders instead of the key encoder. This adds
support for nested and dictionary-encoded payloads, including fixed-size lists,
and addresses #44729.
- **`on`-key normalization:** Preserve the ordering of signed integer and
temporal keys across zero, fixing #45876.
- **Null `on` keys:** Respect validity bitmaps so null LHS timestamps
remain unmatched and null RHS timestamps are never selected, fixing #46780.
- **Input validation:** Allow trailing null `on` values but reject ordered
inputs containing non-null values after a null.
- **Benchmarks:** Expand coverage across threading modes, string-key
sizes, tolerance directions, and input densities. Local results are
approximately 2–109× faster than `main`.
- **Tests:** Re-enable and stabilize the generated-batch backpressure
test, fixing #36248. Add regression coverage for jittered input sequencing
under threaded and serial execution; input- queue and downstream backpressure;
completing or stopping an active LHS batch while paused; preserving LHS
boundaries, indices, selections, arrays, and scalars; exact collision
arbitration; additional key and payload types; scalar hashing; and signed,
null, and invalidly ordered `on` keys.
### Are there any user-facing changes?
- Additional flat `by`-key types are supported, including Boolean,
fixed-size binary, and decimal types. Non-key payload columns are no longer
restricted to types supported by the key encoder, enabling nested and
dictionary- encoded payloads. Dictionary-encoded `by` keys remain unsupported.
- If an LHS `ExecBatch` contains a scalar representing a constant column,
AsofJoin preserves it as a scalar in the output instead of materializing it as
an array.
- AsofJoin emits exactly one output batch for each LHS input batch,
preserving its length, index, and boundary. This includes zero-length LHS
batches.
- With threaded execution, matching and materialization for separate RHS
inputs may run concurrently. With `use_threads=false` or
`ARROW_ENABLE_THREADING=OFF`, the same implementation runs serially without
creating a dedicated processing thread.
- AsofJoin now participates in backpressure. The number of queued batches
is bounded per input, and downstream pause prevents new LHS batches from
starting after the active batch completes.
- Hash collisions can no longer cause different `by` keys to match because
hash matches are verified using exact key equality.
- Signed `on` keys are ordered correctly across zero. Null LHS `on` values
remain unmatched, and null RHS `on` values are never selected.
- Because inputs are ordered by their `on` key, null `on` values must be
trailing. AsofJoin now returns an error if a non-null value appears after a
null.
Technically none of the changes are breaking but it changes some non
contractual behaviors.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
