feat: support batched blob range reads (#3703)

## Summary

Lance can now plan multiple byte ranges for the same blob in one
`read_blob_ranges` operation, but LanceDB users currently cannot expose
a complete set of logical ranges to that planner.

This complements `BlobFile`: file-like consumers such as PyAV can
continue to discover ranges dynamically, while callers that already know
the ranges for a batch can submit them together.

## Motivating example

A training table may store a large video blob together with a small
application-level clip index:

```text
video: blob
clips: [{offset, length}, ...]
```

The caller can select the videos and clips for a batch, obtain their row
IDs from the query, and read all of the selected windows together:

```python
rows = (
    table.search()
    .select(["clips"])
    .with_row_id(True)
    .limit(64)
    .to_arrow()
    .to_pylist()
)

requests = []
for row in rows:
    clip = sample_clip(row["clips"])
    requests.append(
        (row["_rowid"], clip["offset"], clip["length"])
    )

chunks = table.fetch_blob_ranges("video", requests)
```

Here, `_rowid` comes from the LanceDB query, while `offset` and `length`
come from the application's clip index and are relative to that row's
video blob. The caller describes only the logical reads; Lance still
handles validation, source grouping, coalescing, scheduling, and byte
backpressure.

Lance v10.0.0-beta.5 returns one logical result per blob selector or
range request and explicitly distinguishes null blobs from valid empty
values. LanceDB consumes that aligned result contract directly and only
adds a cardinality check for unresolved row IDs.

This PR exposes batched blob-range reads on local Rust and Python
tables. Results preserve request identity, duplicates, null slots, and
valid empty ranges while allowing Lance to execute the physical reads
out of order. Scheduler buffer sizing remains an internal Lance concern,
so the LanceDB API does not expose `io_buffer_size`.

Cloud tables continue to report this operation as unsupported until
there is a corresponding remote API.

python/python/tests/test_blob.py

@@ -184,18 +184,75 @@ def test_fetch_blobs_accepts_query_result():
     assert {blobs[i].as_py() for i in range(len(blobs))} == {b"gamma"}
 
 
-def test_fetch_blobs_null_alignment():
+def test_fetch_blobs_preserves_null_and_empty_values():
     table = _blob_table(
         "nulls",
-        [{"id": 1, "image": b"present"}, {"id": 2, "image": None}],
+        [
+            {"id": 1, "image": b"present"},
+            {"id": 2, "image": None},
+            {"id": 3, "image": b""},
+        ],
     )
     by_id = _row_ids_by_id(table)
-    request = [by_id[1], by_id[2], by_id[1]]
+    request = [by_id[1], by_id[2], by_id[3], by_id[1]]
     blobs = table.fetch_blobs("image", request)
     assert len(blobs) == len(request)
     assert blobs[0].as_py() == b"present"
     assert blobs[1].as_py() is None
-    assert blobs[2].as_py() == b"present"
+    assert blobs[2].as_py() == b""
+    assert blobs[3].as_py() == b"present"
+
+
+def test_fetch_blob_ranges_aligns_repeated_ranges_and_nulls():
+    table = _blob_table(
+        "range_alignment",
+        [{"id": 1, "image": b"abcdefghij"}, {"id": 2, "image": None}],
+    )
+    by_id = _row_ids_by_id(table)
+    requests = [
+        (by_id[1], 2, 3),
+        (by_id[2], 0, 0),
+        (by_id[1], 0, 2),
+        (by_id[1], 2, 3),
+        (by_id[1], 10, 0),
+    ]
+
+    ranges = table.fetch_blob_ranges("image", requests)
+
+    assert ranges.to_pylist() == [b"cde", None, b"ab", b"cde", b""]
+
+
+def test_fetch_blob_ranges_validates_requests():
+    table = _blob_table("range_validation", [{"id": 1, "image": b"abc"}])
+    row_id = _row_ids_by_id(table)[1]
+
+    with pytest.raises(RuntimeError, match="exceeds blob size"):
+        table.fetch_blob_ranges("image", [(row_id, 2, 2)])
+
+    with pytest.raises(RuntimeError, match="offset \\+ length overflowed"):
+        table.fetch_blob_ranges("image", [(row_id, 2**64 - 1, 1)])
+
+    with pytest.raises(ValueError, match="row ids"):
+        table.fetch_blob_ranges("image", [(2**64 - 1, 0, 1)])
+
+
+def test_fetch_blob_ranges_empty_requests_returns_empty_array():
+    table = _blob_table("range_empty", [{"id": 1, "image": b"x"}])
+    assert table.fetch_blob_ranges("image", []).to_pylist() == []
+
+
+@pytest.mark.asyncio
+async def test_async_fetch_blob_ranges():
+    db = await lancedb.connect_async("memory:///")
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = await db.create_table("range_async", schema=schema)
+    await table.add([{"id": 1, "image": b"abcdefghij"}])
+    hits = await table.query().with_row_id().to_arrow()
+    row_id = hits["_rowid"][0].as_py()
+
+    ranges = await table.fetch_blob_ranges("image", [(row_id, 1, 3), (row_id, 6, 2)])
+
+    assert ranges.to_pylist() == [b"bcd", b"gh"]
 
 
 def test_fetch_blobs_nested_path():