From 3f079b7b3253064822e98da10b250e663f1b5543 Mon Sep 17 00:00:00 2001 From: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Date: Thu, 24 Sep 2026 00:34:53 +0000 Subject: [PATCH] fix: identify missing blob row IDs in batch errors --- docs/src/js/classes/Table.md | 4 + docs/src/python/python.md | 2 + nodejs/__test__/table.test.ts | 12 +++ nodejs/lancedb/table.ts | 4 + python/python/lancedb/table.py | 6 ++ python/python/tests/test_blob.py | 4 + rust/lancedb/src/blob.rs | 124 +++++++++++++++++++++---- rust/lancedb/src/table.rs | 9 +- rust/lancedb/tests/blob_integration.rs | 16 ++++ 9 files changed, 158 insertions(+), 23 deletions(-) diff --git a/docs/src/js/classes/Table.md b/docs/src/js/classes/Table.md index a8c382c27..825cdc120 100644 --- a/docs/src/js/classes/Table.md +++ b/docs/src/js/classes/Table.md @@ -524,6 +524,8 @@ table's current checkout. Preserves input order, duplicates, and nulls. Use this for large payloads. See [Table.fetchBlobs](Table.md#fetchblobs) for row-ID validity across versions. +One missing or deleted row ID rejects the entire batch, including on remote +tables; no partial handles are returned. #### Parameters @@ -548,6 +550,8 @@ 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. +One missing or deleted row ID rejects the entire batch, including on remote +tables; no partial results are returned. #### Parameters diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 68cd306c4..847085ab0 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -322,6 +322,8 @@ pylance, LanceDB uses a matching `lance.blob.v2` extension type so blob columns still work. Queries return descriptors. Call [`fetch_blob_files`][lancedb.table.Table.fetch_blob_files] for lazy reads or [`fetch_blobs`][lancedb.table.Table.fetch_blobs] for eager bytes. +One missing or deleted row ID rejects the entire blob-read batch; the APIs do +not return partial results. ::: lancedb.blob diff --git a/nodejs/__test__/table.test.ts b/nodejs/__test__/table.test.ts index ac8904c16..0206d6b1d 100644 --- a/nodejs/__test__/table.test.ts +++ b/nodejs/__test__/table.test.ts @@ -2468,6 +2468,18 @@ describe("when dealing with blob columns", () => { expect(Buffer.from(await files[1]!.read()).toString()).toBe("beta"); }); + it("rejects a whole blob read batch when one row ID was deleted", async () => { + const { table, rowIds } = await openBlobTable(); + await table.delete("id = 2"); + + await expect(table.fetchBlobs("image", rowIds)).rejects.toThrow( + `first missing row ids: [${rowIds[1]}]`, + ); + await expect(table.fetchBlobFiles("image", rowIds)).rejects.toThrow( + `first missing row ids: [${rowIds[1]}]`, + ); + }); + it("reads a half-open range", async () => { const { table, rowIds } = await openBlobTable(); const files = await table.fetchBlobFiles("image", rowIds); diff --git a/nodejs/lancedb/table.ts b/nodejs/lancedb/table.ts index a098322a9..4966c0b48 100644 --- a/nodejs/lancedb/table.ts +++ b/nodejs/lancedb/table.ts @@ -526,6 +526,8 @@ 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. + * One missing or deleted row ID rejects the entire batch, including on remote + * tables; no partial results are returned. */ abstract fetchBlobs( column: string, @@ -538,6 +540,8 @@ export abstract class Table { * * Preserves input order, duplicates, and nulls. Use this for large payloads. * See {@link Table.fetchBlobs} for row-ID validity across versions. + * One missing or deleted row ID rejects the entire batch, including on remote + * tables; no partial handles are returned. */ abstract fetchBlobFiles( column: string, diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 6585eecfb..d068f44b6 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -1937,6 +1937,8 @@ class Table(ABC): ``row_ids`` may be a list, an integer Arrow array or chunked array, or a query table with row ids. ``_rowid`` values stay valid after compaction when the table has stable row ids. + One missing or deleted row ID rejects the entire batch; no partial + results are returned. Convenience for small payloads. For large values use :meth:`fetch_blob_files`. @@ -1958,6 +1960,8 @@ class Table(ABC): ``_rowid`` values stay valid after compaction when the table has stable row ids. + 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. @@ -1979,6 +1983,8 @@ class Table(ABC): ``_rowid`` values stay valid after compaction when the table has stable row ids. + One missing or deleted row ID rejects the entire batch; no partial + handles are returned. """ @abstractmethod diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py index 5e11627ef..6f0caf80b 100644 --- a/python/python/tests/test_blob.py +++ b/python/python/tests/test_blob.py @@ -74,15 +74,19 @@ def _assert_missing_blob_row_ids(exc_info): def _assert_fetch_apis_reject_missing_row_ids(table, row_ids): + expected = f"first missing row ids: [{row_ids[0]}]" with pytest.raises(ValueError) as exc_info: table.fetch_blobs("image", row_ids) _assert_missing_blob_row_ids(exc_info) + assert expected in str(exc_info.value) with pytest.raises(ValueError) as exc_info: table.fetch_blob_files("image", row_ids) _assert_missing_blob_row_ids(exc_info) + assert expected in str(exc_info.value) with pytest.raises(ValueError) as exc_info: table.fetch_blob_ranges("image", [(row_id, 0, 1) for row_id in row_ids]) _assert_missing_blob_row_ids(exc_info) + assert expected in str(exc_info.value) def test_blob_factory_declares_v2_field(): diff --git a/rust/lancedb/src/blob.rs b/rust/lancedb/src/blob.rs index 382211285..4da4fc9c4 100644 --- a/rust/lancedb/src/blob.rs +++ b/rust/lancedb/src/blob.rs @@ -368,25 +368,91 @@ pub(crate) fn ensure_blob_v2_column( } } -fn missing_blob_row_ids_error(column: &str, requested: usize) -> Error { +fn missing_blob_row_ids_error(column: &str, requested: usize, missing: &[u64]) -> Error { + let detail = if missing.is_empty() { + String::new() + } else { + format!("; first missing row ids: {missing:?}") + }; Error::InvalidInput { message: format!( "blob read for column '{column}' requested {requested} row ids but some \ - do not exist in the table; pass row ids collected from this table" + do not exist in the table{detail}; pass row ids collected from this table" ), } } -fn ensure_all_row_ids_resolved(column: &str, requested: usize, resolved: usize) -> Result<()> { - if requested == resolved { +/// Diagnose only after a failed read. Stable row IDs use Lance's deletion-aware +/// index; physical row IDs are fragment/offset pairs. Limit the error to three +/// distinct IDs so a large failed batch stays readable. +async fn find_missing_blob_row_ids(dataset: &Dataset, row_ids: &[u64]) -> Result> { + let mut missing = Vec::new(); + if let Some(index) = lance::dataset::rowids::get_row_id_index(dataset).await? { + for (row_id, address) in row_ids.iter().zip(index.get_many(row_ids)?) { + if address.is_none() && !missing.contains(row_id) { + missing.push(*row_id); + if missing.len() == 3 { + break; + } + } + } + } else { + let mut fragments = std::collections::HashMap::new(); + for &row_id in row_ids { + let fragment_id = row_id >> 32; + let offset = row_id as u32; + if let std::collections::hash_map::Entry::Vacant(entry) = fragments.entry(fragment_id) { + let fragment = dataset.get_fragment(fragment_id as usize); + let state = if let Some(fragment) = fragment { + Some(( + fragment.physical_rows().await?, + fragment.get_deletion_vector().await?, + )) + } else { + None + }; + entry.insert(state); + } + let is_missing = match fragments.get(&fragment_id) { + Some(Some((physical_rows, deletions))) => { + offset as usize >= *physical_rows + || deletions + .as_ref() + .is_some_and(|deletions| deletions.contains(offset)) + } + Some(None) => true, + None => unreachable!(), + }; + if is_missing && !missing.contains(&row_id) { + missing.push(row_id); + if missing.len() == 3 { + break; + } + } + } + } + Ok(missing) +} + +async fn ensure_all_row_ids_resolved( + dataset: &Dataset, + column: &str, + row_ids: &[u64], + resolved: usize, +) -> Result<()> { + if row_ids.len() == resolved { return Ok(()); } - if resolved < requested { - Err(missing_blob_row_ids_error(column, requested)) + if resolved < row_ids.len() { + let missing = find_missing_blob_row_ids(dataset, row_ids) + .await + .unwrap_or_default(); + Err(missing_blob_row_ids_error(column, row_ids.len(), &missing)) } else { Err(Error::Runtime { message: format!( - "blob read for column '{column}' returned {resolved} results for {requested} row ids" + "blob read for column '{column}' returned {resolved} results for {} row ids", + row_ids.len() ), }) } @@ -394,7 +460,12 @@ fn ensure_all_row_ids_resolved(column: &str, requested: usize, resolved: usize) /// Lance reports missing physical and stable row ids through several error paths /// without a shared typed error. -fn map_blob_take_error(column: &str, requested: usize, err: lance::Error) -> Error { +async fn map_blob_take_error( + dataset: &Dataset, + column: &str, + row_ids: &[u64], + err: lance::Error, +) -> Error { let message = err.to_string(); let missing_row_id = message.contains("must not target deleted rows") || message.contains("belongs to non-existent fragment") @@ -402,7 +473,10 @@ fn map_blob_take_error(column: &str, requested: usize, err: lance::Error) -> Err || message.contains("Could not resolve all requested row IDs"); if missing_row_id { - missing_blob_row_ids_error(column, requested) + let missing = find_missing_blob_row_ids(dataset, row_ids) + .await + .unwrap_or_default(); + missing_blob_row_ids_error(column, row_ids.len(), &missing) } else { err.into() } @@ -419,18 +493,25 @@ pub(crate) async fn take_blob_ranges_aligned( return Ok(LargeBinaryBuilder::new().finish()); } + let row_ids = requests + .iter() + .map(|request| request.row_id) + .collect::>(); let lance_requests = requests .iter() .map(|request| LanceBlobRangeRequest::new(request.row_id, request.offset, request.length)) .collect::>(); - let payloads = dataset + let payloads = match dataset .read_blob_ranges(column)? .with_row_ids(lance_requests) .preserve_order(true) .execute() .await - .map_err(|err| map_blob_take_error(column, requests.len(), err))?; - ensure_all_row_ids_resolved(column, requests.len(), payloads.len())?; + { + Ok(payloads) => payloads, + Err(err) => return Err(map_blob_take_error(dataset, column, &row_ids, err).await), + }; + ensure_all_row_ids_resolved(dataset, column, &row_ids, payloads.len()).await?; let mut builder = LargeBinaryBuilder::new(); for payload in payloads { @@ -453,14 +534,17 @@ pub(crate) async fn take_blobs_aligned( return Ok(LargeBinaryBuilder::new().finish()); } - let payloads = dataset + let payloads = match dataset .read_blobs(column)? .with_row_ids(row_ids.to_vec()) .preserve_order(true) .execute() .await - .map_err(|err| map_blob_take_error(column, row_ids.len(), err))?; - ensure_all_row_ids_resolved(column, row_ids.len(), payloads.len())?; + { + Ok(payloads) => payloads, + Err(err) => return Err(map_blob_take_error(dataset, column, row_ids, err).await), + }; + ensure_all_row_ids_resolved(dataset, column, row_ids, payloads.len()).await?; let mut builder = LargeBinaryBuilder::new(); for payload in payloads { @@ -483,11 +567,11 @@ pub(crate) async fn take_blob_files_aligned( return Ok(Vec::new()); } - let handles = dataset - .take_blobs(row_ids, column) - .await - .map_err(|err| map_blob_take_error(column, row_ids.len(), err))?; - ensure_all_row_ids_resolved(column, row_ids.len(), handles.len())?; + let handles = match dataset.take_blobs(row_ids, column).await { + Ok(handles) => handles, + Err(err) => return Err(map_blob_take_error(dataset, column, row_ids, err).await), + }; + ensure_all_row_ids_resolved(dataset, column, row_ids, handles.len()).await?; Ok(handles .into_iter() .map(|handle| handle.map(Into::into)) diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index fbec73dfb..eed32635b 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -1230,7 +1230,8 @@ impl Table { /// [`Self::fetch_blob_files`] for large selections. /// /// `_rowid` values stay valid after compaction when the table has stable - /// row ids. + /// row ids. One missing or deleted row id rejects the whole request; no + /// partial results are returned. /// /// ``` /// use arrow_array::UInt64Array; @@ -1274,7 +1275,8 @@ impl Table { /// non-null blobs produce empty byte strings. /// /// `_rowid` values stay valid after compaction when the table has stable - /// row ids. + /// row ids. One missing or deleted row id rejects the whole request; no + /// partial results are returned. /// /// ``` /// use lancedb::blob::BlobRangeRequest; @@ -1315,7 +1317,8 @@ impl Table { /// read from disk until a call to [`BlobFile::read`]. /// /// `_rowid` values stay valid after compaction when the table has stable - /// row ids. + /// row ids. One missing or deleted row id rejects the whole request; no + /// partial handles are returned. /// /// ``` /// # use lancedb::Table; diff --git a/rust/lancedb/tests/blob_integration.rs b/rust/lancedb/tests/blob_integration.rs index 4a14d2732..0c9942db6 100644 --- a/rust/lancedb/tests/blob_integration.rs +++ b/rust/lancedb/tests/blob_integration.rs @@ -510,11 +510,14 @@ fn assert_missing_blob_row_ids(err: &Error) { } async fn assert_fetch_apis_reject_missing_row_ids(table: &Table, row_ids: &[u64]) -> Result<()> { + let expected = format!("first missing row ids: [{}]", row_ids[0]); let err = table.fetch_blobs("image", row_ids).await.unwrap_err(); assert_missing_blob_row_ids(&err); + assert!(err.to_string().contains(&expected), "{err}"); let err = table.fetch_blob_files("image", row_ids).await.unwrap_err(); assert_missing_blob_row_ids(&err); + assert!(err.to_string().contains(&expected), "{err}"); let err = table .fetch_blob_ranges( @@ -527,6 +530,7 @@ async fn assert_fetch_apis_reject_missing_row_ids(table: &Table, row_ids: &[u64] .await .unwrap_err(); assert_missing_blob_row_ids(&err); + assert!(err.to_string().contains(&expected), "{err}"); Ok(()) } @@ -805,6 +809,18 @@ async fn fetch_blob_apis_reject_row_past_fragment_end() -> Result<()> { assert_fetch_apis_reject_missing_row_ids(&table, &[3, 0]).await } +#[tokio::test] +async fn fetch_blobs_identifies_multiple_missing_ids_without_partial_results() -> Result<()> { + let tmp = tempdir().unwrap(); + let db = connect(tmp.path().to_str().unwrap()).execute().await?; + let table = create_inline_blob_table(&db, "t", &[1, 2], &[Some(b"a"), Some(b"b")]).await?; + + let err = table.fetch_blobs("image", &[0, 2, 3]).await.unwrap_err(); + assert_missing_blob_row_ids(&err); + assert!(err.to_string().contains("first missing row ids: [2, 3]")); + Ok(()) +} + #[tokio::test] async fn fetch_blob_files_rejects_missing_fragment_row_addr() -> Result<()> { let tmp = tempdir().unwrap();