fix(managed): raise when a succeeded run saved no result

A run succeeds with no result id when its rows came back inline but the
result could not be saved for later retrieval. `fetch_table` answered
`None` for that, `fetch_table_rows` turns `None` into `[]`, and `[]` is
also the answer both give for a table that is not synced -- so the two
states were indistinguishable to a caller.

A read-modify-write load would therefore read no existing rows and write
only its new batch, dropping every row already in the table. Silent data
loss is the worst outcome on offer, and raising costs nothing if the
state never occurs. Terminal rather than transient: re-running the query
cannot save a result that was already discarded. The run's
`warning_message` goes into the error, since it is the only thing that
explains why.

The test this replaces asserted the old behaviour -- it pinned
`fetch_table` returning `None` for exactly this case, which committed the
data-loss path to the suite rather than catching it.

Author: anoop-narang

hotdata_framework/managed_client.py

@@ -185,6 +185,22 @@ def _await_query_run(self, query_run_id: str, *, database_id: str) -> str | None
             run = runs.get_query_run(query_run_id, x_database_id=database_id)
             last_status = run.status
             if run.status == "succeeded":
+                if run.result_id is None:
+                    # A run succeeds with no result id when its rows were
+                    # returned inline but the result could not be saved for
+                    # later retrieval. Returning nothing here would surface as
+                    # an empty table -- `fetch_table` answers `None`, and
+                    # `fetch_table_rows` turns that into `[]`, which is the same
+                    # answer it gives for a table that does not exist. A
+                    # read-modify-write load would then read no existing rows
+                    # and write only its new batch, dropping what was there.
+                    # Terminal rather than transient: re-running the query
+                    # cannot save a result that was already discarded.
+                    raise RuntimeError(
+                        f"Query run {query_run_id} succeeded but its result was not "
+                        f"saved, so the table cannot be read"
+                        + (f": {run.warning_message}" if run.warning_message else "")
+                    )
                 return run.result_id
             if run.status == "interrupted":
                 # Terminal, but the server lost the run rather than rejecting

tests/test_managed_client.py

@@ -302,15 +302,21 @@ def get_query_run(self, query_run_id: str, **kwargs: Any) -> Any:
     assert calls.count("query") == 1
 
 
-def test_result_id_is_read_off_the_run_not_the_query_reply(
+def test_a_succeeded_run_that_saved_no_result_raises_rather_than_reading_empty(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """A ``succeeded`` run reports ``result_id: null`` when every row came back
-    inline but the result could not be saved for later retrieval.
-
-    The query reply's own ``result_id`` is optimistic -- it names a result that
-    may never resolve -- so the run's is the one to believe. Believing the reply
-    means fetching Arrow against an id the server will 404.
+    """The one state where an empty answer would be a wrong answer.
+
+    A run succeeds with no `result_id` when its rows came back inline but the
+    result could not be saved. Answering `None` puts that on the same footing as
+    a table that does not exist -- `fetch_table_rows` turns both into `[]` -- so
+    a read-modify-write load would read no existing rows and write only its new
+    batch, dropping the rows already in the table. Silent data loss is the worst
+    outcome available here, so this raises, and raises terminally: re-running
+    cannot save a result that was already discarded.
+
+    The query reply carries an id of its own, and it must not be believed over
+    the run's.
     """
     arrow_calls: list[str] = []
 
@@ -328,7 +334,12 @@ def __init__(self, api: object) -> None:
 
         def get_query_run(self, query_run_id: str, **kwargs: Any) -> Any:
             # ...that the run says was never saved.
-            return SimpleNamespace(status="succeeded", result_id=None, error_message=None)
+            return SimpleNamespace(
+                status="succeeded",
+                result_id=None,
+                error_message=None,
+                warning_message="result row creation failed; result not persisted",
+            )
 
     class FakeArrowResultsApi:
         def __init__(self, api: object) -> None:
@@ -341,20 +352,66 @@ def get_result_arrow(self, result_id: str, **kwargs: Any) -> pa.Table:
     monkeypatch.setattr(mc, "QueryApi", FakeQueryApi)
     monkeypatch.setattr(mc, "QueryRunsApi", FakeQueryRunsApi)
     monkeypatch.setattr(mc, "ArrowResultsApi", FakeArrowResultsApi)
+    monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None)
 
     client = mc.ManagedDatabaseClient(
         api_key="k",
         workspace_id="w",
         api_base_url="https://example.test",
-        max_retries=1,
+        max_retries=3,
         retry_backoff_seconds=0.0,
     )
     client._runtime = _fake_runtime()
 
-    assert client.fetch_table(database="mydb", schema="public", table="orders") is None
+    with pytest.raises(HotdataTerminalError, match="result was not saved"):
+        client.fetch_table(database="mydb", schema="public", table="orders")
+
+    # The reply's id was never used.
     assert arrow_calls == []
 
 
+def test_fetch_table_rows_cannot_turn_an_unsaved_result_into_no_rows(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """`fetch_table_rows` is where an empty answer does the damage.
+
+    It maps `None` to `[]`, which is also its answer for a table that is not
+    synced, so the unsaved-result case must not be able to reach that mapping.
+    """
+
+    class FakeQueryApi:
+        def __init__(self, api: object) -> None:
+            pass
+
+        def query(self, request: object, *, x_database_id: str) -> QueryResponse:
+            return _query_response("rslt1")
+
+    class FakeQueryRunsApi:
+        def __init__(self, api: object) -> None:
+            pass
+
+        def get_query_run(self, query_run_id: str, **kwargs: Any) -> Any:
+            return SimpleNamespace(
+                status="succeeded", result_id=None, error_message=None, warning_message=None
+            )
+
+    monkeypatch.setattr(mc, "QueryApi", FakeQueryApi)
+    monkeypatch.setattr(mc, "QueryRunsApi", FakeQueryRunsApi)
+    monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None)
+
+    client = mc.ManagedDatabaseClient(
+        api_key="k",
+        workspace_id="w",
+        api_base_url="https://example.test",
+        max_retries=1,
+        retry_backoff_seconds=0.0,
+    )
+    client._runtime = _fake_runtime()
+
+    with pytest.raises(HotdataTerminalError):
+        client.fetch_table_rows(database="mydb", schema="public", table="orders")
+
+
 def _load_recording_runtime(calls: list[str], uploads: list[str] | None = None) -> SimpleNamespace:
     """A runtime whose ``load_managed_table`` records each mode and always fails
     with a transient error, so retry behaviour is observable via ``calls``.