fix: read legacy blob columns through blob APIs (#4649)

Legacy blob v1 columns were rejected by LanceDB's blob validation and
omitted from `blob_columns()`, even though the pinned Lance reader
already supports them. Accept both blob versions in read validation and
discovery so `fetch_blobs`, `fetch_blob_files`, and `fetch_blob_ranges`
can read legacy payloads through the existing native reader.

Keep list-nested blobs excluded from row-based reads and document how to
fetch legacy blobs using explicit row IDs. Regression coverage checks
sync and async APIs, independently written Lance datasets, ordering and
duplicates across fragments, null and empty blobs, lazy seeking, ranges,
and deleted rows.

Fixes #4644

Validation:

- Both new Python regressions failed on the original implementation with
the reported legacy-column rejection and empty discovery result, then
passed with the fix.
- Python blob suite: 202 passed, 1 skipped.
- Remote blob tests: 29 passed.
- Focused Rust blob unit tests with `remote`: 13 passed.
- Rust blob integration suite with `remote`: 59 passed, including legacy
storage 2.0 and 2.1.
- Rust compile and Clippy checks with `remote`, tests, and examples
passed.
- Python package Ruff lint and format checks passed; Rust formatting
passed.
- Repository-wide `ruff check .` reports 49 existing errors in unchanged
`docs/web-tests` examples; the Python package is clean.

<!-- lance-gatekeeper-fix:v1 agent=02b7edcbfa8ff0e7fad4615159249b3e
generation=1 -->

---------

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

python/python/lancedb/table.py

@@ -2516,7 +2516,7 @@ class Table(ABC):
 
     @abstractmethod
     def blob_columns(self) -> list[str]:
-        """Names of blob v2 columns that can be fetched by row id.
+        """Names of blob v1 and v2 columns that can be fetched by row id.
 
         Struct-nested blobs use dotted paths. Blobs inside lists are omitted
         because a row id does not identify an individual list element.
@@ -2530,6 +2530,7 @@ class Table(ABC):
     ) -> pa.LargeBinaryArray:
         """Materialize full blob bytes for ``column`` at the given rows.
 
+        Local tables support both legacy blob v1 and blob v2 columns.
         The result has the same length and order as ``row_ids``. Null blobs
         produce null slots; valid empty blobs produce ``b""``.
 
@@ -2553,8 +2554,9 @@ class Table(ABC):
         column: str,
         requests: Sequence[Sequence[int]],
     ) -> pa.LargeBinaryArray:
-        """Materialize row-specific byte ranges from a blob v2 column.
+        """Materialize row-specific byte ranges from a blob column.
 
+        Local tables support both legacy blob v1 and blob v2 columns.
         Each request is a three-item ``(row_id, offset, length)`` sequence.
         Requests may be repeated or reordered, including multiple ranges for
         the same blob.
@@ -2579,6 +2581,7 @@ class Table(ABC):
     ) -> "list[Optional[BlobFile]]":
         """Open lazy, seekable :class:`~lancedb._blob.BlobFile` handles.
 
+        Local tables support both legacy blob v1 and blob v2 columns.
         Prefer this over :meth:`fetch_blobs` for large payloads. ``row_ids`` is
         a ``list[int]``, an integer Arrow array or chunked array, or a query
         ``pyarrow.Table`` carrying row identity via ``_rowid`` or a

python/python/tests/test_blob.py

@@ -1903,10 +1903,74 @@ def test_legacy_v1_query_omits_auto_row_id():
     assert "_rowid" not in hits.column_names
 
 
-def test_fetch_blobs_rejects_legacy_v1_column():
+def test_fetch_legacy_v1_blobs():
     table = _legacy_v1_table("legacy_fetch")
-    with pytest.raises(ValueError, match="legacy blob column.*blob v2"):
-        table.fetch_blobs("legacy", [0])
+    table.add(
+        [
+            {"id": 2, "legacy": None},
+            {"id": 3, "legacy": b""},
+            {"id": 4, "legacy": b"another fragment"},
+        ]
+    )
+    hits = table.search().select(["id", "legacy"]).with_row_id(True).to_arrow()
+    assert pa.types.is_struct(hits["legacy"].type)
+    by_id = dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist()))
+    row_ids = [by_id[i] for i in [4, 2, 1, 3, 4]]
+    expected = [b"another fragment", None, b"old", b"", b"another fragment"]
+    assert table.fetch_blobs("legacy", row_ids).to_pylist() == expected
+    assert table.blob_columns() == ["legacy"]
+    assert table.fetch_blobs("legacy", hits).to_pylist() == [
+        b"old",
+        None,
+        b"",
+        b"another fragment",
+    ]
+    assert table.fetch_blobs("legacy", []).to_pylist() == []
+    assert table.fetch_blob_files("legacy", []) == []
+
+    handles = table.fetch_blob_files("legacy", row_ids)
+    assert len(handles) == len(expected)
+    for handle, payload in zip(handles, expected):
+        if payload is None:
+            assert handle is None
+        else:
+            with handle:
+                assert handle.size() == len(payload)
+                assert handle.read(2) == payload[:2]
+                assert handle.seek(0) == 0
+                assert handle.read() == payload
+
+    assert table.fetch_blob_ranges(
+        "legacy", [(by_id[4], 8, 8), (by_id[2], 0, 1), (by_id[3], 0, 0)]
+    ).to_pylist() == [b"fragment", None, b""]
+
+    table.delete("id = 2")
+    for fetch in [table.fetch_blobs, table.fetch_blob_files]:
+        with pytest.raises(ValueError, match="row ids but some do not exist"):
+            fetch("legacy", [by_id[1], by_id[2]])
+
+
+@pytest.mark.asyncio
+async def test_async_fetch_legacy_v1_blobs(tmp_path):
+    table = _legacy_v1_table("legacy_async_source")
+    # Open a dataset written by Lance independently of the LanceDB connection.
+    lance.write_dataset(
+        pa.Table.from_pylist(
+            [{"id": 1, "legacy": b"old"}, {"id": 2, "legacy": None}],
+            schema=table.schema,
+        ),
+        tmp_path / "legacy.lance",
+        data_storage_version="2.0",
+    )
+    db = await lancedb.connect_async(tmp_path)
+    table = await db.open_table("legacy")
+    assert await table.blob_columns() == ["legacy"]
+    hits = await table.query().with_row_id().to_arrow()
+    assert (await table.fetch_blobs("legacy", hits)).to_pylist() == [b"old", None]
+    handles = await table.fetch_blob_files("legacy", hits)
+    assert handles[0].read() == b"old"
+    handles[0].close()
+    assert handles[1] is None
 
 
 @pytest.mark.asyncio