Skip to content

Commit db0ff57

Browse files
committed
fix(managed_client): wait for result ready on sync query path
fetch_table fetched the persisted result as Arrow using the result_id from a synchronous QueryResponse without waiting for it to reach 'ready'. Against the live async backend the result is often still 'processing', so Arrow fetches failed on every read-modify-write (merge/append) and state read. The async path already waited; the sync path now does too. Adds a regression test driving the sync path with a 'processing' -> 'ready' result, asserting Arrow is fetched only after readiness.
1 parent 5ec8803 commit db0ff57

2 files changed

Lines changed: 98 additions & 1 deletion

File tree

‎hotdata_framework/managed_client.py‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,11 @@ def _query_database_scoped(self, sql: str, *, database_id: str) -> str | None:
108108
x_database_id=database_id,
109109
)
110110
if isinstance(raw, QueryResponse):
111-
return raw.result_id
111+
# A synchronous response still persists its full result out-of-band
112+
# under ``result_id``; that result may be ``processing`` when the
113+
# inline preview returns, so wait for ``ready`` before the caller
114+
# fetches it as Arrow.
115+
return self._wait_result_ready(raw.result_id)
112116

113117
if isinstance(raw, AsyncQueryResponse):
114118
runs = QueryRunsApi(self._runtime.api)

‎tests/test_managed_client.py‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
"""Regression tests for ManagedDatabaseClient result handling."""
2+
3+
from __future__ import annotations
4+
5+
from types import SimpleNamespace
6+
from typing import Any
7+
8+
import pyarrow as pa
9+
import pytest
10+
from hotdata.models.query_response import QueryResponse
11+
12+
import hotdata_framework.managed_client as mc
13+
14+
15+
def _query_response(result_id: str) -> QueryResponse:
16+
return QueryResponse(
17+
columns=[],
18+
rows=[],
19+
row_count=0,
20+
preview_row_count=0,
21+
truncated=False,
22+
nullable=[],
23+
result_id=result_id,
24+
query_run_id="qr",
25+
execution_time_ms=1,
26+
)
27+
28+
29+
def test_fetch_table_waits_for_ready_before_arrow(monkeypatch: pytest.MonkeyPatch) -> None:
30+
"""A synchronous ``QueryResponse`` persists its full result out-of-band, and
31+
that result can still be ``processing`` when the inline preview returns.
32+
33+
``fetch_table`` must poll the result to ``ready`` before fetching it as
34+
Arrow. The earlier bug returned the ``result_id`` immediately on the sync
35+
path, so Arrow was fetched against a ``processing`` result and failed.
36+
"""
37+
calls: list[str] = []
38+
39+
class FakeQueryApi:
40+
def __init__(self, api: object) -> None:
41+
pass
42+
43+
def query(self, request: object, *, x_database_id: str) -> QueryResponse:
44+
calls.append("query")
45+
return _query_response("rslt1")
46+
47+
statuses = iter(["processing", "processing", "ready"])
48+
49+
class FakeResultsApi:
50+
def __init__(self, api: object) -> None:
51+
pass
52+
53+
def get_result(self, result_id: str) -> Any:
54+
status = next(statuses)
55+
calls.append(f"get_result:{status}")
56+
return SimpleNamespace(status=status, result_id=result_id, error_message=None)
57+
58+
class FakeArrowResultsApi:
59+
def __init__(self, api: object) -> None:
60+
pass
61+
62+
def get_result_arrow(self, result_id: str) -> pa.Table:
63+
calls.append("arrow")
64+
return pa.table({"id": [1, 2]})
65+
66+
monkeypatch.setattr(mc, "QueryApi", FakeQueryApi)
67+
monkeypatch.setattr(mc, "ResultsApi", FakeResultsApi)
68+
monkeypatch.setattr(mc, "ArrowResultsApi", FakeArrowResultsApi)
69+
monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None)
70+
71+
client = mc.ManagedDatabaseClient(
72+
api_key="k",
73+
workspace_id="w",
74+
api_base_url="https://example.test",
75+
max_retries=1,
76+
retry_backoff_seconds=0.0,
77+
)
78+
client._runtime = SimpleNamespace( # type: ignore[assignment]
79+
api=object(),
80+
resolve_managed_database=lambda name: SimpleNamespace(id="db1", default_connection_id="c"),
81+
list_managed_tables=lambda database, schema=None: [
82+
SimpleNamespace(table="orders", synced=True)
83+
],
84+
)
85+
86+
table = client.fetch_table(database="mydb", schema="public", table="orders")
87+
88+
assert table is not None
89+
assert table.num_rows == 2
90+
# The result was polled to readiness, and Arrow was fetched only afterwards.
91+
assert "get_result:processing" in calls
92+
assert "get_result:ready" in calls
93+
assert calls.index("arrow") > calls.index("get_result:ready")

0 commit comments

Comments
 (0)