mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-12 14:52:26 +00:00
fix: handle remote blob fetch request caps (#4309)
## Cause Remote `fetch_blobs` sent every row ID in one Cloud request, exceeding the server's 1024-row or 64 MiB byte cap. A single blob over the byte cap could not be read through this method. ## Fix - Send at most 1024 row IDs per request and split batches further when the server reports the blob-byte cap, preserving input order and nulls. - Resolve one exact dataset version before multi-request reads. If an initial single request hits the byte cap, resolve the version before any successful split or Range read; use that version for every resulting request. - Read an individual oversized blob through the existing Range route. - Document the per-request caps and transparent handling in the Rust, Python, and TypeScript APIs and generated JS docs. ## Validation - Focused Rust `fetch_blobs` tests: 26 passed, including row-count, byte-cap, Range fallback, and exact-version assertions. - Rust clippy and formatting passed. - Focused Python blob tests: 13 passed; Ruff lint and formatting passed. - Node build, lint, docs generation, and focused `fetchBlobs` test passed. Fixes #4275 <!-- lance-gatekeeper-fix:v1 agent=6ae94d49bf1d4cae057bba72584e7b90 generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
This commit is contained in:
lancedb-gatefixer[bot]
@@ -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
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -1963,8 +1963,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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -1251,7 +1251,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
|
||||
|
||||
Reference in New Issue
Block a user