mirror of
https://github.com/lancedb/lancedb.git
synced 2026-08-18 12:08:35 +00:00
fix(python): guard concurrent table deletes (#3787)
<!-- lance-gatekeeper-fix:v1 agent=5c80c44c083b3b8ad0da595419d468fc generation=1 --> ## Root cause The legacy synchronous Python table called `delete` on a shared, mutable `lance.Dataset`. Concurrent table operations could hold a PyO3 borrow while delete requested an exclusive borrow, producing `RuntimeError: Already borrowed`. The current async-backed binding fixes this by cloning its thread-safe Rust table handle before awaiting, but that concurrency contract had no regression coverage. ## Fix - Document why delete must clone the Rust table handle before entering its async future. - Add a barrier-synchronized regression test that deletes distinct rows through one shared table from eight Python threads. - Verify every delete commits exactly one row, every commit gets a distinct version, and no rows remain. ## Validation - `cargo check --quiet --features remote --tests --examples` - `cargo fmt --all -- --check` - `uv run --extra tests --extra dev ruff format --check python/tests/test_table.py` - `uv run --extra tests --extra dev ruff check python/tests/test_table.py` - `uv run --extra tests --extra dev pytest python/tests/test_table.py::test_concurrent_deletes_are_thread_safe python/tests/test_table.py::test_delete python/tests/test_table.py::test_delete_expr python/tests/test_table.py::test_delete_expr_async -q` (4 passed) - Manual stress reproduction: 100 concurrent deletes on one table completed at versions 2–101 with zero rows remaining. Fixes #530 Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
This commit is contained in:
committed by
GitHub
parent
624a75edf7
commit
7357d63e87
@@ -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
|
||||
@@ -2124,6 +2125,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",
|
||||
|
||||
@@ -745,6 +745,9 @@ impl Table {
|
||||
|
||||
#[allow(private_interfaces)]
|
||||
pub fn delete(self_: PyRef<'_, Self>, condition: PredicateArg) -> PyResult<Bound<'_, PyAny>> {
|
||||
// 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 {
|
||||
|
||||
Reference in New Issue
Block a user