fix: support remote blob range reads (#4553)

python/python/lancedb/table.py

@@ -2190,8 +2190,9 @@ class Table(ABC):
         One missing or deleted row ID rejects the entire batch; no partial
         results are returned.
 
-        Row IDs can be obtained from a query with ``with_row_id(True)``. This
-        API is currently supported only by local tables.
+        Row IDs can be obtained from a query with ``with_row_id(True)``.
+        Remote tables read the requested ranges through version-pinned blob
+        handles using HTTP byte-range requests.
         """
 
     @abstractmethod

python/python/tests/test_remote_db.py

@@ -1623,7 +1623,6 @@ def remote_unsupported_operations():
     return [
         ("uses_v2_manifest_paths", (), {}, "uses_v2_manifest_paths"),
         ("migrate_v2_manifest_paths", (), {}, "migrate_manifest_paths_v2"),
-        ("fetch_blob_ranges", ("image", [(1, 0, 1)]), {}, "fetch_blob_ranges"),
         ("optimize", (), {}, "optimize"),
         ("create_index", ("vector",), {"config": HnswPq()}, "IVF_HNSW_PQ"),
     ]
@@ -3362,13 +3361,20 @@ def blob_remote_table(*, server_version=Version("0.5.0")):
             else:
                 assert body["with_row_id"] is True
                 if columns == ["image"]:
-                    # fetch_blob_files sizes its handles from a descriptor take.
+                    # Blob handles are sized from a descriptor take.
                     assert body["filter"].startswith("_rowid IN")
                     response_table = blob_descriptor_take_table()
+                    row_ids = [
+                        int(value) for value in re.findall(r"\d+", body["filter"])
+                    ]
+                    response_table = response_table.filter(
+                        pa.compute.is_in(response_table["_rowid"], pa.array(row_ids))
+                    )
                 else:
                     response_table = blob_query_response_table()
             request.send_response(200)
             request.send_header("Content-Type", "application/vnd.apache.arrow.file")
+            request.send_header("x-lancedb-version", "1")
             request.end_headers()
             with pa.ipc.new_file(request.wfile, response_table.schema) as writer:
                 writer.write_table(response_table)
@@ -3400,6 +3406,42 @@ def test_remote_blob_columns_and_fetch():
         assert blobs.to_pylist() == [b"alpha", None, b"gamma"]
 
 
+@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:
+        table = remote_table._table if use_async else remote_table
+        requests = [
+            (30, 2, 3),
+            (10, 1, 3),
+            (20, 0, 1),
+            (10, 0, 2),
+            (40, 0, 0),
+            (10, 5, 0),
+            (30, 2, 3),
+        ]
+        result = table.fetch_blob_ranges("image", requests)
+        if use_async:
+            result = await result
+        assert isinstance(result, pa.LargeBinaryArray)
+        assert result.to_pylist() == [b"mma", b"lph", None, b"al", b"", b"", b"mma"]
+
+        for invalid, message in [
+            ([(10, 0, 1), (99, 0, 1)], "only 1 exist"),
+            ([(10, 4, 2)], "exceeds blob size"),
+            ([(20, 2**64 - 1, 1)], "overflows u64"),
+        ]:
+            with pytest.raises(ValueError, match=message):
+                result = table.fetch_blob_ranges("image", invalid)
+                if use_async:
+                    await result
+
+        result = table.fetch_blob_ranges("image", [])
+        if use_async:
+            result = await result
+        assert result.to_pylist() == []
+
+
 @pytest.mark.asyncio
 async def test_async_remote_to_lance_rejects_storage_access(monkeypatch):
     with blob_remote_table() as remote_table:

rust/lancedb/src/remote/table.rs

@@ -2884,12 +2884,10 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
 
     async fn fetch_blob_ranges(
         &self,
-        _column: &str,
-        _requests: &[crate::blob::BlobRangeRequest],
+        column: &str,
+        requests: &[crate::blob::BlobRangeRequest],
     ) -> Result<LargeBinaryArray> {
-        Err(Error::NotSupported {
-            message: "fetch_blob_ranges is not supported for remote tables.".into(),
-        })
+        self.fetch_blob_ranges_impl(column, requests).await
     }
 
     async fn fetch_blob_files(
@@ -5394,6 +5392,14 @@ mod tests {
                 err.to_string()
                     .contains("blobs inside lists cannot be fetched")
             );
+            let err = table
+                .fetch_blob_ranges(column, [crate::blob::BlobRangeRequest::new(10, 0, 1)])
+                .await
+                .unwrap_err();
+            assert!(
+                err.to_string()
+                    .contains("blobs inside lists cannot be fetched")
+            );
         }
     }
 
@@ -5955,10 +5961,6 @@ mod tests {
                 "optimize",
                 table.optimize(OptimizeAction::All).await.map(|_| ()),
             ),
-            (
-                "fetch_blob_ranges",
-                table.fetch_blob_ranges("image", []).await.map(|_| ()),
-            ),
             (
                 "Wand factor",
                 table

rust/lancedb/src/remote/table/blobs.rs

@@ -9,6 +9,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
+use arrow_array::builder::LargeBinaryBuilder;
 use arrow_array::{Array, LargeBinaryArray, StructArray, UInt8Array, UInt64Array};
 use arrow_schema::DataType;
 use bytes::Bytes;
@@ -20,7 +21,7 @@ use reqwest::{Response, StatusCode, header};
 use tokio::sync::{Mutex, OwnedSemaphorePermit};
 
 use crate::Error;
-use crate::blob::BlobFile;
+use crate::blob::{BlobFile, BlobRangeRequest};
 use crate::error::Result;
 use crate::query::{QueryFilter, QueryRequest, Select};
 use crate::remote::client::{HttpSend, RequestResultExt, RestfulLanceDbClient};
@@ -538,6 +539,37 @@ impl<S: HttpSend> RemoteTable<S> {
         Ok(blobs)
     }
 
+    /// Materialize ranges through version-pinned handles, preserving nulls and order.
+    pub(super) async fn fetch_blob_ranges_impl(
+        &self,
+        column: &str,
+        requests: &[BlobRangeRequest],
+    ) -> Result<LargeBinaryArray> {
+        let row_ids = requests
+            .iter()
+            .map(|request| request.row_id)
+            .collect::<Vec<_>>();
+        let files = self.fetch_blob_files_impl(column, &row_ids).await?;
+        let mut builder = LargeBinaryBuilder::with_capacity(requests.len(), 0);
+        for (file, request) in files.into_iter().zip(requests) {
+            let end =
+                request
+                    .offset
+                    .checked_add(request.length)
+                    .ok_or_else(|| Error::InvalidInput {
+                        message: "blob range offset + length overflows u64".into(),
+                    })?;
+            match file {
+                Some(file) => {
+                    let bytes = file.read_range(request.offset..end).await?;
+                    builder.append_value(bytes);
+                }
+                None => builder.append_null(),
+            }
+        }
+        Ok(builder.finish())
+    }
+
     /// Open seekable handles for `row_ids`.
     ///
     /// Descriptors for the requested rows come from one `_rowid` take query per
@@ -1191,6 +1223,123 @@ mod tests {
         (table, requests)
     }
 
+    #[tokio::test]
+    async fn remote_blob_ranges_preserve_order_nulls_and_empty_ranges() {
+        let (table, requests) = mock_take_blob_table(
+            vec![
+                (10, Some((BlobKind::Inline, PAYLOAD.len() as u64))),
+                (20, None),
+                (30, Some((BlobKind::Packed, PAYLOAD.len() as u64))),
+                (40, Some((BlobKind::Inline, 0))),
+            ],
+            Some("7"),
+        );
+
+        let ranges = table
+            .fetch_blob_ranges(
+                "image",
+                &[
+                    BlobRangeRequest::new(30, 6, 5),
+                    BlobRangeRequest::new(10, 0, 3),
+                    BlobRangeRequest::new(20, 0, 1),
+                    BlobRangeRequest::new(10, 2, 2),
+                    BlobRangeRequest::new(40, 0, 0),
+                    BlobRangeRequest::new(10, PAYLOAD.len() as u64, 0),
+                    BlobRangeRequest::new(30, 6, 5),
+                ],
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(
+            ranges,
+            LargeBinaryArray::from(vec![
+                Some(&PAYLOAD[6..11]),
+                Some(&PAYLOAD[0..3]),
+                None,
+                Some(&PAYLOAD[2..4]),
+                Some(b"".as_slice()),
+                Some(b"".as_slice()),
+                Some(&PAYLOAD[6..11]),
+            ])
+        );
+        let requests = requests.lock().unwrap();
+        assert_eq!(
+            requests[0],
+            TakeMockRequest::Query {
+                row_ids: vec![10, 20, 30, 40],
+                version: None,
+            }
+        );
+        // Only nonempty, non-null ranges are read, at the descriptor take's version.
+        let reads = requests[1..]
+            .iter()
+            .map(|request| match request {
+                TakeMockRequest::Range {
+                    row_id,
+                    range,
+                    version,
+                } => {
+                    assert_eq!(version.as_deref(), Some("7"));
+                    (*row_id, range.as_str())
+                }
+                other => panic!("unexpected request: {other:?}"),
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(
+            reads,
+            [
+                (30, "bytes=6-10"),
+                (10, "bytes=0-2"),
+                (10, "bytes=2-3"),
+                (30, "bytes=6-10")
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn remote_blob_ranges_reject_missing_rows_and_invalid_ranges() {
+        for (ranges, message) in [
+            (
+                vec![
+                    BlobRangeRequest::new(10, 0, 1),
+                    BlobRangeRequest::new(99, 0, 1),
+                ],
+                "only 1 exist",
+            ),
+            (vec![BlobRangeRequest::new(10, 35, 2)], "exceeds blob size"),
+            (vec![BlobRangeRequest::new(10, 37, 0)], "exceeds blob size"),
+            (
+                vec![BlobRangeRequest::new(10, u64::MAX, 1)],
+                "overflows u64",
+            ),
+            (
+                vec![BlobRangeRequest::new(20, u64::MAX, 1)],
+                "overflows u64",
+            ),
+        ] {
+            let (table, requests) = mock_take_blob_table(
+                vec![
+                    (10, Some((BlobKind::Inline, PAYLOAD.len() as u64))),
+                    (20, None),
+                ],
+                Some("7"),
+            );
+            let error = table.fetch_blob_ranges("image", &ranges).await.unwrap_err();
+            assert!(error.to_string().contains(message), "{error}");
+            // Missing rows and invalid ranges fail without reading any bytes.
+            assert_eq!(requests.lock().unwrap().len(), 1);
+        }
+    }
+
+    #[tokio::test]
+    async fn remote_blob_ranges_accept_empty_requests_without_http() {
+        let (table, requests) = mock_take_blob_table(vec![], Some("7"));
+        let ranges = table.fetch_blob_ranges("image", &[]).await.unwrap();
+        assert!(ranges.is_empty());
+        assert!(requests.lock().unwrap().is_empty());
+    }
+
     #[tokio::test]
     async fn remote_blob_files_are_sized_from_one_descriptor_take() {
         let payload_size = PAYLOAD.len() as u64;

rust/lancedb/src/table.rs

@@ -1304,6 +1304,8 @@ impl Table {
     /// ranges for the same blob. The output has the same length and order as
     /// the requests. Null blobs produce null output slots; empty ranges on
     /// non-null blobs produce empty byte strings.
+    /// Remote tables read through version-pinned [`BlobFile`] handles using
+    /// HTTP byte-range requests.
     ///
     /// `_rowid` values stay valid after compaction when the table has stable
     /// row ids. One missing or deleted row id rejects the whole request; no