GH-41364: [GLib][Ruby] Allow passing thread pool to ExecutePlan (#48462)

### Rationale for this change

Aggregators like `first` and `last` are unusable in Ruby because they don't work when the execution plan is executed using multiple threads.

### What changes are included in this PR?

This adds the `ThreadPool` class to be able create a thread pool with a single thread. This can then be passed when creating an `ExecuteContext`, which in turn can be passed when creating an `ExecutePlan`.

### Are these changes tested?

A Ruby test that shows that the `first` aggregator works.

### Are there any user-facing changes?

A new `GArrowThreadPool` class, and changed signatures of the functions `garrow_execute_context_new` and `garrow_execute_plan_new`. However since the new arguments are nullable, it is backwards compatible for the Ruby API.

**This PR includes breaking changes to public APIs.**

* GitHub Issue: #41364

Authored-by: Sten Larsson <[email protected]>
Signed-off-by: Sutou Kouhei <[email protected]>

Author: stenlarsson

c_glib/arrow-glib/arrow-glib.h

@@ -33,6 +33,7 @@
 #include <arrow-glib/decoder.h>
 #include <arrow-glib/enums.h>
 #include <arrow-glib/error.h>
+#include <arrow-glib/executor.h>
 #include <arrow-glib/expression.h>
 #include <arrow-glib/field.h>
 #include <arrow-glib/interval.h>

c_glib/arrow-glib/compute.cpp

@@ -26,6 +26,7 @@
 #include <arrow-glib/datum.hpp>
 #include <arrow-glib/enums.h>
 #include <arrow-glib/error.hpp>
+#include <arrow-glib/executor.hpp>
 #include <arrow-glib/expression.hpp>
 #include <arrow-glib/reader.hpp>
 #include <arrow-glib/record-batch.hpp>
@@ -294,52 +295,132 @@ garrow_compute_initialize(GError **error)
   return garrow::check(error, status, "[compute][initialize]");
 }
 
-typedef struct GArrowExecuteContextPrivate_
+struct GArrowExecuteContextPrivate
 {
-  arrow::compute::ExecContext context;
-} GArrowExecuteContextPrivate;
+  std::shared_ptr<arrow::compute::ExecContext> context;
+  GArrowExecutor *executor;
+};
 
 G_DEFINE_TYPE_WITH_PRIVATE(GArrowExecuteContext, garrow_execute_context, G_TYPE_OBJECT)
 
 #define GARROW_EXECUTE_CONTEXT_GET_PRIVATE(object)                                       \
   static_cast<GArrowExecuteContextPrivate *>(                                            \
     garrow_execute_context_get_instance_private(GARROW_EXECUTE_CONTEXT(object)))
 
+enum {
+  PROP_EXECUTOR = 1,
+};
+
+static void
+garrow_execute_context_dispose(GObject *object)
+{
+  auto priv = GARROW_EXECUTE_CONTEXT_GET_PRIVATE(object);
+
+  if (priv->executor) {
+    g_object_unref(priv->executor);
+    priv->executor = nullptr;
+  }
+
+  G_OBJECT_CLASS(garrow_execute_context_parent_class)->dispose(object);
+}
+
 static void
 garrow_execute_context_finalize(GObject *object)
 {
   auto priv = GARROW_EXECUTE_CONTEXT_GET_PRIVATE(object);
-  priv->context.~ExecContext();
+  priv->context.~shared_ptr();
   G_OBJECT_CLASS(garrow_execute_context_parent_class)->finalize(object);
 }
 
+static void
+garrow_execute_context_set_property(GObject *object,
+                                    guint prop_id,
+                                    const GValue *value,
+                                    GParamSpec *pspec)
+{
+  auto priv = GARROW_EXECUTE_CONTEXT_GET_PRIVATE(object);
+
+  switch (prop_id) {
+  case PROP_EXECUTOR:
+    {
+      priv->executor = GARROW_EXECUTOR(g_value_dup_object(value));
+      auto arrow_executor = garrow_executor_get_raw(priv->executor);
+      priv->context =
+        std::make_shared<arrow::compute::ExecContext>(arrow::default_memory_pool(),
+                                                      arrow_executor.get());
+      break;
+    }
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
+    break;
+  }
+}
+
+static void
+garrow_execute_context_get_property(GObject *object,
+                                    guint prop_id,
+                                    GValue *value,
+                                    GParamSpec *pspec)
+{
+  auto priv = GARROW_EXECUTE_CONTEXT_GET_PRIVATE(object);
+
+  switch (prop_id) {
+  case PROP_EXECUTOR:
+    g_value_set_object(value, priv->executor);
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
+    break;
+  }
+}
+
 static void
 garrow_execute_context_init(GArrowExecuteContext *object)
 {
   auto priv = GARROW_EXECUTE_CONTEXT_GET_PRIVATE(object);
-  new (&priv->context) arrow::compute::ExecContext(arrow::default_memory_pool(), nullptr);
+  new (&priv->context) std::shared_ptr<arrow::compute::ExecContext>;
 }
 
 static void
 garrow_execute_context_class_init(GArrowExecuteContextClass *klass)
 {
   auto gobject_class = G_OBJECT_CLASS(klass);
 
+  gobject_class->dispose = garrow_execute_context_dispose;
   gobject_class->finalize = garrow_execute_context_finalize;
+  gobject_class->set_property = garrow_execute_context_set_property;
+  gobject_class->get_property = garrow_execute_context_get_property;
+
+  GParamSpec *spec;
+  /**
+   * GArrowExecuteContext:executor:
+   *
+   * The executor for execution.
+   *
+   * Since: 23.0.0
+   */
+  spec = g_param_spec_object(
+    "executor",
+    "Executor",
+    "The executor for execution",
+    GARROW_TYPE_EXECUTOR,
+    static_cast<GParamFlags>(G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY));
+  g_object_class_install_property(gobject_class, PROP_EXECUTOR, spec);
 }
 
 /**
  * garrow_execute_context_new:
+ * @executor: (nullable): A #GArrowExecutor or %NULL.
  *
  * Returns: A newly created #GArrowExecuteContext.
  *
  * Since: 1.0.0
  */
 GArrowExecuteContext *
-garrow_execute_context_new(void)
+garrow_execute_context_new(GArrowExecutor *executor)
 {
-  auto execute_context = g_object_new(GARROW_TYPE_EXECUTE_CONTEXT, NULL);
-  return GARROW_EXECUTE_CONTEXT(execute_context);
+  return GARROW_EXECUTE_CONTEXT(
+    g_object_new(GARROW_TYPE_EXECUTE_CONTEXT, "executor", executor, nullptr));
 }
 
 typedef struct GArrowFunctionOptionsPrivate_
@@ -1898,6 +1979,7 @@ garrow_execute_plan_class_init(GArrowExecutePlanClass *klass)
 
 /**
  * garrow_execute_plan_new:
+ * @context: (nullable): A #GArrowExecuteContext or %NULL.
  * @error: (nullable): Return location for a #GError or %NULL.
  *
  * Returns: (nullable): A newly created #GArrowExecutePlan on success,
@@ -1906,9 +1988,15 @@ garrow_execute_plan_class_init(GArrowExecutePlanClass *klass)
  * Since: 6.0.0
  */
 GArrowExecutePlan *
-garrow_execute_plan_new(GError **error)
+garrow_execute_plan_new(GArrowExecuteContext *context, GError **error)
 {
-  auto arrow_plan_result = arrow::acero::ExecPlan::Make();
+  arrow::Result<std::shared_ptr<arrow::acero::ExecPlan>> arrow_plan_result;
+  if (context) {
+    auto arrow_context = garrow_execute_context_get_raw(context);
+    arrow_plan_result = arrow::acero::ExecPlan::Make(*arrow_context);
+  } else {
+    arrow_plan_result = arrow::acero::ExecPlan::Make();
+  }
   if (garrow::check(error, arrow_plan_result, "[execute-plan][new]")) {
     return GARROW_EXECUTE_PLAN(
       g_object_new(GARROW_TYPE_EXECUTE_PLAN, "plan", &(*arrow_plan_result), NULL));
@@ -7207,7 +7295,7 @@ arrow::compute::ExecContext *
 garrow_execute_context_get_raw(GArrowExecuteContext *context)
 {
   auto priv = GARROW_EXECUTE_CONTEXT_GET_PRIVATE(context);
-  return &priv->context;
+  return priv->context.get();
 }
 
 GArrowFunctionOptions *

c_glib/arrow-glib/compute.h

@@ -21,6 +21,7 @@
 
 #include <arrow-glib/compute-definition.h>
 #include <arrow-glib/datum.h>
+#include <arrow-glib/executor.h>
 #include <arrow-glib/reader.h>
 
 G_BEGIN_DECLS
@@ -40,7 +41,7 @@ struct _GArrowExecuteContextClass
 
 GARROW_AVAILABLE_IN_1_0
 GArrowExecuteContext *
-garrow_execute_context_new(void);
+garrow_execute_context_new(GArrowExecutor *executor);
 
 GARROW_AVAILABLE_IN_7_0
 gboolean
@@ -322,7 +323,7 @@ struct _GArrowExecutePlanClass
 
 GARROW_AVAILABLE_IN_6_0
 GArrowExecutePlan *
-garrow_execute_plan_new(GError **error);
+garrow_execute_plan_new(GArrowExecuteContext *context, GError **error);
 GARROW_AVAILABLE_IN_6_0
 GArrowExecuteNode *
 garrow_execute_plan_build_node(GArrowExecutePlan *plan,

c_glib/arrow-glib/executor.cpp

@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <arrow-glib/error.hpp>
+#include <arrow-glib/executor.hpp>
+
+#include <arrow/util/thread_pool.h>
+
+G_BEGIN_DECLS
+
+/**
+ * SECTION: executor
+ * @section_id: executor-classes
+ * @title: Executor classes
+ * @include: arrow-glib/arrow-glib.h
+ *
+ * #GArrowExecutor is the base class for executor implementations.
+ *
+ * #GArrowThreadPool is a class for thread pool management.
+ */
+
+struct GArrowExecutorPrivate
+{
+  std::shared_ptr<arrow::internal::Executor> executor;
+};
+
+G_DEFINE_ABSTRACT_TYPE_WITH_PRIVATE(GArrowExecutor, garrow_executor, G_TYPE_OBJECT)
+
+#define GARROW_EXECUTOR_GET_PRIVATE(obj)                                                 \
+  static_cast<GArrowExecutorPrivate *>(                                                  \
+    garrow_executor_get_instance_private(GARROW_EXECUTOR(obj)))
+
+enum {
+  PROP_EXECUTOR = 1,
+};
+
+static void
+garrow_executor_init(GArrowExecutor *object)
+{
+  auto priv = GARROW_EXECUTOR_GET_PRIVATE(object);
+  new (&priv->executor) std::shared_ptr<arrow::internal::Executor>;
+}
+
+static void
+garrow_executor_finalize(GObject *object)
+{
+  auto priv = GARROW_EXECUTOR_GET_PRIVATE(object);
+  priv->executor.~shared_ptr();
+  G_OBJECT_CLASS(garrow_executor_parent_class)->finalize(object);
+}
+
+static void
+garrow_executor_set_property(GObject *object,
+                             guint prop_id,
+                             const GValue *value,
+                             GParamSpec *pspec)
+{
+  auto priv = GARROW_EXECUTOR_GET_PRIVATE(GARROW_EXECUTOR(object));
+
+  switch (prop_id) {
+  case PROP_EXECUTOR:
+    priv->executor = *static_cast<std::shared_ptr<arrow::internal::Executor> *>(
+      g_value_get_pointer(value));
+    break;
+  default:
+    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
+    break;
+  }
+}
+
+static void
+garrow_executor_class_init(GArrowExecutorClass *klass)
+{
+  auto gobject_class = G_OBJECT_CLASS(klass);
+
+  gobject_class->finalize = garrow_executor_finalize;
+  gobject_class->set_property = garrow_executor_set_property;
+
+  GParamSpec *spec;
+  spec = g_param_spec_pointer(
+    "executor",
+    "Executor",
+    "The raw std::shared_ptr<arrow::internal::Executor> *",
+    static_cast<GParamFlags>(G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY));
+  g_object_class_install_property(gobject_class, PROP_EXECUTOR, spec);
+}
+
+G_DEFINE_TYPE(GArrowThreadPool, garrow_thread_pool, GARROW_TYPE_EXECUTOR)
+
+static void
+garrow_thread_pool_init(GArrowThreadPool *object)
+{
+}
+
+static void
+garrow_thread_pool_class_init(GArrowThreadPoolClass *klass)
+{
+}
+
+/**
+ * garrow_thread_pool_new:
+ * @n_threads: The number of threads in the pool.
+ * @error: (nullable): Return location for a #GError or %NULL.
+ *
+ * Returns: (nullable): A newly created #GArrowThreadPool on success,
+ *   %NULL on error.
+ *
+ * Since: 23.0.0
+ */
+GArrowThreadPool *
+garrow_thread_pool_new(guint n_threads, GError **error)
+{
+  auto arrow_thread_pool_result = arrow::internal::ThreadPool::Make(n_threads);
+  if (garrow::check(error, arrow_thread_pool_result, "[thread-pool][new]")) {
+    auto arrow_thread_pool = *arrow_thread_pool_result;
+    auto thread_pool = GARROW_THREAD_POOL(
+      g_object_new(GARROW_TYPE_THREAD_POOL, "executor", &arrow_thread_pool, nullptr));
+    return thread_pool;
+  } else {
+    return nullptr;
+  }
+}
+
+G_END_DECLS
+
+std::shared_ptr<arrow::internal::Executor>
+garrow_executor_get_raw(GArrowExecutor *executor)
+{
+  if (!executor)
+    return nullptr;
+
+  auto priv = GARROW_EXECUTOR_GET_PRIVATE(executor);
+  return priv->executor;
+}