feat: list a table's per-row Function errors from the client (#4208)

A refresh running under a skip policy records each row it skipped, with
the failing input and the error, but the client could not read that
store: the server exposes it over SQL and, since recently, a REST route.
A user who hit per-row failures still had to open a SQL session.

`Table::function_errors` calls the route. The listing is table-addressed
with optional job and column filters, the same addressing the SQL
surface uses, so the two cannot disagree about what a table's errors
are. The two non-record signals come back as their own fields rather
than as rows: capped-fragment summaries, and whether the listing stopped
at its limit. Local tables refuse rather than answer with an empty list.

Python and Node expose the same call, with the same optional filters.

python/python/lancedb/__init__.py

@@ -12,6 +12,9 @@ __version__ = importlib.metadata.version("lancedb")
 
 from ._lancedb import connect as lancedb_connect
 from ._lancedb import FtsToken
+from ._lancedb import FunctionErrorFragment as FunctionErrorFragment
+from ._lancedb import FunctionErrorRecord as FunctionErrorRecord
+from ._lancedb import FunctionErrors as FunctionErrors
 from ._lancedb import LsmWriteSpec
 from ._lancedb import tokenize as _tokenize
 from .common import URI, sanitize_uri

python/python/lancedb/_lancedb.pyi

@@ -461,6 +461,12 @@ class Table:
     ) -> AddColumnsResult: ...
     async def refresh_column(self, column: str) -> RefreshColumnResult: ...
     async def refresh_column_async(self, column: str) -> Job: ...
+    async def function_errors(
+        self,
+        job_id: Optional[str] = None,
+        column: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> FunctionErrors: ...
     async def refresh_materialized_view(
         self, full: bool = False, source_version: Optional[int] = None
     ) -> RefreshMaterializedViewResult: ...
@@ -820,6 +826,29 @@ class RefreshColumnResult:
     rows_filled: int
     version: int
 
+class FunctionErrorRecord:
+    job_id: str
+    fragment_id: int
+    row_offset: Optional[int]
+    column: str
+    function: str
+    function_version: str
+    table_version: int
+    error_type: str
+    error_message: str
+    created_at_millis: int
+
+class FunctionErrorFragment:
+    job_id: str
+    fragment_id: int
+    rows_skipped: int
+    rows_recorded: int
+
+class FunctionErrors:
+    records: list[FunctionErrorRecord]
+    fragments: list[FunctionErrorFragment]
+    truncated: bool
+
 class RefreshMaterializedViewResult:
     @staticmethod
     def from_json(value: str) -> RefreshMaterializedViewResult: ...

python/python/lancedb/remote/table.py

@@ -994,6 +994,16 @@ class RemoteTable(Table):
     def refresh_column_async(self, column: str) -> Job[RefreshColumnResult]:
         return Job(LOOP.run(self._table.refresh_column_async(column)))
 
+    def function_errors(
+        self,
+        job_id: Optional[str] = None,
+        column: Optional[str] = None,
+        limit: Optional[int] = None,
+    ):
+        return LOOP.run(
+            self._table.function_errors(job_id=job_id, column=column, limit=limit)
+        )
+
     def alter_columns(
         self, *alterations: Iterable[Dict[str, str]]
     ) -> AlterColumnsResult:

python/python/lancedb/table.py

@@ -186,6 +186,7 @@ if TYPE_CHECKING:
         CompactionStats,
         Tag,
         AddColumnsResult,
+        FunctionErrors,
         RefreshColumnResult,
         AddResult,
         AlterColumnsResult,
@@ -2308,6 +2309,47 @@ class Table(ABC):
         'finished'
         """
 
+    @abstractmethod
+    def function_errors(
+        self,
+        job_id: Optional[str] = None,
+        column: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> "FunctionErrors":
+        """
+        The per-row errors Function refreshes recorded on this table.
+
+        A refresh running under a skip policy records each row it skipped
+        with the input that failed and the error. This lists those records,
+        newest job first, plus a summary for any fragment whose per-row
+        detail was capped. LanceDB Cloud and Enterprise only; reading errors
+        needs read access to the table, since a message carries the value
+        that failed.
+
+        Parameters
+        ----------
+        job_id: str, optional
+            Only errors recorded by this job.
+        column: str, optional
+            Only errors on this column.
+        limit: int, optional
+            At most this many records (server default 10000, cap 100000).
+
+        Returns
+        -------
+        FunctionErrors
+            ``records``, ``fragments`` and ``truncated``, the last saying
+            whether the listing stopped at its limit.
+
+        Examples
+        --------
+        >>> errors = table.function_errors(column="embedding")  # doctest: +SKIP
+        >>> for record in errors.records:  # doctest: +SKIP
+        ...     print(record.job_id, record.row_offset, record.error_message)
+        >>> if errors.truncated:  # doctest: +SKIP
+        ...     print("listing stopped at the limit")
+        """
+
     @abstractmethod
     def alter_columns(self, *alterations: Iterable[Dict[str, str]]):
         """
@@ -4362,6 +4404,18 @@ class LanceTable(Table):
         """
         return Job(LOOP.run(self._table.refresh_column_async(column)))
 
+    def function_errors(
+        self,
+        job_id: Optional[str] = None,
+        column: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> "FunctionErrors":
+        """The per-row errors Function refreshes recorded on this table. See
+        [`Table.function_errors`][lancedb.table.Table.function_errors]."""
+        return LOOP.run(
+            self._table.function_errors(job_id=job_id, column=column, limit=limit)
+        )
+
     def alter_columns(
         self, *alterations: Iterable[Dict[str, str]]
     ) -> AlterColumnsResult:
@@ -6428,6 +6482,49 @@ class AsyncTable:
         """
         return await self._inner.refresh_column(column)
 
+    async def function_errors(
+        self,
+        job_id: Optional[str] = None,
+        column: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> "FunctionErrors":
+        """
+        The per-row errors Function refreshes recorded on this table.
+
+        A refresh running under a skip policy records each row it skipped
+        with the input that failed and the error. This lists those records,
+        newest job first, plus a summary for any fragment whose per-row
+        detail was capped. LanceDB Cloud and Enterprise only; reading errors
+        needs read access to the table, since a message carries the value
+        that failed.
+
+        Parameters
+        ----------
+        job_id: str, optional
+            Only errors recorded by this job.
+        column: str, optional
+            Only errors on this column.
+        limit: int, optional
+            At most this many records (server default 10000, cap 100000).
+
+        Returns
+        -------
+        FunctionErrors
+            ``records``, ``fragments`` and ``truncated``, the last saying
+            whether the listing stopped at its limit.
+
+        Examples
+        --------
+        >>> errors = await table.function_errors(column="embedding")  # doctest: +SKIP
+        >>> for record in errors.records:  # doctest: +SKIP
+        ...     print(record.job_id, record.row_offset, record.error_message)
+        >>> if errors.truncated:  # doctest: +SKIP
+        ...     print("listing stopped at the limit")
+        """
+        return await self._inner.function_errors(
+            job_id=job_id, column=column, limit=limit
+        )
+
     async def refresh_column_async(
         self, column: str
     ) -> AsyncJob[RefreshColumnJobResult]:

python/python/tests/test_remote_db.py

@@ -982,6 +982,71 @@ def test_remote_refresh_async_returns_typed_terminal_result():
     assert result.version == 8
 
 
+def test_remote_function_errors_lists_the_rows_a_refresh_skipped():
+    listing = {
+        "records": [
+            {
+                "job_id": "j-7",
+                "fragment_id": 3,
+                "row_offset": 9,
+                "column": "embedding",
+                "function": "embed",
+                "function_version": "2",
+                "table_version": 11,
+                "error_type": "ValueError",
+                "error_message": "bad input 'x'",
+                "created_at_millis": 1700000000000,
+            }
+        ],
+        "fragments": [
+            {
+                "job_id": "j-7",
+                "fragment_id": 4,
+                "rows_skipped": 500,
+                "rows_recorded": 100,
+            }
+        ],
+        "truncated": True,
+    }
+    bodies = []
+
+    def handler(request):
+        content_len = int(request.headers.get("Content-Length", 0))
+        body = request.rfile.read(content_len) if content_len > 0 else b""
+        if request.path == "/v1/table/test/errors":
+            bodies.append(json.loads(body))
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(json.dumps(listing).encode())
+        elif request.path == "/v1/table/test/describe/":
+            request.send_response(200)
+            request.send_header("Content-Type", "application/json")
+            request.end_headers()
+            request.wfile.write(
+                json.dumps({"version": 1, "schema": {"fields": []}}).encode()
+            )
+        else:
+            request.send_response(404)
+            request.end_headers()
+
+    with mock_lancedb_connection(handler) as db:
+        table = db.open_table("test")
+        errors = table.function_errors(job_id="j-7", column="embedding", limit=2)
+        everything = table.function_errors()
+
+    assert bodies == [{"job_id": "j-7", "column": "embedding", "limit": 2}, {}]
+    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].function_version == "2"
+    assert (errors.fragments[0].rows_skipped, errors.fragments[0].rows_recorded) == (
+        500,
+        100,
+    )
+    assert everything.truncated is True
+
+
 def test_remote_job_wait_raises_on_failure():
     from lancedb.exceptions import JobFailedError
     from lancedb.index import BTree

python/python/tests/test_table.py

@@ -4327,6 +4327,13 @@ async def test_computed_column_async(tmp_path):
     assert (await table.to_arrow())["tripled"].to_pylist() == [9]
 
 
+def test_function_errors_are_remote_only(tmp_path):
+    db = lancedb.connect(tmp_path)
+    table = db.create_table("t", [{"x": 1}])
+    with pytest.raises(NotImplementedError, match="LanceDB Cloud and Enterprise"):
+        table.function_errors()
+
+
 def test_refresh_column_async_returns_job(tmp_path):
     db = lancedb.connect(tmp_path)
     table = db.create_table("computed_job", [{"x": 1}, {"x": 2}])

python/src/lib.rs

@@ -16,8 +16,9 @@ use query::{FTSQuery, HybridQuery, Query, VectorQuery};
 use session::Session;
 use table::{
     AddColumnsResult, AddResult, AlterColumnsResult, DeleteResult, DropColumnsResult, FtsToken,
-    LsmWriteSpec, MergeResult, PyBlobFile, RefreshColumnResult, RefreshMaterializedViewResult,
-    Table, UpdateFieldMetadataResult, UpdateResult,
+    FunctionErrorFragment, FunctionErrorRecord, FunctionErrors, LsmWriteSpec, MergeResult,
+    PyBlobFile, RefreshColumnResult, RefreshMaterializedViewResult, Table,
+    UpdateFieldMetadataResult, UpdateResult,
 };
 
 pub mod arrow;
@@ -84,6 +85,9 @@ pub fn _lancedb(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<RecordBatchStream>()?;
     m.add_class::<AddColumnsResult>()?;
     m.add_class::<RefreshColumnResult>()?;
+    m.add_class::<FunctionErrors>()?;
+    m.add_class::<FunctionErrorRecord>()?;
+    m.add_class::<FunctionErrorFragment>()?;
     m.add_class::<RefreshMaterializedViewResult>()?;
     m.add_class::<AlterColumnsResult>()?;
     m.add_class::<UpdateFieldMetadataResult>()?;

python/src/table.rs

@@ -441,6 +441,117 @@ impl From<lancedb::table::RefreshColumnResult> for RefreshColumnResult {
     }
 }
 
+/// One row a Function refresh skipped, as the server recorded it.
+#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)]
+#[derive(Clone, Debug)]
+pub struct FunctionErrorRecord {
+    pub job_id: String,
+    pub fragment_id: u64,
+    pub row_offset: Option<u32>,
+    pub column: String,
+    pub function: String,
+    pub function_version: String,
+    pub table_version: u64,
+    pub error_type: String,
+    pub error_message: String,
+    pub created_at_millis: i64,
+}
+
+#[pymethods]
+impl FunctionErrorRecord {
+    pub fn __repr__(&self) -> String {
+        format!(
+            "FunctionErrorRecord(job_id={:?}, fragment_id={}, row_offset={:?}, column={:?}, \
+             error_type={:?}, error_message={:?})",
+            self.job_id,
+            self.fragment_id,
+            self.row_offset,
+            self.column,
+            self.error_type,
+            self.error_message
+        )
+    }
+}
+
+impl From<lancedb::function::FunctionErrorRecord> for FunctionErrorRecord {
+    fn from(record: lancedb::function::FunctionErrorRecord) -> Self {
+        Self {
+            job_id: record.job_id,
+            fragment_id: record.fragment_id,
+            row_offset: record.row_offset,
+            column: record.column,
+            function: record.function,
+            function_version: record.function_version,
+            table_version: record.table_version,
+            error_type: record.error_type,
+            error_message: record.error_message,
+            created_at_millis: record.created_at_millis,
+        }
+    }
+}
+
+/// A fragment whose per-row error detail was capped.
+#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)]
+#[derive(Clone, Debug)]
+pub struct FunctionErrorFragment {
+    pub job_id: String,
+    pub fragment_id: u64,
+    pub rows_skipped: u64,
+    pub rows_recorded: u64,
+}
+
+#[pymethods]
+impl FunctionErrorFragment {
+    pub fn __repr__(&self) -> String {
+        format!(
+            "FunctionErrorFragment(job_id={:?}, fragment_id={}, rows_skipped={}, rows_recorded={})",
+            self.job_id, self.fragment_id, self.rows_skipped, self.rows_recorded
+        )
+    }
+}
+
+impl From<lancedb::function::FunctionErrorFragment> for FunctionErrorFragment {
+    fn from(fragment: lancedb::function::FunctionErrorFragment) -> Self {
+        Self {
+            job_id: fragment.job_id,
+            fragment_id: fragment.fragment_id,
+            rows_skipped: fragment.rows_skipped,
+            rows_recorded: fragment.rows_recorded,
+        }
+    }
+}
+
+/// A table's per-row Function errors.
+#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)]
+#[derive(Clone, Debug)]
+pub struct FunctionErrors {
+    pub records: Vec<FunctionErrorRecord>,
+    pub fragments: Vec<FunctionErrorFragment>,
+    pub truncated: bool,
+}
+
+#[pymethods]
+impl FunctionErrors {
+    pub fn __repr__(&self) -> String {
+        format!(
+            "FunctionErrors(records={}, fragments={}, truncated={})",
+            self.records.len(),
+            self.fragments.len(),
+            self.truncated
+        )
+    }
+}
+
+impl From<lancedb::function::FunctionErrors> for FunctionErrors {
+    fn from(errors: lancedb::function::FunctionErrors) -> Self {
+        Self {
+            records: errors.records.into_iter().map(Into::into).collect(),
+            fragments: errors.fragments.into_iter().map(Into::into).collect(),
+            truncated: errors.truncated,
+        }
+    }
+}
+
 #[pyclass(get_all, from_py_object)]
 #[derive(Clone, Debug)]
 pub struct RefreshMaterializedViewResult {
@@ -1638,6 +1749,25 @@ impl Table {
         })
     }
 
+    #[pyo3(signature = (job_id=None, column=None, limit=None))]
+    pub fn function_errors(
+        self_: PyRef<'_, Self>,
+        job_id: Option<String>,
+        column: Option<String>,
+        limit: Option<usize>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        let request = lancedb::function::FunctionErrorsRequest {
+            job_id,
+            column,
+            limit,
+        };
+        future_into_py(self_.py(), async move {
+            let errors = inner.function_errors(request).await.infer_error()?;
+            Ok(FunctionErrors::from(errors))
+        })
+    }
+
     #[pyo3(signature = (full=false, source_version=None))]
     pub fn refresh_materialized_view(
         self_: PyRef<'_, Self>,