diff --git a/duckdb/_dbapi_type_object.py b/duckdb/_dbapi_type_object.py index ed73760d..4a58bbb8 100644 --- a/duckdb/_dbapi_type_object.py +++ b/duckdb/_dbapi_type_object.py @@ -25,7 +25,9 @@ - DuckDB Type System: https://duckdb.org/docs/sql/data_types/overview """ -from duckdb import sqltypes +import collections.abc + +from duckdb.sqltypes import DuckDBPyType class DBAPITypeObject: @@ -35,36 +37,40 @@ class DBAPITypeObject: It allows checking whether a specific DuckDB type belongs to a broader category like STRING, NUMBER, DATETIME, etc. - The type object supports equality comparison with DuckDBPyType instances, - returning True if the type belongs to this category. + A category is a set of type ids, as DuckDBPyType.id gives them. A DuckDBPyType + belongs to the category when its id is in the set, whatever its parameters + (such as the width and scale of a DECIMAL) or its alias (such as JSON, a VARCHAR). Args: - types: A list of DuckDBPyType instances that belong to this type category. + type_ids: Lowercase type ids of the types that belong to this category. Example: - >>> string_types = DBAPITypeObject([sqltypes.VARCHAR, sqltypes.CHAR]) - >>> result = sqltypes.VARCHAR == string_types # True - >>> result = sqltypes.INTEGER == string_types # False + >>> string_types = DBAPITypeObject({"varchar"}) + >>> result = duckdb.sqltypes.VARCHAR == string_types # True + >>> result = duckdb.sqltypes.INTEGER == string_types # False Note: This follows the DB API 2.0 specification where type objects are compared using equality operators rather than isinstance() checks. """ - def __init__(self, types: list[sqltypes.DuckDBPyType]) -> None: + def __init__(self, type_ids: collections.abc.Iterable[str]) -> None: """Initialize a DB API type object. Args: - types: List of DuckDB types that belong to this category. + type_ids: Lowercase type ids, as DuckDBPyType.id gives them, of the types that belong to this category. """ - self.types = types + if isinstance(type_ids, str): + msg = "type_ids must be a collection of type ids, not a single str" + raise TypeError(msg) + self.type_ids = frozenset(type_ids) def __eq__(self, other: object) -> bool: """Check if a DuckDB type belongs to this type category. This method implements the DB API 2.0 type checking mechanism. - It returns True if the other object is a DuckDBPyType that - is contained in this type category. + It returns True if the other object is a DuckDBPyType whose id + is in this type category. Args: other: The object to compare, typically a DuckDBPyType instance. @@ -73,34 +79,32 @@ def __eq__(self, other: object) -> bool: True if other is a DuckDBPyType in this category, False otherwise. Example: - >>> NUMBER == sqltypes.INTEGER # True - >>> NUMBER == sqltypes.VARCHAR # False + >>> NUMBER == duckdb.sqltypes.INTEGER # True + >>> NUMBER == duckdb.sqltypes.VARCHAR # False """ - if isinstance(other, sqltypes.DuckDBPyType): - return other in self.types - return False + return isinstance(other, DuckDBPyType) and other.id in self.type_ids def __repr__(self) -> str: """Return a string representation of this type object. Returns: - A string showing the type object and its contained DuckDB types. + A string listing the type ids of this category, uppercase and sorted. Example: >>> repr(STRING) '' """ - return f"" + return f"" # Define the standard DB API 2.0 type objects for DuckDB -STRING = DBAPITypeObject([sqltypes.VARCHAR]) +STRING = DBAPITypeObject({"varchar"}) """ STRING type object for text-based database columns. This type object represents all string/text types in DuckDB. Currently includes: -- VARCHAR: Variable-length character strings +- VARCHAR: Variable-length character strings, and the types that alias it, such as JSON Use this to check if a column contains textual data that should be handled as Python strings. @@ -113,22 +117,22 @@ def __repr__(self) -> str: """ NUMBER = DBAPITypeObject( - [ - sqltypes.TINYINT, - sqltypes.UTINYINT, - sqltypes.SMALLINT, - sqltypes.USMALLINT, - sqltypes.INTEGER, - sqltypes.UINTEGER, - sqltypes.BIGINT, - sqltypes.UBIGINT, - sqltypes.HUGEINT, - sqltypes.UHUGEINT, - sqltypes.DuckDBPyType("BIGNUM"), - sqltypes.DuckDBPyType("DECIMAL"), - sqltypes.FLOAT, - sqltypes.DOUBLE, - ] + { + "tinyint", + "utinyint", + "smallint", + "usmallint", + "integer", + "uinteger", + "bigint", + "ubigint", + "hugeint", + "uhugeint", + "bignum", + "decimal", + "float", + "double", + } ) """ NUMBER type object for numeric database columns. @@ -144,7 +148,7 @@ def __repr__(self) -> str: Decimal Types: - BIGNUM: Arbitrary precision integers -- DECIMAL: Fixed-point decimal numbers +- DECIMAL: Fixed-point decimal numbers, of every width and scale Floating Point Types: - FLOAT: 32-bit floating point @@ -161,16 +165,16 @@ def __repr__(self) -> str: """ DATETIME = DBAPITypeObject( - [ - sqltypes.DATE, - sqltypes.TIME, - sqltypes.TIME_TZ, - sqltypes.TIMESTAMP, - sqltypes.TIMESTAMP_TZ, - sqltypes.TIMESTAMP_NS, - sqltypes.TIMESTAMP_MS, - sqltypes.TIMESTAMP_S, - ] + { + "date", + "time", + "time with time zone", + "timestamp", + "timestamp with time zone", + "timestamp_ns", + "timestamp_ms", + "timestamp_s", + } ) """ DATETIME type object for date and time database columns. @@ -201,7 +205,7 @@ def __repr__(self) -> str: >>> cursor.description[2][1] == DATETIME # Check if third column is date/time """ -BINARY = DBAPITypeObject([sqltypes.BLOB]) +BINARY = DBAPITypeObject({"blob"}) """ BINARY type object for binary data database columns. diff --git a/src/include/duckdb_python/pyconnection/pyconnection.hpp b/src/include/duckdb_python/pyconnection/pyconnection.hpp index cd1c3575..0c38bffa 100644 --- a/src/include/duckdb_python/pyconnection/pyconnection.hpp +++ b/src/include/duckdb_python/pyconnection/pyconnection.hpp @@ -49,9 +49,13 @@ struct DefaultConnectionHolder { public: std::shared_ptr Get(); void Set(std::shared_ptr conn); + //! Once the module's cleanup ran and the interpreter is finalizing, the holder neither opens nor keeps a + //! connection, because nothing would close it + void Release(); private: std::shared_ptr connection; + bool released = false; mutex l; }; diff --git a/src/pyconnection.cpp b/src/pyconnection.cpp index 3543d719..d861f8c2 100644 --- a/src/pyconnection.cpp +++ b/src/pyconnection.cpp @@ -2155,18 +2155,77 @@ void DuckDBPyConnection::LoadExtension(const string &extension) { ExtensionHelper::LoadExternalExtension(*connection.context, extension_opts); } +// sys.is_finalizing is asked because Py_IsFinalizing is public only from Python 3.13; without sys the interpreter is +// finalizing. +static bool InterpreterIsFinalizing() { + PyObject *is_finalizing = PySys_GetObject("is_finalizing"); + if (!is_finalizing) { + return true; + } + PyObject *result = PyObject_CallNoArgs(is_finalizing); + if (!result) { + PyErr_Clear(); + return true; + } + const int truth = PyObject_IsTrue(result); + Py_DECREF(result); + if (truth < 0) { + PyErr_Clear(); + return true; + } + return truth != 0; +} + +// The mutex guards only reads and swaps of the pointer: opening or destroying a connection releases the GIL and can run +// Python code that fetches the default connection again, so both happen after the mutex is released. std::shared_ptr DefaultConnectionHolder::Get() { - lock_guard guard(l); - if (!connection || connection->con.ConnectionIsClosed()) { - nb::dict config_dict; - connection = DuckDBPyConnection::Connect(nb::str(":memory:"), false, config_dict); + bool was_released; + { + lock_guard guard(l); + if (connection && !connection->con.ConnectionIsClosed()) { + return connection; + } + was_released = released; + } + if (was_released && InterpreterIsFinalizing()) { + throw ConnectionException("the default connection is not available during interpreter shutdown"); + } + nb::dict config_dict; + auto opened = DuckDBPyConnection::Connect(nb::str(":memory:"), false, config_dict); + std::shared_ptr replaced; + { + lock_guard guard(l); + if (connection && !connection->con.ConnectionIsClosed()) { + return connection; + } + replaced = std::move(connection); + connection = opened; } - return connection; + return opened; } void DefaultConnectionHolder::Set(std::shared_ptr conn) { - lock_guard guard(l); - connection = conn; + bool was_released; + { + lock_guard guard(l); + was_released = released; + } + if (was_released && InterpreterIsFinalizing()) { + return; + } + { + lock_guard guard(l); + connection.swap(conn); + } +} + +void DefaultConnectionHolder::Release() { + std::shared_ptr released_connection; + { + lock_guard guard(l); + released = true; + connection.swap(released_connection); + } } void DuckDBPyConnection::Cursors::AddCursor(std::shared_ptr conn) { @@ -2519,7 +2578,7 @@ void DuckDBPyConnection::Exit(DuckDBPyConnection &self, const nb::object &exc_ty } void DuckDBPyConnection::Cleanup() { - GetModuleState().default_connection.Set(nullptr); + GetModuleState().default_connection.Release(); GetModuleState().import_cache.reset(); GetModuleState().consumed_arrow_readers.reset(); } diff --git a/src/typing/pytype.cpp b/src/typing/pytype.cpp index 36561006..e385ebc8 100644 --- a/src/typing/pytype.cpp +++ b/src/typing/pytype.cpp @@ -364,6 +364,19 @@ void DuckDBPyType::Initialize(nb::handle &m) { nb::is_operator()); type_module.def("__eq__", &DuckDBPyType::EqualsString, "Compare two types for equality", nb::arg("other"), nb::is_operator()); + // Binds every other operand in nanobind's strict first pass, because its implicit conversion pass prints to + // stderr each time the conversion to DuckDBPyType fails; registered after the str overload, so a plain str + // operand is compared by its text and never parsed. + type_module.def( + "__eq__", + [](const DuckDBPyType &self, const nb::object &other) -> nb::object { + std::unique_ptr converted; + if (!DuckDBPyType::TryConvert(other, converted)) { + return nb::borrow(nb::handle(Py_NotImplemented)); + } + return nb::bool_(self.Equals(*converted)); + }, + "Compare two types for equality", nb::arg("other").none(), nb::is_operator()); type_module.def("__hash__", [](const DuckDBPyType &type) { auto s = type.ToString(); return nb::hash(nb::str(s.c_str(), s.size())); diff --git a/tests/fast/api/test_dbapi_type_object.py b/tests/fast/api/test_dbapi_type_object.py new file mode 100644 index 00000000..4331b4fb --- /dev/null +++ b/tests/fast/api/test_dbapi_type_object.py @@ -0,0 +1,506 @@ +import subprocess +import sys +import textwrap + +import pytest + +import duckdb +from duckdb import sqltypes + +THREAD_COUNTS_SCRIPT = """ + import os + + def thread_count(): + return len(os.listdir("/proc/self/task")) + + before_import = thread_count() + import duckdb + from duckdb import sqltypes + + sqltypes.BIGINT == "int8" + after_import = thread_count() + duckdb.execute("SELECT 42").fetchall() + print(before_import, after_import, thread_count()) +""" + + +def run_bounded(script, *args): + # A deadlock can hold the GIL, so only faulthandler's own thread can still report where every thread is stuck. + watchdog = "import faulthandler\nfaulthandler.dump_traceback_later(45, exit=True)\n" + try: + proc = subprocess.run( + [sys.executable, "-c", watchdog + textwrap.dedent(script), *args], + capture_output=True, + text=True, + timeout=60, + check=False, + ) + except subprocess.TimeoutExpired as hang: + stderr = hang.stderr.decode() if isinstance(hang.stderr, bytes) else hang.stderr or "" + msg = f"the child process hung for 60 seconds and was killed\n{stderr}" + raise AssertionError(msg) from hang + assert proc.returncode == 0, proc.stdout + proc.stderr + # An editable install prints its rebuild progress to stdout before the script runs. + return proc.stdout.splitlines()[-1] + + +linux_only = pytest.mark.skipif(not sys.platform.startswith("linux"), reason="counts threads through /proc/self/task") + + +@pytest.fixture(scope="module") +def thread_counts(): + before_import, after_import, after_first_use = map(int, run_bounded(THREAD_COUNTS_SCRIPT).split()) + return before_import, after_import, after_first_use + + +# Only the OS task list sees the engine's thread pools, and only while a database with threads stays open, so a +# database opened with one thread and no async threads, or opened and closed again, goes unnoticed. +@linux_only +class TestImportOpensNoDatabase: + def test_import_starts_no_threads(self, thread_counts): + before_import, after_import, _ = thread_counts + assert after_import == before_import + + def test_first_use_opens_the_default_database(self, thread_counts): + _, after_import, after_first_use = thread_counts + assert after_first_use > after_import + + +class TestTypeObjectMatching: + @pytest.mark.parametrize( + ("type_object", "sql_type", "expected"), + [ + ("NUMBER", "BIGNUM", True), + ("NUMBER", "DECIMAL(4,1)", True), + ("NUMBER", "DECIMAL(10,2)", True), + ("NUMBER", "DECIMAL(18,3)", True), + ("NUMBER", "DECIMAL(38,0)", True), + ("NUMBER", "HUGEINT", True), + ("NUMBER", "FLOAT", True), + ("NUMBER", "VARCHAR", False), + ("STRING", "VARCHAR", True), + ("STRING", "JSON", True), + ("STRING", "INTEGER", False), + ("DATETIME", "TIMESTAMP", True), + ("DATETIME", "VARCHAR", False), + ("BINARY", "BLOB", True), + ("BINARY", "VARCHAR", False), + ("ROWID", "BIGINT", False), + ], + ) + def test_description_type_matches(self, duckdb_cursor, type_object, sql_type, expected): + duckdb_cursor.execute(f"SELECT NULL::{sql_type} AS c") + column_type = duckdb_cursor.description[0][1] + assert str(column_type) == sql_type + assert (column_type == getattr(duckdb, type_object)) is expected + assert (getattr(duckdb, type_object) == column_type) is expected + + @pytest.mark.parametrize( + ("sql_type", "category"), + [ + *[ + (sql_type, "NUMBER") + for sql_type in ( + "TINYINT", + "UTINYINT", + "SMALLINT", + "USMALLINT", + "INTEGER", + "UINTEGER", + "BIGINT", + "UBIGINT", + "HUGEINT", + "UHUGEINT", + "BIGNUM", + "DECIMAL(4,1)", + "FLOAT", + "DOUBLE", + ) + ], + *[ + (sql_type, "DATETIME") + for sql_type in ( + "DATE", + "TIME", + "TIMETZ", + "TIMESTAMP", + "TIMESTAMPTZ", + "TIMESTAMP_NS", + "TIMESTAMP_MS", + "TIMESTAMP_S", + ) + ], + ("VARCHAR", "STRING"), + ("JSON", "STRING"), + ("BLOB", "BINARY"), + *[ + (sql_type, None) + for sql_type in ( + "BOOLEAN", + "UUID", + "INTERVAL", + "TIME_NS", + "BIT", + "VARIANT", + "INTEGER[]", + "STRUCT(a INTEGER)", + "ENUM('a')", + ) + ], + ], + ) + def test_each_type_belongs_to_exactly_its_category(self, duckdb_cursor, sql_type, category): + duckdb_cursor.execute(f"SELECT NULL::{sql_type} AS c") + column_type = duckdb_cursor.description[0][1] + matches = {name for name in ("STRING", "NUMBER", "DATETIME", "BINARY") if column_type == getattr(duckdb, name)} + assert matches == ({category} if category else set()) + assert column_type != duckdb.ROWID + + def test_an_aggregate_state_matches_by_the_id_of_its_value(self, duckdb_cursor): + duckdb_cursor.execute("SELECT count(*) EXPORT_STATE AS c FROM range(3)") + column_type = duckdb_cursor.description[0][1] + assert str(column_type) == "AGGREGATE_STATE" + assert column_type == duckdb.NUMBER + + def test_the_categories_hold_no_type_instances(self): + from duckdb import _dbapi_type_object + + categories = [duckdb.STRING, duckdb.NUMBER, duckdb.DATETIME, duckdb.BINARY] + held = [*vars(_dbapi_type_object).values(), *(value for c in categories for value in vars(c).values())] + assert not any(isinstance(value, sqltypes.DuckDBPyType) for value in held) + assert all(isinstance(type_id, str) for c in categories for type_id in c.type_ids) + + def test_type_ids_are_copied(self, duckdb_cursor): + type_ids = {"decimal"} + category = duckdb.DBAPITypeObject(type_ids) + type_ids.add("varchar") + duckdb_cursor.execute("SELECT NULL::DECIMAL(10,2) AS c") + assert category != sqltypes.VARCHAR + assert category == duckdb_cursor.description[0][1] + + def test_type_ids_cannot_be_a_bare_string(self): + with pytest.raises(TypeError): + duckdb.DBAPITypeObject("decimal") + + def test_number_repr(self): + assert repr(duckdb.NUMBER) == ( + "" + ) + + +class TestTypeComparisonIsQuiet: + def test_type_object_comparison_writes_nothing_to_stderr(self, duckdb_cursor, capfd): + duckdb_cursor.execute("SELECT 'a'::VARCHAR AS c") + column_type = duckdb_cursor.description[0][1] + capfd.readouterr() + assert column_type == duckdb.STRING + assert duckdb.STRING == column_type # noqa: SIM300 + assert capfd.readouterr().err == "" + + @pytest.mark.parametrize( + ("left", "right", "expected"), + [ + (sqltypes.BIGINT, "bigint", True), + (sqltypes.BIGINT, "INT8", False), + (sqltypes.BIGINT, "int8", False), + (sqltypes.BIGINT, int, True), + (sqltypes.BIGINT, int | None, True), + (sqltypes.VARCHAR, str, True), + (sqltypes.BIGINT, str, False), + (sqltypes.DuckDBPyType(list[int]), list[int], True), + (sqltypes.BIGINT, 42, False), + (sqltypes.BIGINT, object(), False), + (sqltypes.BIGINT, None, False), + ], + ids=[ + "BIGINT-bigint_string", + "BIGINT-INT8_string", + "BIGINT-int8_string", + "BIGINT-int", + "BIGINT-int_or_None", + "VARCHAR-str", + "BIGINT-str", + "BIGINT_list-list_of_int", + "BIGINT-42", + "BIGINT-object", + "BIGINT-None", + ], + ) + def test_converted_operand_comparison(self, capfd, left, right, expected): + capfd.readouterr() + assert (left == right) is expected + assert (left != right) is not expected + assert capfd.readouterr().err == "" + + def test_numpy_type_operand_comparison(self, capfd): + np = pytest.importorskip("numpy") + left = sqltypes.BIGINT + capfd.readouterr() + assert (left == np.int64) is True + assert (left != np.int64) is False + assert capfd.readouterr().err == "" + + +class TestTypeEquality: + def test_typed_equality_is_exact_and_text_equality_ignores_case(self, duckdb_cursor): + lower = sqltypes.DuckDBPyType("ENUM('a')", connection=duckdb_cursor) + upper = sqltypes.DuckDBPyType("ENUM('A')", connection=duckdb_cursor) + assert lower != upper + assert upper != lower + assert lower == str(upper) + assert str(upper) == lower + assert lower not in [upper] + + +TWO_THREADS_FIRST_USE_SCRIPT = """ + import sys + import threading + import duckdb + + if "close_first" in sys.argv: + duckdb.close() + barrier = threading.Barrier(2) + errors = [] + + def worker(): + barrier.wait() + try: + assert duckdb.cursor().execute("SELECT 1").fetchall() == [(1,)] + except BaseException as error: + errors.append(repr(error)) + + threads = [threading.Thread(target=worker, daemon=True) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert not errors, errors + print("SUCCESS") +""" + +EIGHT_THREADS_FIRST_USE_SCRIPT = """ + import os + import sys + import threading + import time + import duckdb + + count_threads = "count_threads" in sys.argv + + def thread_count(): + return len(os.listdir("/proc/self/task")) + + def settled_thread_count(): + # A joined or closed thread can stay listed for a moment while the OS reaps it. + deadline = time.monotonic() + 10 + last, since = thread_count(), time.monotonic() + while time.monotonic() < deadline: + time.sleep(0.01) + now = thread_count() + if now != last: + last, since = now, time.monotonic() + elif time.monotonic() - since >= 0.2: + break + return last + + if count_threads: + before_probe = thread_count() + probe = duckdb.connect() + one_database = thread_count() - before_probe + probe.close() + del probe + idle = settled_thread_count() + + count = 8 + barrier = threading.Barrier(count) + connections = [None] * count + results = [None] * count + errors = [] + + def worker(i): + barrier.wait() + try: + connections[i] = duckdb.default_connection() + results[i] = connections[i].cursor().execute("SELECT 1").fetchall() + except BaseException as error: + errors.append(repr(error)) + + threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(count)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert not errors, errors + assert results == [[(1,)]] * count, results + default = duckdb.default_connection() + assert all(connection is default for connection in connections) + if count_threads: + remaining = settled_thread_count() + assert abs(remaining - (idle + one_database)) <= 4, (remaining, idle, one_database) + print("SUCCESS") +""" + +REPLACE_WHILE_FETCHING_SCRIPT = """ + import threading + import duckdb + + errors = [] + for _ in range(20): + duckdb.execute("SELECT 1").fetchall() + replacement = duckdb.connect() + stop = threading.Event() + fetching = threading.Event() + + def fetch_default(stop=stop, fetching=fetching): + try: + while not stop.is_set(): + duckdb.default_connection() + fetching.set() + except BaseException as error: + errors.append(repr(error)) + fetching.set() + + fetcher = threading.Thread(target=fetch_default, daemon=True) + fetcher.start() + fetching.wait() + duckdb.set_default_connection(replacement) + del replacement + stop.set() + fetcher.join() + assert not errors, errors + print("SUCCESS") +""" + +EXIT_WHILE_A_DAEMON_FETCHES_SCRIPT = """ + import threading + import duckdb + + duckdb.set_default_connection(duckdb.connect(config={"threads": 1, "async_threads": 0})) + looping = threading.Event() + + def fetch_forever(): + while True: + duckdb.default_connection() + looping.set() + + threading.Thread(target=fetch_forever, daemon=True).start() + looping.wait() + print("SUCCESS") +""" + +FINALIZER_FETCHES_THE_DEFAULT_SCRIPT = """ + import duckdb + from duckdb import sqltypes + + fetched = [] + + class Identity: + def __call__(self, x): + return x + + def __del__(self): + fetched.append(duckdb.default_connection()) + + con = duckdb.connect() + con.create_function("identity", Identity(), [sqltypes.BIGINT], sqltypes.BIGINT) + duckdb.set_default_connection(con) + del con + replacement = duckdb.connect() + duckdb.set_default_connection(replacement) + assert len(fetched) == 1, fetched + assert fetched[0] is replacement + print("SUCCESS") +""" + +SHUTDOWN_FINALIZER_SCRIPT = """ + import os + import time + import duckdb + from duckdb import sqltypes + + def thread_count(listdir=os.listdir): + return len(listdir("/proc/self/task")) + + def fetch(default_connection=duckdb.default_connection, refused=duckdb.ConnectionException): + try: + default_connection() + except refused: + return "refused" + return "opened" + + class Identity: + def __call__(self, x): + return x + + # Module globals may already be gone when this runs at exit, so everything it needs is bound here. + def __del__( + self, + write=os.write, + fetch=fetch, + thread_count=thread_count, + connect=duckdb.connect, + set_default_connection=duckdb.set_default_connection, + sleep=time.sleep, + monotonic=time.monotonic, + ): + before = thread_count() + first = fetch() + set_default_connection(connect()) + second = fetch() + # A joined thread can stay listed in /proc for a moment, so the count gets two seconds to settle. + after = thread_count() + deadline = monotonic() + 2.0 + while after > before and monotonic() < deadline: + sleep(0.001) + after = thread_count() + write(1, f"{first} {second} {before} {after}\\n".encode()) + + duckdb.default_connection().create_function("identity", Identity(), [sqltypes.BIGINT], sqltypes.BIGINT) + print("STARTED") +""" + +RECREATED_AFTER_RELEASE_SCRIPT = """ + import sys + import duckdb + + before = duckdb.default_connection() + del duckdb._clean_default_connection + del sys.modules["_duckdb"]._clean_default_connection + after = duckdb.default_connection() + assert after is not before + assert duckdb.sql("SELECT 42").fetchall() == [(42,)] + print("SUCCESS") +""" + + +class TestDefaultConnectionRace: + @pytest.mark.parametrize("args", [(), ("close_first",)], ids=["first_use", "first_use_after_close"]) + def test_two_threads_open_the_default_connection(self, args): + assert run_bounded(TWO_THREADS_FIRST_USE_SCRIPT, *args) == "SUCCESS" + + @pytest.mark.parametrize( + "args", + [(), pytest.param(("count_threads",), marks=linux_only)], + ids=["shared_connection", "duplicates_torn_down"], + ) + def test_eight_threads_share_one_default_connection(self, args): + assert run_bounded(EIGHT_THREADS_FIRST_USE_SCRIPT, *args) == "SUCCESS" + + def test_replacing_the_default_connection_while_another_thread_fetches_it(self): + assert run_bounded(REPLACE_WHILE_FETCHING_SCRIPT) == "SUCCESS" + + def test_exit_while_a_daemon_thread_fetches_the_default_connection(self): + assert run_bounded(EXIT_WHILE_A_DAEMON_FETCHES_SCRIPT) == "SUCCESS" + + def test_replacing_the_default_connection_runs_a_finalizer_that_fetches_it(self): + assert run_bounded(FINALIZER_FETCHES_THE_DEFAULT_SCRIPT) == "SUCCESS" + + def test_the_default_connection_is_recreated_after_its_cleanup_ran(self): + assert run_bounded(RECREATED_AFTER_RELEASE_SCRIPT) == "SUCCESS" + + @linux_only + def test_a_finalizer_at_exit_cannot_reopen_the_default_connection(self): + first, second, before, after = run_bounded(SHUTDOWN_FINALIZER_SCRIPT).split() + assert (first, second) == ("refused", "refused") + assert int(after) <= int(before)