diff --git a/CHANGELOG.md b/CHANGELOG.md index 877e101..174cc4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,21 @@ Changelog style — terse "bullet + brief why": **Breaking:**. See the released entries below for the target density. --> +### Fixed + +- `HotdataSourceClient` now defines its own `_poll` rather than inheriting one from + `ManagedDatabaseClient`, which no longer provides it: the framework's managed + client waits on the query run instead of on the result body, and dropped the + shared helper with it. Both readiness waits here would otherwise raise + `AttributeError` on the next `hotdata-framework` release this pin accepts. The + local copy treats `interrupted` as terminal — it was absent, so an interrupted + run waited out the full 300s bound and then reported a timeout — and drops + `cancelled`, which this API does not send; an unrecognised status keeps waiting + and the timeout names it. The wait reads its own `_POLL_TIMEOUT_SECONDS` / + `_POLL_SLEEP_SECONDS` rather than the base class's constants, so it rents + nothing -- and under distinct names, which avoids shadowing the base values for + inherited waits. + ## [0.15.0] - 2026-08-27 ### Changed diff --git a/src/hotdata_dlt_destination/source_client.py b/src/hotdata_dlt_destination/source_client.py index e77980b..35cbc5a 100644 --- a/src/hotdata_dlt_destination/source_client.py +++ b/src/hotdata_dlt_destination/source_client.py @@ -44,13 +44,14 @@ trusted the persisted result is taken regardless of cost: slower and complete beats faster and silently short. The count is a second request rather than folded into the readiness poll because -the poll lives in the shared client, and duplicating its sync/async branching to -save ~12% of the latency would risk the two drifting. +the poll would then have to branch on sync/async the way the read above it does, +and duplicating that to save ~12% of the latency would risk the two drifting. """ from __future__ import annotations -from typing import TYPE_CHECKING, Any, TypeVar +import time +from typing import TYPE_CHECKING, Any, Protocol, TypeVar import pyarrow as pa import pyarrow.types as pat @@ -64,7 +65,7 @@ from hotdata_framework.managed_client import ManagedDatabaseClient if TYPE_CHECKING: - from collections.abc import Iterator + from collections.abc import Callable, Iterator # Rows per Arrow batch handed to dlt. Bounds peak memory for one batch; it does # not bound the query, whose size the caller controls with its own LIMIT. @@ -75,6 +76,16 @@ T = TypeVar("T", pa.Table, pa.RecordBatch) +class _HasStatus(Protocol): + """What `_poll` needs off a run or a result: a status, and a reason.""" + + status: str + error_message: str | None + + +_S = TypeVar("_S", bound=_HasStatus) + + class IncompleteReadError(RuntimeError): """A read returned fewer rows than the result holds. @@ -320,6 +331,58 @@ def _resolve(self, response: object, *, database_id: str) -> tuple[str, int] | N "so whether it produced rows cannot be established" ) + # This wait's own bounds. `_poll` reading two private attributes off a base + # class in another distribution is the coupling that broke these waits once + # already, so it reads its own -- but under names of their own rather than + # the base class's, which the first attempt reused. Matching those names + # would shadow them for every *inherited* method too, so a framework release + # that retuned either value would silently stop reaching this client. + _POLL_TIMEOUT_SECONDS = 300.0 + _POLL_SLEEP_SECONDS = 0.4 + + # Statuses that end a wait without satisfying it. `interrupted` is terminal + # and belongs here -- omitting it made an interrupted run wait out the whole + # 300s bound and then report a timeout, which names the wrong problem. + # `cancelled`, which this poll used to list, is not a status the API sends. + _POLL_FAILURES = frozenset({"failed", "interrupted"}) + + def _poll( + self, + fetch: Callable[[], _S], + *, + is_ready: Callable[[_S], bool], + describe: str, + ) -> _S: + """Poll ``fetch`` until ``is_ready`` holds, or raise on failure/timeout. + + Owned here rather than inherited. The base class used to provide this and + no longer does: it now waits on the query run instead, so the generic + helper had no second caller left there. This client still needs a wait of + its own shape -- retry per request, not per wait, and ``limit=0`` on the + readiness check -- so it keeps one, and a private method on a base class + in another package was never a safe thing to build on. + + An unrecognised status keeps polling rather than ending the wait. Calling + an unknown status terminal would be easier to diagnose and much worse to + live with: one status added upstream would fail every read at once, where + waiting costs a single slow call. So the timeout names the status it last + saw, which is what makes an omission findable. + """ + deadline = time.monotonic() + self._POLL_TIMEOUT_SECONDS + last_status: str | None = None + while time.monotonic() < deadline: + obj = fetch() + last_status = obj.status + if is_ready(obj): + return obj + if obj.status in self._POLL_FAILURES: + raise RuntimeError(obj.error_message or f"{describe} {obj.status}") + time.sleep(self._POLL_SLEEP_SECONDS) + raise TimeoutError( + f"{describe} timed out after {self._POLL_TIMEOUT_SECONDS}s " + f"(last status: {last_status})" + ) + def _await_result(self, result_id: str, *, database_id: str) -> tuple[str, int]: """Wait for a result to be ready, and take its row count off the way out. @@ -338,9 +401,8 @@ def _await_result(self, result_id: str, *, database_id: str) -> tuple[str, int]: response because the total is a HEADER: reading it means the ``_with_http_info`` variant, whose wrapper puts the model one level down at ``.data``, and ``_poll`` inspects ``obj.status`` directly to spot a - failed or cancelled result. Passing the wrapper in breaks that check, so - the poll keeps the plain model and the header is fetched once, after - ready. + failed result. Passing the wrapper in breaks that check, so the poll + keeps the plain model and the header is fetched once, after ready. """ results = ResultsApi(self._runtime.api) self._poll( diff --git a/tests/test_client.py b/tests/test_client.py index 74e5890..903a9fb 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -292,7 +292,7 @@ def test_list_managed_tables_by_id(monkeypatch) -> None: def _patch_query_apis(monkeypatch, arrow_table, *, arrow_in_hotdata_client: bool) -> dict: from hotdata.models.query_response import QueryResponse as _QR - scopes: dict[str, list] = {"result": [], "arrow": []} + scopes: dict[str, list] = {"run": [], "result": [], "arrow": []} class FakeQueryApi: def __init__(self, api): @@ -311,11 +311,28 @@ def query(self, request, *, x_database_id): execution_time_ms=1, ) + # Which endpoint answers "is it ready?" depends on the installed framework: + # up to 0.13 it polls `GET /results/{id}`, after that it waits on the query + # run. Both are faked so this suite does not silently bind to one of them -- + # the assertion below accepts whichever was used. + class FakeQueryRunsApi: + def __init__(self, api): + pass + + def get_query_run(self, query_run_id, *, x_database_id=None): + scopes["run"].append(x_database_id) + return SimpleNamespace( + status="succeeded", + result_id="r1", + error_message=None, + warning_message=None, + ) + class FakeResultsApi: def __init__(self, api): pass - def get_result(self, result_id, *, x_database_id=None): + def get_result(self, result_id, *, x_database_id=None, limit=None): scopes["result"].append(x_database_id) return SimpleNamespace(status="ready", result_id=result_id, error_message=None) @@ -328,7 +345,12 @@ def get_result_arrow(self, result_id, *, x_database_id): return arrow_table monkeypatch.setattr("hotdata_framework.managed_client.QueryApi", FakeQueryApi) - monkeypatch.setattr("hotdata_framework.managed_client.ResultsApi", FakeResultsApi) + monkeypatch.setattr("hotdata_framework.managed_client.QueryRunsApi", FakeQueryRunsApi) + # `raising=False`: the symbol is gone from the module in later framework + # versions, which stop reading a result body to learn its status. + monkeypatch.setattr( + "hotdata_framework.managed_client.ResultsApi", FakeResultsApi, raising=False + ) monkeypatch.setattr("hotdata_framework.managed_client.ArrowResultsApi", FakeArrowResultsApi) if arrow_in_hotdata_client: monkeypatch.setattr( @@ -346,7 +368,7 @@ def test_execute_sql_carries_database_scope(monkeypatch) -> None: client = _client(_Runtime(), config=_cfg(database_id="db_99")) table = client.execute_sql('SELECT * FROM "default"."public"."spans"') assert table.num_rows == 2 - assert scopes["result"] == ["db_99"] + assert (scopes["run"] or scopes["result"]) == ["db_99"] assert scopes["arrow"] == ["db_99"] client.close() @@ -360,7 +382,7 @@ def test_fetch_table_carries_database_scope(monkeypatch) -> None: client = _client(rt, config=_cfg(database_id="db_42")) table = client.fetch_table(schema="public", table="orders") assert table is not None and table.num_rows == 1 - assert scopes["result"] == ["db_42"] + assert (scopes["run"] or scopes["result"]) == ["db_42"] assert scopes["arrow"] == ["db_42"] client.close() diff --git a/tests/test_e2e_inmemory.py b/tests/test_e2e_inmemory.py index a3e1f0b..e181640 100644 --- a/tests/test_e2e_inmemory.py +++ b/tests/test_e2e_inmemory.py @@ -44,6 +44,9 @@ def __init__(self) -> None: # (db, schema, table) -> (partition_by, sorted_by) as declared at creation self.layouts: dict[tuple, tuple[list, list]] = {} self.tables: dict[tuple, pa.Table] = {} + # The id minted by the most recent query, so the query-run fake can name + # the result the run produced. + self.last_result_id: str | None = None self.uploads: dict[str, pa.Table] = {} self.results: dict[str, pa.Table] = {} self.load_counts: dict[tuple, int] = {} @@ -210,6 +213,10 @@ def query(self, request, *, x_database_id): be = _ACTIVE["backend"] database = be.id_to_name[x_database_id] result = be.execute_sql(database, request.sql) + # The run is what names the result now, so the id has to be reachable + # from the run fake as well as from this reply. + rid = be.store_result(result) + be.last_result_id = rid return QueryResponse( columns=[], rows=[], @@ -217,20 +224,42 @@ def query(self, request, *, x_database_id): preview_row_count=0, truncated=False, nullable=[], - result_id=be.store_result(result), + result_id=rid, query_run_id="qr", execution_time_ms=1, ) class _FakeResultsApi: + """Readiness, as frameworks up to 0.13 ask for it.""" + def __init__(self, api): pass - def get_result(self, result_id, *, x_database_id=None): + def get_result(self, result_id, *, x_database_id=None, limit=None): return SimpleNamespace(status="ready", result_id=result_id, error_message=None) +class _FakeQueryRunsApi: + """Readiness, and the id of the result the run produced. + + Later frameworks wait on the query run rather than polling the result body + for a status. Both fakes are installed so this suite works either side of + that change instead of binding to one framework version. + """ + + def __init__(self, api): + pass + + def get_query_run(self, query_run_id, *, x_database_id=None): + return SimpleNamespace( + status="succeeded", + result_id=_ACTIVE["backend"].last_result_id, + error_message=None, + warning_message=None, + ) + + class _FakeArrowResultsApi: def __init__(self, api): pass @@ -265,7 +294,10 @@ def backend(monkeypatch): be = InMemoryBackend() _ACTIVE["backend"] = be monkeypatch.setattr(mc, "QueryApi", _FakeQueryApi) - monkeypatch.setattr(mc, "ResultsApi", _FakeResultsApi) + monkeypatch.setattr(mc, "QueryRunsApi", _FakeQueryRunsApi) + # `raising=False`: later frameworks drop this import, having stopped reading + # a result body to learn its status. + monkeypatch.setattr(mc, "ResultsApi", _FakeResultsApi, raising=False) monkeypatch.setattr(mc, "ArrowResultsApi", _FakeArrowResultsApi) # hotdata_client.py also references ArrowResultsApi (for execute_sql) and # DatabasesApi (for the id-first bind). diff --git a/tests/test_source_client.py b/tests/test_source_client.py index 8794f14..39a6026 100644 --- a/tests/test_source_client.py +++ b/tests/test_source_client.py @@ -1139,3 +1139,72 @@ def test_read_rows_returns_empty_for_an_unpersisted_but_verified_zero() -> None: client = FakeSourceClient(table=empty, truncated=False, result_id=None) assert client.read_rows("SELECT * FROM public.t", database_id=DB) == [] assert client.fetch_calls == 0 + + +def test_the_readiness_poll_is_owned_here_not_inherited() -> None: + """`_poll` must be defined on this class, not merely reachable through it. + + It used to live on `ManagedDatabaseClient`, and this client called it without + defining one. When that base stopped waiting on the result and started + waiting on the query run, the generic helper lost its second caller there and + was removed -- and both readiness waits here would have become + `AttributeError` on the next framework release that this package's pin + accepted. Nothing in this suite would have caught it, because the method + resolved fine against the version installed at the time. + + Reaching into a private method on a base class from another distribution is + the thing that made that possible, so the assertion is about ownership rather + than about the attribute merely existing. + """ + for name in ("_poll", "_POLL_TIMEOUT_SECONDS", "_POLL_SLEEP_SECONDS"): + assert name in HotdataSourceClient.__dict__, ( + f"{name} must be defined on HotdataSourceClient; inheriting it from " + "hotdata-framework couples this package to a private name it does not own" + ) + + +def test_poll_stops_on_a_run_that_was_interrupted() -> None: + """`interrupted` is terminal, and was in neither status this poll knew. + + An interrupted run therefore satisfied no branch, so the wait ran to its full + 300-second bound and then reported a timeout -- naming the wrong problem, and + charging five minutes for it. + """ + client = HotdataSourceClient.__new__(HotdataSourceClient) + calls: list[str] = [] + + def fetch() -> SimpleNamespace: + calls.append("poll") + return SimpleNamespace(status="interrupted", error_message="instance lost") + + with pytest.raises(RuntimeError, match="instance lost"): + client._poll(fetch, is_ready=lambda r: r.status == "ready", describe="Result r1") + + # One request, not a timeout's worth. + assert calls == ["poll"] + + +def test_poll_keeps_waiting_on_an_unknown_status_and_names_it_on_timeout() -> None: + """An unrecognised status waits rather than ending the wait. + + Treating an unknown status as terminal is the easier failure to diagnose and + much the worse one to suffer: a single status added upstream would fail every + read at once, where waiting costs one slow call. What makes the omission + findable is the timeout naming the status it last saw -- which is exactly + what was missing while `interrupted` went unrecognised. + """ + client = HotdataSourceClient.__new__(HotdataSourceClient) + + def fetch() -> SimpleNamespace: + return SimpleNamespace(status="something_new", error_message=None) + + with ( + pytest.MonkeyPatch.context() as mp, + pytest.raises(TimeoutError, match="something_new"), + ): + # Both constants belong to this class -- and are deliberately not named + # after the base class's, which these would otherwise shadow for every + # inherited wait as well. + mp.setattr(HotdataSourceClient, "_POLL_TIMEOUT_SECONDS", 0.05) + mp.setattr(HotdataSourceClient, "_POLL_SLEEP_SECONDS", 0.0) + client._poll(fetch, is_ready=lambda r: r.status == "ready", describe="Result r1")