From 40af9f350e785d762cc887e029718fc9904bbe0b Mon Sep 17 00:00:00 2001 From: "lancedb-gatefixer[bot]" <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Date: Fri, 23 Oct 2026 01:09:32 +0800 Subject: [PATCH] fix(python): avoid deadlock for query-backed iterators (#3820) ## Summary - move synchronous iterator preparation off LanceDB's singleton background event loop - keep the loop available when an input iterator executes a synchronous table query - cover query-backed iterators for both `Table.add` and `create_table` ## Root cause The async implementations prepared input data on the singleton background loop and peeked at the first item of iterators there. If that iterator called a synchronous LanceDB query, `LOOP.run()` scheduled the query on the same loop and blocked waiting for it, so neither operation could progress. ## Validation - `ruff format` on the changed Python files - `ruff check .` - `cd python && uv run --frozen --extra tests pytest python/tests/test_table.py -q` (138 passed) - the broader non-slow suite was attempted but hit a native segmentation fault in the external DataFusion `register_table_provider` path; an isolated control using the pre-change inline preparation behavior reproduced the same unrelated crash Fixes #2107 Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> --- python/python/lancedb/db.py | 13 +++++++++++-- python/python/lancedb/table.py | 16 ++++++++++++---- python/python/tests/test_table.py | 24 ++++++++++++++++++++++++ 3 files changed, 47 insertions(+), 6 deletions(-) diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 5def89b01..e1e5d270a 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -5,6 +5,7 @@ from __future__ import annotations from abc import abstractmethod +import asyncio from datetime import timedelta from pathlib import Path import sys @@ -2652,8 +2653,16 @@ class AsyncConnection(object): if fill_value is None: fill_value = 0.0 - data, schema = sanitize_create_table( - data, schema, metadata, on_bad_vectors, fill_value + # Input preparation may advance a user-provided iterator. Keep that work + # off the background event loop so an iterator can use the synchronous + # LanceDB API without blocking the loop that API needs to make progress. + data, schema = await asyncio.to_thread( + sanitize_create_table, + data, + schema, + metadata, + on_bad_vectors, + fill_value, ) validate_schema(schema) diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index d5872e6f0..006ef8cb9 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -6213,8 +6213,12 @@ class AsyncTable: if mode == "overwrite": # For overwrite, apply the same preprocessing as create_table # so vector columns are inferred as FixedSizeList. - data, _ = sanitize_create_table( - data, None, on_bad_vectors=on_bad_vectors, fill_value=fill_value + data, _ = await asyncio.to_thread( + sanitize_create_table, + data, + None, + on_bad_vectors=on_bad_vectors, + fill_value=fill_value, ) if blob_v2_column_paths(schema): # Plain bytes cannot be inferred as blob v2. Keep matching blob @@ -6224,7 +6228,8 @@ class AsyncTable: elif on_bad_vectors != "error" or ( schema.metadata is not None and b"embedding_functions" in schema.metadata ): - data = _sanitize_data( + data = await asyncio.to_thread( + _sanitize_data, data, schema, metadata=schema.metadata, @@ -6233,7 +6238,10 @@ class AsyncTable: allow_subschema=True, ) _register_optional_converters() - data = to_scannable(data) + # Converting an iterator peeks at its first item. A synchronous query in + # that iterator schedules work on LOOP, so peeking on LOOP's own thread + # would deadlock waiting for itself. + data = await asyncio.to_thread(to_scannable, data) progress, owns = _normalize_progress(progress) try: return await self._inner.add( diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index 755e94d7f..095f0b911 100644 --- a/python/python/tests/test_table.py +++ b/python/python/tests/test_table.py @@ -518,6 +518,30 @@ def test_add(mem_db: DBConnection): _add(table, schema) +def test_add_from_iterator_that_queries_table(mem_db: DBConnection): + source = mem_db.create_table("source", data=pa.table({"id": range(16)})) + target = mem_db.create_table("target", schema=source.schema) + + def batches(): + for _ in range(5): + yield source.search().limit(10).to_arrow() + + target.add(batches()) + + assert target.count_rows() == 50 + + +def test_create_table_from_iterator_that_queries_table(mem_db: DBConnection): + source = mem_db.create_table("source", data=pa.table({"id": range(16)})) + + def batches(): + yield source.search().limit(10).to_arrow() + + target = mem_db.create_table("target", data=batches()) + + assert target.count_rows() == 10 + + def test_add_releases_arrow_buffers_without_gc(mem_db: DBConnection): """Regression test for https://github.com/lancedb/lancedb/issues/2512.""" schema = pa.schema([pa.field("x", pa.int64())])