Merge branch 'SchedulerKeepAlive' into combined2

Author: Rafał Hibner

.github/workflows/comment_bot.yml

@@ -49,7 +49,7 @@ jobs:
         run: pip install -e arrow/dev/archery[bot]
       - name: Handle GitHub comment event
         env:
-          ARROW_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           CROSSBOW_GITHUB_TOKEN: ${{ secrets.CROSSBOW_GITHUB_TOKEN }}
         run: |
           archery --debug trigger-bot \

.github/workflows/dev.yml

@@ -107,10 +107,9 @@ jobs:
           gem install test-unit
           pip install "cython>=3" setuptools pytest requests setuptools-scm
       - name: Run Release Test
-        env:
-          ARROW_GITHUB_API_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         shell: bash
         run: |
+          echo "GH_TOKEN=${{ secrets.GITHUB_TOKEN }}" > dev/release/.env
           ci/scripts/release_test.sh $(pwd)
       - name: Run Merge Script Test
         shell: bash

.github/workflows/pr_bot.yml

@@ -89,7 +89,7 @@ jobs:
         run: pip install -e arrow/dev/archery[bot]
       - name: Handle PR workflow event
         env:
-          ARROW_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         run: |
           if [ "${GITHUB_EVENT_NAME}" = "workflow_run" ]; then
             # workflow_run is executed on PR review. Update to original event.

c_glib/arrow-glib/basic-array.cpp

@@ -27,6 +27,7 @@
 
 #include <arrow/c/bridge.h>
 
+#include <cmath>
 #include <sstream>
 
 G_BEGIN_DECLS
@@ -491,6 +492,22 @@ garrow_array_statistics_has_distinct_count(GArrowArrayStatistics *statistics)
   return priv->statistics.distinct_count.has_value();
 }
 
+/**
+ * garrow_array_statistics_is_distinct_count_exact:
+ * @statistics: A #GArrowArrayStatistics.
+ *
+ * Returns: %TRUE if the distinct count is available and exact, %FALSE otherwise.
+ *
+ * Since: 22.0.0
+ */
+gboolean
+garrow_array_statistics_is_distinct_count_exact(GArrowArrayStatistics *statistics)
+{
+  auto priv = GARROW_ARRAY_STATISTICS_GET_PRIVATE(statistics);
+  return priv->statistics.distinct_count.has_value() &&
+         std::holds_alternative<int64_t>(priv->statistics.distinct_count.value());
+}
+
 /**
  * garrow_array_statistics_get_distinct_count:
  * @statistics: A #GArrowArrayStatistics.
@@ -499,19 +516,59 @@ garrow_array_statistics_has_distinct_count(GArrowArrayStatistics *statistics)
  *   -1 otherwise.
  *
  * Since: 21.0.0
+ *
+ * Deprecated: 22.0.0. Use garrow_array_statistics_is_distinct_count_exact(),
+ *   garrow_array_statistics_get_distinct_count_exact() and
+ *   garrow_array_statistics_get_distinct_count_approximate() instead.
  */
 gint64
 garrow_array_statistics_get_distinct_count(GArrowArrayStatistics *statistics)
+{
+  return garrow_array_statistics_get_distinct_count_exact(statistics);
+}
+
+/**
+ * garrow_array_statistics_get_distinct_count_exact:
+ * @statistics: A #GArrowArrayStatistics.
+ *
+ * Returns: 0 or larger value if @statistics has a valid exact distinct count
+ *   value, -1 otherwise.
+ *
+ * Since: 22.0.0
+ */
+gint64
+garrow_array_statistics_get_distinct_count_exact(GArrowArrayStatistics *statistics)
 {
   auto priv = GARROW_ARRAY_STATISTICS_GET_PRIVATE(statistics);
   const auto &distinct_count = priv->statistics.distinct_count;
-  if (distinct_count) {
-    return distinct_count.value();
+  if (distinct_count && std::holds_alternative<int64_t>(distinct_count.value())) {
+    return std::get<int64_t>(distinct_count.value());
   } else {
     return -1;
   }
 }
 
+/**
+ * garrow_array_statistics_get_distinct_count_approximate:
+ * @statistics: A #GArrowArrayStatistics.
+ *
+ * Returns: Non `NaN` value if @statistics has a valid approximate distinct count
+ *   value, `NaN` otherwise.
+ *
+ * Since: 22.0.0
+ */
+gdouble
+garrow_array_statistics_get_distinct_count_approximate(GArrowArrayStatistics *statistics)
+{
+  auto priv = GARROW_ARRAY_STATISTICS_GET_PRIVATE(statistics);
+  const auto &distinct_count = priv->statistics.distinct_count;
+  if (distinct_count && std::holds_alternative<double>(distinct_count.value())) {
+    return std::get<double>(distinct_count.value());
+  } else {
+    return std::nan("");
+  }
+}
+
 typedef struct GArrowArrayPrivate_
 {
   std::shared_ptr<arrow::Array> array;

c_glib/arrow-glib/basic-array.h

@@ -61,9 +61,21 @@ garrow_array_statistics_get_null_count(GArrowArrayStatistics *statistics);
 GARROW_AVAILABLE_IN_21_0
 gboolean
 garrow_array_statistics_has_distinct_count(GArrowArrayStatistics *statistics);
+GARROW_AVAILABLE_IN_22_0
+gboolean
+garrow_array_statistics_is_distinct_count_exact(GArrowArrayStatistics *statistics);
+#ifndef GARROW_DISABLE_DEPRECATED
 GARROW_AVAILABLE_IN_21_0
+GARROW_DEPRECATED_IN_22_0_FOR(garrow_array_statistics_get_distinct_count_exact)
 gint64
 garrow_array_statistics_get_distinct_count(GArrowArrayStatistics *statistics);
+#endif
+GARROW_AVAILABLE_IN_22_0
+gint64
+garrow_array_statistics_get_distinct_count_exact(GArrowArrayStatistics *statistics);
+GARROW_AVAILABLE_IN_22_0
+gdouble
+garrow_array_statistics_get_distinct_count_approximate(GArrowArrayStatistics *statistics);
 
 GARROW_AVAILABLE_IN_6_0
 GArrowArray *

c_glib/test/test-array-statistics.rb

@@ -51,11 +51,23 @@ def setup
 
   test("#has_distinct_count?") do
     assert do
-      !@statistics.has_distinct_count?
+      not @statistics.has_distinct_count?
     end
   end
 
-  test ("#distinct_count") do
-    assert_equal(-1, @statistics.distinct_count)
+  test("#distinct_count_exact?") do
+    assert do
+      not @statistics.distinct_count_exact?
+    end
+  end
+
+  test ("#distinct_count_exact") do
+    assert_equal(-1, @statistics.distinct_count_exact)
+  end
+
+  test ("#distinct_count_approximate") do
+    assert do
+      @statistics.distinct_count_approximate.nan?
+    end
   end
 end

cpp/cmake_modules/ThirdpartyToolchain.cmake

@@ -1767,6 +1767,10 @@ function(build_thrift)
   set(WITH_QT5 OFF)
   set(WITH_ZLIB OFF)
 
+  # Apache Thrift may change CMAKE_DEBUG_POSTFIX. So we'll restore the
+  # original CMAKE_DEBUG_POSTFIX later.
+  set(CMAKE_DEBUG_POSTFIX_KEEP ${CMAKE_DEBUG_POSTFIX})
+
   # Remove Apache Arrow's CMAKE_MODULE_PATH to ensure using Apache
   # Thrift's cmake_modules/.
   #
@@ -1775,6 +1779,12 @@ function(build_thrift)
   list(POP_FRONT CMAKE_MODULE_PATH)
   fetchcontent_makeavailable(thrift)
 
+  # Apache Thrift may change CMAKE_DEBUG_POSTFIX. So we restore
+  # CMAKE_DEBUG_POSTFIX.
+  set(CMAKE_DEBUG_POSTFIX
+      ${CMAKE_DEBUG_POSTFIX_KEEP}
+      CACHE BOOL "" FORCE)
+
   if(CMAKE_VERSION VERSION_LESS 3.28)
     set_property(DIRECTORY ${thrift_SOURCE_DIR} PROPERTY EXCLUDE_FROM_ALL TRUE)
   endif()
@@ -2549,7 +2559,7 @@ if(ARROW_USE_XSIMD)
                      IS_RUNTIME_DEPENDENCY
                      FALSE
                      REQUIRED_VERSION
-                     "8.1.0")
+                     "13.0.0")
 
   if(xsimd_SOURCE STREQUAL "BUNDLED")
     set(ARROW_XSIMD arrow::xsimd)

cpp/src/arrow/acero/exec_plan.cc

@@ -127,10 +127,11 @@ struct ExecPlanImpl : public ExecPlan {
     // call.
     auto scope = START_SCOPED_SPAN(span_, "ExecPlan", {{"plan", ToString()}});
     Future<> scheduler_finished = arrow::util::AsyncTaskScheduler::Make(
-        [this](arrow::util::AsyncTaskScheduler* async_scheduler) {
+        [this](std::shared_ptr<arrow::util::AsyncTaskScheduler> async_scheduler) {
           QueryContext* ctx = query_context();
-          RETURN_NOT_OK(ctx->Init(async_scheduler));
 
+          RETURN_NOT_OK(ctx->Init(async_scheduler.get()));
+          async_scheduler_ = async_scheduler;
 #ifdef ARROW_WITH_OPENTELEMETRY
           if (HasMetadata()) {
             auto pairs = metadata().get()->sorted_pairs();
@@ -231,8 +232,8 @@ struct ExecPlanImpl : public ExecPlan {
         // If an error occurs during StopProducing then we submit a task to fail.  If we
         // have already aborted then this will be ignored.  This way the failing status
         // will get communicated to finished_.
-        query_context()->async_scheduler()->AddSimpleTask(
-            [st] { return st; }, "ExecPlan::StopProducingErrorReporter"sv);
+        async_scheduler_->AddSimpleTask([st] { return st; },
+                                        "ExecPlan::StopProducingErrorReporter"sv);
       }
     }
   }
@@ -341,6 +342,7 @@ struct ExecPlanImpl : public ExecPlan {
   arrow::util::tracing::Span span_;
   std::shared_ptr<const KeyValueMetadata> metadata_;
   QueryContext query_context_;
+  std::shared_ptr<arrow::util::AsyncTaskScheduler> async_scheduler_;
   // This field only exists for backwards compatibility.  Remove once the deprecated
   // ExecPlan::Make overloads have been removed.
   std::shared_ptr<ThreadPool> owned_thread_pool_;

cpp/src/arrow/array/array_test.cc

@@ -3900,6 +3900,7 @@ class TestArrayDataStatistics : public ::testing::Test {
   void SetUp() {
     valids_ = {1, 0, 1, 1};
     null_count_ = std::count(valids_.begin(), valids_.end(), 0);
+    distinct_count_ = 3.0;
     average_byte_width_ = 4.0;
     null_buffer_ = *internal::BytesToBits(valids_);
     values_ = {1, 0, 3, -4};
@@ -3910,6 +3911,7 @@ class TestArrayDataStatistics : public ::testing::Test {
                             null_count_);
     data_->statistics = std::make_shared<ArrayStatistics>();
     data_->statistics->null_count = null_count_;
+    data_->statistics->distinct_count = distinct_count_;
     data_->statistics->average_byte_width = average_byte_width_;
     data_->statistics->is_average_byte_width_exact = true;
     data_->statistics->min = min_;
@@ -3921,6 +3923,7 @@ class TestArrayDataStatistics : public ::testing::Test {
  protected:
   std::vector<uint8_t> valids_;
   size_t null_count_;
+  double distinct_count_;
   double average_byte_width_;
   std::shared_ptr<Buffer> null_buffer_;
   std::vector<int32_t> values_;
@@ -3937,6 +3940,10 @@ TEST_F(TestArrayDataStatistics, MoveConstructor) {
   ASSERT_TRUE(moved_data.statistics->null_count.has_value());
   ASSERT_EQ(null_count_, moved_data.statistics->null_count.value());
 
+  ASSERT_TRUE(moved_data.statistics->distinct_count.has_value());
+  ASSERT_DOUBLE_EQ(distinct_count_,
+                   std::get<double>(moved_data.statistics->distinct_count.value()));
+
   ASSERT_TRUE(moved_data.statistics->average_byte_width.has_value());
   ASSERT_DOUBLE_EQ(average_byte_width_,
                    moved_data.statistics->average_byte_width.value());
@@ -3959,6 +3966,10 @@ TEST_F(TestArrayDataStatistics, CopyConstructor) {
   ASSERT_TRUE(copied_data.statistics->null_count.has_value());
   ASSERT_EQ(null_count_, copied_data.statistics->null_count.value());
 
+  ASSERT_TRUE(copied_data.statistics->distinct_count.has_value());
+  ASSERT_DOUBLE_EQ(distinct_count_,
+                   std::get<double>(copied_data.statistics->distinct_count.value()));
+
   ASSERT_TRUE(copied_data.statistics->average_byte_width.has_value());
   ASSERT_DOUBLE_EQ(average_byte_width_,
                    copied_data.statistics->average_byte_width.value());
@@ -3983,6 +3994,10 @@ TEST_F(TestArrayDataStatistics, MoveAssignment) {
   ASSERT_TRUE(moved_data.statistics->null_count.has_value());
   ASSERT_EQ(null_count_, moved_data.statistics->null_count.value());
 
+  ASSERT_TRUE(moved_data.statistics->distinct_count.has_value());
+  ASSERT_DOUBLE_EQ(distinct_count_,
+                   std::get<double>(moved_data.statistics->distinct_count.value()));
+
   ASSERT_TRUE(moved_data.statistics->average_byte_width.has_value());
   ASSERT_DOUBLE_EQ(average_byte_width_,
                    moved_data.statistics->average_byte_width.value());
@@ -4006,6 +4021,10 @@ TEST_F(TestArrayDataStatistics, CopyAssignment) {
   ASSERT_TRUE(copied_data.statistics->null_count.has_value());
   ASSERT_EQ(null_count_, copied_data.statistics->null_count.value());
 
+  ASSERT_TRUE(copied_data.statistics->distinct_count.has_value());
+  ASSERT_DOUBLE_EQ(distinct_count_,
+                   std::get<double>(copied_data.statistics->distinct_count.value()));
+
   ASSERT_TRUE(copied_data.statistics->average_byte_width.has_value());
   ASSERT_DOUBLE_EQ(average_byte_width_,
                    copied_data.statistics->average_byte_width.value());

cpp/src/arrow/array/statistics.h

@@ -39,6 +39,8 @@ struct ARROW_EXPORT ArrayStatistics {
   /// value exists, one of them is used. `std::nullopt` is used
   /// otherwise.
   using ValueType = std::variant<bool, int64_t, uint64_t, double, std::string>;
+  using NumericType = std::variant<int64_t, double>;
+  using CountType = NumericType;
 
   static const std::shared_ptr<DataType>& ValueToArrowType(
       const std::optional<ValueType>& value,
@@ -76,7 +78,9 @@ struct ARROW_EXPORT ArrayStatistics {
   std::optional<int64_t> null_count = std::nullopt;
 
   /// \brief The number of distinct values, may not be set
-  std::optional<int64_t> distinct_count = std::nullopt;
+  /// Note: when set to `int64_t`, it represents `exact_distinct_count`,
+  /// and when set to `double`, it represents `approximate_distinct_count`.
+  std::optional<CountType> distinct_count = std::nullopt;
 
   /// \brief The average size in bytes of a row in an array, may not be set.
   std::optional<double> average_byte_width = std::nullopt;