diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 3c32c974a..5b4bdbdb6 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -709,6 +709,9 @@ class LanceDBConnection(DBConnection): self._namespace_client_properties = namespace_client_properties if _inner is not None: self._conn = _inner + # Native-derived wrappers resolve this in their async reconstruction + # path so construction never synchronously re-enters LOOP. + self._read_consistency_interval = read_consistency_interval self._cached_namespace_client = None return @@ -758,11 +761,14 @@ class LanceDBConnection(DBConnection): # storage_options. Also, this class really shouldn't be holding any state # beyond _conn. self._conn = AsyncConnection(LOOP.run(do_connect())) + # Keep property access synchronous so debugger introspection cannot wait on + # the background loop while that thread is suspended at a breakpoint. + self._read_consistency_interval = read_consistency_interval self._cached_namespace_client: Optional[LanceNamespace] = None @property def read_consistency_interval(self) -> Optional[timedelta]: - return LOOP.run(self._conn.get_read_consistency_interval()) + return self._read_consistency_interval @property def session(self) -> Optional[Session]: @@ -773,8 +779,16 @@ class LanceDBConnection(DBConnection): return self._conn.uri @classmethod - def from_inner(cls, inner: LanceDbConnection): - return cls(None, _inner=inner) + def from_inner( + cls, + inner: LanceDbConnection, + read_consistency_interval: Optional[timedelta], + ): + return cls( + None, + read_consistency_interval=read_consistency_interval, + _inner=inner, + ) def __repr__(self) -> str: return f"{self.__class__.__name__}(uri={self._conn.uri!r})" diff --git a/python/python/lancedb/permutation.py b/python/python/lancedb/permutation.py index b3d95c352..bcf84bf3a 100644 --- a/python/python/lancedb/permutation.py +++ b/python/python/lancedb/permutation.py @@ -226,7 +226,7 @@ class PermutationBuilder: async def do_execute(): inner_tbl = await self._async.execute() - return LanceTable.from_inner(inner_tbl) + return await LanceTable.from_inner(inner_tbl) return LOOP.run(do_execute()) diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 8b1a83bd4..0afc60bc5 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -2186,11 +2186,15 @@ class LanceTable(Table): return self.name @classmethod - def from_inner(cls, tbl: LanceDBTable): - from .db import LanceDBConnection + async def from_inner(cls, tbl: LanceDBTable): + from .db import AsyncConnection, LanceDBConnection async_tbl = AsyncTable(tbl) - conn = LanceDBConnection.from_inner(tbl.database()) + inner_conn = tbl.database() + read_consistency_interval = await AsyncConnection( + inner_conn + ).get_read_consistency_interval() + conn = LanceDBConnection.from_inner(inner_conn, read_consistency_interval) return cls( conn, async_tbl.name, diff --git a/python/python/tests/test_db.py b/python/python/tests/test_db.py index 86b70b75d..93b791650 100644 --- a/python/python/tests/test_db.py +++ b/python/python/tests/test_db.py @@ -77,6 +77,23 @@ def test_sync_repr_does_not_use_background_loop(tmp_path, monkeypatch): assert repr(table) == f"LanceTable(name='test', _conn={db!r})" +def test_read_consistency_interval_does_not_use_background_loop(tmp_path, monkeypatch): + from lancedb.background_loop import LOOP + from lancedb.db import LanceDBConnection + + consistency_interval = timedelta(seconds=5) + db = lancedb.connect(tmp_path, read_consistency_interval=consistency_interval) + db_from_inner = LanceDBConnection.from_inner(db._inner, consistency_interval) + + def fail_run(*args, **kwargs): + raise AssertionError("properties should not use the Python background loop") + + monkeypatch.setattr(LOOP, "run", fail_run) + + assert db.read_consistency_interval == consistency_interval + assert db_from_inner.read_consistency_interval == consistency_interval + + def test_ingest_pd(tmp_path): db = lancedb.connect(tmp_path) diff --git a/python/python/tests/test_permutation.py b/python/python/tests/test_permutation.py index e23d2da6b..6d8f6f431 100644 --- a/python/python/tests/test_permutation.py +++ b/python/python/tests/test_permutation.py @@ -6,6 +6,7 @@ import math import pytest from lancedb import DBConnection, Table, connect +from lancedb.background_loop import LOOP from lancedb.permutation import Permutation, Permutations, permutation_builder @@ -31,6 +32,25 @@ def test_split_random_ratios(mem_db): assert 65 <= split_1_count <= 75 # ~70% ± tolerance +def test_execute_does_not_reenter_background_loop(tmp_path, monkeypatch): + import threading + + db = connect(tmp_path) + tbl = db.create_table("test_table", pa.table({"x": range(10)})) + original_run = LOOP.run + + def fail_on_reentry(future): + assert threading.current_thread() is not LOOP.thread + return original_run(future) + + monkeypatch.setattr(LOOP, "run", fail_on_reentry) + + permutation_tbl = permutation_builder(tbl).execute() + + assert permutation_tbl.count_rows() == 10 + assert permutation_tbl._conn.read_consistency_interval is None + + def test_split_random_counts(mem_db): """Test random splitting with absolute counts.""" tbl = mem_db.create_table( diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index c4d85ea4f..248e13622 100644 --- a/python/python/tests/test_table.py +++ b/python/python/tests/test_table.py @@ -6,6 +6,7 @@ import os import sys import threading import warnings +from concurrent.futures import ThreadPoolExecutor from datetime import date, datetime, timedelta from time import sleep from typing import List @@ -2158,6 +2159,27 @@ def test_delete(mem_db: DBConnection): assert table.to_arrow()["id"].to_pylist() == [1] +def test_concurrent_deletes_are_thread_safe(mem_db: DBConnection): + num_workers = 8 + table = mem_db.create_table( + "my_table", data=[{"id": row_id} for row_id in range(num_workers)] + ) + barrier = threading.Barrier(num_workers) + + def delete(row_id: int): + barrier.wait() + return table.delete(f"id = {row_id}") + + with ThreadPoolExecutor(max_workers=num_workers) as pool: + results = list(pool.map(delete, range(num_workers))) + + assert all(result.num_deleted_rows == 1 for result in results) + assert sorted(result.version for result in results) == list( + range(2, num_workers + 2) + ) + assert table.count_rows() == 0 + + def test_delete_expr(mem_db: DBConnection): table = mem_db.create_table( "my_table", diff --git a/python/src/table.rs b/python/src/table.rs index 3835afe53..4ad1fe42a 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -759,6 +759,9 @@ impl Table { #[allow(private_interfaces)] pub fn delete(self_: PyRef<'_, Self>, condition: PredicateArg) -> PyResult> { + // Do not hold the Python borrow across the await. The cloned Rust table + // handle is thread-safe and allows deletes on the same Python table to + // run concurrently without PyO3 reporting "Already borrowed". let inner = self_.inner_ref()?.clone(); future_into_py(self_.py(), async move { let result = match &condition {