Merge remote-tracking branch 'origin/main' into gatekeeper/fix-2325-1

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
 

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,

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

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<u64> = (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<u8>) -> Table {
         table_with_fetch_blobs_content_type(Some(ARROW_STREAM_CONTENT_TYPE), body)
     }

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<S: HttpSend> RemoteTable<S> {
                 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::<LargeBinaryArray>()
+            .expect("concatenating LargeBinary arrays returns LargeBinary")
+            .clone())
+    }
+
+    async fn fetch_blobs_chunk(
+        &self,
+        column: &str,
+        row_ids: &[u64],
+        read_snapshot: ReadSnapshot,
+    ) -> Result<LargeBinaryArray> {
         let mut body = serde_json::json!({
             "version": read_snapshot.version,
             "column": column,
@@ -460,6 +530,16 @@ impl<S: HttpSend> RemoteTable<S> {
         }
 
         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<Vec<Option<BlobFile>>> {
         let encoded_column = urlencoding::encode(column);
         let requesters = row_ids
             .iter()

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