Skip to content

GH-46421: [C++][Acero] Asofjoin respect PauseProducing from downstream. Version 2 - #51094

Draft
gitmodimo wants to merge 1 commit into
apache:mainfrom
gitmodimo:asof_join_rework
Draft

GH-46421: [C++][Acero] Asofjoin respect PauseProducing from downstream. Version 2#51094
gitmodimo wants to merge 1 commit into
apache:mainfrom
gitmodimo:asof_join_rework

Conversation

@gitmodimo

@gitmodimo gitmodimo commented Aug 31, 2026

Copy link
Copy Markdown
Contributor

Rationale for this change

This is a second attempt at solving #46421. The first attempt in #46140 caused a performance regression because output batches merged multiple input batches and could grow very large. This repeatedly grew, reallocated, and copied the unmaterialized table. This PR instead redesigns the node's data flow and addresses the related correctness and execution issues together.

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 [C++] Make AsofJoinNode robust to input batch scheduling #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 [C++] Parallel asof join node #34135. Once every RHS input completes, the coordinator reuses the original LHS arrays and scalars, resolving the unnecessary copying described in [C++] Asof-joins inefficiently copy the left hand side  #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 [C++][Acero] Asofjoin does not propagate pause upstream #46421.

  • Key matching: Use hashes for lookup followed by exact by-key comparison, preventing collisions from producing incorrect joins and fixing [C++] AsofJoinNode 128-bit hashing #32894 without switching to 128-bit hashes. Support additional flat key types and scalar by keys. 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 [C++][Acero] Not support type like Fixed Size List for non-key column in asof join node #44729.

  • on-key normalization: Preserve the ordering of signed integer and temporal keys across zero, fixing [C++][Acero] Negative values in NormalizeTime are mapped to the highest uint64_t values #45876.

  • Null on keys: Respect validity bitmaps so null LHS timestamps remain unmatched and null RHS timestamps are never selected, fixing [C++][Acero][Python] Asof join does not detect null times #46780.

  • Input ordering: Validate that each input provides ordering compatible with its on key and preserve the LHS ordering in the output.

  • 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 [C++] Fix and re-enable Asof Join Backpresure test flakiness #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, arrays, scalars, and ordering; exact collision arbitration; additional key and payload types; mixed scalar and array by keys; signed and null on keys; and input-ordering validation.

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.

  • Plans with unordered or incompatible input ordering are now rejected during construction.

@gitmodimo

Copy link
Copy Markdown
Contributor Author

@zanmato1984 I do have this PR ready but I first want to establish expanded benchmark baseline. Can you please run benchmarks for this PR in current form?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant