fix(python): explain explicit blob reads and subset copies (#4940)

Unsupported remote blob materialization errors now point to row-ID
selection, fetch_blobs and fetch_blob_files. Document descriptor
structs, fetched bytes and Function BlobReader inputs, and show a
bounded subset-copy recipe that keeps the source unchanged between
selection and fetching.

Adds guidance assertions and copy coverage preserving payloads, empty
blobs and nulls.

No new blob materialization API is introduced.

Formatting applied; no local builds or tests were run. CI will execute
the regression coverage.

docs/web/management/table.mdx

@@ -477,6 +477,51 @@ the default `"lazy"` mode because it would open `BlobFile` handles. Query with
 `blob_mode="descriptions"`, then read bytes with `fetch_blobs` or
 `fetch_blob_files`.
 
+###### Copy a subset of blob rows {#multimodal-copy-subset}
+
+Blob query results describe the stored payload; they are not the bytes to
+insert into another table. These APIs return different values:
+
+| Read path | Value |
+| --- | --- |
+| SQL or an Arrow query in `"descriptions"` mode | Descriptor struct identifying the stored blob. Treat it as opaque. |
+| `table.fetch_blobs(column, row_ids)` | Arrow `large_binary` array; `.to_pylist()` gives `bytes` or `None` in the requested order. |
+| A Function parameter annotated `lancedb.BlobReader` | Lazy, seekable reader over the input blob, available inside the Function. |
+
+For an Enterprise source table, select the rows and fetch their bytes explicitly.
+The following example copies up to 100 rows from an existing `images` table with
+an integer `id` and a blob `image` column. Keep the source unchanged during this
+operation: selection and fetching are separate requests.
+
+```python
+import lancedb
+import pyarrow as pa
+
+source = db.open_table("images")
+hits = (
+    source.search()
+    .where("id < 100")
+    .select(["id"])
+    .with_row_id(True)
+    .limit(100)
+    .to_arrow()
+)
+payloads = source.fetch_blobs("image", hits["_rowid"]).to_pylist()
+subset = db.create_table(
+    "image_subset",
+    schema=pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")]),
+)
+if hits.num_rows:
+    subset.add([
+        {"id": image_id, "image": payload}
+        for image_id, payload in zip(hits["id"].to_pylist(), payloads, strict=True)
+    ])
+```
+
+Empty blobs remain `b""` and null blobs remain `None`. Fetch bounded batches
+for larger subsets; `fetch_blobs` materializes all requested bytes in memory.
+For incremental decoding, use `fetch_blob_files` instead.
+
 `blob_mode="bytes"` materializes top-level and struct-nested blob columns. It
 raises `ValueError` when the selected columns include blobs
 inside a list. Select those columns with `blob_mode="descriptions"`, or leave

python/python/lancedb/query.py

@@ -206,7 +206,10 @@ def _blob_bytes_without_snapshot_error() -> NotImplementedError:
         "blob_mode='bytes' cannot safely materialize blob columns without a "
         "stable table snapshot. LanceDB Enterprise and namespace QueryTable pushdown "
         "run the query on a server, so the blob fetch could read another table "
-        "version. Use blob_mode='descriptions'."
+        "version. Use blob_mode='descriptions' to inspect descriptors. To read "
+        "payloads, select row IDs with with_row_id(True), then call "
+        "table.fetch_blobs(column, row_ids) or table.fetch_blob_files(column, "
+        "row_ids). Keep the source unchanged between selection and fetching."
     )
 
 
@@ -453,7 +456,10 @@ def _finish_remote_pandas(
     raise NotImplementedError(
         f"remote to_pandas(blob_mode={blob_mode!r}) cannot safely materialize "
         "blob columns without a stable table snapshot; "
-        "use blob_mode='descriptions'"
+        "use blob_mode='descriptions' to inspect descriptors. To read payloads, "
+        "select row IDs with with_row_id(True), then call "
+        "table.fetch_blobs(column, row_ids) or table.fetch_blob_files(column, "
+        "row_ids). Keep the source unchanged between selection and fetching."
     )
 
 

python/python/lancedb/table.py

@@ -2626,6 +2626,12 @@ class Table(ABC):
         bytes. The client splits requests automatically and reads an individual
         larger blob through the Range route. This method still materializes all
         bytes in memory; for large values use :meth:`fetch_blob_files`.
+
+        The returned Arrow ``large_binary`` array contains payload bytes, not
+        SQL/query descriptor dictionaries. Use ``.to_pylist()`` to obtain
+        Python ``bytes`` or ``None`` values for copying into a blob column.
+        Select row IDs with ``search().with_row_id(True)`` first. Selection and
+        fetching are separate requests: keep the source unchanged between them.
         """
 
     @abstractmethod

python/python/tests/test_blob.py

@@ -1070,8 +1070,10 @@ def test_bytes_mode_refused_under_namespace_query_pushdown(tmp_path):
     )
     table.add([{"id": 1, "image": b"one"}])
 
-    with pytest.raises(NotImplementedError, match="stable table snapshot"):
+    with pytest.raises(NotImplementedError, match="stable table snapshot") as exc:
         table.to_arrow(blob_mode="bytes")
+    assert "fetch_blobs(column, row_ids)" in str(exc.value)
+    assert "with_row_id(True)" in str(exc.value)
     with pytest.raises(NotImplementedError, match="stable table snapshot"):
         table.search().to_list(blob_mode="bytes")
     with pytest.raises(NotImplementedError, match="stable table snapshot"):
@@ -1103,6 +1105,36 @@ async def test_async_bytes_mode_refused_under_namespace_query_pushdown(tmp_path)
         await table.query().to_batches(blob_mode="bytes")
 
 
+def test_copy_blob_subset_preserves_payloads_nulls_and_empty_values(tmp_path):
+    db = lancedb.connect(tmp_path)
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    rows = [
+        {"id": 1, "image": b"one"},
+        {"id": 2, "image": b""},
+        {"id": 3, "image": None},
+        {"id": 4, "image": b"excluded"},
+    ]
+    source = db.create_table("source", data=rows, schema=schema)
+    hits = (
+        source.search()
+        .where("id < 4")
+        .select(["id"])
+        .with_row_id(True)
+        .limit(100)
+        .to_arrow()
+    )
+    payloads = source.fetch_blobs("image", hits["_rowid"]).to_pylist()
+    subset = db.create_table("subset", schema=schema)
+    subset.add(
+        [
+            {"id": image_id, "image": payload}
+            for image_id, payload in zip(hits["id"].to_pylist(), payloads, strict=True)
+        ]
+    )
+    copied = subset.search().to_list(blob_mode="bytes")
+    assert sorted(copied, key=lambda row: row["id"]) == rows[:3]
+
+
 def test_fetch_blobs_round_trip():
     table = _blob_table(
         "round_trip",

python/python/tests/test_remote_db.py

@@ -4958,8 +4958,12 @@ def test_sync_remote_blob_pandas_avoids_storage_access(
             assert "_rowid" not in df.columns
             assert df["image"][0]["size"] == 5
         else:
-            with pytest.raises(NotImplementedError, match="stable table snapshot"):
+            with pytest.raises(
+                NotImplementedError, match="stable table snapshot"
+            ) as exc:
                 query.to_pandas(blob_mode=blob_mode)
+            assert "fetch_blobs(column, row_ids)" in str(exc.value)
+            assert "with_row_id(True)" in str(exc.value)
 
 
 @pytest.mark.asyncio