mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-13 14:16:54 +00:00
fix: expose stable row IDs in Function errors (#4841)
Function error records expose physical row locations, which can change when a table is compacted. Expose the server-provided optional stable `row_id` through Rust and Python so callers can locate the same row after compaction. Document stable-ID retries and retain physical addresses for tables without stable row IDs. Older server responses continue to deserialize with `row_id=None`. The Python remote response test and Rust remote deserialization test passed on verification SDK `4e1725fa352035e2079dfd235b8271cd05adf433`, including decoding a legacy response without the new field. This branch applies the same SDK patch to current main; intervening main changes leave the affected error types and assertions unchanged. Formatting and lint checks passed after correcting the retry examples to use the `table.name` property. In a separate Linux server experiment, a real skipped Function row retained `_rowid=506` after compaction changed its physical address. The SDK field requires a server that emits stable row IDs; this PR does not change server dependencies.
This commit is contained in:
@@ -1006,6 +1006,7 @@ class FunctionErrorRecord:
|
||||
job_id: str
|
||||
fragment_id: int
|
||||
row_offset: Optional[int]
|
||||
row_id: Optional[int]
|
||||
row_address: Optional[int]
|
||||
column: str
|
||||
function: str
|
||||
|
||||
@@ -3301,9 +3301,31 @@ class Table(ABC):
|
||||
``records``, ``fragments`` and ``truncated``, the last saying
|
||||
whether the listing stopped at its limit.
|
||||
|
||||
Each record identifies its row with ``fragment_id``, ``row_offset``
|
||||
and the physical ``row_address`` (``(fragment_id << 32) |
|
||||
row_offset``). To retry the recorded rows, collect the non-null
|
||||
Each record identifies its row with ``fragment_id``, ``row_offset``,
|
||||
the physical ``row_address`` (``(fragment_id << 32) |
|
||||
row_offset``), and, when the table uses stable row IDs, the stable
|
||||
``row_id`` read from the refresh's pinned table version. Prefer
|
||||
``row_id`` for retries: compaction and rewrite can change the
|
||||
physical fields, while a stable row ID survives them.
|
||||
|
||||
To retry the recorded rows with stable row IDs:
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
errors = table.function_errors(column="embedding")
|
||||
row_ids = sorted(
|
||||
record.row_id
|
||||
for record in errors.records
|
||||
if record.row_id is not None
|
||||
)
|
||||
if row_ids:
|
||||
row_id_list = ", ".join(str(row_id) for row_id in row_ids)
|
||||
database.execute_query(
|
||||
f"REFRESH COLUMN embedding ON {table.name} "
|
||||
f"WHERE _rowid IN ({row_id_list})"
|
||||
)
|
||||
|
||||
For a table without stable row IDs, collect the non-null physical
|
||||
addresses before the table is compacted or rewritten and run the
|
||||
SQL refresh:
|
||||
|
||||
@@ -3318,7 +3340,7 @@ class Table(ABC):
|
||||
if addresses:
|
||||
address_list = ", ".join(str(address) for address in addresses)
|
||||
database.execute_query(
|
||||
f"REFRESH COLUMN embedding ON {table.name()} "
|
||||
f"REFRESH COLUMN embedding ON {table.name} "
|
||||
f"WHERE _rowaddr IN ({address_list})"
|
||||
)
|
||||
|
||||
@@ -7786,9 +7808,31 @@ class AsyncTable:
|
||||
``records``, ``fragments`` and ``truncated``, the last saying
|
||||
whether the listing stopped at its limit.
|
||||
|
||||
Each record identifies its row with ``fragment_id``, ``row_offset``
|
||||
and the physical ``row_address`` (``(fragment_id << 32) |
|
||||
row_offset``). To retry the recorded rows, collect the non-null
|
||||
Each record identifies its row with ``fragment_id``, ``row_offset``,
|
||||
the physical ``row_address`` (``(fragment_id << 32) |
|
||||
row_offset``), and, when the table uses stable row IDs, the stable
|
||||
``row_id`` read from the refresh's pinned table version. Prefer
|
||||
``row_id`` for retries: compaction and rewrite can change the
|
||||
physical fields, while a stable row ID survives them.
|
||||
|
||||
To retry the recorded rows with stable row IDs:
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
errors = await table.function_errors(column="embedding")
|
||||
row_ids = sorted(
|
||||
record.row_id
|
||||
for record in errors.records
|
||||
if record.row_id is not None
|
||||
)
|
||||
if row_ids:
|
||||
row_id_list = ", ".join(str(row_id) for row_id in row_ids)
|
||||
await database.execute_query(
|
||||
f"REFRESH COLUMN embedding ON {table.name} "
|
||||
f"WHERE _rowid IN ({row_id_list})"
|
||||
)
|
||||
|
||||
For a table without stable row IDs, collect the non-null physical
|
||||
addresses before the table is compacted or rewritten and run the
|
||||
SQL refresh:
|
||||
|
||||
@@ -7803,7 +7847,7 @@ class AsyncTable:
|
||||
if addresses:
|
||||
address_list = ", ".join(str(address) for address in addresses)
|
||||
await database.execute_query(
|
||||
f"REFRESH COLUMN embedding ON {table.name()} "
|
||||
f"REFRESH COLUMN embedding ON {table.name} "
|
||||
f"WHERE _rowaddr IN ({address_list})"
|
||||
)
|
||||
|
||||
|
||||
@@ -2047,6 +2047,7 @@ def test_remote_function_errors_lists_the_rows_a_refresh_skipped():
|
||||
"job_id": "j-7",
|
||||
"fragment_id": 3,
|
||||
"row_offset": 9,
|
||||
"row_id": 42,
|
||||
"row_address": 12884901897,
|
||||
"column": "embedding",
|
||||
"function": "embed",
|
||||
@@ -2098,6 +2099,7 @@ def test_remote_function_errors_lists_the_rows_a_refresh_skipped():
|
||||
assert errors.truncated is True
|
||||
assert [r.error_message for r in errors.records] == ["bad input 'x'"]
|
||||
assert errors.records[0].row_offset == 9
|
||||
assert errors.records[0].row_id == 42
|
||||
assert errors.records[0].row_address == 12884901897
|
||||
assert errors.records[0].function_version == "2"
|
||||
assert repr(errors.records[0]) == (
|
||||
|
||||
@@ -614,6 +614,7 @@ pub struct FunctionErrorRecord {
|
||||
pub job_id: String,
|
||||
pub fragment_id: u64,
|
||||
pub row_offset: Option<u32>,
|
||||
pub row_id: Option<u64>,
|
||||
pub row_address: Option<u64>,
|
||||
pub column: String,
|
||||
pub function: String,
|
||||
@@ -653,6 +654,7 @@ impl From<lancedb::function::FunctionErrorRecord> for FunctionErrorRecord {
|
||||
job_id: record.job_id,
|
||||
fragment_id: record.fragment_id,
|
||||
row_offset: record.row_offset,
|
||||
row_id: record.row_id,
|
||||
row_address: record.row_address,
|
||||
column: record.column,
|
||||
function: record.function,
|
||||
|
||||
@@ -1043,6 +1043,13 @@ pub struct FunctionErrorRecord {
|
||||
/// detail was capped and only the fragment summary remains.
|
||||
#[serde(default)]
|
||||
pub row_offset: Option<u32>,
|
||||
/// Stable row identity at the table version the refresh read.
|
||||
///
|
||||
/// Present only when that version used stable row IDs. It remains valid
|
||||
/// across later compaction and rewrite, unlike `row_address`. `None` on
|
||||
/// legacy tables and on records written before the server reported it.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub row_id: Option<u64>,
|
||||
/// Portable physical row address, `(fragment_id << 32) | row_offset`.
|
||||
///
|
||||
/// `None` when the record covers a fragment rather than one row, or when
|
||||
|
||||
@@ -11026,7 +11026,7 @@ mod tests {
|
||||
.status(200)
|
||||
.body(
|
||||
r#"{"records": [{"job_id": "j-7", "fragment_id": 3, "row_offset": 9,
|
||||
"row_address": 12884901897,
|
||||
"row_id": 42, "row_address": 12884901897,
|
||||
"column": "embedding", "function": "embed", "function_version": "2",
|
||||
"table_version": 11, "error_type": "ValueError",
|
||||
"error_message": "bad input 'x'", "created_at_millis": 1700000000000}],
|
||||
@@ -11052,6 +11052,7 @@ mod tests {
|
||||
job_id: "j-7".into(),
|
||||
fragment_id: 3,
|
||||
row_offset: Some(9),
|
||||
row_id: Some(42),
|
||||
row_address: Some(12_884_901_897),
|
||||
column: "embedding".into(),
|
||||
function: "embed".into(),
|
||||
@@ -11073,7 +11074,8 @@ mod tests {
|
||||
);
|
||||
assert!(errors.truncated);
|
||||
|
||||
// Servers before row_address omit it; the record still decodes.
|
||||
// Servers before row_address or row_id omit them; the record still
|
||||
// decodes.
|
||||
let legacy: FunctionErrorRecord = serde_json::from_value(serde_json::json!({
|
||||
"job_id": "j-6",
|
||||
"fragment_id": 2,
|
||||
@@ -11087,6 +11089,7 @@ mod tests {
|
||||
"created_at_millis": 1_700_000_000_000_i64,
|
||||
}))
|
||||
.unwrap();
|
||||
assert!(legacy.row_id.is_none());
|
||||
assert!(legacy.row_address.is_none());
|
||||
|
||||
// No filter sends no filter, and an empty listing reads as such.
|
||||
|
||||
@@ -2141,6 +2141,10 @@ impl Table {
|
||||
/// capped. Filter by job or column through the request; a listing that
|
||||
/// hit its limit reports [`FunctionErrors::truncated`].
|
||||
///
|
||||
/// A record carries a stable `row_id` when the refresh source used stable
|
||||
/// row IDs. Prefer it over the physical `row_address` for retries:
|
||||
/// compaction and rewrite can change the address.
|
||||
///
|
||||
/// LanceDB Enterprise only, and the caller needs read access
|
||||
/// to the table, since a message carries the value that failed.
|
||||
///
|
||||
|
||||
Reference in New Issue
Block a user