feat(python)!: expose typed query errors and request metadata (#4832)

Remote SQL failures currently become generic runtime errors, and HTTP
table lookup can discard server error metadata. Preserve transport
status, current request ID, SQL job ID and worker request ID across Rust
and Python. Select typed column, table, reference and schema exceptions
only when an explicit server code matches the transport status.

Remote `HttpError` and `RuntimeError` handlers remain compatible. Failed
SQL jobs retain `JobFailedError` with a typed SQL cause when available.
Missing metadata stays absent; message text never determines the
exception class. Local missing-column errors retain their type and
qualified-field suggestions, and `JobFailure.source` documents
structured remote Flight SQL failures.

This branch integrates main at `a9ad4997`, including its released Lance
15.0.0-beta.9 dependency update and remote optimize support. The only
textual conflict was two independent additions to `test_remote_db.py`;
both test groups and all original assertions are retained. The feature
has no dependency or lockfile delta against current main.

SDK runtime validation on `03ee633a` (runtime source unchanged in
`f3f2d0b8`), macOS ARM64, pinned Rust 1.97.0 and the frozen uv Python
3.14 environment:

- Rebuilt the native extension from the integrated source. All 22
targeted Python tests passed: native local errors, HTTP/Flight metadata,
legacy sync/async table-not-found behavior, and the three main
remote-optimize regressions.
- All 52 Rust SQL, metadata/source and nested-field tests passed,
including the new main optimize cases.
- Root and Python-binding Rust Clippy with `-D warnings`, Rust
formatting, Python package Ruff and conflict-file formatting passed.
- Repository-wide Ruff reports 59 existing documentation findings, and
its format check identifies six documentation files. Every affected file
is byte-identical to current main; these checks are not reported as
passing.
- Direct integration review and AST comparison confirm both conflicting
test groups were preserved. Earlier exception-constructor/pickling
evidence remains applicable to the unchanged Python error modules.

The unlocked build exhausted disk again on a runner with the same label
but only 9.4 GiB free on a 72 GiB root filesystem. Keep the existing
runner and dependency caches, serialize Cargo compilation, and remove
only completed test/bench executables reported by Cargo. This job never
runs those binaries; normal binaries, libraries and build-script outputs
remain available. Fresh resolution, the ci profile, all features, and
test/bench compilation and linking are retained.

Three probes using Cargo and target-path validation pass with `-D
warnings`: all fixture targets link while ordinary outputs remain
usable; an undefined symbol still produces Cargo exit 101 and its linker
diagnostic; files outside the target directory are never removed. The
[current-head build-no-lock
job](https://github.com/lancedb/lancedb/actions/runs/41911613851/job/126860364065)
passed on `f3f2d0b8`: all compilation/linking completed, 20 completed
test executables were removed, and `target` retained about 7.3 GiB. This
successful run had a 145 GiB root with about 78 GiB free afterward; it
was not an identical rerun of the failed 72 GiB image. The driver tests
also passed in this job. All previous-head jobs ended naturally before
the push; new-head CI and review gates still apply.

Typed remote input exceptions require the server metadata contract.
Older servers retain generic transport errors and existing missing-table
compatibility. The optimize tests use mock servers and do not establish
downstream server acceptance.

BREAKING CHANGE: Rust callers with exhaustive matches on
`lancedb::Error` must handle the new `FieldNotFound` and `Sql` variants.
Local `Error::ColumnNotFound` failures now raise
`lancedb.exceptions.ColumnNotFoundError`, a `ValueError` subclass.
Python callers that previously caught only `RuntimeError` for this local
failure must catch `ColumnNotFoundError` or `ValueError`.

python/python/lancedb/exceptions.py

@@ -27,12 +27,44 @@ class MissingColumnError(KeyError):
         )
 
 
+class ColumnNotFoundError(ValueError):
+    """A requested table column does not exist.
+
+    Remote HTTP and SQL errors inherit this type and carry request metadata.
+    Locally rejected queries have no request or job identifiers.
+    """
+
+    def __init__(self, message: str):
+        super().__init__(message)
+        self.request_id: Optional[str] = None
+        self.job_id: Optional[str] = None
+        self.job_request_id: Optional[str] = None
+
+
+class ReferenceNotFoundError(ValueError):
+    """A requested branch or table reference does not exist."""
+
+    request_id: Optional[str] = None
+    job_id: Optional[str] = None
+    job_request_id: Optional[str] = None
+
+
+class SchemaMismatchError(ValueError):
+    """Input data does not match the table's schema."""
+
+    request_id: Optional[str] = None
+    job_id: Optional[str] = None
+    job_request_id: Optional[str] = None
+
+
 class JobFailedError(RuntimeError):
     """Exception raised when an asynchronous job reaches the failed state.
 
     ``job_id`` identifies the failed job when available. ``sql_job_id`` preserves
     the SQL execution identity reported by the server, which may differ from
-    the outer job's identity.
+    the outer job's identity. ``request_id`` and ``job_request_id`` identify the
+    current request and isolated worker request. A typed SQL cause is available
+    through ``__cause__`` when the server reports one.
     """
 
     def __init__(
@@ -40,10 +72,14 @@ class JobFailedError(RuntimeError):
         message: str,
         job_id: Optional[str] = None,
         sql_job_id: Optional[str] = None,
+        request_id: Optional[str] = None,
+        job_request_id: Optional[str] = None,
     ):
         super().__init__(message)
         self.job_id = job_id
         self.sql_job_id = sql_job_id
+        self.request_id = request_id
+        self.job_request_id = job_request_id
 
 
 class JobCancelledError(RuntimeError):

python/python/lancedb/remote/errors.py

@@ -6,6 +6,12 @@ from typing import Optional
 
 from lance_namespace.errors import TableNotFoundError as _NamespaceTableNotFoundError
 
+from lancedb.exceptions import (
+    ColumnNotFoundError as _ColumnNotFoundError,
+    ReferenceNotFoundError as _ReferenceNotFoundError,
+    SchemaMismatchError as _SchemaMismatchError,
+)
+
 
 class LanceDBClientError(RuntimeError):
     """An error that occurred in the LanceDB client.
@@ -57,6 +63,10 @@ class HttpError(LanceDBClientError):
     is_transport_error: bool
         Whether the request failed while connecting, sending, or reading, rather
         than because of an HTTP status or an invalid response payload.
+    job_id: str or None
+        SQL job reported by the server, when available.
+    job_request_id: str or None
+        Original request executed by the isolated SQL worker, when available.
     """
 
     def __new__(
@@ -69,9 +79,11 @@ class HttpError(LanceDBClientError):
         resource_name: Optional[str] = None,
         namespace_path: Optional[tuple[str, ...]] = None,
         is_transport_error: bool = False,
+        job_id: Optional[str] = None,
+        job_request_id: Optional[str] = None,
     ):
         if cls is HttpError:
-            mapped_cls = _http_error_class(error_code, resource_kind)
+            mapped_cls = _http_error_class(error_code, resource_kind, status_code)
             if mapped_cls is not None:
                 cls = mapped_cls
         return RuntimeError.__new__(cls)
@@ -86,9 +98,13 @@ class HttpError(LanceDBClientError):
         resource_name: Optional[str] = None,
         namespace_path: Optional[tuple[str, ...]] = None,
         is_transport_error: bool = False,
+        job_id: Optional[str] = None,
+        job_request_id: Optional[str] = None,
     ):
         super().__init__(message, request_id, status_code)
         self.is_transport_error = is_transport_error
+        self.job_id = job_id
+        self.job_request_id = job_request_id
         self.error_code = error_code
         self.resource_kind = resource_kind
         self.resource_name = resource_name
@@ -108,13 +124,29 @@ class HttpError(LanceDBClientError):
                 self.resource_name,
                 self.namespace_path,
                 self.is_transport_error,
+                self.job_id,
+                self.job_request_id,
             ),
         )
 
 
 def _http_error_class(
-    error_code: Optional[str], resource_kind: Optional[str]
+    error_code: Optional[str], resource_kind: Optional[str], status_code: Optional[int]
 ) -> Optional[type[HttpError]]:
+    if status_code == 400 and error_code is not None:
+        input_class = {
+            "column_not_found": ColumnNotFoundError,
+            "schema_mismatch": SchemaMismatchError,
+        }.get(error_code)
+        if input_class is not None:
+            return input_class
+    if status_code == 404 and error_code is not None:
+        input_class = {
+            "table_not_found": TableNotFoundError,
+            "reference_not_found": ReferenceNotFoundError,
+        }.get(error_code)
+        if input_class is not None:
+            return input_class
     if error_code == "already_exists":
         return AlreadyExistsError
     if error_code != "not_found" or resource_kind is None:
@@ -126,6 +158,100 @@ def _http_error_class(
     }.get(resource_kind)
 
 
+class ColumnNotFoundError(HttpError, _ColumnNotFoundError):
+    """A remote HTTP request named a missing table column."""
+
+
+class ReferenceNotFoundError(HttpError, _ReferenceNotFoundError):
+    """A remote HTTP request named a missing branch or table reference."""
+
+
+class SchemaMismatchError(HttpError, _SchemaMismatchError):
+    """Input to a remote HTTP request does not match the table schema."""
+
+
+class SqlError(LanceDBClientError):
+    """A SQL response with its gRPC status and correlation fields.
+
+    ``grpc_code`` is the numeric gRPC code. ``error_code`` is an optional
+    machine-readable classification supplied by the server; message text is
+    never used to choose a subclass. ``job_id`` and ``job_request_id`` are absent
+    unless the server reports an isolated SQL job. ``request_id`` identifies the
+    current request. Older servers still produce SqlError without a category.
+
+    Examples
+    --------
+    >>> error = SqlError("Column 'missing' was not found", "req-1", 3,
+    ...                  "column_not_found", "job-1", "worker-1")
+    >>> isinstance(error, SqlColumnNotFoundError)
+    True
+    >>> error.job_request_id
+    'worker-1'
+    """
+
+    def __new__(
+        cls,
+        message: str,
+        request_id: str,
+        grpc_code: int,
+        error_code: Optional[str] = None,
+        job_id: Optional[str] = None,
+        job_request_id: Optional[str] = None,
+    ):
+        if cls is SqlError and error_code is not None:
+            cls = {
+                (3, "column_not_found"): SqlColumnNotFoundError,
+                (3, "schema_mismatch"): SqlSchemaMismatchError,
+                (5, "table_not_found"): SqlTableNotFoundError,
+                (5, "reference_not_found"): SqlReferenceNotFoundError,
+            }.get((grpc_code, error_code), cls)
+        return RuntimeError.__new__(cls)
+
+    def __init__(
+        self,
+        message: str,
+        request_id: str,
+        grpc_code: int,
+        error_code: Optional[str] = None,
+        job_id: Optional[str] = None,
+        job_request_id: Optional[str] = None,
+    ):
+        super().__init__(message, request_id)
+        self.grpc_code = grpc_code
+        self.error_code = error_code
+        self.job_id = job_id
+        self.job_request_id = job_request_id
+
+    def __reduce__(self):
+        return (
+            self.__class__,
+            (
+                str(self),
+                self.request_id,
+                self.grpc_code,
+                self.error_code,
+                self.job_id,
+                self.job_request_id,
+            ),
+        )
+
+
+class SqlColumnNotFoundError(SqlError, _ColumnNotFoundError):
+    """A SQL request named a missing column."""
+
+
+class SqlReferenceNotFoundError(SqlError, _ReferenceNotFoundError):
+    """A SQL request named a missing branch or table reference."""
+
+
+class SqlSchemaMismatchError(SqlError, _SchemaMismatchError):
+    """SQL input does not match the table schema."""
+
+
+class SqlTableNotFoundError(SqlError, ValueError, _NamespaceTableNotFoundError):
+    """A SQL request named a missing table."""
+
+
 class SecretNotFoundError(HttpError):
     """A remote secret was not found."""
 

python/python/tests/test_errors.py

@@ -205,3 +205,84 @@ def test_pickle_retry_error():
     assert restored.max_connect_failures == 3
     assert restored.max_read_failures == 3
     assert restored.status_code == 503
+
+
+@pytest.mark.parametrize(
+    "code,category,type_name",
+    [
+        (3, "column_not_found", "SqlColumnNotFoundError"),
+        (3, "schema_mismatch", "SqlSchemaMismatchError"),
+        (5, "table_not_found", "SqlTableNotFoundError"),
+        (5, "reference_not_found", "SqlReferenceNotFoundError"),
+        (13, "column_not_found", "SqlError"),
+        (14, "table_not_found", "SqlError"),
+        (3, None, "SqlError"),
+        (5, "future_code", "SqlError"),
+    ],
+)
+def test_sql_errors_use_only_server_codes(code, category, type_name):
+    from lancedb.remote import errors
+
+    original = errors.SqlError(
+        "column_not_found: user/file.rs:12",
+        "current",
+        code,
+        category,
+        "sql-job",
+        "worker",
+    )
+    for error in [original, pickle.loads(pickle.dumps(original))]:
+        assert type(error) is getattr(errors, type_name)
+        assert str(error) == "column_not_found: user/file.rs:12"
+        assert error.request_id == "current"
+        assert error.grpc_code == code
+        assert error.error_code == category
+        assert error.job_id == "sql-job"
+        assert error.job_request_id == "worker"
+        assert error.status_code is None
+        if type_name != "SqlError":
+            assert isinstance(error, ValueError)
+
+
+@pytest.mark.parametrize(
+    "status,category,type_name",
+    [
+        (400, "column_not_found", "ColumnNotFoundError"),
+        (400, "schema_mismatch", "SchemaMismatchError"),
+        (404, "table_not_found", "TableNotFoundError"),
+        (404, "reference_not_found", "ReferenceNotFoundError"),
+        (500, "column_not_found", "HttpError"),
+        (503, "table_not_found", "HttpError"),
+    ],
+)
+def test_http_input_errors_keep_status_and_job_metadata(status, category, type_name):
+    from lancedb.remote import errors
+
+    original = errors.HttpError(
+        "original diagnostic",
+        "current",
+        status,
+        error_code=category,
+        job_id="sql-job",
+        job_request_id="worker",
+    )
+    for error in [original, pickle.loads(pickle.dumps(original))]:
+        assert type(error) is getattr(errors, type_name)
+        assert error.status_code == status
+        assert error.job_id == "sql-job"
+        assert error.job_request_id == "worker"
+        assert error.request_id == "current"
+
+
+def test_local_query_keeps_missing_column_type(tmp_path):
+    import lancedb
+    from lancedb.exceptions import ColumnNotFoundError
+
+    db = lancedb.connect(tmp_path)
+    table = db.create_table("events", [{"id": 1}])
+    with pytest.raises(ColumnNotFoundError) as exc_info:
+        table.search().where("missing > 0").to_arrow()
+    assert "missing" in str(exc_info.value)
+    assert exc_info.value.request_id is None
+    assert exc_info.value.job_id is None
+    assert exc_info.value.job_request_id is None

python/python/tests/test_remote_db.py

@@ -5627,6 +5627,49 @@ def test_local_connections_refuse_property_graphs(tmp_path):
         db.drop_property_graph("social")
 
 
+@pytest.mark.parametrize(
+    "status,category,type_name",
+    [
+        (400, "column_not_found", "ColumnNotFoundError"),
+        (400, "schema_mismatch", "SchemaMismatchError"),
+        (404, "table_not_found", "TableNotFoundError"),
+        (404, "reference_not_found", "ReferenceNotFoundError"),
+    ],
+)
+@pytest.mark.parametrize("operation", ["describe_namespace", "open_table"])
+def test_remote_input_error_metadata_crosses_native_boundary(
+    status, category, type_name, operation
+):
+    from lancedb.remote import errors
+
+    request_ids = []
+
+    def handler(request):
+        request_ids.append(request.headers["x-request-id"])
+        request.send_response(status)
+        request.send_header("Content-Type", "application/json")
+        request.send_header("x-lancedb-error-code", category)
+        request.send_header("x-lancedb-sql-job-id", "sql-job-1")
+        request.send_header("x-job-request-id", "worker-request")
+        request.end_headers()
+        request.wfile.write(json.dumps({"error": "original diagnostic"}).encode())
+
+    with mock_lancedb_connection(handler) as db:
+        with pytest.raises(getattr(errors, type_name)) as exc_info:
+            if operation == "open_table":
+                db.open_table("analytics")
+            else:
+                db.describe_namespace(["analytics"])
+    error = exc_info.value
+    assert isinstance(error, HttpError)
+    assert error.request_id == request_ids[-1]
+    assert error.status_code == status
+    assert error.error_code == category
+    assert error.job_id == "sql-job-1"
+    assert error.job_request_id == "worker-request"
+    assert str(error) == "original diagnostic"
+
+
 def _create_table_handler(request):
     if request.path == "/v1/table/test/create/?mode=create":
         request.send_response(200)

python/python/tests/test_sql.py

@@ -281,3 +281,63 @@ def test_parameters_reach_the_server_as_the_values_that_were_sent():
     received = echoed.column("vector").combine_chunks().flatten().to_numpy()
     assert received.view(np.uint32).tolist() == vector.view(np.uint32).tolist()
     assert echoed.column("k").to_pylist() == [10]
+
+
+@pytest.mark.parametrize("job_id", [None, "sql-job-1"])
+def test_sql_input_error_metadata_crosses_native_boundary(job_id):
+    flight = pytest.importorskip("pyarrow.flight")
+    from lancedb.exceptions import JobFailedError, ColumnNotFoundError
+    from lancedb.remote.errors import SqlColumnNotFoundError
+
+    request_ids = []
+
+    class Metadata(flight.ServerMiddleware):
+        def sending_headers(self):
+            headers = {
+                "x-lancedb-error-code": "column_not_found",
+                "x-job-request-id": "worker-request",
+            }
+            if job_id is not None:
+                headers["x-lancedb-sql-job-id"] = job_id
+            return headers
+
+    class Factory(flight.ServerMiddlewareFactory):
+        def start_call(self, info, headers):
+            request_ids.append(headers["x-request-id"][0])
+            return Metadata()
+
+    class RejectingServer(flight.FlightServerBase):
+        def __init__(self):
+            super().__init__("grpc://127.0.0.1:0", middleware={"metadata": Factory()})
+
+        def do_get(self, context, ticket):
+            context.add_trailer("x-lancedb-error-code", "column_not_found")
+            context.add_trailer("x-job-request-id", "worker-request")
+            if job_id is not None:
+                context.add_trailer("x-lancedb-sql-job-id", job_id)
+            raise pa.ArrowInvalid("No column 'missing' in table events")
+
+    with RejectingServer() as server:
+        db = lancedb.connect(
+            "db://analytics",
+            api_key="test-key",
+            host_override="http://localhost:10024",
+            sql_host_override=f"grpc://127.0.0.1:{server.port}",
+        )
+        expected = JobFailedError if job_id is not None else SqlColumnNotFoundError
+        with pytest.raises(expected) as exc_info:
+            db.execute_query("SELECT missing FROM events").read_all()
+    error = exc_info.value
+    if job_id is not None:
+        assert error.job_id == job_id
+        assert error.request_id == request_ids[-1]
+        assert error.job_request_id == "worker-request"
+        error = error.__cause__
+    assert isinstance(error, SqlColumnNotFoundError)
+    assert isinstance(error, ColumnNotFoundError)
+    assert error.request_id == request_ids[-1]
+    assert error.job_id == job_id
+    assert error.job_request_id == "worker-request"
+    assert error.grpc_code == 3
+    assert str(error).startswith("No column 'missing' in table events")
+    assert not str(error).startswith("SQL error")

python/src/error.rs

@@ -8,7 +8,7 @@ use pyo3::{
     types::{PyAnyMethods, PyNone},
 };
 
-use lancedb::error::{Error as LanceError, HttpErrorDetails};
+use lancedb::error::{Error as LanceError, HttpErrorDetails, RemoteErrorMetadata};
 
 pub trait PythonErrorExt<T> {
     /// Convert to a python error based on the Lance error type
@@ -34,26 +34,61 @@ impl<T> PythonErrorExt<T> for std::result::Result<T, LanceError> {
                 | LanceError::DatabaseAlreadyExists { .. }
                 | LanceError::TableAlreadyExists { .. } => self.value_error(),
                 LanceError::TableNotFound { source, .. } => {
-                    let Some(LanceError::Http {
-                        request_id,
-                        status_code,
-                        ..
-                    }) = source.downcast_ref::<LanceError>()
+                    let Some(
+                        request_error @ LanceError::Http {
+                            request_id,
+                            status_code,
+                            source: http_source,
+                        },
+                    ) = source.downcast_ref::<LanceError>()
                     else {
                         return self.value_error();
                     };
                     Python::attach(|py| {
-                        let cls = py
-                            .import(intern!(py, "lancedb.remote.errors"))?
-                            .getattr(intern!(py, "TableNotFoundError"))?;
                         let status_code = status_code.map(|s| s.as_u16());
-                        let py_err = cls.call1((err.to_string(), request_id, status_code))?;
+                        let metadata = request_error.remote_error_metadata();
+                        let details = request_error.http_error_details();
+                        let py_err = if details.is_some()
+                            || metadata.is_some_and(|metadata| metadata.error_code.is_some())
+                        {
+                            new_py_http_error(
+                                py,
+                                http_source.to_string(),
+                                request_id,
+                                status_code,
+                                details,
+                                metadata,
+                                false,
+                            )?
+                        } else {
+                            let cls = py
+                                .import(intern!(py, "lancedb.remote.errors"))?
+                                .getattr(intern!(py, "TableNotFoundError"))?;
+                            let error = cls.call1((err.to_string(), request_id, status_code))?;
+                            error.setattr(
+                                "job_id",
+                                metadata.and_then(|value| value.job_id.as_deref()),
+                            )?;
+                            error.setattr(
+                                "job_request_id",
+                                metadata.and_then(|value| value.job_request_id.as_deref()),
+                            )?;
+                            error
+                        };
                         let cause =
                             http_from_rust_error(py, source.as_ref(), request_id, status_code)?;
                         py_err.setattr(intern!(py, "__cause__"), cause)?;
                         Err(PyErr::from_value(py_err))
                     })
                 }
+                LanceError::ColumnNotFound { .. } | LanceError::FieldNotFound { .. } => {
+                    Python::attach(|py| {
+                        let cls = py
+                            .import(intern!(py, "lancedb.exceptions"))?
+                            .getattr(intern!(py, "ColumnNotFoundError"))?;
+                        Err(PyErr::from_value(cls.call1((err.to_string(),))?))
+                    })
+                }
                 LanceError::CreateDir { .. } => self.os_error(),
                 LanceError::ObjectStore { .. } => Err(PyIOError::new_err(err.to_string())),
                 LanceError::NotSupported { .. } => {
@@ -64,13 +99,21 @@ impl<T> PythonErrorExt<T> for std::result::Result<T, LanceError> {
                     source,
                     status_code,
                 } => Python::attach(|py| {
-                    let message = err.to_string();
+                    let message = if err
+                        .remote_error_metadata()
+                        .is_some_and(|metadata| metadata.error_code.is_some())
+                    {
+                        source.to_string()
+                    } else {
+                        err.to_string()
+                    };
                     let err = new_py_http_error(
                         py,
                         message,
                         request_id,
                         status_code.map(|s| s.as_u16()),
                         err.http_error_details(),
+                        err.remote_error_metadata(),
                         is_transport_error(source.as_ref()),
                     )?;
 
@@ -125,15 +168,46 @@ impl<T> PythonErrorExt<T> for std::result::Result<T, LanceError> {
                     err.setattr(intern!(py, "__cause__"), cause_err)?;
                     Err(PyErr::from_value(err))
                 }),
+                LanceError::Sql {
+                    message,
+                    request_id,
+                    grpc_code,
+                    metadata,
+                } => Python::attach(|py| {
+                    Err(PyErr::from_value(new_py_sql_error(
+                        py, message, request_id, *grpc_code, metadata,
+                    )?))
+                }),
                 LanceError::JobFailed { job_id, failure } => Python::attach(|py| {
                     let cls = py
                         .import(intern!(py, "lancedb.exceptions"))?
                         .getattr(intern!(py, "JobFailedError"))?;
-                    Err(PyErr::from_value(cls.call1((
+                    let sql_source = match failure.source.as_deref() {
+                        Some(LanceError::Sql {
+                            message,
+                            request_id,
+                            grpc_code,
+                            metadata,
+                        }) => Some((
+                            new_py_sql_error(py, message, request_id, *grpc_code, metadata)?,
+                            request_id.as_str(),
+                            metadata.job_request_id.as_deref(),
+                        )),
+                        _ => None,
+                    };
+                    let error = cls.call1((
                         err.to_string(),
                         job_id.clone(),
                         failure.sql_job_id.clone(),
-                    ))?))
+                        sql_source.as_ref().map(|(_, request_id, _)| *request_id),
+                        sql_source
+                            .as_ref()
+                            .and_then(|(_, _, request_id)| *request_id),
+                    ))?;
+                    if let Some((cause, _, _)) = sql_source {
+                        error.setattr(intern!(py, "__cause__"), cause)?;
+                    }
+                    Err(PyErr::from_value(error))
                 }),
                 LanceError::JobCancelled { .. } => Python::attach(|py| {
                     let cls = py
@@ -181,6 +255,8 @@ fn http_from_rust_error(
         request_id,
         status_code,
         details,
+        err.downcast_ref::<LanceError>()
+            .and_then(LanceError::remote_error_metadata),
         is_transport_error(err),
     )?;
 
@@ -201,9 +277,12 @@ fn new_py_http_error<'py>(
     request_id: &str,
     status_code: Option<u16>,
     details: Option<&HttpErrorDetails>,
+    metadata: Option<&RemoteErrorMetadata>,
     is_transport_error: bool,
 ) -> PyResult<Bound<'py, PyAny>> {
-    let error_code = details.map(|details| details.error_code.as_str());
+    let error_code = metadata
+        .and_then(|metadata| metadata.error_code.as_deref())
+        .or_else(|| details.map(|details| details.error_code.as_str()));
     let resource_kind = details.map(|details| details.resource_kind.as_str());
     let resource_name = details.map(|details| details.resource_name.as_str());
     let namespace_path = details.map(|details| details.namespace_path.clone());
@@ -219,6 +298,27 @@ fn new_py_http_error<'py>(
             resource_name,
             namespace_path,
             is_transport_error,
+            metadata.and_then(|metadata| metadata.job_id.as_deref()),
+            metadata.and_then(|metadata| metadata.job_request_id.as_deref()),
+        ))
+}
+
+fn new_py_sql_error<'py>(
+    py: Python<'py>,
+    message: &str,
+    request_id: &str,
+    grpc_code: i32,
+    metadata: &RemoteErrorMetadata,
+) -> PyResult<Bound<'py, PyAny>> {
+    py.import("lancedb.remote.errors")?
+        .getattr("SqlError")?
+        .call1((
+            message,
+            request_id,
+            grpc_code,
+            metadata.error_code.as_deref(),
+            metadata.job_id.as_deref(),
+            metadata.job_request_id.as_deref(),
         ))
 }