[Ruby] Allow disabling threads for execution plans

[stenlarsson]

Describe the enhancement requested

The aggregators first and last are unusable in Ruby because they don't work when the execution plan is executed using threads:

require 'arrow'

table = Arrow::Table.new(
 'foo' => [1, 1],
 'bar' => ['first', 'second'],
)
pp table.group('foo').aggregate('first(bar)')

Executing this script results in:

[execute-plan][build-node]: NotImplemented: Using ordered aggregator in multiple threaded execution is not supported (Arrow::Error::NotImplemented)

This seems like the same issue that was fixed for Python in #36768, and similar solution is needed for Ruby.

Component(s)

Ruby

[kou]

It makes sense.

We need to implement the bindings of arrow::acero::QueryOptions bindings as the first for this:

arrow/cpp/src/arrow/acero/exec_plan.h

Lines 509 to 605 in d7a5777

	/// \brief plan-wide options that can be specified when executing an execution plan
	struct ARROW_ACERO_EXPORT QueryOptions {
	/// \brief Should the plan use a legacy batching strategy
	///
	/// This is currently in place only to support the Scanner::ToTable
	/// method. This method relies on batch indices from the scanner
	/// remaining consistent. This is impractical in the ExecPlan which
	/// might slice batches as needed (e.g. for a join)
	///
	/// However, it still works for simple plans and this is the only way
	/// we have at the moment for maintaining implicit order.
	bool use_legacy_batching = false;
	/// If the output has a meaningful order then sequence the output of the plan
	///
	/// The default behavior (std::nullopt) will sequence output batches if there
	/// is a meaningful ordering in the final node and will emit batches immediately
	/// otherwise.
	///
	/// If explicitly set to true then plan execution will fail if there is no
	/// meaningful ordering. This can be useful to validate a query that should
	/// be emitting ordered results.
	///
	/// If explicitly set to false then batches will be emit immediately even if there
	/// is a meaningful ordering. This could cause batches to be emit out of order but
	/// may offer a small decrease to latency.
	std::optional<bool> sequence_output = std::nullopt;
	/// \brief should the plan use multiple background threads for CPU-intensive work
	///
	/// If this is false then all CPU work will be done on the calling thread. I/O tasks
	/// will still happen on the I/O executor and may be multi-threaded (but should not use
	/// significant CPU resources).
	///
	/// Will be ignored if custom_cpu_executor is set
	bool use_threads = true;
	/// \brief custom executor to use for CPU-intensive work
	///
	/// Must be null or remain valid for the duration of the plan. If this is null then
	/// a default thread pool will be chosen whose behavior will be controlled by
	/// the `use_threads` option.
	::arrow::internal::Executor* custom_cpu_executor = NULLPTR;
	/// \brief custom executor to use for IO work
	///
	/// Must be null or remain valid for the duration of the plan. If this is null then
	/// the global io thread pool will be chosen whose behavior will be controlled by
	/// the "ARROW_IO_THREADS" environment.
	::arrow::internal::Executor* custom_io_executor = NULLPTR;
	/// \brief a memory pool to use for allocations
	///
	/// Must remain valid for the duration of the plan.
	MemoryPool* memory_pool = default_memory_pool();
	/// \brief a function registry to use for the plan
	///
	/// Must remain valid for the duration of the plan.
	FunctionRegistry* function_registry = GetFunctionRegistry();
	/// \brief the names of the output columns
	///
	/// If this is empty then names will be generated based on the input columns
	///
	/// If set then the number of names must equal the number of output columns
	std::vector<std::string> field_names;
	/// \brief Policy for unaligned buffers in source data
	///
	/// Various compute functions and acero internals will type pun array
	/// buffers from uint8_t* to some kind of value type (e.g. we might
	/// cast to int32_t* to add two int32 arrays)
	///
	/// If the buffer is poorly aligned (e.g. an int32 array is not aligned
	/// on a 4-byte boundary) then this is technically undefined behavior in C++.
	/// However, most modern compilers and CPUs are fairly tolerant of this
	/// behavior and nothing bad (beyond a small hit to performance) is likely
	/// to happen.
	///
	/// Note that this only applies to source buffers. All buffers allocated internally
	/// by Acero will be suitably aligned.
	///
	/// If this field is set to kWarn then Acero will check if any buffers are unaligned
	/// and, if they are, will emit a warning.
	///
	/// If this field is set to kReallocate then Acero will allocate a new, suitably aligned
	/// buffer and copy the contents from the old buffer into this new buffer.
	///
	/// If this field is set to kError then Acero will gracefully abort the plan instead.
	///
	/// If this field is set to kIgnore then Acero will not even check if the buffers are
	/// unaligned.
	///
	/// If this field is not set then it will be treated as kWarn unless overridden
	/// by the ACERO_ALIGNMENT_HANDLING environment variable
	std::optional<UnalignedBufferHandling> unaligned_buffer_handling;
	};

We should create a separated issue for this to work on this step by step.

[stenlarsson]

Wouldn't that mean we have to implement bindings for the whole "declaration" API?

I don't fully understand the code, but it seems that when use_threads is false it creates a thread pool with only one thread:

arrow/cpp/src/arrow/acero/exec_plan.cc

Lines 769 to 776 in de3cdc0

	if (use_threads) {
	return DeclarationToTableImpl(std::move(declaration), query_options,
	::arrow::internal::GetCpuThreadPool());
	} else {
	ARROW_ASSIGN_OR_RAISE(std::shared_ptr<ThreadPool> tpool, ThreadPool::Make(1));
	return DeclarationToTableImpl(std::move(declaration), query_options, tpool.get())
	.Then([tpool](const std::shared_ptr<Table>& table) { return table; });
	}

The size of the thread pool is then checked by the GroupByNode:

arrow/cpp/src/arrow/acero/groupby_aggregate_node.cc

Line 189 in 32885c9

bool is_cpu_parallel = plan->query_context()->executor()->GetCapacity() > 1;

I think if we were able to create an ExecuteContext with such a thread pool in Ruby, and then pass that to the constructor of ExecutePlan we should be good, right?

[kou]

Ah, ExecPlan doesn't see arrow::acero::QueryOptions::use_threads...

Our options:

1. Add support for ordered aggregation in parallel in C++

   GH-32884: [C++] Add ordered aggregation #34311 is the PR that implements ordered aggregation.

2. Add support for the declaration API

3. Add Arrow::ThreadPool and use it with Arrow::ExecutePlan

   We also need to improve Arrow::ExecuteContext and so on.

[thisisnic]

This issue has been marked as stale because it has had no activity in the past 365 days. Please remove the stale label or comment below, or this issue will be closed in 14 days. If this improvement is still desired but has no current owner, please add the 'Status: needs champion' label.

[stenlarsson]

I will try to implement this myself.

[kou]

Issue resolved by pull request 48462
#48462