diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 02d9c9e87..351933ed5 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -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 diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 23bc25b97..4724776a1 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -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})" ) diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index 1ee6d464b..8aec4d185 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -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]) == ( diff --git a/python/src/table.rs b/python/src/table.rs index 6ef524310..4ccc76f89 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -614,6 +614,7 @@ pub struct FunctionErrorRecord { pub job_id: String, pub fragment_id: u64, pub row_offset: Option, + pub row_id: Option, pub row_address: Option, pub column: String, pub function: String, @@ -653,6 +654,7 @@ impl From 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, diff --git a/rust/lancedb/src/function.rs b/rust/lancedb/src/function.rs index 8933d0b37..ffa61f158 100644 --- a/rust/lancedb/src/function.rs +++ b/rust/lancedb/src/function.rs @@ -1043,6 +1043,13 @@ pub struct FunctionErrorRecord { /// detail was capped and only the fragment summary remains. #[serde(default)] pub row_offset: Option, + /// 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, /// Portable physical row address, `(fragment_id << 32) | row_offset`. /// /// `None` when the record covers a fragment rather than one row, or when diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 3b241dd3a..b95227db6 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -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. diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 6bc845580..1bd3a711a 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -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. ///