feat: per-worker proxy for table function execute callback

Wire the execute path to per-worker proxy threads on DuckDB >= 1.5.0.
A local_init callback registered via duckdb_table_function_set_local_init
runs once per worker thread, creates a proxy (allocating its Ruby thread
under the GVL through the global executor, since local_init runs on a
non-Ruby thread), and stores it as thread-local init data. The execute
callback retrieves that proxy and dispatches through it via
rbduckdb_function_executor_dispatch_via_proxy, so callbacks from
different workers run concurrently instead of serializing on the single
global executor. DuckDB frees each proxy through rbduckdb_worker_proxy_destroy.

bind and init stay on the global executor (not on the hot path). On
DuckDB < 1.5.0 the local_init hook is absent and the execute callback
keeps using the global executor unchanged.

Verified: with SET threads=4 plus cardinality/max_threads hints, a
GVL-releasing callback reaches max_concurrent=4 (vs 2 on the global
executor) for a ~2x speedup; results are identical. The added test
asserts correctness of the local_init -> proxy -> destroy lifecycle
under multi-threaded execution (throughput is checked manually to avoid
CI flakiness).

Author: otegami

ext/duckdb/table_function.c

@@ -21,6 +21,11 @@ static VALUE rbduckdb_table_function_set_init(VALUE self);
 static void table_function_init_callback(duckdb_init_info info);
 static VALUE rbduckdb_table_function_set_execute(VALUE self);
 static void table_function_execute_callback(duckdb_function_info info, duckdb_data_chunk output);
+#ifdef HAVE_DUCKDB_H_GE_V1_5_0
+/* Thread detection (declared in function_executor.c); used to skip the proxy on Ruby threads. */
+extern int ruby_native_thread_p(void);
+static void table_function_local_init_callback(duckdb_init_info info);
+#endif
 
 static const rb_data_type_t table_function_data_type = {
     "DuckDB/TableFunction",
@@ -358,6 +363,10 @@ static VALUE rbduckdb_table_function_set_execute(VALUE self) {
 
     ctx->execute_proc = rb_block_proc();
     duckdb_table_function_set_function(ctx->table_function, table_function_execute_callback);
+#ifdef HAVE_DUCKDB_H_GE_V1_5_0
+    /* Per-worker proxy threads for the execute path (DuckDB >= 1.5.0). */
+    duckdb_table_function_set_local_init(ctx->table_function, table_function_local_init_callback);
+#endif
 
     rbduckdb_function_executor_ensure_started();
 
@@ -405,6 +414,7 @@ static void execute_execute_callback_protected(void *user_data) {
 static void table_function_execute_callback(duckdb_function_info info, duckdb_data_chunk output) {
     rubyDuckDBTableFunction *ctx;
     struct execute_dispatch_arg darg;
+    struct worker_proxy *proxy = NULL;
 
     ctx = (rubyDuckDBTableFunction *)duckdb_function_get_extra_info(info);
     if (!ctx || ctx->execute_proc == Qnil) return;
@@ -413,9 +423,48 @@ static void table_function_execute_callback(duckdb_function_info info, duckdb_da
     darg.info = info;
     darg.output = output;
 
-    rbduckdb_function_executor_dispatch(execute_execute_callback_protected, &darg);
+#ifdef HAVE_DUCKDB_H_GE_V1_5_0
+    /* On DuckDB >= 1.5.0 each worker thread carries its own proxy (see local_init). */
+    proxy = (struct worker_proxy *)duckdb_function_get_local_init_data(info);
+#endif
+    rbduckdb_function_executor_dispatch_via_proxy(execute_execute_callback_protected, &darg, proxy);
 }
 
+#ifdef HAVE_DUCKDB_H_GE_V1_5_0
+/*
+ * Per-worker init for the execute path (DuckDB >= 1.5.0).
+ *
+ * DuckDB calls this once on each worker thread that will run the execute
+ * callback. We create a per-worker proxy (allocating its Ruby thread under the
+ * GVL via the global executor, since this runs on a non-Ruby thread) and store
+ * it as thread-local init data. The execute callback then dispatches through it
+ * instead of the shared global executor, so workers run callbacks concurrently.
+ * DuckDB invokes rbduckdb_worker_proxy_destroy when the local state is freed.
+ */
+struct table_proxy_create_arg {
+    struct worker_proxy *proxy;
+};
+
+static void table_create_proxy_callback(void *user_data) {
+    struct table_proxy_create_arg *arg = (struct table_proxy_create_arg *)user_data;
+    arg->proxy = rbduckdb_worker_proxy_create();
+}
+
+static void table_function_local_init_callback(duckdb_init_info info) {
+    struct table_proxy_create_arg arg;
+
+    /* A Ruby calling thread runs the callback inline (Case 1/2); no proxy needed. */
+    if (ruby_native_thread_p()) return;
+
+    arg.proxy = NULL;
+    rbduckdb_function_executor_dispatch(table_create_proxy_callback, &arg);
+
+    if (arg.proxy != NULL) {
+        duckdb_init_set_init_data(info, arg.proxy, rbduckdb_worker_proxy_destroy);
+    }
+}
+#endif
+
 rubyDuckDBTableFunction *get_struct_table_function(VALUE self) {
     rubyDuckDBTableFunction *ctx;
     TypedData_Get_Struct(self, rubyDuckDBTableFunction, &table_function_data_type, ctx);

test/duckdb_test/table_function_test.rb

@@ -199,6 +199,66 @@ def test_symbol_columns
       db.close
     end
 
+    # Per-worker proxy (GH-1136): exercises the local_init -> proxy -> destroy
+    # lifecycle under real multi-threaded execution (SET threads=4). We assert
+    # correctness rather than a concurrency count: the proxy path must neither
+    # deadlock nor corrupt results when DuckDB distributes the scan across
+    # workers. Parallel *throughput* is verified separately by
+    # letters/issue-1136/table_function_parallel_check.rb (kept out of CI to
+    # avoid scheduler-dependent flakiness). Requires DuckDB >= 1.5.0
+    # (duckdb_table_function_set_local_init).
+    def test_execute_runs_correctly_under_multiple_threads
+      if ::DuckDBTest.duckdb_library_version < Gem::Version.new('1.5.0')
+        skip 'per-worker proxy requires DuckDB >= 1.5.0'
+      end
+
+      chunks = 64
+      rows_per_chunk = 100
+      remaining = chunks
+      mutex = Mutex.new
+
+      db = DuckDB::Database.open
+      conn = db.connect
+      conn.execute('SET threads=4')
+
+      tf = DuckDB::TableFunction.new
+      tf.name = 'parallel_emitter'
+      tf.bind do |bind_info|
+        bind_info.add_result_column('v', DuckDB::LogicalType::BIGINT)
+        # Tell the planner there is real work so it distributes across workers.
+        bind_info.set_cardinality(chunks * rows_per_chunk, false)
+      end
+      tf.init do |init_info|
+        # Without this DuckDB assigns a single worker and the proxy never fires.
+        init_info.max_threads = 4
+      end
+      tf.execute do |_info, output|
+        has_work = mutex.synchronize do
+          next false if remaining.zero?
+
+          remaining -= 1
+          true
+        end
+
+        unless has_work
+          output.size = 0
+          next
+        end
+
+        rows_per_chunk.times { |i| output.set_value(0, i, 1) }
+        output.size = rows_per_chunk
+        sleep 0.001 # release the GVL so workers can overlap
+      end
+
+      conn.register_table_function(tf)
+      result = conn.query('SELECT COUNT(*), SUM(v) FROM parallel_emitter()').each.to_a
+
+      assert_equal [chunks * rows_per_chunk, chunks * rows_per_chunk], result.first
+
+      conn.disconnect
+      db.close
+    end
+
     private
 
     def setup_incomplete_function