diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 1c64a2cad..dbdcc3b54 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -14,6 +14,8 @@ on: - docs/web-tests/blob-v2/rs/** - nodejs/Cargo.toml - python/Cargo.toml + - ci/build_no_lock.py + - ci/tests/test_build_no_lock.py - .github/actions/setup-protoc/** - .github/workflows/rust.yml @@ -135,7 +137,8 @@ jobs: command: check advisories bans licenses sources build-no-lock: - runs-on: ubuntu-24.04 + # Keep the existing Linux runner; its available disk varies by image. + runs-on: ubuntu-2404-4x-x64 timeout-minutes: 30 env: # Need up-to-date compilers for kernels @@ -162,9 +165,22 @@ jobs: run: | sudo apt update sudo apt install -y protobuf-compiler libssl-dev + - name: Test build output cleanup + run: python3 -m unittest discover -s ci/tests -p test_build_no_lock.py -v - name: Build all run: | - cargo build --profile ci --benches --all-features --tests + df -h . + # Every target is linked, but this job never executes the test binaries. + python3 ci/build_no_lock.py + - name: Report build disk usage + if: always() + run: | + df -h . + for build_path in target "$HOME/.cargo/registry" "$HOME/.cargo/git"; do + if [ -d "$build_path" ]; then + du -h -d 2 "$build_path" + fi + done linux: timeout-minutes: 60 diff --git a/ci/build_no_lock.py b/ci/build_no_lock.py new file mode 100644 index 000000000..6ec156432 --- /dev/null +++ b/ci/build_no_lock.py @@ -0,0 +1,111 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The LanceDB Authors + +"""Link all CI test/bench targets after the workflow removes Cargo.lock. + +This job never runs the test binaries. Discard their completed +executables while retaining normal binaries, libraries and build-script outputs. +""" + +import json +import shutil +import subprocess +import sys +from pathlib import Path + + +def discard_test_executable(message: dict, target_dir: Path) -> bool: + if message.get("reason") != "compiler-artifact": + return False + target = message["target"] + kinds = target["kind"] + # profile.test describes this build; target.test is only a manifest capability. + # Ordinary binaries may still be needed through CARGO_BIN_EXE by later targets. + if "custom-build" in kinds or not ( + message["profile"]["test"] or "test" in kinds or "bench" in kinds + ): + return False + executable = message.get("executable") + if executable is None: + return False + path = Path(executable) + if not path.resolve().is_relative_to(target_dir.resolve()): + raise ValueError(f"executable {path} is outside target directory {target_dir}") + if not path.is_file(): + return False + size = path.stat().st_size + path.unlink() + print( + json.dumps( + { + "removed_test_executable": str(path), + "target": target["name"], + "logical_bytes": size, + "filesystem_free_bytes": shutil.disk_usage(path.parent).free, + } + ), + flush=True, + ) + return True + + +def main() -> int: + metadata = subprocess.run( + ["cargo", "metadata", "--format-version=1", "--no-deps", "--all-features"], + stdout=subprocess.PIPE, + text=True, + ) + if metadata.returncode: + return metadata.returncode + target_dir = Path(json.loads(metadata.stdout)["target_directory"]) + command = [ + "cargo", + "build", + "--profile", + "ci", + "--benches", + "--all-features", + "--tests", + # Bound simultaneous compiler/linker temporary outputs as well as retained files. + "--jobs", + "1", + "--message-format=json,json-diagnostic-rendered-ansi", + ] + cleanup_failed = False + with subprocess.Popen(command, stdout=subprocess.PIPE, text=True) as process: + assert process.stdout is not None + for line in process.stdout: + try: + message = json.loads(line) + except json.JSONDecodeError: + print(line, end="", flush=True) + continue + if not isinstance(message, dict): + print(line, end="", flush=True) + continue + if message.get("reason") == "compiler-message": + diagnostic = message["message"] + sys.stderr.write( + diagnostic.get("rendered") or json.dumps(diagnostic) + "\n" + ) + sys.stderr.flush() + elif message.get("reason") == "compiler-artifact": + try: + discard_test_executable(message, target_dir) + except (OSError, ValueError) as error: + print( + f"Cannot discard completed test executable: {error}", + file=sys.stderr, + ) + cleanup_failed = True + elif message.get("reason") not in { + "build-script-executed", + "build-finished", + }: + print(line, end="", flush=True) + result = process.wait() + return result if result else int(cleanup_failed) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/ci/tests/test_build_no_lock.py b/ci/tests/test_build_no_lock.py new file mode 100644 index 000000000..634e8d333 --- /dev/null +++ b/ci/tests/test_build_no_lock.py @@ -0,0 +1,127 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The LanceDB Authors + +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +SCRIPT = Path(__file__).resolve().parents[1] / "build_no_lock.py" +SPEC = importlib.util.spec_from_file_location("build_no_lock", SCRIPT) +BUILD = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(BUILD) + + +class BuildNoLockTest(unittest.TestCase): + def fixture(self, directory): + root = Path(directory) + (root / "src").mkdir() + (root / "tests").mkdir() + (root / "benches").mkdir() + (root / "Cargo.toml").write_text( + '[package]\nname = "artifact_fixture"\nversion = "0.0.0"\nedition = "2024"\n' + "[workspace]\n[features]\nprobe = []\n" + '[[bin]]\nname = "fixture_cli"\npath = "src/main.rs"\n' + '[[bench]]\nname = "fixture_bench"\nharness = false\n' + '[profile.ci]\ninherits = "dev"\ndebug = 0\nincremental = false\n' + ) + (root / "build.rs").write_text( + 'fn main() { let out = std::path::PathBuf::from(std::env::var_os("OUT_DIR").unwrap());' + ' std::fs::write(out.join("generated.rs"), "pub fn value() -> u32 { 7 }").unwrap(); }' + ) + (root / "src/lib.rs").write_text( + '#[cfg(not(feature = "probe"))] compile_error!("all features must be built");\n' + 'include!(concat!(env!("OUT_DIR"), "/generated.rs"));\n' + "#[test] fn library_test() { assert_eq!(value(), 7); }\n" + ) + (root / "src/main.rs").write_text( + "fn main() { assert_eq!(artifact_fixture::value(), 7); }\n" + ) + (root / "tests/retains_binary.rs").write_text( + "#[test] fn ordinary_binary_is_available() {\n" + ' assert!(std::process::Command::new(env!("CARGO_BIN_EXE_fixture_cli"))\n' + " .status().unwrap().success()); }\n" + ) + (root / "benches/fixture_bench.rs").write_text( + "fn main() { assert_eq!(artifact_fixture::value(), 7); }\n" + ) + return root + + def run_build(self, root): + env = os.environ | {"CARGO_TARGET_DIR": str(root / "target")} + return subprocess.run( + [sys.executable, str(SCRIPT)], + cwd=root, + env=env, + text=True, + capture_output=True, + ) + + def test_links_all_targets_and_retains_compiler_inputs(self): + # Stay below the repository so rustup keeps its configured toolchain. + with tempfile.TemporaryDirectory(prefix=".no-lock-", dir=SCRIPT.parent) as temp: + root = self.fixture(temp) + result = self.run_build(root) + self.assertEqual(result.returncode, 0, result.stderr) + messages = [ + json.loads(line) + for line in result.stdout.splitlines() + if line.startswith("{") + ] + removed = [item for item in messages if "removed_test_executable" in item] + self.assertTrue( + {"artifact_fixture", "fixture_cli", "retains_binary", "fixture_bench"} + <= {item["target"] for item in removed} + ) + self.assertTrue( + all( + not Path(item["removed_test_executable"]).exists() + for item in removed + ) + ) + target = root / "target/ci" + self.assertTrue((target / "fixture_cli").is_file()) + # Cargo guarantees this binary for test execution, not compile-time reads. + subprocess.run([str(target / "fixture_cli")], check=True) + self.assertTrue(list((target / "deps").glob("libartifact_fixture*.rlib"))) + self.assertTrue(list((target / "build").glob("*/build-script-build"))) + self.assertTrue(list((target / "build").glob("*/out/generated.rs"))) + + def test_link_failure_preserves_cargo_status_and_diagnostic(self): + with tempfile.TemporaryDirectory(prefix=".no-lock-", dir=SCRIPT.parent) as temp: + root = self.fixture(temp) + with (root / "src/lib.rs").open("a") as file: + file.write( + '#[cfg(test)] unsafe extern "C" { fn missing_no_lock_probe_symbol(); }\n' + "#[test] fn linker_must_run() { unsafe { missing_no_lock_probe_symbol() } }\n" + ) + result = self.run_build(root) + self.assertEqual(result.returncode, 101, result.stderr) + self.assertIn("missing_no_lock_probe_symbol", result.stderr) + self.assertIn("linking with", result.stderr) + + def test_never_deletes_an_executable_outside_the_target_directory(self): + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + target = root / "target" + target.mkdir() + outside = root / "other-owner-binary" + outside.write_bytes(b"keep") + message = { + "reason": "compiler-artifact", + "target": {"kind": ["test"], "name": "foreign"}, + "profile": {"test": True}, + "executable": str(outside), + } + with self.assertRaisesRegex(ValueError, "outside target directory"): + BUILD.discard_test_executable(message, target) + self.assertEqual(outside.read_bytes(), b"keep") + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 6340a4e4c..efb57605b 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -833,12 +833,61 @@ except HttpError as error: print("request failed:", error.request_id, error.status_code) ``` -Secret and function lifecycle APIs currently use Flight SQL and may still raise -`RuntimeError`. The typed subclasses cover HTTP failures, including dependency -errors during column refresh. +SQL response failures raise `SqlError`, retaining the numeric `grpc_code` and +`request_id`. Servers that report a recognized `x-lancedb-error-code` select a +specific input-error subclass. HTTP failures use the corresponding `HttpError` +subclass. Missing metadata stays absent; diagnostic text is never used to guess +the error type. An Internal or Unavailable response remains a server failure, +even if its message or metadata resembles an input error. + +For errors that can arrive over either transport, catch the shared input type: + +```python +from lancedb.exceptions import ColumnNotFoundError, SchemaMismatchError +from lancedb.remote.errors import HttpError, SqlError + +try: + db.execute_query("SELECT missing FROM events").read_all() +except (ColumnNotFoundError, SchemaMismatchError) as error: + print(error.request_id, error.job_id, error.job_request_id) +except (HttpError, SqlError) as error: + print(error.request_id) +``` + +`ReferenceNotFoundError` identifies a missing branch or table reference. Table +errors are `TableNotFoundError` for HTTP and `SqlTableNotFoundError` for SQL; +both remain compatible with `lance_namespace.errors.TableNotFoundError`. + +A background SQL job still raises `JobFailedError`. Its `__cause__` is the typed +`SqlError` when available, and both carry the current `request_id` and original +worker `job_request_id`. The outer job ID can differ from `sql_job_id`. These +fields are `None` when the server does not report them. Older servers produce +generic HTTP/SQL exceptions with their transport status and current request ID. ::: lancedb.remote.errors.HttpError +::: lancedb.remote.errors.ColumnNotFoundError + +::: lancedb.remote.errors.ReferenceNotFoundError + +::: lancedb.remote.errors.SchemaMismatchError + +::: lancedb.remote.errors.SqlError + +::: lancedb.remote.errors.SqlColumnNotFoundError + +::: lancedb.remote.errors.SqlTableNotFoundError + +::: lancedb.remote.errors.SqlReferenceNotFoundError + +::: lancedb.remote.errors.SqlSchemaMismatchError + +::: lancedb.exceptions.ColumnNotFoundError + +::: lancedb.exceptions.ReferenceNotFoundError + +::: lancedb.exceptions.SchemaMismatchError + ::: lancedb.remote.errors.TableNotFoundError ::: lancedb.remote.errors.SecretNotFoundError diff --git a/python/python/lancedb/exceptions.py b/python/python/lancedb/exceptions.py index 516b54eed..e5f7c9a47 100644 --- a/python/python/lancedb/exceptions.py +++ b/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): diff --git a/python/python/lancedb/remote/errors.py b/python/python/lancedb/remote/errors.py index 7315fefe7..56b5cc34d 100644 --- a/python/python/lancedb/remote/errors.py +++ b/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.""" diff --git a/python/python/tests/test_errors.py b/python/python/tests/test_errors.py index 30e4b7d9a..8721a2bb7 100644 --- a/python/python/tests/test_errors.py +++ b/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 diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index 300c4f849..1ee6d464b 100644 --- a/python/python/tests/test_remote_db.py +++ b/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) diff --git a/python/python/tests/test_sql.py b/python/python/tests/test_sql.py index 539b88647..0ed382d93 100644 --- a/python/python/tests/test_sql.py +++ b/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") diff --git a/python/src/error.rs b/python/src/error.rs index 369b68c47..a86051e5d 100644 --- a/python/src/error.rs +++ b/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 { /// Convert to a python error based on the Lance error type @@ -34,26 +34,61 @@ impl PythonErrorExt for std::result::Result { | LanceError::DatabaseAlreadyExists { .. } | LanceError::TableAlreadyExists { .. } => self.value_error(), LanceError::TableNotFound { source, .. } => { - let Some(LanceError::Http { - request_id, - status_code, - .. - }) = source.downcast_ref::() + let Some( + request_error @ LanceError::Http { + request_id, + status_code, + source: http_source, + }, + ) = source.downcast_ref::() 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 PythonErrorExt for std::result::Result { 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 PythonErrorExt for std::result::Result { 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::() + .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, details: Option<&HttpErrorDetails>, + metadata: Option<&RemoteErrorMetadata>, is_transport_error: bool, ) -> PyResult> { - 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> { + 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(), )) } diff --git a/rust/lancedb/src/error.rs b/rust/lancedb/src/error.rs index b607bd9b8..c31167148 100644 --- a/rust/lancedb/src/error.rs +++ b/rust/lancedb/src/error.rs @@ -27,18 +27,68 @@ pub struct HttpErrorDetails { pub namespace_path: Vec, } +/// Correlation and classification metadata reported by a remote server. +/// +/// Fields are absent when the server did not supply them. Diagnostic text is +/// never used to infer an error code or a job identity. +#[cfg(feature = "remote")] +#[derive(Debug, Clone, Default, PartialEq, Eq)] +#[non_exhaustive] +pub struct RemoteErrorMetadata { + /// Machine-readable input error code, when available. + pub error_code: Option, + /// SQL job that failed, when the request ran in an isolated job. + pub job_id: Option, + /// Request executed by the isolated SQL worker. + pub job_request_id: Option, +} + +#[cfg(feature = "remote")] +impl RemoteErrorMetadata { + pub(crate) fn from_headers(headers: &reqwest::header::HeaderMap) -> Self { + let header = |name| { + headers + .get(name) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned) + }; + Self { + error_code: header("x-lancedb-error-code"), + job_id: header("x-lancedb-sql-job-id"), + job_request_id: header("x-job-request-id"), + } + } +} + #[cfg(feature = "remote")] #[derive(Debug)] pub(crate) struct HttpErrorSource { message: String, details: Option, + metadata: RemoteErrorMetadata, } #[cfg(feature = "remote")] impl HttpErrorSource { - pub(crate) fn new(message: String, response_body: Option<&str>) -> Self { + pub(crate) fn new( + message: String, + response_body: Option<&str>, + metadata: RemoteErrorMetadata, + ) -> Self { let details = response_body.and_then(parse_http_error_details); - Self { message, details } + let message = if metadata.error_code.is_some() { + response_body + .and_then(|body| serde_json::from_str::(body).ok()) + .and_then(|body| body.get("error")?.as_str().map(str::to_owned)) + .unwrap_or(message) + } else { + message + }; + Self { + message, + details, + metadata, + } } } @@ -114,7 +164,8 @@ pub struct JobFailure { /// Isolated SQL job that raised this failure, when the server supplies /// one across the Flight SQL boundary. pub sql_job_id: Option, - /// The error the job failed with, when it ran in this process. + /// The underlying error from a job run in this process or a structured + /// remote Flight SQL failure. pub source: Option>, } @@ -163,6 +214,13 @@ pub enum Error { EmbeddingFunctionNotFound { name: String, reason: String }, #[snafu(display("Column '{name}' was not found"))] ColumnNotFound { name: String }, + /// A missing query field with schema-qualified candidate paths. + #[snafu(display("{message}"))] + FieldNotFound { + name: String, + message: String, + source: BoxError, + }, #[snafu(display("Column '{name}' already exists"))] ColumnAlreadyExists { name: String }, #[snafu(display("Column '{name}' is not a computed column"))] @@ -181,6 +239,15 @@ pub enum Error { }, #[snafu(display("Schema Error: {message}"))] Schema { message: String }, + /// A SQL response retaining its gRPC status and server metadata. + #[cfg(feature = "remote")] + #[snafu(display("{message} (request_id={request_id})"))] + Sql { + message: String, + request_id: String, + grpc_code: i32, + metadata: Box, + }, #[snafu(display("Runtime error: {message}"))] Runtime { message: String }, #[snafu(display("Timeout error: {message}"))] @@ -285,9 +352,27 @@ impl From for Error { match source { lance::Error::Wrapped { error, .. } => Self::from_box_error(error), lance::Error::External { source } => Self::from_box_error(source), - lance::Error::InvalidInput { source, .. } => Self::InvalidInput { - message: source.to_string(), - }, + lance::Error::InvalidInput { + source, location, .. + } => { + if let Some(DataFusionError::SchemaError(error, _)) = source + .downcast_ref::() + .map(|error| error.find_root()) + && let datafusion_common::SchemaError::FieldNotFound { field, .. } = + error.as_ref() + { + // Preserve the known field identity before this boundary + // turns other input diagnostics into plain messages. + log::info!("Query input rejected at {location}: {source}"); + Self::ColumnNotFound { + name: field.name.clone(), + } + } else { + Self::InvalidInput { + message: source.to_string(), + } + } + } _ => Self::Lance { source }, } } @@ -332,6 +417,38 @@ impl Error { } } + /// Classification and correlation metadata supplied by a remote server. + /// + /// Unlike parsing [`Self::to_string`], this preserves absence and does not + /// mistake a user-provided identifier for an error classification. + /// + /// ``` + /// # #[cfg(feature = "remote")] + /// fn reported_job(error: &lancedb::Error) -> Option<&str> { + /// error.remote_error_metadata()?.job_id.as_deref() + /// } + /// ``` + #[cfg(feature = "remote")] + pub fn remote_error_metadata(&self) -> Option<&RemoteErrorMetadata> { + match self { + Self::Http { source, .. } => source + .downcast_ref::() + .map(|source| &source.metadata), + Self::Retry { source, .. } => source + .downcast_ref::() + .and_then(Self::remote_error_metadata), + Self::TableNotFound { source, .. } => source + .downcast_ref::() + .and_then(Self::remote_error_metadata), + Self::JobFailed { failure, .. } => failure + .source + .as_deref() + .and_then(Self::remote_error_metadata), + Self::Sql { metadata, .. } => Some(metadata), + _ => None, + } + } + fn from_box_error(mut source: Box) -> Self { source = match source.downcast::() { Ok(e) => match *e { @@ -425,6 +542,23 @@ impl From for Error { mod tests { use super::*; + #[test] + fn missing_filter_column_retains_its_type_across_lance_conversion() { + let field_error = DataFusionError::SchemaError( + Box::new(datafusion_common::SchemaError::FieldNotFound { + field: Box::new(datafusion_common::Column::from_name("missing")), + valid_fields: vec![datafusion_common::Column::from_name("id")], + }), + Box::new(None), + ); + let source = + lance::Error::invalid_input_source(Box::new(field_error.context("binding filter"))); + assert!(matches!(Error::from(source), Error::ColumnNotFound { name } if name == "missing")); + + let source = lance::Error::invalid_input("No field named missing"); + assert!(matches!(Error::from(source), Error::InvalidInput { .. })); + } + #[cfg(feature = "remote")] #[test] fn non_timeout_http_error_message_is_unchanged() { diff --git a/rust/lancedb/src/remote/client.rs b/rust/lancedb/src/remote/client.rs index 38104645d..e0722bfbc 100644 --- a/rust/lancedb/src/remote/client.rs +++ b/rust/lancedb/src/remote/client.rs @@ -10,7 +10,7 @@ use reqwest::{ use std::{collections::HashMap, future::Future, str::FromStr, sync::Arc, time::Duration}; use tokio::sync::Semaphore; -use crate::error::{Error, HttpErrorSource, Result}; +use crate::error::{Error, HttpErrorSource, RemoteErrorMetadata, Result}; use crate::remote::db::RemoteOptions; use crate::remote::retry::{ResolvedRetryConfig, RetryCounter}; use crate::utils::validate_database_name; @@ -1037,13 +1037,15 @@ impl RestfulLanceDbClient { if status.is_success() { Ok(response) } else { + let metadata = RemoteErrorMetadata::from_headers(response.headers()); let response_text = response.text().await.ok(); let source = match response_text { Some(response_text) => HttpErrorSource::new( format!("{}: {}", status, response_text), Some(&response_text), + metadata, ), - None => HttpErrorSource::new(status.to_string(), None), + None => HttpErrorSource::new(status.to_string(), None, metadata), }; Err(Error::Http { source: Box::new(source), diff --git a/rust/lancedb/src/remote/sql.rs b/rust/lancedb/src/remote/sql.rs index cd19f733e..94d8bf063 100644 --- a/rust/lancedb/src/remote/sql.rs +++ b/rust/lancedb/src/remote/sql.rs @@ -1643,25 +1643,39 @@ pub(super) fn sql_flight_error(request_id: &str, error: &FlightError) -> Error { } pub(super) fn sql_status_error(request_id: &str, status: &tonic::Status) -> Error { - let sql_job_id = status - .metadata() - .get(SQL_JOB_ID_METADATA_KEY) - .and_then(|value| value.to_str().ok()) - .filter(|value| !value.is_empty()) - .map(str::to_string); + let header = |name| { + status + .metadata() + .get(name) + .and_then(|value| value.to_str().ok()) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + }; + let metadata = crate::error::RemoteErrorMetadata { + error_code: header("x-lancedb-error-code"), + job_id: header(SQL_JOB_ID_METADATA_KEY), + job_request_id: header("x-job-request-id"), + }; + let sql_job_id = metadata.job_id.clone(); + let error = Error::Sql { + message: status.message().into(), + request_id: request_id.into(), + grpc_code: status.code() as i32, + metadata: Box::new(metadata), + }; if let Some(sql_job_id) = sql_job_id { return Error::JobFailed { job_id: Some(sql_job_id.clone()), failure: JobFailure { phase: None, - message: Some(status.message().to_string()), + message: Some(status.message().into()), retryable: None, sql_job_id: Some(sql_job_id), - source: None, + source: Some(Arc::new(error)), }, }; } - sql_error(request_id, status) + error } #[cfg(test)] diff --git a/rust/lancedb/src/remote/sql_test.rs b/rust/lancedb/src/remote/sql_test.rs index ff06ae9d6..b25ab4ddb 100644 --- a/rust/lancedb/src/remote/sql_test.rs +++ b/rust/lancedb/src/remote/sql_test.rs @@ -3459,3 +3459,65 @@ async fn view_listing_uses_bounded_sql_and_scoped_cursors() { shutdown.send(()).unwrap(); } } + +#[test] +fn sql_status_keeps_input_category_and_request_ids() { + for (code, category) in [ + (tonic::Code::InvalidArgument, "column_not_found"), + (tonic::Code::InvalidArgument, "schema_mismatch"), + (tonic::Code::NotFound, "table_not_found"), + (tonic::Code::NotFound, "reference_not_found"), + (tonic::Code::Internal, "column_not_found"), + ] { + let mut status = tonic::Status::new(code, "original diagnostic"); + status + .metadata_mut() + .insert("x-lancedb-error-code", category.parse().unwrap()); + status + .metadata_mut() + .insert("x-job-request-id", "worker-request".parse().unwrap()); + for job_id in [None, Some("sql-job-1")] { + if let Some(job_id) = job_id { + status + .metadata_mut() + .insert("x-lancedb-sql-job-id", job_id.parse().unwrap()); + } + let error = super::sql_status_error("current-request", &status); + assert_eq!( + error.remote_error_metadata().unwrap().job_id.as_deref(), + job_id + ); + let source = match &error { + Error::JobFailed { + job_id: actual_job_id, + failure, + } => { + assert_eq!(actual_job_id.as_deref(), job_id); + assert_eq!(failure.sql_job_id.as_deref(), job_id); + failure.source.as_deref().unwrap() + } + error => { + assert!(job_id.is_none()); + error + } + }; + let Error::Sql { + message, + request_id, + grpc_code, + metadata, + } = source + else { + panic!("SQL status must remain structured: {source}"); + }; + assert_eq!(message, "original diagnostic"); + assert_eq!(request_id, "current-request"); + assert_eq!(*grpc_code, code as i32); + assert_eq!(metadata.error_code.as_deref(), Some(category)); + assert_eq!(metadata.job_id.as_deref(), job_id); + assert_eq!(metadata.job_request_id.as_deref(), Some("worker-request")); + } + } + let error = super::sql_status_error("request", &tonic::Status::not_found("column_not_found")); + assert!(error.remote_error_metadata().unwrap().error_code.is_none()); +} diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 590f4866f..3b241dd3a 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -12,6 +12,7 @@ use super::sql::SqlClient; use super::{ARROW_FILE_CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE, extract_job_id}; use crate::blob::BlobFile; use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions}; +use crate::error::{HttpErrorSource, RemoteErrorMetadata}; use crate::expr::expr_to_sql_string; use crate::index::Index; use crate::index::IndexStatistics; @@ -1086,9 +1087,10 @@ impl RemoteTable { ) -> Result { let status = response.status(); if status == StatusCode::NOT_FOUND { + let metadata = RemoteErrorMetadata::from_headers(response.headers()); let body = response.text().await.ok().unwrap_or_default(); let request_error = Error::Http { - source: body.into(), + source: Box::new(HttpErrorSource::new(body.clone(), Some(&body), metadata)), request_id: request_id.into(), status_code: Some(status), }; diff --git a/rust/lancedb/src/table/query.rs b/rust/lancedb/src/table/query.rs index 2f5c75766..8a72872e8 100644 --- a/rust/lancedb/src/table/query.rs +++ b/rust/lancedb/src/table/query.rs @@ -460,6 +460,7 @@ fn enrich_lance_field_not_found( let Some(field) = find_missing_field(&error) else { return error.into(); }; + log::info!("Query field rejected: {error}"); field_not_found_error(field, &Schema::from(schema)) } @@ -480,7 +481,14 @@ fn field_not_found_error(field: &Column, schema: &Schema) -> Error { } message.push('.'); - Error::InvalidInput { message } + Error::FieldNotFound { + name: field.name.clone(), + message, + source: Box::new(lance::Error::field_not_found( + field.name.clone(), + valid_fields, + )), + } } fn find_missing_field<'a>(error: &'a (dyn std::error::Error + 'static)) -> Option<&'a Column> { @@ -1603,6 +1611,13 @@ mod tests { Box::new(None), ); let error = field_not_found_diagnostic(&source, &schema).unwrap(); + assert!(matches!(&error, Error::FieldNotFound { name, .. } if name == "missing")); + assert!(matches!( + std::error::Error::source(&error) + .unwrap() + .downcast_ref::(), + Some(lance::Error::FieldNotFound { .. }) + )); assert!( error.to_string().contains( "Valid fields are `a`.`b`.`c`.`d`.`e`, `metadata`.`child.with.dot`, `metadata`.`Title`, `metadata`.`123child`, `metadata`.`child``tick`"