diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index a8c382c27..17e2d4f34 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -548,6 +548,10 @@ Bytes for `column` at row IDs from [Query.withRowId](Query.md#withrowid). Reads the table's current checkout. IDs from another version can fail after compaction unless stable row ids are enabled. Results keep input order and duplicates. Null blobs are `null`. Empty blobs are empty buffers. +Remote servers limit each request to 1024 row IDs and 64 MiB of blob bytes. +The client splits requests automatically and reads an individual larger +blob through the Range route. This method still materializes all bytes in +memory; use [Table.fetchBlobFiles](Table.md#fetchblobfiles) for large values. #### Parameters diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index cabcf88aa..132e8bc20 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -537,6 +537,10 @@ export abstract class Table { * Reads the table's current checkout. IDs from another version can fail after * compaction unless stable row ids are enabled. Results keep input order and * duplicates. Null blobs are `null`. Empty blobs are empty buffers. + * Remote servers limit each request to 1024 row IDs and 64 MiB of blob bytes. + * The client splits requests automatically and reads an individual larger + * blob through the Range route. This method still materializes all bytes in + * memory; use {@link Table.fetchBlobFiles} for large values. */ abstract fetchBlobs( column: string, diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 08149c434..661fff75c 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -2049,8 +2049,10 @@ class Table(ABC): ``_rowid`` values stay valid after compaction when the table has stable row ids. - Convenience for small payloads. For large values use - :meth:`fetch_blob_files`. + Remote servers limit each request to 1024 row IDs and 64 MiB of blob + bytes. The client splits requests automatically and reads an individual + larger blob through the Range route. This method still materializes all + bytes in memory; for large values use :meth:`fetch_blob_files`. """ @abstractmethod diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index cabddbe7b..195c18ea5 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -5187,6 +5187,192 @@ mod tests { assert_eq!(blobs.value(2), b"gamma"); } + #[tokio::test] + async fn test_fetch_blobs_splits_row_ids_at_one_version_and_preserves_order() { + let request_sizes = Arc::new(std::sync::Mutex::new(Vec::new())); + let seen = request_sizes.clone(); + let table = Table::new_with_handler_version( + "my_table", + semver::Version::new(0, 5, 0), + move |request| { + if request.url().path() == "/v1/table/my_table/describe/" { + return http::Response::builder() + .status(200) + .body(br#"{"version":7,"schema":{"fields":[]}}"#.to_vec()) + .unwrap(); + } + assert_eq!(request.url().path(), "/v1/table/my_table/fetch_blobs/"); + let body = request_body_json(&request); + assert_eq!(body["version"], 7); + let ids = body["row_ids"].as_array().unwrap(); + seen.lock().unwrap().push(ids.len()); + if ids.len() > 1024 { + return http::Response::builder() + .status(400) + .body(b"fetch_blobs accepts at most 1024 row IDs".to_vec()) + .unwrap(); + } + let mut builder = LargeBinaryBuilder::new(); + for id in ids { + let id = id.as_u64().unwrap(); + if id == 1023 { + builder.append_null(); + } else { + builder.append_value(id.to_string().as_bytes()); + } + } + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new( + "image", + DataType::LargeBinary, + true, + )])), + vec![Arc::new(builder.finish())], + ) + .unwrap(); + http::Response::builder() + .status(200) + .header(CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE) + .body(write_ipc_stream_uncompressed(&batch)) + .unwrap() + }, + ); + + let ids: Vec = (0..1024).chain([42]).collect(); + let blobs = table.fetch_blobs("image", &ids).await.unwrap(); + assert_eq!(blobs.len(), 1025); + assert_eq!(blobs.value(0), b"0"); + assert_eq!(blobs.value(1022), b"1022"); + assert!(blobs.is_null(1023)); + assert_eq!(blobs.value(1024), b"42"); + assert_eq!(request_sizes.lock().unwrap().as_slice(), &[1024, 1]); + } + + #[tokio::test] + async fn test_fetch_blobs_splits_byte_limited_requests_and_reads_large_blob_by_range() { + // Simulate a lower byte cap so this test exercises the same 400 response + // without allocating 64 MiB of blob data. + let requests = Arc::new(std::sync::Mutex::new(Vec::new())); + let seen = requests.clone(); + let table = Table::new_with_handler_version( + "my_table", + semver::Version::new(0, 5, 0), + move |request| { + let path = request.url().path(); + if path == "/v1/table/my_table/describe/" { + return http::Response::builder() + .status(200) + .body(br#"{"version":42,"schema":{"fields":[]}}"#.to_vec()) + .unwrap(); + } + if path == "/v1/table/my_table/fetch_blobs/" { + let body = request_body_json(&request); + let ids = body["row_ids"].as_array().unwrap(); + if body["version"].is_null() { + // Only the initial failed request may read live latest. + assert_eq!(ids.len(), 5); + } else { + assert_eq!(body["version"], 42); + } + seen.lock().unwrap().push(format!("POST {}", ids.len())); + let mut builder = LargeBinaryBuilder::new(); + let mut total_bytes = 0; + for id in ids { + let value: Option<&[u8]> = match id.as_u64().unwrap() { + 10 => Some(b"aaaa"), + 20 => Some(b"bbb"), + 30 => None, + 40 => Some(b"0123456789"), + id => panic!("unexpected row id {id}"), + }; + if let Some(value) = value { + total_bytes += value.len(); + builder.append_value(value); + } else { + builder.append_null(); + } + } + if total_bytes > 6 { + return http::Response::builder() + .status(400) + .body(br#"{"error":"Bad request: fetch_blobs accepts at most 67108864 total blob bytes"}"#.to_vec()) + .unwrap(); + } + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new( + "image", + DataType::LargeBinary, + true, + )])), + vec![Arc::new(builder.finish())], + ) + .unwrap(); + return http::Response::builder() + .status(200) + .header(CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE) + .body(write_ipc_stream_uncompressed(&batch)) + .unwrap(); + } + assert_eq!(path, "/v1/table/my_table/blob/image/40/bytes"); + assert!(request.url().query().unwrap().contains("version=42")); + let range = request + .headers() + .get(reqwest::header::RANGE) + .unwrap() + .to_str() + .unwrap(); + seen.lock().unwrap().push(format!("GET {range}")); + match range { + "bytes=0-0" => http::Response::builder() + .status(206) + .header(reqwest::header::CONTENT_RANGE, "bytes 0-0/10") + .body(b"0".to_vec()) + .unwrap(), + "bytes=0-" => http::Response::builder() + .status(206) + .header(reqwest::header::CONTENT_RANGE, "bytes 0-9/10") + .body(b"0123456789".to_vec()) + .unwrap(), + _ => panic!("unexpected range: {range}"), + } + }, + ); + + let blobs = table + .fetch_blobs("image", &[10, 20, 30, 40, 20]) + .await + .unwrap(); + assert_eq!(blobs.len(), 5); + assert_eq!(blobs.value(0), b"aaaa"); + assert_eq!(blobs.value(1), b"bbb"); + assert!(blobs.is_null(2)); + assert_eq!(blobs.value(3), b"0123456789"); + assert_eq!(blobs.value(4), b"bbb"); + let requests = requests.lock().unwrap(); + assert!(requests.contains(&"GET bytes=0-0".to_string())); + assert!(requests.contains(&"GET bytes=0-".to_string())); + } + + #[tokio::test] + async fn test_fetch_blobs_does_not_split_unrelated_bad_requests() { + let requests = Arc::new(AtomicUsize::new(0)); + let seen = requests.clone(); + let table = + Table::new_with_handler_version("my_table", semver::Version::new(0, 5, 0), move |_| { + seen.fetch_add(1, Ordering::SeqCst); + http::Response::builder() + .status(400) + .body(b"unknown blob column".to_vec()) + .unwrap() + }); + + assert_fetch_blobs_http_error( + table.fetch_blobs("missing", &[10, 20]).await.unwrap_err(), + "unknown blob column", + ); + assert_eq!(requests.load(Ordering::SeqCst), 1); + } + fn table_with_fetch_blobs_response(body: Vec) -> Table { table_with_fetch_blobs_content_type(Some(ARROW_STREAM_CONTENT_TYPE), body) } diff --git a/rust/lancedb/src/remote/table/blobs.rs b/rust/lancedb/src/remote/table/blobs.rs index 597b41782..29023eeed 100644 --- a/rust/lancedb/src/remote/table/blobs.rs +++ b/rust/lancedb/src/remote/table/blobs.rs @@ -21,7 +21,22 @@ use crate::error::Result; use crate::remote::client::{HttpSend, RequestResultExt, RestfulLanceDbClient}; use crate::table::BaseTable; -use super::{FreshnessHeaders, FreshnessState, RemoteTable, freshness_headers_snapshot}; +use super::{ + FreshnessHeaders, FreshnessState, ReadSnapshot, RemoteTable, freshness_headers_snapshot, +}; + +// The Cloud route rejects larger row-id lists. Its separate 64 MiB byte limit +// is handled by splitting a rejected request below. +const MAX_FETCH_BLOBS_ROW_IDS: usize = 1024; + +fn is_fetch_blobs_byte_limit_error(error: &Error) -> bool { + matches!(error, Error::Http { + source, + status_code: Some(StatusCode::BAD_REQUEST), + .. + } if source.to_string().contains("fetch_blobs accepts at most") + && source.to_string().contains("total blob bytes")) +} #[derive(Debug, Clone, Copy)] enum RangeRequestMode { @@ -369,7 +384,62 @@ impl RemoteTable { message: "fetch_blobs is not supported on this LanceDB Cloud server".into(), }); } - let read_snapshot = self.snapshot_read_state().await; + let mut read_snapshot = self.snapshot_read_state().await; + // A selection spanning requests must use one exact dataset version. + // Resolve latest before the first chunk; a checked-out version is exact already. + if row_ids.len() > MAX_FETCH_BLOBS_ROW_IDS && read_snapshot.version.is_none() { + read_snapshot.version = Some(self.describe_read_snapshot(read_snapshot).await?.version); + } + let mut pending: Vec<&[u64]> = row_ids.chunks(MAX_FETCH_BLOBS_ROW_IDS).rev().collect(); + let mut chunks = Vec::new(); + while let Some(ids) = pending.pop() { + match self.fetch_blobs_chunk(column, ids, read_snapshot).await { + Ok(blobs) => chunks.push(blobs), + Err(error) if is_fetch_blobs_byte_limit_error(&error) => { + // A single-request call may turn into several requests after a + // byte-cap error. No bytes from the failed request were used. + if read_snapshot.version.is_none() { + read_snapshot.version = + Some(self.describe_read_snapshot(read_snapshot).await?.version); + } + if ids.len() == 1 { + // The whole-byte route cannot serve this blob. The Range route + // has no aggregate response limit and preserves null alignment. + let mut files = self + .fetch_blob_files_with_snapshot(column, ids, read_snapshot) + .await?; + let blob = match files.pop().unwrap() { + Some(file) => Some(file.read().await?), + None => None, + }; + chunks.push(LargeBinaryArray::from(vec![blob.as_deref()])); + } else { + let mid = ids.len() / 2; + pending.push(&ids[mid..]); + pending.push(&ids[..mid]); + } + } + Err(error) => return Err(error), + } + } + + if chunks.len() == 1 { + return Ok(chunks.pop().unwrap()); + } + let chunk_refs: Vec<&dyn Array> = chunks.iter().map(|chunk| chunk as &dyn Array).collect(); + Ok(arrow::compute::concat(&chunk_refs)? + .as_any() + .downcast_ref::() + .expect("concatenating LargeBinary arrays returns LargeBinary") + .clone()) + } + + async fn fetch_blobs_chunk( + &self, + column: &str, + row_ids: &[u64], + read_snapshot: ReadSnapshot, + ) -> Result { let mut body = serde_json::json!({ "version": read_snapshot.version, "column": column, @@ -460,6 +530,16 @@ impl RemoteTable { } let read_snapshot = self.snapshot_read_state().await; + self.fetch_blob_files_with_snapshot(column, row_ids, read_snapshot) + .await + } + + async fn fetch_blob_files_with_snapshot( + &self, + column: &str, + row_ids: &[u64], + read_snapshot: ReadSnapshot, + ) -> Result>> { let encoded_column = urlencoding::encode(column); let requesters = row_ids .iter() diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index ba4adc294..7bcc7b5d6 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -1253,7 +1253,10 @@ impl Table { /// Materialize blob bytes for the given row ids. /// /// Output matches `row_ids` in length and order. Null blobs are null; - /// valid empty blobs contain empty byte strings. Prefer + /// valid empty blobs contain empty byte strings. Cloud limits individual + /// requests to 1024 row ids and 64 MiB of blob bytes; the remote client + /// splits requests and reads a single larger blob through the Range route. + /// This method still materializes all bytes in memory, so prefer /// [`Self::fetch_blob_files`] for large selections. /// /// `_rowid` values stay valid after compaction when the table has stable