From 44ba9c6dee530516cd00cb4e86477715c6a57ca1 Mon Sep 17 00:00:00 2001 From: "lancedb-gatefixer[bot]" <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 13:43:55 -0700 Subject: [PATCH] fix: support remote blob range reads (#4553) --- python/python/lancedb/table.py | 5 +- python/python/tests/test_remote_db.py | 46 +++++++- rust/lancedb/src/remote/table.rs | 20 ++-- rust/lancedb/src/remote/table/blobs.rs | 151 ++++++++++++++++++++++++- rust/lancedb/src/table.rs | 2 + 5 files changed, 210 insertions(+), 14 deletions(-) diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index f0cb05427..a1a54ee8e 100644 --- a/python/python/lancedb/table.py +++ b/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 diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index 6ee41aa0c..e9768b0bf 100644 --- a/python/python/tests/test_remote_db.py +++ b/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: diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 57843ca55..d21583c91 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -2884,12 +2884,10 @@ impl BaseTable for RemoteTable { async fn fetch_blob_ranges( &self, - _column: &str, - _requests: &[crate::blob::BlobRangeRequest], + column: &str, + requests: &[crate::blob::BlobRangeRequest], ) -> Result { - 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 diff --git a/rust/lancedb/src/remote/table/blobs.rs b/rust/lancedb/src/remote/table/blobs.rs index fdcb8f440..e7219280c 100644 --- a/rust/lancedb/src/remote/table/blobs.rs +++ b/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 RemoteTable { 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 { + let row_ids = requests + .iter() + .map(|request| request.row_id) + .collect::>(); + 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::>(); + 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; diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index c6193d750..802c2f517 100644 --- a/rust/lancedb/src/table.rs +++ b/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