tests: add a variety of tests relating to threading. Tests should pass whether GIL or free threaded. Tested with pytest-run-parallel: pytest --parallel-threads=10 --iterations=5

Author: paultiq

pyproject.toml

@@ -227,6 +227,7 @@ test = [ # dependencies used for running tests
     "pytest",
     "pytest-reraise",
     "pytest-timeout",
+    "pytest-run-parallel",
     "mypy",
     "coverage",
     "gcovr; python_version < '3.14'",

tests/conftest.py

@@ -336,3 +336,13 @@ def finalizer():
 
     duckdb.connect(test_dbfarm)
     return test_dbfarm
+
+
+@pytest.fixture(scope="function")
+def num_threads_testing():
+    """Get thread count: enough to load the system, but still as fast test."""
+    import multiprocessing
+
+    cpu_count = multiprocessing.cpu_count()
+    # Use 1.5x CPU count, max 12 for CI compatibility
+    return min(12, max(4, int(cpu_count * 1.5)))

tests/fast/threading/README.md

@@ -0,0 +1,10 @@
+Tests in this directory are intended to be run with [pytest-run-parallel](https://github.com/Quansight-Labs/pytest-run-parallel) to exercise thread safety. 
+
+Example usage: `pytest --parallel-threads=10 --iterations=5 --verbose tests/fast/threading -n 4 --durations=5`
+
+#### Thread Safety and DuckDB
+
+Not all duckdb operations are thread safe - cursors are not thread safe, so some care must be considered to avoid running tests that concurrently hit the same tests.
+
+Tests can be marked as single threaded with:
+- `pytest.mark.thread_unsafe` or the equivalent `pytest.mark.parallel_threads(1)`

tests/fast/threading/test_basic_operations.py

@@ -0,0 +1,117 @@
+import gc
+import random
+import time
+import weakref
+from threading import get_ident
+
+import uuid
+
+import pytest
+
+import duckdb
+
+
+def test_basic():
+    with duckdb.connect(":memory:") as conn:
+        result = conn.execute("SELECT 1").fetchone()
+        assert result[0] == 1
+        int_type = duckdb.type("INTEGER")
+        assert int_type is not None, "type creation failed"
+
+
+def test_connection_instance_cache(tmp_path):
+    thread_id = get_ident()
+    for i in range(10):
+        with duckdb.connect(tmp_path / f"{thread_id}_{uuid.uuid4()}.db") as conn:
+            conn.execute(
+                f"CREATE TABLE IF NOT EXISTS thread_{thread_id}_data_{i} (x BIGINT)"
+            )
+            conn.execute(f"INSERT INTO thread_{thread_id}_data_{i} VALUES (100), (100)")
+
+            time.sleep(random.uniform(0.0001, 0.001))
+
+            result = conn.execute(
+                f"SELECT COUNT(*) FROM thread_{thread_id}_data_{i}"
+            ).fetchone()[0]
+            assert result == 2, f"Iteration {i}: expected 2 rows, got {result}"
+
+
+def test_cleanup():
+    weak_refs = []
+
+    for i in range(5):
+        conn = duckdb.connect(":memory:")
+        weak_refs.append(weakref.ref(conn))
+        try:
+            conn.execute("CREATE TABLE test (x INTEGER)")
+            conn.execute("INSERT INTO test VALUES (1), (2), (3)")
+            result = conn.execute("SELECT COUNT(*) FROM test").fetchone()
+            assert result[0] == 3
+        finally:
+            conn.close()
+            conn = None
+
+        if i % 3 == 0:
+            with duckdb.connect(":memory:") as new_conn:
+                result = new_conn.execute("SELECT 1").fetchone()
+                assert result[0] == 1
+
+        if i % 10 == 0:
+            gc.collect()
+            time.sleep(random.uniform(0.0001, 0.0005))
+
+    gc.collect()
+    time.sleep(0.1)
+    gc.collect()
+
+    alive_refs = [ref for ref in weak_refs if ref() is not None]
+    assert len(alive_refs) <= 10, (
+        f"{len(alive_refs)} connections still alive (expected <= 10)"
+    )
+
+
+def test_default_connection():
+    with duckdb.connect() as conn1:
+        r1 = conn1.execute("SELECT 1").fetchone()[0]
+        assert r1 == 1, f"expected 1, got {r1}"
+
+    with duckdb.connect(":memory:") as conn2:
+        r2 = conn2.execute("SELECT 2").fetchone()[0]
+        assert r2 == 2, f"expected 2, got {r2}"
+
+
+def test_type_system():
+    for i in range(20):
+        types = [
+            duckdb.type("INTEGER"),
+            duckdb.type("VARCHAR"),
+            duckdb.type("DOUBLE"),
+            duckdb.type("BOOLEAN"),
+            duckdb.list_type(duckdb.type("INTEGER")),
+            duckdb.struct_type(
+                {"a": duckdb.type("INTEGER"), "b": duckdb.type("VARCHAR")}
+            ),
+        ]
+
+        for t in types:
+            assert t is not None, "type creation failed"
+
+        if i % 5 == 0:
+            with duckdb.connect(":memory:") as conn:
+                conn.execute(
+                    "CREATE TABLE test (a INTEGER, b VARCHAR, c DOUBLE, d BOOLEAN)"
+                )
+                result = conn.execute("SELECT COUNT(*) FROM test").fetchone()
+                assert result[0] == 0
+
+
+def test_import_cache():
+    with duckdb.connect(":memory:") as conn:
+        conn.execute("CREATE TABLE test AS SELECT range as x FROM range(10)")
+        result = conn.fetchdf()
+        assert len(result) > 0, "fetchdf failed"
+
+        result = conn.execute("SELECT range as x FROM range(5)").fetchnumpy()
+        assert len(result["x"]) == 5, "fetchnumpy failed"
+
+        conn.execute("DROP TABLE test")

tests/fast/threading/test_concurrent_access.py

@@ -0,0 +1,111 @@
+"""
+Concurrent access tests for DuckDB Python bindings with free threading support.
+
+These tests verify that the DuckDB Python module can handle concurrent access
+from multiple threads safely, testing module state isolation, memory management,
+and connection handling under various stress conditions.
+"""
+
+import gc
+import random
+import time
+import concurrent.futures
+
+import pytest
+
+import duckdb
+
+
+def test_concurrent_connections():
+    with duckdb.connect() as conn:
+        result = conn.execute("SELECT random() as id, random()*2 as doubled").fetchone()
+        assert result is not None
+
+
+@pytest.mark.parallel_threads(1)
+def test_shared_connection_stress(num_threads_testing):
+    """Test concurrent operations on shared connection using cursors."""
+    iterations = 10
+
+    with duckdb.connect(":memory:") as connection:
+        connection.execute(
+            "CREATE TABLE stress_test (id INTEGER, thread_id INTEGER, value TEXT)"
+        )
+
+        def worker_thread(thread_id: int) -> None:
+            cursor = connection.cursor()
+            for i in range(iterations):
+                cursor.execute(
+                    "INSERT INTO stress_test VALUES (?, ?, ?)",
+                    [i, thread_id, f"thread_{thread_id}_value_{i}"],
+                )
+                cursor.execute(
+                    "SELECT COUNT(*) FROM stress_test WHERE thread_id = ?", [thread_id]
+                ).fetchone()
+                time.sleep(random.uniform(0.0001, 0.001))
+
+        with concurrent.futures.ThreadPoolExecutor(
+            max_workers=num_threads_testing
+        ) as executor:
+            futures = [
+                executor.submit(worker_thread, i) for i in range(num_threads_testing)
+            ]
+            # Wait for all to complete, will raise if any fail
+            for future in concurrent.futures.as_completed(futures):
+                future.result()
+
+        total_rows = connection.execute("SELECT COUNT(*) FROM stress_test").fetchone()[
+            0
+        ]
+        expected_rows = num_threads_testing * iterations
+        assert total_rows == expected_rows
+
+
+@pytest.mark.parallel_threads(1)
+def test_module_state_isolation():
+    """Test that module state is properly accessible."""
+    with duckdb.connect(":memory:"):
+        assert hasattr(duckdb, "__version__")
+
+        with duckdb.connect() as default_conn:
+            result = default_conn.execute("SELECT 'default' as type").fetchone()
+            assert result[0] == "default"
+
+        int_type = duckdb.type("INTEGER")
+        string_type = duckdb.type("VARCHAR")
+        assert int_type is not None
+        assert string_type is not None
+
+
+def test_rapid_connect_disconnect():
+    connections_count = 10
+    """Test rapid connection creation and destruction."""
+    for i in range(connections_count):
+        conn = duckdb.connect(":memory:")
+        try:
+            result = conn.execute("SELECT 1").fetchone()[0]
+            assert result == 1
+        finally:
+            conn.close()
+
+        # Sometimes force GC to increase pressure
+        if i % 3 == 0:
+            gc.collect()
+
+
+def test_exception_handling():
+    """Test exception handling doesn't affect module state."""
+    conn = duckdb.connect(":memory:")
+    try:
+        conn.execute("CREATE TABLE test (x INTEGER)")
+        conn.execute("INSERT INTO test VALUES (1), (2), (3)")
+
+        for i in range(10):
+            if i % 3 == 0:
+                with pytest.raises(duckdb.CatalogException):
+                    conn.execute("SELECT * FROM nonexistent_table")
+            else:
+                result = conn.execute("SELECT COUNT(*) FROM test").fetchone()[0]
+                assert result == 3
+    finally:
+        conn.close()

tests/fast/threading/test_connection_lifecycle_races.py

@@ -0,0 +1,105 @@
+"""
+Test connection lifecycle races.
+
+Focused on DuckDBPyConnection constructor and Close
+"""
+
+import gc
+import concurrent.futures
+
+import pytest
+
+import duckdb
+
+
+def test_concurrent_connection_creation_destruction():
+    conn = duckdb.connect()
+    try:
+        result = conn.execute("SELECT 1").fetchone()
+        assert result[0] == 1
+    finally:
+        conn.close()
+
+
+def test_connection_destructor_race():
+    conn = duckdb.connect()
+    result = conn.execute("SELECT COUNT(*) FROM range(1)").fetchone()
+    assert result[0] == 1
+
+    del conn
+    gc.collect()
+
+
+@pytest.mark.parallel_threads(1)
+def test_concurrent_close_operations(num_threads_testing):
+    with duckdb.connect(":memory:") as conn:
+        conn.execute("CREATE TABLE shared_table (id INTEGER, data VARCHAR)")
+        conn.execute("INSERT INTO shared_table VALUES (1, 'test')")
+
+        def attempt_close_connection(cursor, thread_id):
+            _result = cursor.execute("SELECT COUNT(*) FROM shared_table").fetchone()
+
+            cursor.close()
+
+            return True
+
+        with concurrent.futures.ThreadPoolExecutor(
+            max_workers=num_threads_testing
+        ) as executor:
+            futures = [
+                executor.submit(attempt_close_connection, conn.cursor(), i)
+                for i in range(num_threads_testing)
+            ]
+            results = [
+                future.result() for future in concurrent.futures.as_completed(futures)
+            ]
+
+        assert all(results)
+
+
+@pytest.mark.parallel_threads(1)
+def test_cursor_operations_race(num_threads_testing):
+    conn = duckdb.connect(":memory:")
+    try:
+        conn.execute("CREATE TABLE cursor_test (id INTEGER, name VARCHAR)")
+        conn.execute(
+            "INSERT INTO cursor_test SELECT i, 'name_' || i FROM range(100) t(i)"
+        )
+
+        def cursor_operations(thread_id):
+            """Perform cursor operations concurrently."""
+            # Get a cursor
+            cursor = conn.cursor()
+            cursor.execute(
+                f"SELECT * FROM cursor_test WHERE id % {num_threads_testing} = {thread_id}"
+            )
+            results = cursor.fetchall()
+
+            return True
+
+        with concurrent.futures.ThreadPoolExecutor(
+            max_workers=num_threads_testing
+        ) as executor:
+            futures = [
+                executor.submit(cursor_operations, i)
+                for i in range(num_threads_testing)
+            ]
+            results = [
+                future.result() for future in concurrent.futures.as_completed(futures)
+            ]
+
+        assert all(results)
+    finally:
+        conn.close()
+
+
+def test_rapid_connection_cycling():
+    """Test rapid connection creation and destruction cycles."""
+    num_cycles = 5
+    for cycle in range(num_cycles):
+        conn = duckdb.connect(":memory:")
+        try:
+            result = conn.execute(f"SELECT 1 + {cycle}").fetchone()
+            assert result[0] == 1 + cycle
+        finally:
+            conn.close()