mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-14 15:52:26 +00:00
fix(python): normalize blob row ID inputs and errors (#4293)
This commit is contained in:
@@ -527,6 +527,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
|
||||
|
||||
@@ -556,6 +558,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.
|
||||
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
|
||||
|
||||
@@ -342,6 +342,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
|
||||
|
||||
|
||||
@@ -2817,6 +2817,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("adds a LargeBinary Arrow column to a blob table", async () => {
|
||||
const db = await connect(tmpDir.name);
|
||||
const schema = new Schema([new Field("id", new Int64()), blob("image")]);
|
||||
|
||||
@@ -536,6 +536,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.
|
||||
* 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
|
||||
@@ -553,6 +555,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,
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import operator
|
||||
from collections.abc import Awaitable, Callable, Iterable
|
||||
from typing import TYPE_CHECKING, Optional, Union
|
||||
|
||||
@@ -432,16 +433,30 @@ def _leaf_struct_column(tbl: pa.Table, path: str) -> pa.StructArray:
|
||||
|
||||
|
||||
def _normalize_blob_row_ids(
|
||||
row_ids: Union[list[int], pa.Table], blob_column: str
|
||||
row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table], blob_column: str
|
||||
) -> list[int]:
|
||||
if isinstance(row_ids, pa.Table):
|
||||
return read_row_ids_from_hits(row_ids, blob_column)
|
||||
if isinstance(row_ids, (pa.Array, pa.ChunkedArray)):
|
||||
raise ValueError(
|
||||
"pass a query table with _rowid, not a column array "
|
||||
"(use fetch_blobs('image', hits), not fetch_blobs('image', hits['image']))"
|
||||
)
|
||||
return list(row_ids)
|
||||
values = read_row_ids_from_hits(row_ids, blob_column)
|
||||
elif isinstance(row_ids, (pa.Array, pa.ChunkedArray)):
|
||||
if not pa.types.is_integer(row_ids.type):
|
||||
raise ValueError(
|
||||
"row id arrays must have an integer type; pass a query table with "
|
||||
"_rowid when using a blob descriptor column"
|
||||
)
|
||||
values = row_ids.to_pylist()
|
||||
else:
|
||||
values = list(row_ids)
|
||||
|
||||
normalized = []
|
||||
for value in values:
|
||||
try:
|
||||
row_id = operator.index(value)
|
||||
except TypeError as exc:
|
||||
raise ValueError("row ids must be integers without nulls") from exc
|
||||
if row_id < 0:
|
||||
raise ValueError("row ids must be non-negative")
|
||||
normalized.append(row_id)
|
||||
return normalized
|
||||
|
||||
|
||||
def _wrap_blob_files(handles: Iterable[object]) -> list[Optional[BlobFile]]:
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from datetime import date, datetime, timedelta
|
||||
from decimal import Decimal
|
||||
from typing import Dict, List, Optional, Tuple, Any, TypedDict, Union, Literal
|
||||
from typing import Dict, List, Optional, Sequence, Tuple, Any, TypedDict, Union, Literal
|
||||
from uuid import UUID
|
||||
|
||||
import pyarrow as pa
|
||||
@@ -563,7 +563,7 @@ class Table:
|
||||
async def fetch_blob_ranges(
|
||||
self,
|
||||
column: str,
|
||||
requests: List[Tuple[int, int, int]],
|
||||
requests: Sequence[Sequence[int]],
|
||||
) -> pa.LargeBinaryArray: ...
|
||||
async def fetch_blob_files(
|
||||
self, column: str, row_ids: list[int]
|
||||
|
||||
@@ -1154,7 +1154,9 @@ class RemoteTable(Table):
|
||||
return LOOP.run(self._table.blob_columns())
|
||||
|
||||
def fetch_blobs(
|
||||
self, column: str, row_ids: Union[list[int], pa.Table]
|
||||
self,
|
||||
column: str,
|
||||
row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
|
||||
) -> pa.LargeBinaryArray:
|
||||
return LOOP.run(self._table.fetch_blobs(column, row_ids))
|
||||
|
||||
@@ -1162,7 +1164,9 @@ class RemoteTable(Table):
|
||||
return LOOP.run(self._table.fetch_blob_ranges(column, list(requests)))
|
||||
|
||||
def fetch_blob_files(
|
||||
self, column: str, row_ids: Union[list[int], pa.Table]
|
||||
self,
|
||||
column: str,
|
||||
row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
|
||||
) -> "list[Optional[BlobFile]]":
|
||||
return LOOP.run(self._table.fetch_blob_files(column, row_ids))
|
||||
|
||||
|
||||
@@ -2058,15 +2058,20 @@ class Table(ABC):
|
||||
|
||||
@abstractmethod
|
||||
def fetch_blobs(
|
||||
self, column: str, row_ids: Union[list[int], pa.Table]
|
||||
self,
|
||||
column: str,
|
||||
row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
|
||||
) -> pa.LargeBinaryArray:
|
||||
"""Materialize full blob bytes for ``column`` at the given rows.
|
||||
|
||||
The result has the same length and order as ``row_ids``. Null blobs
|
||||
produce null slots; valid empty blobs produce ``b""``.
|
||||
|
||||
``_rowid`` values stay valid after compaction when the table has stable
|
||||
row ids.
|
||||
``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.
|
||||
|
||||
Remote servers limit each request to 1024 row IDs and 64 MiB of blob
|
||||
bytes. The client splits requests automatically and reads an individual
|
||||
@@ -2078,17 +2083,20 @@ class Table(ABC):
|
||||
def fetch_blob_ranges(
|
||||
self,
|
||||
column: str,
|
||||
requests: Sequence[Tuple[int, int, int]],
|
||||
requests: Sequence[Sequence[int]],
|
||||
) -> pa.LargeBinaryArray:
|
||||
"""Materialize row-specific byte ranges from a blob v2 column.
|
||||
|
||||
Each request is a ``(row_id, offset, length)`` tuple. Requests may be
|
||||
repeated or reordered, including multiple ranges for the same blob.
|
||||
Each request is a three-item ``(row_id, offset, length)`` sequence.
|
||||
Requests may be repeated or reordered, including multiple ranges for
|
||||
the same blob.
|
||||
The result has the same length and order as ``requests``; null blobs
|
||||
produce null slots and empty ranges on non-null blobs produce ``b""``.
|
||||
|
||||
``_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.
|
||||
@@ -2096,18 +2104,22 @@ class Table(ABC):
|
||||
|
||||
@abstractmethod
|
||||
def fetch_blob_files(
|
||||
self, column: str, row_ids: Union[list[int], pa.Table]
|
||||
self,
|
||||
column: str,
|
||||
row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
|
||||
) -> "list[Optional[BlobFile]]":
|
||||
"""Open lazy, seekable :class:`~lancedb._blob.BlobFile` handles.
|
||||
|
||||
Prefer this over :meth:`fetch_blobs` for large payloads. ``row_ids`` is
|
||||
a ``list[int]`` or a query ``pyarrow.Table`` carrying row identity via
|
||||
``_rowid`` or a ``_lance_row_id`` field on the blob descriptor. Null
|
||||
rows are ``None``. Remote tables require LanceDB Cloud server 0.5.0 or
|
||||
newer.
|
||||
a ``list[int]``, an integer Arrow array or chunked array, or a query
|
||||
``pyarrow.Table`` carrying row identity via ``_rowid`` or a
|
||||
``_lance_row_id`` field on the blob descriptor. Null blob values return
|
||||
``None``. Remote tables require LanceDB Cloud server 0.5.0 or newer.
|
||||
|
||||
``_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
|
||||
@@ -3013,19 +3025,23 @@ class LanceTable(Table):
|
||||
return LOOP.run(self._table.blob_columns())
|
||||
|
||||
def fetch_blobs(
|
||||
self, column: str, row_ids: Union[list[int], pa.Table]
|
||||
self,
|
||||
column: str,
|
||||
row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
|
||||
) -> pa.LargeBinaryArray:
|
||||
return LOOP.run(self._table.fetch_blobs(column, row_ids))
|
||||
|
||||
def fetch_blob_ranges(
|
||||
self,
|
||||
column: str,
|
||||
requests: Sequence[Tuple[int, int, int]],
|
||||
requests: Sequence[Sequence[int]],
|
||||
) -> pa.LargeBinaryArray:
|
||||
return LOOP.run(self._table.fetch_blob_ranges(column, list(requests)))
|
||||
|
||||
def fetch_blob_files(
|
||||
self, column: str, row_ids: Union[list[int], pa.Table]
|
||||
self,
|
||||
column: str,
|
||||
row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
|
||||
) -> "list[Optional[BlobFile]]":
|
||||
return LOOP.run(self._table.fetch_blob_files(column, row_ids))
|
||||
|
||||
@@ -7079,7 +7095,9 @@ class AsyncTable:
|
||||
return await self._inner.blob_columns()
|
||||
|
||||
async def fetch_blobs(
|
||||
self, column: str, row_ids: Union[list[int], pa.Table]
|
||||
self,
|
||||
column: str,
|
||||
row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
|
||||
) -> pa.LargeBinaryArray:
|
||||
return await self._inner.fetch_blobs(
|
||||
column, _normalize_blob_row_ids(row_ids, column)
|
||||
@@ -7088,12 +7106,14 @@ class AsyncTable:
|
||||
async def fetch_blob_ranges(
|
||||
self,
|
||||
column: str,
|
||||
requests: Sequence[Tuple[int, int, int]],
|
||||
requests: Sequence[Sequence[int]],
|
||||
) -> pa.LargeBinaryArray:
|
||||
return await self._inner.fetch_blob_ranges(column, list(requests))
|
||||
|
||||
async def fetch_blob_files(
|
||||
self, column: str, row_ids: Union[list[int], pa.Table]
|
||||
self,
|
||||
column: str,
|
||||
row_ids: Union[list[int], pa.Array, pa.ChunkedArray, pa.Table],
|
||||
) -> "list[Optional[BlobFile]]":
|
||||
handles = await self._inner.fetch_blob_files(
|
||||
column, _normalize_blob_row_ids(row_ids, column)
|
||||
|
||||
@@ -70,21 +70,25 @@ def _row_ids_by_id(table):
|
||||
|
||||
def _assert_missing_blob_row_ids(exc_info):
|
||||
message = str(exc_info.value)
|
||||
assert "row ids" in message
|
||||
assert "row ids but some do not exist in the table" in message
|
||||
assert "rowaddr" not in message
|
||||
assert "fragment" not in message
|
||||
|
||||
|
||||
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():
|
||||
@@ -861,6 +865,38 @@ def test_fetch_blobs_accepts_query_result():
|
||||
assert {blobs[i].as_py() for i in range(len(blobs))} == {b"gamma"}
|
||||
|
||||
|
||||
def test_fetch_blob_apis_accept_integer_arrow_row_ids():
|
||||
table = _blob_table(
|
||||
"arrow_row_ids",
|
||||
[{"id": 1, "image": b"alpha"}, {"id": 2, "image": b"beta"}],
|
||||
)
|
||||
by_id = _row_ids_by_id(table)
|
||||
ids = [by_id[2], by_id[1]]
|
||||
|
||||
for row_ids in (
|
||||
pa.array(ids, type=pa.uint64()),
|
||||
pa.chunked_array([ids[:1], ids[1:]], type=pa.int64()),
|
||||
):
|
||||
assert table.fetch_blobs("image", row_ids).to_pylist() == [b"beta", b"alpha"]
|
||||
files = table.fetch_blob_files("image", row_ids)
|
||||
assert [file.read() for file in files] == [b"beta", b"alpha"]
|
||||
|
||||
|
||||
def test_fetch_blob_apis_validate_arrow_and_negative_row_ids():
|
||||
table = _blob_table("invalid_arrow_row_ids", [{"id": 1, "image": b"alpha"}])
|
||||
hits = table.search().with_row_id(True).to_arrow()
|
||||
|
||||
for fetch in (table.fetch_blobs, table.fetch_blob_files):
|
||||
with pytest.raises(ValueError, match="non-negative"):
|
||||
fetch("image", [-1])
|
||||
with pytest.raises(ValueError, match="non-negative"):
|
||||
fetch("image", pa.array([-1], type=pa.int64()))
|
||||
with pytest.raises(ValueError, match="without nulls"):
|
||||
fetch("image", pa.array([None], type=pa.int64()))
|
||||
with pytest.raises(ValueError, match="integer type"):
|
||||
fetch("image", hits["image"])
|
||||
|
||||
|
||||
def test_fetch_blobs_after_compact_with_stable_row_ids(tmp_path):
|
||||
db = lancedb.connect(
|
||||
tmp_path, storage_options={"new_table_enable_stable_row_ids": "true"}
|
||||
@@ -992,6 +1028,15 @@ def test_fetch_blob_ranges_aligns_repeated_ranges_and_nulls():
|
||||
assert ranges.to_pylist() == [b"cde", None, b"ab", b"cde", b""]
|
||||
|
||||
|
||||
def test_fetch_blob_ranges_accepts_three_item_sequences():
|
||||
table = _blob_table("range_sequences", [{"id": 1, "image": b"abcdef"}])
|
||||
row_id = _row_ids_by_id(table)[1]
|
||||
|
||||
ranges = table.fetch_blob_ranges("image", [[row_id, 1, 2], (row_id, 4, 2)])
|
||||
|
||||
assert ranges.to_pylist() == [b"bc", b"ef"]
|
||||
|
||||
|
||||
def test_fetch_blob_ranges_validates_requests():
|
||||
table = _blob_table("range_validation", [{"id": 1, "image": b"abc"}])
|
||||
row_id = _row_ids_by_id(table)[1]
|
||||
@@ -1006,6 +1051,30 @@ def test_fetch_blob_ranges_validates_requests():
|
||||
table.fetch_blob_ranges("image", [(2**64 - 1, 0, 1)])
|
||||
_assert_missing_blob_row_ids(exc_info)
|
||||
|
||||
for index, name in enumerate(("row id", "offset", "length")):
|
||||
request = [row_id, 0, 1]
|
||||
request[index] = -1
|
||||
with pytest.raises(
|
||||
ValueError, match=f"blob range {name} must be a non-negative"
|
||||
):
|
||||
table.fetch_blob_ranges("image", [request])
|
||||
|
||||
with pytest.raises(ValueError, match="three values"):
|
||||
table.fetch_blob_ranges("image", [[row_id, 0]])
|
||||
|
||||
|
||||
def test_fetch_blob_apis_reject_row_past_fragment_end():
|
||||
table = _blob_table(
|
||||
"past_fragment_end",
|
||||
[
|
||||
{"id": 1, "image": b"a"},
|
||||
{"id": 2, "image": b"b"},
|
||||
{"id": 3, "image": b"c"},
|
||||
],
|
||||
)
|
||||
_assert_fetch_apis_reject_missing_row_ids(table, [3])
|
||||
_assert_fetch_apis_reject_missing_row_ids(table, [3, 0])
|
||||
|
||||
|
||||
def test_fetch_blob_apis_reject_missing_fragment_row_addr():
|
||||
table = _blob_table("missing_frag", [{"id": 1, "image": b"x"}])
|
||||
@@ -1023,6 +1092,19 @@ def test_fetch_blob_apis_reject_deleted_row_ids():
|
||||
_assert_fetch_apis_reject_missing_row_ids(table, [by_id[2], by_id[1]])
|
||||
|
||||
|
||||
def test_fetch_blob_apis_reject_deleted_stable_row_ids(tmp_path):
|
||||
db = lancedb.connect(
|
||||
tmp_path, storage_options={"new_table_enable_stable_row_ids": "true"}
|
||||
)
|
||||
schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
|
||||
table = db.create_table("stable_deleted", schema=schema)
|
||||
table.add([{"id": 1, "image": b"one"}, {"id": 2, "image": b"two"}])
|
||||
by_id = _row_ids_by_id(table)
|
||||
table.delete("id = 2")
|
||||
|
||||
_assert_fetch_apis_reject_missing_row_ids(table, [by_id[2], by_id[1]])
|
||||
|
||||
|
||||
def test_fetch_blob_ranges_empty_requests_returns_empty_array():
|
||||
table = _blob_table("range_empty", [{"id": 1, "image": b"x"}])
|
||||
assert table.fetch_blob_ranges("image", []).to_pylist() == []
|
||||
|
||||
+21
-5
@@ -1462,14 +1462,30 @@ impl Table {
|
||||
pub fn fetch_blob_ranges(
|
||||
self_: PyRef<'_, Self>,
|
||||
column: String,
|
||||
requests: Vec<(u64, u64, u64)>,
|
||||
requests: Vec<Vec<i128>>,
|
||||
) -> PyResult<Bound<'_, PyAny>> {
|
||||
let inner = self_.inner_ref()?.clone();
|
||||
let requests = requests
|
||||
.into_iter()
|
||||
.map(|values| {
|
||||
let [row_id, offset, length] = <[i128; 3]>::try_from(values).map_err(|_| {
|
||||
PyValueError::new_err("each blob range request must have three values")
|
||||
})?;
|
||||
let to_u64 = |value, name| {
|
||||
u64::try_from(value).map_err(|_| {
|
||||
PyValueError::new_err(format!(
|
||||
"blob range {name} must be a non-negative 64-bit integer"
|
||||
))
|
||||
})
|
||||
};
|
||||
Ok(BlobRangeRequest::new(
|
||||
to_u64(row_id, "row id")?,
|
||||
to_u64(offset, "offset")?,
|
||||
to_u64(length, "length")?,
|
||||
))
|
||||
})
|
||||
.collect::<PyResult<Vec<_>>>()?;
|
||||
future_into_py(self_.py(), async move {
|
||||
let requests = requests
|
||||
.into_iter()
|
||||
.map(|(row_id, offset, length)| BlobRangeRequest::new(row_id, offset, length))
|
||||
.collect::<Vec<_>>();
|
||||
let blobs: LargeBinaryArray = inner
|
||||
.fetch_blob_ranges(column, requests)
|
||||
.await
|
||||
|
||||
+119
-39
@@ -399,45 +399,115 @@ pub(crate) fn ensure_blob_v2_column(
|
||||
}
|
||||
}
|
||||
|
||||
fn ensure_all_row_ids_resolved(column: &str, requested: usize, resolved: usize) -> Result<()> {
|
||||
if requested == resolved {
|
||||
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{detail}; pass row ids collected from this table"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// 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(Error::InvalidInput {
|
||||
message: format!(
|
||||
"blob read for column '{column}' requested {requested} row ids but only {resolved} \
|
||||
exist in the table; pass row ids collected from this table"
|
||||
),
|
||||
})
|
||||
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()
|
||||
),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Lance take reports a missing physical row address as NotSupported or InvalidInput.
|
||||
fn map_blob_take_error(column: &str, requested: usize, err: lance::Error) -> Error {
|
||||
let missing_row_addr = match &err {
|
||||
lance::Error::NotSupported { source, .. } => {
|
||||
source.to_string().contains("must not target deleted rows")
|
||||
}
|
||||
lance::Error::InvalidInput { source, .. } => source
|
||||
.to_string()
|
||||
.contains("belongs to non-existent fragment"),
|
||||
_ => false,
|
||||
};
|
||||
/// Lance reports missing physical and stable row ids through several error paths
|
||||
/// without a shared typed 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")
|
||||
|| (message.contains("Invalid read params ") && message.contains("addressable rows"))
|
||||
|| message.contains("Could not resolve all requested row IDs");
|
||||
|
||||
if missing_row_addr {
|
||||
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"
|
||||
),
|
||||
}
|
||||
if missing_row_id {
|
||||
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()
|
||||
}
|
||||
@@ -454,18 +524,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 {
|
||||
@@ -488,14 +565,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 {
|
||||
@@ -518,11 +598,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))
|
||||
|
||||
@@ -1259,7 +1259,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;
|
||||
@@ -1303,7 +1304,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;
|
||||
@@ -1344,7 +1346,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;
|
||||
|
||||
@@ -501,17 +501,23 @@ async fn collect_id_rowid(table: &Table) -> Result<Vec<(i64, u64)>> {
|
||||
fn assert_missing_blob_row_ids(err: &Error) {
|
||||
assert!(matches!(err, Error::InvalidInput { .. }), "got {err:?}");
|
||||
let message = err.to_string();
|
||||
assert!(message.contains("row ids"), "{message}");
|
||||
assert!(
|
||||
message.contains("row ids but some do not exist in the table"),
|
||||
"{message}"
|
||||
);
|
||||
assert!(!message.contains("rowaddr"), "{message}");
|
||||
assert!(!message.contains("fragment"), "{message}");
|
||||
}
|
||||
|
||||
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(
|
||||
@@ -524,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(())
|
||||
}
|
||||
|
||||
@@ -790,6 +797,30 @@ async fn fetch_blobs_out_of_range_id_errors_without_panic() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fetch_blob_apis_reject_row_past_fragment_end() -> 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, 3], &[Some(b"a"), Some(b"b"), Some(b"c")])
|
||||
.await?;
|
||||
|
||||
assert_fetch_apis_reject_missing_row_ids(&table, &[3]).await?;
|
||||
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();
|
||||
@@ -831,6 +862,23 @@ async fn fetch_blob_apis_reject_deleted_row_ids() -> Result<()> {
|
||||
assert_fetch_apis_reject_missing_row_ids(&table, &[deleted_row_addr, live_row_addr]).await
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fetch_blob_apis_reject_deleted_stable_row_ids() -> Result<()> {
|
||||
let tmp = tempdir().unwrap();
|
||||
let db = connect(tmp.path().to_str().unwrap())
|
||||
.storage_option(OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, "true")
|
||||
.execute()
|
||||
.await?;
|
||||
let table = create_inline_blob_table(&db, "t", &[1, 2], &[Some(b"one"), Some(b"two")]).await?;
|
||||
let pairs = collect_id_rowid(&table).await?;
|
||||
let deleted_row_id = pairs.iter().find(|(id, _)| *id == 2).unwrap().1;
|
||||
let live_row_id = pairs.iter().find(|(id, _)| *id == 1).unwrap().1;
|
||||
|
||||
table.delete("id = 2").await?;
|
||||
|
||||
assert_fetch_apis_reject_missing_row_ids(&table, &[deleted_row_id, live_row_id]).await
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fetch_blobs_rejects_non_blob_column() -> Result<()> {
|
||||
let tmp = tempdir().unwrap();
|
||||
|
||||
Reference in New Issue
Block a user