test: Add free-threading test suite and fix thread-safety issues

.github/workflows/pytest.yml

@@ -150,6 +150,8 @@ jobs:
         python-version: ["3.14t"]
         os: [ubuntu-latest]
     runs-on: ${{ matrix.os }}
+    # Backstop against deadlocks in multithreaded tests: fail instead of hanging.
+    timeout-minutes: 10
     env:
         PYTHON_GIL: 0
         UV_LOCKED: 0
@@ -164,5 +166,18 @@ jobs:
           enable-cache: "true"
           cache-suffix: python-314t-${{ matrix.python-version }}
           cache-dependency-glob: "pyproject.toml"
-      - name: Run pytest
-        run: make run-ci DEPS="--pre --group tests --group plugins --extra pandas --extra pyarrow" CMD="pytest tests --cov=src --cov=tests --cov-fail-under=50 --runslow --durations=30 --constructors=pandas,pandas[nullable],pandas[pyarrow],pyarrow"
+      # Run the suite again with every test executed by 4 threads at once
+      # with pytest-run-parallel to surface thread-safety issues.
+      # This catches global-state races across the whole suite; shared-object races
+      # are covered by the dedicated tests in tests/free_threading_test.py.
+      # Tests marked `thread_unsafe` run single-threaded.
+      - name: Run pytest (parallel threads)
+        env:
+          # NOTE: Clear PYTEST_ADDOPTS since xdist forks processes,
+          # and the point here is threads within one process
+          PYTEST_ADDOPTS: ""
+        run: |
+          make run-ci \
+          DEPS="--pre --group tests --group plugins --extra pandas --extra pyarrow" \
+          RUN_ONLY="--with pytest-run-parallel" \
+          CMD="pytest tests --parallel-threads=4 --constructors=pandas,pandas[nullable],pandas[pyarrow],pyarrow"

_typos.toml

@@ -5,6 +5,7 @@ ba = "ba"  # Used as column name in docstring examples (way too much?)
 iy = "iy"  # Used as column name (once in a test)
 pn = "pn"  # Used in docs: pn = PandasLikeNamespace(...)
 TYP = "TYP"  # Used in flake8 rule
+tpe = "tpe"  # Alias for ThreadPoolExecutor
 
 [files]
 extend-exclude = ["tests/data/*"]

pyproject.toml

@@ -415,7 +415,12 @@ filterwarnings = [
   "ignore:.*interchange protocol is deprecated.*:DeprecationWarning"
 ]
 xfail_strict = true
-markers = ["slow: marks tests as slow (deselect with '-m \"not slow\"')"]
+markers = [
+  "slow: marks tests as slow (deselect with '-m \"not slow\"')",
+  # Registered manually so the mark is valid even when pytest-run-parallel is not installed;
+  # the plugin (python-314t job) uses it to run marked tests single-threaded.
+  "thread_unsafe(reason): marks tests that must not run in multiple threads at once",
+]
 env = [
   "MODIN_ENGINE=python",
   "PYARROW_IGNORE_TIMEZONE=1",

src/narwhals/_duckdb/dataframe.py

@@ -16,6 +16,7 @@
     join_column_names,
     lit,
     native_to_narwhals_dtype,
+    temporary_view_name,
     window_expression,
 )
 from narwhals._sql.dataframe import SQLLazyFrame
@@ -388,13 +389,19 @@ def join_asof(
             else f'rhs."{name}"'
             for name in keep_cols
         )
-        query = f"""
+        # `rhs` is resolved via replacement scan.
+        # `rel.query` creates a view, so this fails on read-only databases (#3567);
+        # unavoidable until the relational API gains ASOF join support.
+        view = temporary_view_name()
+        joined = lhs.query(
+            view,
+            f"""
             SELECT {",".join(rhs_select)}
-            FROM lhs
+            FROM {view} AS lhs
             ASOF LEFT JOIN rhs
             ON {condition}
-            """  # noqa: S608
-        joined = duckdb.sql(query)
+            """,  # noqa: S608
+        )
 
         select = [col(name) for name in lhs.columns]
         select.extend(
@@ -457,25 +464,23 @@ def sort(self, *by: str, descending: bool | Sequence[bool], nulls_last: bool) ->
         return self._with_native(self.native.sort(*it))
 
     def top_k(self, k: int, *, by: Iterable[str], reverse: bool | Sequence[bool]) -> Self:
-        _rel = self.native
         by = list(by)
         if isinstance(reverse, bool):
             descending = extend_bool(not reverse, len(by))
         else:
             descending = tuple(not rev for rev in reverse)
+        tmp_name = generate_temporary_column_name(8, self.columns, prefix="row_number_")
         expr = window_expression(
             F("row_number"),
             order_by=by,
             descending=descending,
             nulls_last=extend_bool(True, len(by)),
         )
-        condition = expr <= lit(k)
-        query = f"""
-            SELECT *
-            FROM _rel
-            QUALIFY {condition}
-        """  # noqa: S608
-        return self._with_native(duckdb.sql(query))
+        return self._with_native(
+            self.native.select(StarExpression(), expr.alias(tmp_name)).filter(
+                col(tmp_name) <= lit(k)
+            )
+        ).drop([tmp_name], strict=False)
 
     def drop_nulls(self, subset: Sequence[str] | None) -> Self:
         subset_ = subset if subset is not None else self.columns
@@ -542,19 +547,21 @@ def unpivot(
             raise NotImplementedError(msg)
 
         unpivot_on = join_column_names(*on_)
-        _rel = self.native
+        # `rel.query` creates a view, so this fails on read-only databases (#3567).
         # Replace with Python API once
         # https://github.com/duckdb/duckdb/discussions/16980 is addressed.
-        query = f"""
-            unpivot _rel
+        view = temporary_view_name()
+        unpivoted = self.native.query(
+            view,
+            f"""
+            unpivot {view}
             on {unpivot_on}
             into
                 name {col(variable_name)}
                 value {col(value_name)}
-            """
-        return self._with_native(
-            duckdb.sql(query).select(*[*index_, variable_name, value_name])
+            """,
         )
+        return self._with_native(unpivoted.select(*index_, variable_name, value_name))
 
     @requires.backend_version((1, 3))
     def with_row_index(self, name: str, order_by: Sequence[str]) -> Self:

src/narwhals/_duckdb/namespace.py

@@ -20,6 +20,7 @@
     lit,
     narwhals_to_native_dtype,
     sql_expression,
+    temporary_view_name,
     when,
     window_expression,
 )
@@ -73,12 +74,18 @@ def scan_csv(
         self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
     ) -> DuckDBLazyFrame:
         validate_separators(separator, ("delimiter", "delim", "sep"), kwds)
-        native = duckdb.read_csv(source, delimiter=separator, **kwds)
-        return self._lazyframe.from_native(native, context=self)
+        # Without an explicit `connection`, DuckDB reads through the
+        # process-global default connection.
+        reader = kwds.pop("connection", None) or duckdb
+        native_frame = reader.read_csv(source, delimiter=separator, **kwds)
+        return self._lazyframe.from_native(native_frame, context=self)
 
     def scan_parquet(self, source: NormalizedPath, **kwds: Any) -> DuckDBLazyFrame:
-        native = duckdb.read_parquet(source, **kwds)
-        return self._lazyframe.from_native(native, context=self)
+        # Without an explicit `connection`, DuckDB reads through the
+        # process-global default connection.
+        reader = kwds.pop("connection", None) or duckdb
+        native_frame = reader.read_parquet(source, **kwds)
+        return self._lazyframe.from_native(native_frame, context=self)
 
     def _function(self, name: str, *args: Expression) -> Expression:  # type: ignore[override]
         return function(name, *args)
@@ -113,9 +120,13 @@ def concat(
             res = first.native
             for _item in native_items[1:]:
                 # TODO(unassigned): use relational API when available https://github.com/duckdb/duckdb/discussions/16996
-                res = duckdb.sql("""
-                    from res select * union all by name from _item select *
-                """)
+                # `_item` is resolved via replacement scan.
+                # `rel.query` creates a view, so this fails on read-only databases (#3567).
+                view = temporary_view_name()
+                res = res.query(
+                    view,
+                    f"from {view} select * union all by name from _item select *",  # noqa: S608
+                )
             return first._with_native(res)
         res = reduce(lambda x, y: x.union(y), native_items)
         return first._with_native(res)

src/narwhals/_duckdb/utils.py

@@ -7,7 +7,13 @@
 import duckdb
 from duckdb import Expression
 
-from narwhals._utils import Implementation, Version, extend_bool, isinstance_or_issubclass
+from narwhals._utils import (
+    Implementation,
+    Version,
+    extend_bool,
+    generate_temporary_column_name,
+    isinstance_or_issubclass,
+)
 from narwhals.exceptions import ColumnNotFoundError
 
 if TYPE_CHECKING:
@@ -337,6 +343,26 @@ def join_column_names(*names: str) -> str:
     return ", ".join(str(col(name)) for name in names)
 
 
+def temporary_view_name() -> str:
+    """Unique name for `DuckDBPyRelation.query`'s `virtual_table_name`.
+
+    `rel.query(view, sql)` registers `rel` as a view named `view` and runs `sql` on
+    `rel`'s own connection (see [relational api]). We prefer it to `duckdb.sql(statement)`,
+    which uses the global default connection and cannot see relations from other
+    connections (see [using connections in parallel pythonprograms]).
+
+    The name must be unique per call: views persist and the lazy result re-binds by
+    name on execution, so a reused name shadows earlier views and corrupts their plans.
+    `rel.query` must also run in the frame whose locals the SQL references, as
+    [replacement scans] only see the caller's frame [3].
+
+    [relational api]: https://duckdb.org/docs/current/clients/python/relational_api
+    [using connections in parallel pythonprograms]: https://duckdb.org/docs/current/clients/python/overview#using-connections-in-parallel-python-programs
+    [replacement scans]: https://duckdb.org/docs/current/clients/c/replacement_scans
+    """
+    return generate_temporary_column_name(8, [], prefix="_narwhals_")
+
+
 def generate_order_by_sql(
     *order_by: str | Expression, descending: Sequence[bool], nulls_last: Sequence[bool]
 ) -> str:

src/narwhals/functions.py

@@ -692,6 +692,23 @@ def scan_csv(
     For the libraries that do not support lazy dataframes, the function reads
     a csv file eagerly and then converts the resulting dataframe to a lazyframe.
 
+    Note:
+        Spark-like backends require a `session` object to be passed in `kwargs`.
+
+        For instance:
+
+        ```py
+        import narwhals as nw
+        from sqlframe.duckdb import DuckDBSession
+
+        nw.scan_csv(source, backend="sqlframe", session=DuckDBSession())
+        ```
+
+    Note:
+        For the DuckDB backend, a `connection` object can be passed in `kwargs`
+        to read through a specific `DuckDBPyConnection` instead of the
+        process-global default connection.
+
     Arguments:
         source: Path to a file.
         backend: The eager backend for DataFrame creation.
@@ -806,7 +823,7 @@ def scan_parquet(
     a parquet file eagerly and then converts the resulting dataframe to a lazyframe.
 
     Note:
-        Spark like backends require a session object to be passed in `kwargs`.
+        Spark like backends require a `session` object to be passed in `kwargs`.
 
         For instance:
 
@@ -817,6 +834,11 @@ def scan_parquet(
         nw.scan_parquet(source, backend="sqlframe", session=DuckDBSession())
         ```
 
+    Note:
+        For the DuckDB backend, a `connection` object can be passed in `kwargs`
+        to read through a specific `DuckDBPyConnection` instead of the
+        process-global default connection.
+
     Arguments:
         source: Path to a file.
         backend: The eager backend for DataFrame creation.

src/narwhals/sql.py

@@ -22,9 +22,15 @@
     raise ModuleNotFoundError(msg) from _exc
 
 CONN = duckdb.connect()
-TZ = DeferredTimeZone(
-    CONN.sql("select value from duckdb_settings() where name = 'TimeZone'")
-)
+
+
+def _cursor() -> duckdb.DuckDBPyConnection:
+    # DuckDB connections must not be used concurrently from multiple threads.
+    # Each cursor shares `CONN`'s catalog but executes independently.
+    # Citing from [Multiple Python Threads](https://duckdb.org/docs/current/guides/python/multiple_threads)
+    # > Each thread must use the `.cursor()` method to create a thread-local
+    # > connection to the same DuckDB file based on the original connection
+    return CONN.cursor()
 
 
 class SQLTable(LazyFrame[duckdb.DuckDBPyRelation]):
@@ -90,16 +96,20 @@ def table(name: str, schema: IntoSchema) -> SQLTable:
         |           0 rows           |
         └────────────────────────────┘
     """
+    cursor = _cursor()
+    tz = DeferredTimeZone(
+        cursor.sql("select value from duckdb_settings() where name = 'TimeZone'")
+    )
     column_mapping = {
-        col: narwhals_to_native_dtype(dtype, Version.MAIN, TZ)
+        col: narwhals_to_native_dtype(dtype, Version.MAIN, tz)
         for col, dtype in Schema(schema).items()
     }
     dtypes = ", ".join(f'"{col}" {dtype}' for col, dtype in column_mapping.items())
-    CONN.sql(f"""
+    cursor.sql(f"""
         CREATE TABLE "{name}"
         ({dtypes});
         """)
-    lf = from_native(CONN.table(name))
+    lf = from_native(cursor.table(name))
     return SQLTable(lf._compliant_frame, level=lf._level)
 
 

src/narwhals/stable/v2/__init__.py

@@ -1079,6 +1079,23 @@ def scan_csv(
     For the libraries that do not support lazy dataframes, the function reads
     a csv file eagerly and then converts the resulting dataframe to a lazyframe.
 
+    Note:
+        Spark-like backends require a `session` object to be passed in `kwargs`.
+
+        For instance:
+
+        ```py
+        import narwhals as nw
+        from sqlframe.duckdb import DuckDBSession
+
+        nw.scan_csv(source, backend="sqlframe", session=DuckDBSession())
+        ```
+
+    Note:
+        For the DuckDB backend, a `connection` object can be passed in `kwargs`
+        to read through a specific `DuckDBPyConnection` instead of the
+        process-global default connection.
+
     Arguments:
         source: Path to a file.
         backend: The eager backend for DataFrame creation.
@@ -1128,7 +1145,7 @@ def scan_parquet(
     a parquet file eagerly and then converts the resulting dataframe to a lazyframe.
 
     Note:
-        Spark like backends require a session object to be passed in `kwargs`.
+        Spark like backends require a `session` object to be passed in `kwargs`.
 
         For instance:
 
@@ -1139,6 +1156,11 @@ def scan_parquet(
         nw.scan_parquet(source, backend="sqlframe", session=DuckDBSession())
         ```
 
+    Note:
+        For the DuckDB backend, a `connection` object can be passed in `kwargs`
+        to read through a specific `DuckDBPyConnection` instead of the
+        process-global default connection.
+
     Arguments:
         source: Path to a file.
         backend: The eager backend for DataFrame creation.