GH-46224: [C++][Acero] Fix the hang in asof join

[zanmato1984]

Rationale for this change

A hang of asof join is reported in #46224 . To explain the cause of the hang, I shall first brief the basic processing of asof join.

There is an outstanding processing thread for the asof join, whose lifespan is bound to the asof node. This thread will wait on a command queue. Once a command is pushed into the queue, the thread pops it and do one round of processing. The command is issued as the batches are received from either left or right side input of the asof join node. Each round of processing will advance the accumulated left and right inputs as much as possible to see if it is sufficient to output a batch (if yes, i.e., either side of the asof join can be concluded by the latest input timestamp, then emit it, otherwise adjust the internal state according to the latest input timestamp).

Notably, the processing is by design to advance the inputs "as much as possible". However there is a bug (seemingly since day 1) when advancing the right side input in a tricky condition: if all right side rows are at the times before the least time of the left rows, then advancing the right side input won't cross batches in one round of processing. This is OK (i.e., no hang) as long as there are enough commands issued into the queue to trigger enough rounds of processing. However if lucky enough, say, the left input batches come late, then the processing with consume some commands issued by the right side input but advance nothing (because the left side input is empty), resulting in insufficient commands to fully advance both the input.

What changes are included in this PR?

Fix the issue that advancing the right side input won't cross batches when the minimal left time is bigger than all the right times.

Are these changes tested?

Yes, added a dedicated case.

Are there any user-facing changes?

None.

• GitHub Issue: [C++][Acero] Apparent deadlock in Table.join_asof #46224

[github-actions]

⚠️ GitHub issue #46224has been automatically assigned in GitHub to PR creator.

[zanmato1984]

@github-actions crossbow submit -g cpp -g r

[github-actions]

Revision: aa5a9be

Submitted crossbow builds: ursacomputing/crossbow @ actions-d733af0ab3

Task	Status
example-cpp-minimal-build-static
example-cpp-minimal-build-static-system-dependency
example-cpp-tutorial
r-binary-packages
r-recheck-most
test-alpine-linux-cpp
test-build-cpp-fuzz
test-conda-cpp
test-conda-cpp-meson
test-conda-cpp-valgrind
test-cuda-cpp-ubuntu-22.04-cuda-11.7.1
test-debian-12-cpp-amd64
test-debian-12-cpp-i386
test-fedora-39-cpp
test-r-arrow-backwards-compatibility
test-r-clang-sanitizer
test-r-depsource-bundled
test-r-depsource-system
test-r-dev-duckdb
test-r-devdocs
test-r-extra-packages
test-r-gcc-11
test-r-gcc-12
test-r-install-local
test-r-install-local-minsizerel
test-r-linux-as-cran
test-r-linux-rchk
test-r-linux-sanitizer
test-r-linux-valgrind
test-r-macos-as-cran
test-r-minimal-build
test-r-offline-maximal
test-r-offline-minimal
test-r-rhub-debian-gcc-devel-lto-latest
test-r-rhub-debian-gcc-release-custom-ccache
test-r-rhub-ubuntu-release-latest
test-r-rocker-r-ver-latest
test-r-rstudio-r-base-4.1-opensuse155
test-r-rstudio-r-base-4.2-focal
test-r-ubuntu-22.04
test-r-versions
test-ubuntu-22.04-cpp
test-ubuntu-22.04-cpp-20
test-ubuntu-22.04-cpp-bundled
test-ubuntu-22.04-cpp-emscripten
test-ubuntu-22.04-cpp-no-threading
test-ubuntu-24.04-cpp
test-ubuntu-24.04-cpp-bundled-offline
test-ubuntu-24.04-cpp-gcc-13-bundled
test-ubuntu-24.04-cpp-gcc-14
test-ubuntu-24.04-cpp-minimal-with-formats
test-ubuntu-24.04-cpp-thread-sanitizer

[zanmato1984]

@github-actions crossbow submit -g cpp -g r

[github-actions]

Revision: 6ea4624a8ec7c27ab7c57bdcacf295995d0e12a5

Submitted crossbow builds: ursacomputing/crossbow @ actions-a2fa793a58

Task	Status
example-cpp-minimal-build-static
example-cpp-minimal-build-static-system-dependency
example-cpp-tutorial
r-binary-packages
r-recheck-most
test-alpine-linux-cpp
test-build-cpp-fuzz
test-conda-cpp
test-conda-cpp-meson
test-conda-cpp-valgrind
test-cuda-cpp-ubuntu-22.04-cuda-11.7.1
test-debian-12-cpp-amd64
test-debian-12-cpp-i386
test-fedora-39-cpp
test-r-arrow-backwards-compatibility
test-r-clang-sanitizer
test-r-depsource-bundled
test-r-depsource-system
test-r-dev-duckdb
test-r-devdocs
test-r-extra-packages
test-r-gcc-11
test-r-gcc-12
test-r-install-local
test-r-install-local-minsizerel
test-r-linux-as-cran
test-r-linux-rchk
test-r-linux-sanitizer
test-r-linux-valgrind
test-r-macos-as-cran
test-r-minimal-build
test-r-offline-maximal
test-r-offline-minimal
test-r-rhub-debian-gcc-devel-lto-latest
test-r-rhub-debian-gcc-release-custom-ccache
test-r-rhub-ubuntu-release-latest
test-r-rocker-r-ver-latest
test-r-rstudio-r-base-4.1-opensuse155
test-r-rstudio-r-base-4.2-focal
test-r-ubuntu-22.04
test-r-versions
test-ubuntu-22.04-cpp
test-ubuntu-22.04-cpp-20
test-ubuntu-22.04-cpp-bundled
test-ubuntu-22.04-cpp-emscripten
test-ubuntu-22.04-cpp-no-threading
test-ubuntu-24.04-cpp
test-ubuntu-24.04-cpp-bundled-offline
test-ubuntu-24.04-cpp-gcc-13-bundled
test-ubuntu-24.04-cpp-gcc-14
test-ubuntu-24.04-cpp-minimal-with-formats
test-ubuntu-24.04-cpp-thread-sanitizer

[zanmato1984]

IMO this code is apparently NOT doing what it is supposed to do: by here the have_active_batch must be true, meaning queue_ must be non-empty. Then popping the queue always gets a valid std::optional and effectively setting have_active_batch to false. The if branch below won't be executed.

[zanmato1984]

I noticed there are two fixes made before in the following if branch #36094 and #36499. Given that this have_active_batch &= !queue_.TryPop() exists since #13028 where it was introduced, and the lack of corresponding tests, I'm not sure how they fixed anything and what the bugs even were. So I just remove them.

[zanmato1984]

Hi, @westonpace@icexelloss@pitrou would you help to review? Thanks.

[zanmato1984]

@github-actions crossbow submit -g cpp -g python -g r

[github-actions]

Revision: af70071

Submitted crossbow builds: ursacomputing/crossbow @ actions-ccf7708cfd

Task	Status
example-cpp-minimal-build-static
example-cpp-minimal-build-static-system-dependency
example-cpp-tutorial
example-python-minimal-build-fedora-conda
example-python-minimal-build-ubuntu-venv
r-binary-packages
r-recheck-most
test-alpine-linux-cpp
test-build-cpp-fuzz
test-conda-cpp
test-conda-cpp-meson
test-conda-cpp-valgrind
test-conda-python-3.10
test-conda-python-3.10-hdfs-2.9.2
test-conda-python-3.10-hdfs-3.2.1
test-conda-python-3.10-pandas-latest-numpy-latest
test-conda-python-3.11
test-conda-python-3.11-dask-latest
test-conda-python-3.11-dask-upstream_devel
test-conda-python-3.11-hypothesis
test-conda-python-3.11-pandas-latest-numpy-1.26
test-conda-python-3.11-pandas-latest-numpy-latest
test-conda-python-3.11-pandas-nightly-numpy-nightly
test-conda-python-3.11-pandas-upstream_devel-numpy-nightly
test-conda-python-3.11-spark-master
test-conda-python-3.12
test-conda-python-3.12-cpython-debug
test-conda-python-3.13
test-conda-python-3.9
test-conda-python-3.9-pandas-1.1.3-numpy-1.19.5
test-conda-python-emscripten
test-cuda-cpp-ubuntu-22.04-cuda-11.7.1
test-cuda-python-ubuntu-22.04-cuda-11.7.1
test-debian-12-cpp-amd64
test-debian-12-cpp-i386
test-debian-12-python-3-amd64
test-debian-12-python-3-i386
test-fedora-39-cpp
test-fedora-39-python-3
test-r-arrow-backwards-compatibility
test-r-clang-asan
test-r-clang-ubsan
test-r-depsource-bundled
test-r-depsource-system
test-r-dev-duckdb
test-r-devdocs
test-r-extra-packages
test-r-gcc-11
test-r-gcc-12
test-r-install-local
test-r-install-local-minsizerel
test-r-linux-as-cran
test-r-linux-rchk
test-r-linux-sanitizer
test-r-linux-valgrind
test-r-m1-san
test-r-macos-as-cran
test-r-minimal-build
test-r-offline-maximal
test-r-offline-minimal
test-r-rhub-debian-gcc-devel-lto-latest
test-r-rhub-debian-gcc-release-custom-ccache
test-r-rhub-ubuntu-release-latest
test-r-rocker-r-ver-latest
test-r-rstudio-r-base-4.1-opensuse155
test-r-rstudio-r-base-4.2-focal
test-r-ubuntu-22.04
test-r-versions
test-ubuntu-22.04-cpp
test-ubuntu-22.04-cpp-20
test-ubuntu-22.04-cpp-bundled
test-ubuntu-22.04-cpp-emscripten
test-ubuntu-22.04-cpp-no-threading
test-ubuntu-22.04-python-3
test-ubuntu-22.04-python-313-freethreading
test-ubuntu-24.04-cpp
test-ubuntu-24.04-cpp-bundled-offline
test-ubuntu-24.04-cpp-gcc-13-bundled
test-ubuntu-24.04-cpp-gcc-14
test-ubuntu-24.04-cpp-minimal-with-formats
test-ubuntu-24.04-cpp-thread-sanitizer
test-ubuntu-24.04-python-3

[pitrou]

I'm curious, this does only happen when the right side is concerned? The update logic is not symmetric?

[zanmato1984]

I'm curious, this does only happen when the right side is concerned? The update logic is not symmetric?

Right, not symmetric as one might would naturally expect (the same situation for hash join as well - right side to build the hash table and left side to probe the hash table).

The left VS. right differs in the following aspects:

1. There is only one left side input but possibly multiple right side inputs. The right side inputs are flattened and equally iterated during the whole processing (as opposed to traversed in a tree-like fashion).
2. Several processing logic, and in addition some methods are right side inputs only, e.g.:

   arrow/cpp/src/arrow/acero/asof_join_node.cc

   Line 949 in 03bdb2a

   Result<RhsUpdateState> UpdateRhs() {

   
   and

   arrow/cpp/src/arrow/acero/asof_join_node.cc

   Line 963 in 03bdb2a

   rhs.AdvanceAndMemoize(lhs_latest_time, rhs_empty));

So it's normal that some issues are right side only.

But it does remind me that the method I was fixing, namely ‎InputState::Advance, also applies to the left side input, so I think I should see if the issue exists for the left side as well, probably by adding a symmetric case you suggested.

Thanks for the question!

[zanmato1984]

Hi @pitrou , comments all addressed. Would you take a look again? Thanks.

[zanmato1984]

@github-actions crossbow submit -g cpp

[github-actions]

Revision: dd7a66b

Submitted crossbow builds: ursacomputing/crossbow @ actions-7d095c8a25

Task	Status
example-cpp-minimal-build-static
example-cpp-minimal-build-static-system-dependency
example-cpp-tutorial
test-alpine-linux-cpp
test-build-cpp-fuzz
test-conda-cpp
test-conda-cpp-meson
test-conda-cpp-valgrind
test-cuda-cpp-ubuntu-22.04-cuda-11.7.1
test-debian-12-cpp-amd64
test-debian-12-cpp-i386
test-fedora-39-cpp
test-ubuntu-22.04-cpp
test-ubuntu-22.04-cpp-20
test-ubuntu-22.04-cpp-bundled
test-ubuntu-22.04-cpp-emscripten
test-ubuntu-22.04-cpp-no-threading
test-ubuntu-24.04-cpp
test-ubuntu-24.04-cpp-bundled-offline
test-ubuntu-24.04-cpp-gcc-13-bundled
test-ubuntu-24.04-cpp-gcc-14
test-ubuntu-24.04-cpp-minimal-with-formats
test-ubuntu-24.04-cpp-thread-sanitizer

[zanmato1984]

@github-actions crossbow submit -g cpp

[github-actions]

Revision: 5f5b895

Submitted crossbow builds: ursacomputing/crossbow @ actions-973e5e92ed

Task	Status
example-cpp-minimal-build-static
example-cpp-minimal-build-static-system-dependency
example-cpp-tutorial
test-alpine-linux-cpp
test-build-cpp-fuzz
test-conda-cpp
test-conda-cpp-meson
test-conda-cpp-valgrind
test-cuda-cpp-ubuntu-22.04-cuda-11.7.1
test-debian-12-cpp-amd64
test-debian-12-cpp-i386
test-fedora-39-cpp
test-ubuntu-22.04-cpp
test-ubuntu-22.04-cpp-20
test-ubuntu-22.04-cpp-bundled
test-ubuntu-22.04-cpp-emscripten
test-ubuntu-22.04-cpp-no-threading
test-ubuntu-24.04-cpp
test-ubuntu-24.04-cpp-bundled-offline
test-ubuntu-24.04-cpp-gcc-13-bundled
test-ubuntu-24.04-cpp-gcc-14
test-ubuntu-24.04-cpp-minimal-with-formats
test-ubuntu-24.04-cpp-thread-sanitizer

[zanmato1984]

Thanks @pitrou for reviewing. Is there anything else to address?

[pitrou]

LGTM

[conbench-apache-arrow]

After merging your PR, Conbench analyzed the 4 benchmarking runs that have been run so far on merge-commit fe29b7d.

There were no benchmark performance regressions. 🎉

The full Conbench report has more details. It also includes information about 19 possible false positives for unstable benchmarks that are known to sometimes produce them.