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

<!-- lance-gatekeeper-fix:v1 agent=8b4aaf556014ac0e457c9d63495697e4
generation=1 -->

Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>

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)
 

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(

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())])