mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-13 14:16:54 +00:00
feat(sql): return durable materialized view job handles (#4824)
Materialized-view creation and refresh can outlive the submitting connection, but their SQL query IDs cannot be reopened through the Job Registry. On servers that negotiate early handles, return the durable registry job ID and use the existing job API to describe, reopen, wait for, or cancel it. Older servers retain the connection-scoped query handle. Registry refresh handles forward running and terminal progress observations and establish the committed-version read-freshness fence before wait returns. Preserve a nested SQL failure ID separately from the outer job ID through Rust and the Python native bindings, so Function failures remain traceable. Rust callers constructing `JobFailure` directly must supply the new optional `sql_job_id` field or use `..Default::default()`. Validation: seven SDK regression cases and the docs-link test pass, including plain wait and running-to-terminal progress observations; the observer regression fails before the fix. Formatting and strict Clippy pass. Earlier accepted coverage includes a real Flight test service exercising materialized-view creation followed by registry describe/reopen/cancel, and two Python tests after rebuilding the native extension. The published product tree matches the validated candidate. Refs ENT-3562. The matching Sophon server change is being validated separately before publication.
This commit is contained in:
@@ -466,6 +466,8 @@ class JobFailureInfo:
|
||||
def message(self) -> Optional[str]: ...
|
||||
@property
|
||||
def retryable(self) -> Optional[bool]: ...
|
||||
@property
|
||||
def sql_job_id(self) -> Optional[str]: ...
|
||||
|
||||
class JobDescription:
|
||||
@property
|
||||
|
||||
@@ -739,11 +739,14 @@ class DBConnection(EnforceOverrides):
|
||||
"""Submit materialized-view creation and return its job.
|
||||
|
||||
The job may already be complete in LanceDB Embedded. In LanceDB
|
||||
Enterprise, creation
|
||||
submits CREATE MATERIALIZED VIEW through the SQL server. Its job has
|
||||
``job_type == "query"`` and ``id`` identifies the connection-scoped
|
||||
query. Registry event history is unavailable.
|
||||
Distributed, creation submits CREATE MATERIALIZED VIEW through the SQL
|
||||
server and the returned ``id`` is the registry job ID. That ID can be
|
||||
reopened with :meth:`open_job`, and the handle supports registry
|
||||
progress, events, cancellation, and ``wait`` while the job runs.
|
||||
Wait for the job before opening or querying the view.
|
||||
|
||||
Servers without early registry-handle support return a connection-scoped
|
||||
query job instead. Its ID cannot be reopened with :meth:`open_job`.
|
||||
"""
|
||||
raise NotImplementedError(
|
||||
_append_doc_url(
|
||||
|
||||
@@ -3,6 +3,8 @@
|
||||
|
||||
"""Custom exception handling"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class MissingValueError(ValueError):
|
||||
"""Exception raised when a required value is missing."""
|
||||
@@ -26,9 +28,22 @@ class MissingColumnError(KeyError):
|
||||
|
||||
|
||||
class JobFailedError(RuntimeError):
|
||||
"""Exception raised when an asynchronous job reaches the failed state."""
|
||||
"""Exception raised when an asynchronous job reaches the failed state.
|
||||
|
||||
pass
|
||||
``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.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
message: str,
|
||||
job_id: Optional[str] = None,
|
||||
sql_job_id: Optional[str] = None,
|
||||
):
|
||||
super().__init__(message)
|
||||
self.job_id = job_id
|
||||
self.sql_job_id = sql_job_id
|
||||
|
||||
|
||||
class JobCancelledError(RuntimeError):
|
||||
|
||||
@@ -208,9 +208,14 @@ class AsyncMaterializedView:
|
||||
"""Submit a refresh and return its job without waiting.
|
||||
|
||||
The job may already be complete in LanceDB Embedded. In LanceDB
|
||||
Enterprise, refreshes are submitted through the SQL service. Options and
|
||||
Distributed, refreshes are submitted through the SQL service and the
|
||||
returned ``id`` is the registry job ID, usable with ``open_job`` for
|
||||
progress, events, and cancellation while the refresh runs. Options and
|
||||
result metrics follow
|
||||
[refresh][lancedb.materialized_view.AsyncMaterializedView.refresh].
|
||||
|
||||
Servers without early registry-handle support return a connection-scoped
|
||||
query job instead. Its ID cannot be reopened with ``open_job``.
|
||||
"""
|
||||
from ._lancedb import RefreshMaterializedViewResult
|
||||
|
||||
|
||||
@@ -2268,8 +2268,10 @@ def test_remote_job_wait_raises_on_failure():
|
||||
with mock_lancedb_connection(handler) as db:
|
||||
table = db.create_table("test", [{"id": 1}])
|
||||
job = table.create_index_async("id", config=BTree())
|
||||
with pytest.raises(JobFailedError, match="job-2"):
|
||||
with pytest.raises(JobFailedError, match="job-2") as exc_info:
|
||||
job.wait()
|
||||
assert exc_info.value.job_id == "job-2"
|
||||
assert exc_info.value.sql_job_id is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("server_version", ["0.2.0", "0.6.0"])
|
||||
@@ -5123,7 +5125,10 @@ def test_remote_connection_jobs_surface():
|
||||
waiting_for_job_id="job-maintenance",
|
||||
paused_at_millis=1200,
|
||||
failure=dict(
|
||||
phase="execute", message="worker died", retryable=True
|
||||
phase="execute",
|
||||
message="worker died",
|
||||
retryable=True,
|
||||
sql_job_id="sql-job-1",
|
||||
),
|
||||
)
|
||||
).encode()
|
||||
@@ -5211,13 +5216,17 @@ def test_remote_connection_jobs_surface():
|
||||
assert progress["waiting_for_job_id"] == "job-maintenance"
|
||||
assert progress["paused_at_millis"] == 1200
|
||||
assert job.failure.message == "worker died"
|
||||
assert job.failure.sql_job_id == "sql-job-1"
|
||||
assert repr(job.failure) == (
|
||||
'JobFailureInfo(phase="execute", message="worker died", retryable=True)'
|
||||
'JobFailureInfo(phase="execute", message="worker died", retryable=True, '
|
||||
'sql_job_id="sql-job-1")'
|
||||
)
|
||||
assert job.status() == "failed"
|
||||
observed = []
|
||||
with pytest.raises(JobFailedError, match="worker died"):
|
||||
with pytest.raises(JobFailedError, match="worker died") as exc_info:
|
||||
job.wait(timeout=timedelta(seconds=5), progress=observed.append)
|
||||
assert exc_info.value.job_id == "job-1"
|
||||
assert exc_info.value.sql_job_id == "sql-job-1"
|
||||
assert observed[0]["state"] == "failed"
|
||||
assert observed[0]["rows_computed"] == 4
|
||||
assert finished.progress()["paused"] is False
|
||||
|
||||
Reference in New Issue
Block a user