feat: add negotiated batched blob range reads across SDKs (#4994)

Existing Rust and Python blob range calls must keep working when only the SDK is upgraded. Add negotiated batched range reads across Rust, Python sync/async and TypeScript, preserving ordered NULLs, duplicates and branch/version reads with splitting and reassembly at 1,024 ranges or 64 MiB per request. Only a successful describe response advertising `x-lancedb-blob-ranges: 1` selects the batched endpoint. Missing or unknown capability values select the existing descriptor-query plus BlobFile Range GET route before the operation. Endpoint failures never switch protocols. HTTP 400 preserves the server message and request ID as Rust `InvalidInput` / Python `ValueError`; other errors retain their existing semantics. The operation binds both numeric version and immutable snapshot identity from its producing query or first range response. Every later partition and oversized-range piece sends that original identity and requires an equal response identity before decoding. Same-name table recreation therefore fails explicitly instead of returning mixed old/new payloads; no partial array is returned on later failure. Targets `main`. The read-buffering prerequisite #4979, safe-write prerequisite #4977 and deferred-query identity prerequisite #4973 are merged into main; #5000 also supplies the merged SQL timeout-test prerequisite. Server support is [Sophon #10219](https://github.com/lancedb/sophon/pull/10219), including #10211. Exact SDK `1497b508324662180086a4d1e0a6eb9e0b3970db` and server `f39712c2c150b220eb831e00f6663c9eef32dce5` passed a real local SDK → REST → PE experiment: one Rust harness test, 14 assertion groups, 38 REST range requests and 34 PE RPCs. All four sync/async × warm/cold table-recreation cases failed explicitly before the second descriptor lookup or PE dispatch, with no mixed payload or partial result. The other groups retained 1,025-range success, 70 MiB reassembly, branch/version reads, overwrite snapshot consistency, error behavior and TypeScript buffer ownership. Bindings were rebuilt and their loaded hashes verified. This is local interoperability evidence, not Cloud, mixed-fleet or deployed header-forwarding acceptance. The server dependency upgrade, External/cache/URI-policy acceptance and validated rollout remain required. Enforcing PE support must precede QN and SDK enablement; identity does not retain versions or freeze External objects. This PR does not claim deployed support or complete ENT-3100/GA acceptance. Published documentation retains the Early access badge and released Enterprise instructions, including explicit fetches for copying. The new Enterprise `blob_mode="bytes"` guidance, snapshot-bound copy example, future streaming restrictions and GA wording remain source-only MDX drafts until release. The correction distinguishing released Enterprise SQL bytes from SDK descriptors remains visible. Documentation validation checked comment boundaries, visible instructions, Python example syntax and whitespace; the site was not built. The main integration keeps these release gates unchanged. Main integration at `0491d9bdcc65ce5d52ea93e1083c89e951ea4f9d` normally merges main `9b13e8f72530c8b1950c09685f6920711154a539`, including its Lance beta.18 dependency. Focused current-source validation passed: 493 remote-table/query/Blob/IPC/write-progress Rust tests, 23 shared create/write-body tests, 59 native Blob integration tests and 7 public HTTP range tests. Rebuilt Python passed 468 selected tests (one optional PyAV test skipped), and rebuilt TypeScript passed 194 selected tests across four suites; lint/format, TypeScript compilation and API documentation generation passed. Shared tests include actual HTTP multipart failures, acknowledgment boundaries, dictionary compaction and native Blob compaction. The 598 unrelated Jest cases excluded by the focused shared-suite selector were not run. The retained 14-group joint REST/PE experiment used the earlier exact candidates stated above. Operation-identity/range/IPC code is unchanged, but dependencies and shared write interfaces changed with main; the new local checks validate the SDK combination and do not constitute a fresh final-head SDK/server joint, Cloud or rollout run. Early access and every remaining server/External/cache/policy/dependency/rollout condition remain in force. CI snapshot on this head (2026-11-12 18:30 UTC): `build-no-lock` failed in the standalone `ci/tests/test_build_no_lock.py` fixture at the retained-rlib assertion, before the actual SDK no-lock build. The fixture, helper and Rust workflow are byte-identical to main; the fixture does not load SDK source. The underlying CI/environment cause is not diagnosed here. Other jobs remained pending/running at the snapshot, so CI is not reported as passing. Failure: https://github.com/lancedb/lancedb/actions/runs/43280277756/job/131399418284.
2026-11-13 14:16:54 +00:00 · 2026-11-13 04:03:04 +08:00
parent 1806dc2323
commit 6d24446a30
21 changed files with 2473 additions and 70 deletions
@@ -2661,8 +2661,12 @@ class Table(ABC):
        results are returned.

        Row IDs can be obtained from a query with ``with_row_id(True)``.
-        LanceDB Enterprise reads the requested ranges through version-pinned blob
-        handles using HTTP byte-range requests.
+        LanceDB Enterprise uses the batched endpoint when the server advertises
+        support, partitioning at 1024 ranges and 64 MiB, including individual
+        larger ranges. Older servers use descriptor queries and BlobFile range
+        reads. Both protocols read one exact branch and version without changing
+        the table's checkout. Errors from an advertised endpoint propagate without
+        switching protocols. The final output occupies memory.
        """

    @abstractmethod
@@ -8652,6 +8656,10 @@ class AsyncTable:
        column: str,
        requests: Sequence[Sequence[int]],
    ) -> pa.LargeBinaryArray:
+        """Materialize row-specific byte ranges.
+
+        See [Table.fetch_blob_ranges][lancedb.table.Table.fetch_blob_ranges].
+        """
        return await self._inner.fetch_blob_ranges(column, list(requests))

    async def fetch_blob_files(
@@ -0,0 +1,380 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Exercise the public sync/async range APIs over bounded loopback HTTP."""
+
+import asyncio
+import hashlib
+import json
+from pathlib import Path
+import socket
+import threading
+
+import lancedb._lancedb as native
+import pyarrow as pa
+import pytest
+
+from lancedb.remote.errors import HttpError, TableNotFoundError
+from test_remote_db import mock_lancedb_connection
+
+MIB = 1024 * 1024
+IDENTITY = "v1:" + "a" * 64
+ROUTE = "/v1/table/test/fetch_blob_ranges/"
+
+
+def payload(row, offset, length, version):
+    pattern = bytes((row * 17 + i + version) % 251 for i in range(251))
+    start = offset % 251
+    return (pattern * ((start + length + 250) // 251))[start : start + length]
+
+
+class RangeServer:
+    def __init__(self, fault=None):
+        self.fault = fault
+        self.calls = []
+        self.fetches = []
+        self.identities = []
+        self.stalled = threading.Event()
+        self.release = threading.Event()
+
+    def __call__(self, request):
+        request.connection.settimeout(15)
+        size = int(request.headers.get("Content-Length", 0))
+        assert size <= 256 * 1024
+        body = json.loads(request.rfile.read(size) or b"{}")
+        self.calls.append((request.command, request.path, body))
+        if request.path in (
+            "/v1/table/test/describe/",
+            "/v1/table/test/branches/list/",
+        ):
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.send_header("phalanx-version", "1.13.0-beta.0")
+            if request.path.endswith("/describe/"):
+                request.send_header("x-lancedb-blob-ranges", "1")
+            request.end_headers()
+            response = (
+                {
+                    "branches": {
+                        "experiment": {
+                            "parentBranch": None,
+                            "parentVersion": 1,
+                            "createAt": 1,
+                            "manifestSize": 1,
+                        }
+                    }
+                }
+                if request.path.endswith("branches/list/")
+                else {"version": 8, "schema": {"fields": []}}
+            )
+            request.wfile.write(json.dumps(response).encode())
+            return
+        assert (request.command, request.path) == ("POST", ROUTE)
+        self.fetches.append(body)
+        self.identities.append(request.headers.get("x-lancedb-query-snapshot"))
+        assert body["column"] == "image"
+        assert body["branch"] == "experiment"
+        ranges = body["ranges"]
+        assert len(ranges) <= 1024
+        assert sum(r["length"] for r in ranges) <= 64 * MIB
+        version = body["version"] or (7 if len(self.fetches) == 1 else 8)
+        values = [
+            None
+            if r["row_id"] == 20
+            else payload(r["row_id"], r["offset"], r["length"], version)
+            for r in ranges
+        ]
+        fault = (
+            self.fault
+            if len(self.fetches) > 1 or str(self.fault).startswith("first-")
+            else None
+        )
+        if fault in ("recreated", "gone"):
+            request.send_response(400 if fault == "recreated" else 404)
+            request.end_headers()
+            request.wfile.write(
+                b"query snapshot identity mismatch: source gone or replaced"
+            )
+            return
+        if fault == "stall" and len(self.fetches) == 2:
+            request.send_response(200)
+            request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
+            request.send_header("x-lancedb-version", str(version))
+            request.send_header("x-lancedb-query-snapshot", IDENTITY)
+            request.end_headers()
+            request.wfile.flush()
+            self.stalled.set()
+            assert self.release.wait(15), "test did not release stalled response"
+            return
+        if fault == "structured400":
+            request.send_response(400)
+            request.send_header("x-lancedb-error-code", "invalid_input")
+            request.end_headers()
+            request.wfile.write(b'{"error":"range exceeds blob size"}')
+            return
+        if isinstance(fault, int):
+            request.send_response(fault)
+            request.end_headers()
+            request.wfile.write(b"ordinary server failure")
+            return
+        if fault == "disconnect":
+            request.connection.shutdown(socket.SHUT_RDWR)
+            return
+        column = "wrong" if fault == "column" else "value"
+        data_type = pa.binary() if fault == "type" else pa.large_binary()
+        if fault == "count":
+            values = []
+        elif fault == "length":
+            values = [b""]
+        batch = pa.table({column: pa.array(values, type=data_type)})
+        sink = pa.BufferOutputStream()
+        with pa.ipc.new_stream(sink, batch.schema) as writer:
+            writer.write_table(batch, max_chunksize=257)
+        ipc = sink.getvalue().to_pybytes()
+        if fault == "eos":
+            ipc = ipc[:-8]
+        elif fault == "body":
+            ipc = ipc[: len(ipc) // 2]
+        request.send_response(200)
+        request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
+        if fault != "version":
+            request.send_header("x-lancedb-version", str(version))
+        if fault not in ("missing-identity", "first-missing-identity"):
+            identity = (
+                "broken"
+                if fault == "first-invalid-identity"
+                else "v1:" + "b" * 64
+                if fault == "wrong-identity"
+                else IDENTITY
+            )
+            request.send_header("x-lancedb-query-snapshot", identity)
+        request.send_header(
+            "Content-Length", str(len(ipc) + (10 if fault == "transport" else 0))
+        )
+        request.end_headers()
+        for i in range(0, len(ipc), 8192):
+            request.wfile.write(ipc[i : i + 8192])
+
+
+async def fetch(table, requests, use_async):
+    result = table.fetch_blob_ranges("image", requests)
+    return await result if use_async else result
+
+
+def api(db, use_async, version=None):
+    table = db.open_table("test", branch="experiment", version=version)
+    return table._table if use_async else table
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize("version", [None, 3])
+async def test_range_sdk_order_snapshot_and_overwrite(use_async, version):
+    handler = RangeServer()
+    with mock_lancedb_connection(handler) as db:
+        table = api(db, use_async, version)
+        requests = [(20 if i % 9 == 0 else 10 + i % 7, i, i % 4) for i in range(1025)]
+        requests += [(10, 0, 0), (10, 2, 3), (10, 2, 3)]
+        result = await fetch(table, requests, use_async)
+        assert isinstance(result, pa.LargeBinaryArray)
+        assert result.to_pylist() == [
+            None if row == 20 else payload(row, offset, length, version or 7)
+            for row, offset, length in requests
+        ]
+        assert [b["version"] for b in handler.fetches] == [version, version or 7]
+        result = await fetch(table, [(10, 0, 1)], use_async)
+        assert result.to_pylist() == [payload(10, 0, 1, version or 8)]
+        assert handler.fetches[-1]["version"] == version
+        assert len(handler.fetches) == 3
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_range_sdk_cached_reopen_negotiates_capability(use_async):
+    handler = RangeServer()
+    with mock_lancedb_connection(handler) as db:
+        for version in (7, 8):
+            table = api(db, use_async)
+            result = await fetch(table, [(10, 0, 1)], use_async)
+            assert result.to_pylist() == [payload(10, 0, 1, version)]
+        assert len(handler.fetches) == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_range_sdk_large_single_and_aggregate_budget(use_async):
+    handler = RangeServer()
+    with mock_lancedb_connection(handler) as db:
+        table = api(db, use_async)
+        requests = [
+            (10, 3, 33 * MIB),
+            (30, 7, 33 * MIB),
+            (40, 11, 70 * MIB),
+            (20, 0, 2**64 - 1),
+        ]
+        # The loopback server advances latest after the first partition.
+        result = await fetch(table, requests, use_async)
+        assert len(result) == 4
+        for i, (row, offset, length) in enumerate(requests[:3]):
+            expected = payload(row, offset, length, 7)
+            assert (
+                hashlib.sha256(result[i].as_buffer()).digest()
+                == hashlib.sha256(expected).digest()
+            )
+            assert len(result[i].as_buffer()) == length
+        assert result[3].as_py() is None
+        assert len(handler.fetches) == 3
+        assert [b["version"] for b in handler.fetches] == [None, 7, 7]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize(
+    "fault",
+    [
+        "type",
+        "column",
+        "count",
+        "length",
+        "eos",
+        "body",
+        "transport",
+        "version",
+        "disconnect",
+        401,
+        403,
+        404,
+        429,
+        500,
+    ],
+)
+async def test_range_sdk_later_failure_rejects_entire_call(use_async, fault):
+    handler = RangeServer(fault=fault)
+    with mock_lancedb_connection(
+        handler, retry_config={"retries": 2, "connect_retries": 2, "read_retries": 2}
+    ) as db:
+        table = api(db, use_async)
+        with pytest.raises(Exception) as exc:
+            await fetch(table, [(10, 0, 1)] * 1025, use_async)
+        if fault == 404:
+            assert isinstance(exc.value, TableNotFoundError)
+            assert isinstance(exc.value.__cause__, HttpError)
+            assert exc.value.__cause__.status_code == 404
+        else:
+            assert not isinstance(exc.value, ValueError)
+        if isinstance(fault, int):
+            assert exc.value.status_code == fault
+        assert len(handler.fetches) == 2
+        assert all(
+            path in ("/v1/table/test/describe/", "/v1/table/test/branches/list/", ROUTE)
+            for _, path, _ in handler.calls
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_range_sdk_structured_400_preserves_value_error(use_async):
+    handler = RangeServer(fault="structured400")
+    with mock_lancedb_connection(handler) as db:
+        table = api(db, use_async)
+        with pytest.raises(ValueError, match="range exceeds blob size"):
+            await fetch(table, [(10, 0, 1)] * 1025, use_async)
+        assert len(handler.fetches) == 2
+
+
+@pytest.mark.asyncio
+async def test_range_sdk_async_cancel_releases_only_request_permit():
+    handler = RangeServer(fault="stall")
+    with mock_lancedb_connection(handler, blob_request_concurrency=1) as db:
+        table = api(db, True)
+        pending = asyncio.create_task(fetch(table, [(10, 0, 1)] * 2049, True))
+        try:
+            assert await asyncio.to_thread(handler.stalled.wait, 10)
+            pending.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await pending
+            assert len(handler.fetches) == 2
+            result = await asyncio.wait_for(fetch(table, [(10, 0, 1)], True), 5)
+            assert result.to_pylist() == [payload(10, 0, 1, 8)]
+            assert len(handler.fetches) == 3
+        finally:
+            pending.cancel()
+            handler.release.set()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_range_sdk_integer_validation_and_empty_zero_io(use_async):
+    handler = RangeServer()
+    with mock_lancedb_connection(handler) as db:
+        table = api(db, use_async)
+        before = len(handler.calls)
+        assert (await fetch(table, [], use_async)).to_pylist() == []
+        for request in [
+            (-1, 0, 1),
+            (2**64, 0, 1),
+            (10, -1, 1),
+            (10, 0, -1),
+            (10, 0, 2**64),
+            (10, 0.5, 1),
+            (20, 2**64 - 1, 1),
+            (10, 0),
+        ]:
+            with pytest.raises((ValueError, OverflowError, TypeError)):
+                await fetch(table, [request], use_async)
+        assert len(handler.calls) == before
+
+
+def test_range_sdk_loaded_artifact_identity():
+    artifact = Path(native.__file__).resolve()
+    digest = hashlib.sha256()
+    with artifact.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    print(f"loaded native artifact={artifact} sha256={digest.hexdigest()}")
+    assert artifact.is_file()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize(
+    "fault", ["missing-identity", "wrong-identity", "recreated", "gone"]
+)
+@pytest.mark.parametrize("huge", [False, True])
+async def test_range_sdk_original_identity_survives_every_partition(
+    use_async, fault, huge
+):
+    handler = RangeServer(fault=fault)
+    with mock_lancedb_connection(handler) as db:
+        table = api(db, use_async)
+        requests = [(10, 0, 64 * MIB + 1)] if huge else [(10, 0, 1)] * 1025
+        error = (
+            ValueError
+            if fault == "recreated"
+            else TableNotFoundError
+            if fault == "gone"
+            else NotImplementedError
+        )
+        with pytest.raises(
+            error, match="snapshot identity" if fault != "gone" else None
+        ):
+            await fetch(table, requests, use_async)
+        assert handler.identities == [None, IDENTITY]
+        assert len(handler.fetches) == 2
+        assert [body["version"] for body in handler.fetches] == [None, 7]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize("fault", ["first-missing-identity", "first-invalid-identity"])
+async def test_range_sdk_bulk_requires_identity_before_payload_acceptance(
+    use_async, fault
+):
+    handler = RangeServer(fault=fault)
+    with mock_lancedb_connection(handler) as db:
+        table = api(db, use_async)
+        before = len(handler.calls)
+        with pytest.raises(NotImplementedError, match="snapshot identity"):
+            await fetch(table, [(10, 0, 1)] * 1025, use_async)
+        assert len(handler.fetches) == 1
+        assert len(handler.calls) == before + 1
@@ -5466,6 +5466,7 @@ def blob_remote_table(
    *,
    server_version=Version("0.5.0"),
    location=None,
+    range_capability=None,
    version_header="1",
    framing="file",
    requests=None,
@@ -5487,6 +5488,8 @@ def blob_remote_table(
            request.send_response(200)
            request.send_header("Content-Type", "application/json")
            request.send_header("phalanx-version", str(server_version))
+            if range_capability is not None:
+                request.send_header("x-lancedb-blob-ranges", range_capability)
            request.end_headers()
            response = dict(BLOB_DESCRIBE_RESPONSE)
            response["version"] = body.get("version") or state["version"]
@@ -5594,6 +5597,42 @@ def blob_remote_table(
            writer_factory = pa.ipc.new_file if framing == "file" else pa.ipc.new_stream
            with writer_factory(request.wfile, response_table.schema) as writer:
                writer.write_table(response_table, max_chunksize=1)
+        elif (
+            range_capability == "1"
+            and request.path == "/v1/table/test/fetch_blob_ranges/"
+        ):
+            size = int(request.headers["Content-Length"])
+            body = json.loads(request.rfile.read(size))
+            payloads = {10: b"alpha", 20: None, 30: b"gamma", 40: b""}
+            values = []
+            for item in body["ranges"]:
+                row, offset, length = (
+                    item[key] for key in ("row_id", "offset", "length")
+                )
+                error = None
+                if row not in payloads:
+                    error = "only 1 exist"
+                elif payloads[row] is not None and offset + length > len(payloads[row]):
+                    error = "range exceeds blob size"
+                if error:
+                    request.send_response(400)
+                    request.end_headers()
+                    request.wfile.write(error.encode())
+                    return
+                payload = payloads[row]
+                values.append(
+                    None if payload is None else payload[offset : offset + length]
+                )
+            response_table = pa.table(
+                {"value": pa.array(values, type=pa.large_binary())}
+            )
+            request.send_response(200)
+            request.send_header("Content-Type", "application/vnd.apache.arrow.stream")
+            request.send_header("x-lancedb-version", "1")
+            request.send_header("x-lancedb-query-snapshot", "v1:" + "a" * 64)
+            request.end_headers()
+            with pa.ipc.new_stream(request.wfile, response_table.schema) as writer:
+                writer.write_table(response_table)
        elif request.path == "/v1/table/test/fetch_blobs/":
            content_len = int(request.headers.get("Content-Length", 0))
            body = json.loads(request.rfile.read(content_len))
@@ -5901,8 +5940,11 @@ def test_sync_remote_blob_pandas_avoids_storage_access(

@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
-async def test_remote_fetch_blob_ranges(use_async):
-    with blob_remote_table() as remote_table:
+@pytest.mark.parametrize("range_capability", [None, "0", "2", "1"])
+async def test_remote_fetch_blob_ranges(use_async, range_capability):
+    with blob_remote_table(
+        server_version="1.13.0-beta.0", range_capability=range_capability
+    ) as remote_table:
        table = remote_table._table if use_async else remote_table
        requests = [
            (30, 2, 3),