mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-12 23:02:28 +00:00
fix: identify missing blob row IDs in batch errors
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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():
|
||||
|
||||
+104
-20
@@ -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<Vec<u64>> {
|
||||
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::<Vec<_>>();
|
||||
let lance_requests = requests
|
||||
.iter()
|
||||
.map(|request| LanceBlobRangeRequest::new(request.row_id, request.offset, request.length))
|
||||
.collect::<Vec<_>>();
|
||||
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))
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user