Merge branch 'SourceNodeFixPauseAfterStop' into combined3

Author: Rafał Hibner

.github/workflows/cpp.yml

@@ -93,8 +93,8 @@ jobs:
             image: ubuntu-cpp-sanitizer
             llvm: 14
             runs-on: ubuntu-latest
-            title: AMD64 Ubuntu 22.04 C++ ASAN UBSAN
-            ubuntu: 22.04
+            title: AMD64 Ubuntu 24.04 C++ ASAN UBSAN
+            ubuntu: 24.04
           - arch: arm64v8
             clang-tools: 14
             image: ubuntu-cpp

ci/docker/python-free-threaded-wheel-manylinux-test-unittests.dockerfile

@@ -41,11 +41,9 @@ ENV PATH "${ARROW_PYTHON_VENV}/bin:${PATH}"
 
 # pandas doesn't provide wheels for aarch64 yet, so we have to install nightly Cython
 # along with the rest of pandas' build dependencies and disable build isolation
-COPY python/requirements-wheel-test.txt /arrow/python/
 RUN python -m pip install \
     --pre \
     --prefer-binary \
     --extra-index-url "https://pypi.anaconda.org/scientific-python-nightly-wheels/simple" \
     Cython numpy
 RUN python -m pip install "meson-python==0.13.1" "meson==1.2.1" wheel "versioneer[toml]" ninja
-RUN python -m pip install --no-build-isolation -r /arrow/python/requirements-wheel-test.txt

ci/docker/python-free-threaded-wheel-musllinux-test-unittests.dockerfile

@@ -56,11 +56,9 @@ RUN cp /usr/share/zoneinfo/Etc/UTC /etc/localtime
 
 # pandas doesn't provide wheels for aarch64 yet, so we have to install nightly Cython
 # along with the rest of pandas' build dependencies and disable build isolation
-COPY python/requirements-wheel-test.txt /arrow/python/
 RUN python -m pip install \
     --pre \
     --prefer-binary \
     --extra-index-url "https://pypi.anaconda.org/scientific-python-nightly-wheels/simple" \
     Cython numpy
 RUN python -m pip install "meson-python==0.13.1" "meson==1.2.1" wheel "versioneer[toml]" ninja
-RUN python -m pip install --no-build-isolation -r /arrow/python/requirements-wheel-test.txt

ci/docker/ubuntu-22.04-cpp.dockerfile

@@ -173,9 +173,6 @@ RUN /arrow/ci/scripts/install_minio.sh latest /usr/local
 COPY ci/scripts/install_gcs_testbench.sh /arrow/ci/scripts/
 RUN /arrow/ci/scripts/install_gcs_testbench.sh default
 
-COPY ci/scripts/install_azurite.sh /arrow/ci/scripts/
-RUN /arrow/ci/scripts/install_azurite.sh
-
 COPY ci/scripts/install_ceph.sh /arrow/ci/scripts/
 RUN /arrow/ci/scripts/install_ceph.sh
 

ci/scripts/cpp_test.sh

@@ -42,43 +42,50 @@ if [ -z "${ARROW_DEBUG_MEMORY_POOL}" ]; then
   export ARROW_DEBUG_MEMORY_POOL=trap
 fi
 
+exclude_tests=()
 ctest_options=()
+if ! type azurite >/dev/null 2>&1; then
+  exclude_tests+=("arrow-azurefs-test")
+fi
 case "$(uname)" in
   Linux)
     n_jobs=$(nproc)
     ;;
   Darwin)
     n_jobs=$(sysctl -n hw.ncpu)
     # TODO: https://github.com/apache/arrow/issues/40410
-    exclude_tests="arrow-s3fs-test"
-    ctest_options+=(--exclude-regex "${exclude_tests}")
+    exclude_tests+=("arrow-s3fs-test")
     ;;
   MINGW*)
     n_jobs=${NUMBER_OF_PROCESSORS:-1}
     # TODO: Enable these crashed tests.
     # https://issues.apache.org/jira/browse/ARROW-9072
-    exclude_tests="gandiva-binary-test"
-    exclude_tests="${exclude_tests}|gandiva-boolean-expr-test"
-    exclude_tests="${exclude_tests}|gandiva-date-time-test"
-    exclude_tests="${exclude_tests}|gandiva-decimal-single-test"
-    exclude_tests="${exclude_tests}|gandiva-decimal-test"
-    exclude_tests="${exclude_tests}|gandiva-filter-project-test"
-    exclude_tests="${exclude_tests}|gandiva-filter-test"
-    exclude_tests="${exclude_tests}|gandiva-hash-test"
-    exclude_tests="${exclude_tests}|gandiva-if-expr-test"
-    exclude_tests="${exclude_tests}|gandiva-in-expr-test"
-    exclude_tests="${exclude_tests}|gandiva-internals-test"
-    exclude_tests="${exclude_tests}|gandiva-literal-test"
-    exclude_tests="${exclude_tests}|gandiva-null-validity-test"
-    exclude_tests="${exclude_tests}|gandiva-precompiled-test"
-    exclude_tests="${exclude_tests}|gandiva-projector-test"
-    exclude_tests="${exclude_tests}|gandiva-utf8-test"
-    ctest_options+=(--exclude-regex "${exclude_tests}")
+    exclude_tests+=("gandiva-binary-test")
+    exclude_tests+=("gandiva-boolean-expr-test")
+    exclude_tests+=("gandiva-date-time-test")
+    exclude_tests+=("gandiva-decimal-single-test")
+    exclude_tests+=("gandiva-decimal-test")
+    exclude_tests+=("gandiva-filter-project-test")
+    exclude_tests+=("gandiva-filter-test")
+    exclude_tests+=("gandiva-hash-test")
+    exclude_tests+=("gandiva-if-expr-test")
+    exclude_tests+=("gandiva-in-expr-test")
+    exclude_tests+=("gandiva-internals-test")
+    exclude_tests+=("gandiva-literal-test")
+    exclude_tests+=("gandiva-null-validity-test")
+    exclude_tests+=("gandiva-precompiled-test")
+    exclude_tests+=("gandiva-projector-test")
+    exclude_tests+=("gandiva-utf8-test")
     ;;
   *)
     n_jobs=${NPROC:-1}
     ;;
 esac
+if [ "${#exclude_tests[@]}" -gt 0 ]; then
+  IFS="|"
+  ctest_options+=(--exclude-regex "${exclude_tests[*]}")
+  unset IFS
+fi
 
 if [ "${ARROW_EMSCRIPTEN:-OFF}" = "ON" ]; then
   n_jobs=1 # avoid spurious fails on emscripten due to loading too many big executables

ci/scripts/install_azurite.sh

@@ -20,18 +20,9 @@
 set -e
 
 node_version="$(node --version)"
-echo "node version = ${node_version}"
-
-case "${node_version}" in
-  v12*)
-    # Pin azurite to 3.29.0 due to https://github.com/apache/arrow/issues/41505
-    azurite_version=v3.29.0
-    ;;
-  *)
-    azurite_version=latest
-    ;;
-esac
+echo "Node.js version = ${node_version}"
 
+azurite_version=latest
 case "$(uname)" in
   Darwin)
     npm install -g azurite@${azurite_version}
@@ -46,5 +37,4 @@ case "$(uname)" in
     which azurite
     ;;
 esac
-
-echo "azurite version = $(azurite --version)"
+echo "Azurite version = $(azurite --version)"

ci/scripts/python_wheel_unix_test.sh

@@ -100,9 +100,19 @@ if [ "${CHECK_WHEEL_CONTENT}" == "ON" ]; then
     --path "${source_dir}/python/repaired_wheels"
 fi
 
+is_free_threaded() {
+  python -c "import sysconfig; print('ON' if sysconfig.get_config_var('Py_GIL_DISABLED') else 'OFF')"
+}
+
 if [ "${CHECK_UNITTESTS}" == "ON" ]; then
   # Install testing dependencies
-  python -m pip install -U -r "${source_dir}/python/requirements-wheel-test.txt"
+  if [ "$(is_free_threaded)" = "ON" ]; then
+    echo "Free-threaded Python build detected"
+    python -m pip install -U -r "${source_dir}/python/requirements-wheel-test-3.13t.txt"
+  elif [ "$(is_free_threaded)" = "OFF" ]; then
+    echo "Regular Python build detected"
+    python -m pip install -U -r "${source_dir}/python/requirements-wheel-test.txt"
+  fi
 
   # Execute unittest, test dependencies must be installed
   python -c 'import pyarrow; pyarrow.create_library_symlinks()'

ci/scripts/r_windows_build.sh

@@ -35,7 +35,7 @@ printenv
 makepkg-mingw --noconfirm --noprogressbar --skippgpcheck --nocheck --syncdeps --cleanbuild
 
 VERSION=$(grep Version $ARROW_HOME/r/DESCRIPTION | cut -d " " -f 2)
-DST_DIR="arrow-$VERSION"
+DST_DIR="r-libarrow-windows-x86_64-$VERSION"
 
 # Collect the build artifacts and make the shape of zip file that rwinlib expects
 ls

cpp/src/arrow/acero/meson.build

@@ -86,7 +86,11 @@ arrow_acero_lib = library(
     gnu_symbol_visibility: 'inlineshidden',
 )
 
-arrow_acero_dep = declare_dependency(link_with: [arrow_acero_lib])
+arrow_acero_dep = declare_dependency(
+    dependencies: [arrow_compute_dep],
+    link_with: [arrow_acero_lib],
+)
+meson.override_dependency('arrow-acero', arrow_acero_dep)
 
 arrow_acero_testing_sources = ['test_nodes.cc', 'test_util_internal.cc']
 
@@ -148,6 +152,7 @@ foreach key, val : arrow_acero_benchmarks
 endforeach
 
 pkg.generate(
+    arrow_acero_lib,
     filebase: 'arrow-acero',
     name: 'Apache Arrow Acero Engine',
     description: 'Apache Arrow\'s Acero Engine',

cpp/src/arrow/acero/source_node.cc

@@ -238,7 +238,7 @@ struct SourceNode : ExecNode, public TracedNode {
 
   void PauseProducing(ExecNode* output, int32_t counter) override {
     std::lock_guard<std::mutex> lg(mutex_);
-    if (counter <= backpressure_counter_) {
+    if (counter <= backpressure_counter_ || stop_requested_) {
       return;
     }
     backpressure_counter_ = counter;
@@ -266,11 +266,11 @@ struct SourceNode : ExecNode, public TracedNode {
     to_finish.MarkFinished();
   }
 
-  Status StopProducing() override {
-    // GH-35837: ensure node is not paused
+  Status StopProducingImpl() override {
     Future<> to_finish;
     {
       std::lock_guard<std::mutex> lg(mutex_);
+      stop_requested_ = true;
       if (!backpressure_future_.is_finished()) {
         to_finish = backpressure_future_;
         backpressure_future_ = Future<>::MakeFinished();
@@ -279,13 +279,6 @@ struct SourceNode : ExecNode, public TracedNode {
     if (to_finish.is_valid()) {
       to_finish.MarkFinished();
     }
-    // only then stop
-    return ExecNode::StopProducing();
-  }
-
-  Status StopProducingImpl() override {
-    std::unique_lock<std::mutex> lock(mutex_);
-    stop_requested_ = true;
     return Status::OK();
   }