feat: remove SET threads=1 requirement for table functions on DuckDB >= 1.5.0

Author: otegami

lib/duckdb/connection.rb

@@ -236,12 +236,11 @@ def register_table_function(table_function)
     # @param columns [Hash{String => DuckDB::LogicalType}, nil] optional column schema override;
     #   if omitted, the adapter determines the columns (e.g. from headers or inference)
     # @raise [ArgumentError] if no adapter is registered for the object's class
-    # @raise [DuckDB::Error] if threads setting is not 1
+    # @raise [DuckDB::Error] if threads > 1 on DuckDB < 1.5.0
     # @return [void]
     #
     # @example Expose a CSV as a table
     #   require 'csv'
-    #   con.execute('SET threads=1')
     #   DuckDB::TableFunction.add_table_adapter(CSV, CSVTableAdapter.new)
     #   csv = CSV.new(File.read('data.csv'), headers: true)
     #   con.expose_as_table(csv, 'csv_table')
@@ -263,16 +262,22 @@ def expose_as_table(object, name, columns: nil)
 
     private
 
+    # DuckDB >= 1.5.0 provides per-worker proxy threads via init_local_state,
+    # making table function callbacks thread-safe with multiple DuckDB threads.
+    # On older versions, the global executor serializes all callbacks and can
+    # deadlock under concurrent workloads, so we enforce threads=1.
     def check_threads
+      return if Gem::Version.new(LIBRARY_VERSION) >= Gem::Version.new('1.5.0')
+
       result = execute("SELECT current_setting('threads')")
       thread_count = result.first.first.to_i
 
       return unless thread_count > 1
 
       raise DuckDB::Error,
-            'Functions with Ruby callbacks require single-threaded execution. ' \
-            "Current threads setting: #{thread_count}. " \
-            "Execute 'SET threads=1' before registering functions."
+            'Table functions with Ruby callbacks require single-threaded execution ' \
+            "on DuckDB < 1.5.0. Current threads setting: #{thread_count}. " \
+            "Execute 'SET threads=1' before registering table functions."
     end
 
     def run_appender_block(appender, &)

test/duckdb_test/data_chunk_test.rb

@@ -56,7 +56,7 @@ def test_data_chunk_get_vector
 
     # Test 4: Vector#logical_type returns LogicalType
     def test_vector_logical_type # rubocop:disable Metrics/AbcSize, Metrics/MethodLength, Minitest/MultipleAssertions
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       table_function = DuckDB::TableFunction.new
       table_function.name = 'test_vector_type'
@@ -98,7 +98,7 @@ def test_vector_logical_type # rubocop:disable Metrics/AbcSize, Metrics/MethodLe
 
     # Test 5: DataChunk#set_value with INTEGER
     def test_data_chunk_set_value_integer # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       done = false
       table_function = DuckDB::TableFunction.new
@@ -133,7 +133,7 @@ def test_data_chunk_set_value_integer # rubocop:disable Metrics/AbcSize, Metrics
 
     # Test 6: DataChunk#set_value with BIGINT
     def test_data_chunk_set_value_bigint # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       done = false
       table_function = DuckDB::TableFunction.new
@@ -165,7 +165,7 @@ def test_data_chunk_set_value_bigint # rubocop:disable Metrics/AbcSize, Metrics/
 
     # Test 7: DataChunk#set_value with VARCHAR
     def test_data_chunk_set_value_varchar # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       done = false
       table_function = DuckDB::TableFunction.new
@@ -199,7 +199,7 @@ def test_data_chunk_set_value_varchar # rubocop:disable Metrics/AbcSize, Metrics
 
     # Test 8: DataChunk#set_value with DOUBLE
     def test_data_chunk_set_value_double # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       done = false
       table_function = DuckDB::TableFunction.new
@@ -233,7 +233,7 @@ def test_data_chunk_set_value_double # rubocop:disable Metrics/AbcSize, Metrics/
 
     # Test 9: DataChunk#set_value with NULL
     def test_data_chunk_set_value_null # rubocop:disable Metrics/AbcSize, Metrics/MethodLength, Minitest/MultipleAssertions
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       done = false
       table_function = DuckDB::TableFunction.new
@@ -269,7 +269,7 @@ def test_data_chunk_set_value_null # rubocop:disable Metrics/AbcSize, Metrics/Me
 
     # Test 10: DataChunk#set_value with BLOB
     def test_data_chunk_set_value_blob # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       done = false
       table_function = DuckDB::TableFunction.new
@@ -306,7 +306,7 @@ def test_data_chunk_set_value_blob # rubocop:disable Metrics/AbcSize, Metrics/Me
     # Test 11: DataChunk#set_value with multiple columns
     # rubocop:disable Metrics/AbcSize, Metrics/MethodLength, Minitest/MultipleAssertions
     def test_data_chunk_set_value_multiple_columns
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       done = false
       table_function = DuckDB::TableFunction.new
@@ -361,7 +361,7 @@ def test_data_chunk_set_value_multiple_columns
 
     # Test 12: DataChunk#set_value with TIMESTAMP
     def test_data_chunk_set_value_timestamp # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       done = false
       table_function = DuckDB::TableFunction.new
@@ -398,7 +398,7 @@ def test_data_chunk_set_value_timestamp # rubocop:disable Metrics/AbcSize, Metri
 
     # Test 13: DataChunk#set_value with TIMESTAMP_TZ
     def test_data_chunk_set_value_timestamp_tz # rubocop:disable Metrics/AbcSize, Metrics/MethodLength, Minitest/MultipleAssertions
-      @conn.execute('SET threads=1')
+      @conn.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       done = false
       table_function = DuckDB::TableFunction.new

test/duckdb_test/gc_stress_test.rb

@@ -83,7 +83,7 @@ def test_table_function_with_gc_compaction
       skip 'GC.compact not available' unless GC.respond_to?(:compact)
       skip 'GC.compact hangs on Windows in parallel test execution' if Gem.win_platform?
 
-      @con.execute('SET threads=1') # Table functions still require single-threaded execution
+      @con.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       # Capture local variables
       multiplier = 3
@@ -136,7 +136,7 @@ def test_mixed_functions_gc_stress
       skip 'GC.compact not available' unless GC.respond_to?(:compact)
       skip 'GC.compact hangs on Windows in parallel test execution' if Gem.win_platform?
 
-      @con.execute('SET threads=1') # Table functions still require single-threaded execution
+      @con.execute('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       # Register both scalar and table functions
       @con.register_scalar_function(DuckDB::ScalarFunction.new.tap do |sf|

test/duckdb_test/table_function_csv_test.rb

@@ -50,7 +50,9 @@ def csv_to_duckdb_data(csv, output)
     def setup
       @db = DuckDB::Database.open
       @con = @db.connect
-      @con.execute('SET threads=1') # Required for Ruby callbacks
+      return unless Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
+
+      @con.execute('SET threads=1')
     end
 
     def teardown

test/duckdb_test/table_function_integration_test.rb

@@ -7,7 +7,9 @@ class TableFunctionIntegrationTest < Minitest::Test
     def setup
       @database = DuckDB::Database.open
       @connection = @database.connect
-      @connection.execute('SET threads=1') # Required for Ruby callbacks
+      return unless Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
+
+      @connection.execute('SET threads=1')
     end
 
     def teardown

test/duckdb_test/table_function_test.rb

@@ -16,7 +16,7 @@ def test_new
     def test_create_with_set_value
       db = DuckDB::Database.open
       conn = db.connect
-      conn.query('SET threads=1')
+      conn.query('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       called = 0
 
@@ -92,7 +92,7 @@ def test_gc_compaction_safety
 
       db = DuckDB::Database.open
       conn = db.connect
-      conn.query('SET threads=1')
+      conn.query('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       # Capture local variable in callbacks
       row_multiplier = 2
@@ -158,7 +158,7 @@ def test_gc_compaction_safety
     def test_symbol_columns
       db = DuckDB::Database.open
       conn = db.connect
-      conn.query('SET threads=1')
+      conn.query('SET threads=1') if Gem::Version.new(DuckDB::LIBRARY_VERSION) < Gem::Version.new('1.5.0')
 
       # Capture local variable in callbacks
       row_multiplier = 2
@@ -206,6 +206,42 @@ def test_symbol_columns
     end
     # rubocop:enable Metrics/AbcSize, Metrics/MethodLength
 
+    def test_table_function_with_multithread # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
+      unless Gem::Version.new(DuckDB::LIBRARY_VERSION) >= Gem::Version.new('1.5.0')
+        skip 'per-worker proxy requires duckdb >= 1.5.0'
+      end
+
+      db = DuckDB::Database.open
+      conn = db.connect
+      conn.execute('SET threads=4')
+
+      called = 0
+      tf = DuckDB::TableFunction.new
+      tf.name = 'mt_generate'
+      tf.bind do |bind_info|
+        bind_info.add_result_column('value', DuckDB::LogicalType::INTEGER)
+      end
+      tf.init { |_init_info| } # rubocop:disable Lint/EmptyBlock
+      tf.execute do |_func_info, output|
+        if called.zero?
+          100.times { |i| output.set_value(0, i, i * 2) }
+          output.size = 100
+          called += 1
+        else
+          output.size = 0
+        end
+      end
+
+      conn.register_table_function(tf)
+      result = conn.execute('SELECT SUM(value) FROM mt_generate()')
+
+      # sum(0, 2, 4, ..., 198) = 2 * sum(0..99) = 2 * 4950 = 9900
+      assert_equal 9900, result.first.first
+
+      conn.disconnect
+      db.close
+    end
+
     private
 
     def setup_incomplete_function