From d0e6bb8b54a96b4591ae4320fff6c0c57d4f9303 Mon Sep 17 00:00:00 2001 From: jallegri <18682329+jallegri@users.noreply.github.com> Date: Tue, 27 Oct 2026 21:56:43 -0300 Subject: [PATCH] fix(query): preserve result metadata and delete diagnostics --- CHANGELOG.md | 2 + src/arc_client/models/delete.py | 32 +++- src/arc_client/models/query.py | 11 +- src/arc_client/query/executor.py | 4 +- src/arc_client/query/msgpack_response.py | 35 +++- tests/unit/http/test_delete_diagnostics.py | 162 +++++++++++++++++ .../http/test_query_metadata_contracts.py | 168 ++++++++++++++++++ 7 files changed, 405 insertions(+), 9 deletions(-) create mode 100644 tests/unit/http/test_delete_diagnostics.py create mode 100644 tests/unit/http/test_query_metadata_contracts.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4baaefc..bc56ffe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,6 +66,8 @@ plus bug fixes and the first real HTTP-layer test coverage. ### Fixed +- MessagePack queries preserve cap metadata and exact timestamp nanoseconds. + DELETE results retain partial-failure, cold-storage and peer diagnostics. - Retention and continuous-query clients accept Arc's list/null/flat response shapes. Sparse updates remain sparse; explicit full replacements support released servers, and CLEAR requests explicit null values on newer servers. diff --git a/src/arc_client/models/delete.py b/src/arc_client/models/delete.py index 9196127..1640424 100644 --- a/src/arc_client/models/delete.py +++ b/src/arc_client/models/delete.py @@ -2,9 +2,9 @@ from __future__ import annotations -from typing import Optional +from typing import Any, Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator class DeleteResponse(BaseModel): @@ -18,6 +18,13 @@ class DeleteResponse(BaseModel): execution_time_ms: Execution time in milliseconds. dry_run: Whether this was a dry run preview. files_processed: List of processed file names. + failed_files: Files whose deletion could not be completed. + unreachable_files: Manifest files unavailable on this node. + unflushed_nodes: Nodes whose buffered rows could not be flushed. + cold_deleted_files: Whole cold files deleted or selected by a dry run. + cold_files: Partially matching cold files left unchanged. + cold_rows_not_deleted: Matching rows left in cold storage. + cold_not_examined: Reason cold storage could not be examined. error: Error message if operation failed. """ @@ -28,8 +35,29 @@ class DeleteResponse(BaseModel): execution_time_ms: float = 0.0 dry_run: bool = False files_processed: list[str] = Field(default_factory=list) + failed_files: list[str] = Field(default_factory=list) + unreachable_files: list[str] = Field(default_factory=list) + unflushed_nodes: list[str] = Field(default_factory=list) + cold_deleted_files: list[str] = Field(default_factory=list) + cold_files: list[str] = Field(default_factory=list) + cold_rows_not_deleted: int = 0 + cold_not_examined: str = "" error: Optional[str] = None + @field_validator( + "files_processed", + "failed_files", + "unreachable_files", + "unflushed_nodes", + "cold_deleted_files", + "cold_files", + mode="before", + ) + @classmethod + def _normalize_null_arrays(cls, value: Any) -> Any: + """Go nil slices encode as null; expose them as empty lists.""" + return [] if value is None else value + class DeleteConfigResponse(BaseModel): """Response containing delete configuration. diff --git a/src/arc_client/models/query.py b/src/arc_client/models/query.py index ed8f742..770e4cf 100644 --- a/src/arc_client/models/query.py +++ b/src/arc_client/models/query.py @@ -15,6 +15,10 @@ class QueryResponse(BaseModel): columns: List of column names in the result. data: List of rows, where each row is a list of values. row_count: Number of rows returned. + truncated: Whether the result was cut short by a query failure. + truncation_reason: Reason the result was truncated, if available. + rows_capped: Whether the result reached a configured row cap. + row_cap: Configured row cap, if available. execution_time_ms: Query execution time in milliseconds. timestamp: Server timestamp when the query was executed. error: Error message if the query failed. @@ -24,11 +28,14 @@ class QueryResponse(BaseModel): columns: list[str] = Field(default_factory=list) data: list[list[Any]] = Field(default_factory=list) row_count: int = 0 + truncated: bool = False + truncation_reason: Optional[str] = None + rows_capped: bool = False + row_cap: Optional[int] = None execution_time_ms: float = 0.0 timestamp: Optional[str] = None error: Optional[str] = None - # Populated only by the msgpack query path (``query_msgpack``); the JSON - # path leaves it empty. Parallel to ``columns``: Arrow type name per column. + # Parallel to ``columns``: Arrow type name per column, when supplied. types: list[str] = Field(default_factory=list) diff --git a/src/arc_client/query/executor.py b/src/arc_client/query/executor.py index 0f260d7..9941e07 100644 --- a/src/arc_client/query/executor.py +++ b/src/arc_client/query/executor.py @@ -94,7 +94,9 @@ def query_msgpack(self, sql: str, database: Optional[str] = None) -> QueryRespon **Experimental.** This uses Arc's ``/api/v1/query/msgpack`` endpoint, which returns a compact columnar MessagePack envelope with native - timestamp encoding. The result is transposed to row-major and returned + timestamp encoding without silently discarding nanoseconds. Exactly representable + microseconds become UTC datetime; other values remain msgpack.Timestamp. + The result is transposed to row-major and returned as a :class:`QueryResponse` (same shape as :meth:`query`), with the per-column Arrow types available in ``QueryResponse.types``. diff --git a/src/arc_client/query/msgpack_response.py b/src/arc_client/query/msgpack_response.py index 66871f6..6385f8e 100644 --- a/src/arc_client/query/msgpack_response.py +++ b/src/arc_client/query/msgpack_response.py @@ -4,8 +4,10 @@ (``application/msgpack``): a map with keys ``success, columns, types, data, row_count, execution_time_ms, timestamp``. ``data`` is an array-of-columns (one inner array per column), not row-major, and timestamp columns arrive as -native MessagePack timestamp extensions (decoded here to ``datetime``). Per-cell -nulls are possible. +native MessagePack timestamp extensions. Values exactly representable by Python +``datetime`` are decoded to UTC datetimes; submicrosecond or out-of-range values +remain ``msgpack.Timestamp`` objects, preserving their seconds and nanoseconds. +Per-cell nulls are possible. The error envelope is a 4-key map ``{success: false, error, execution_time_ms, timestamp}``. @@ -30,6 +32,9 @@ def decode_msgpack_query_response(content: bytes) -> QueryResponse: The columnar ``data`` is transposed to row-major so the result matches the JSON query path (``QueryResponse.data`` is a list of rows). + Timestamp cells, including those nested in lists or maps, become UTC + ``datetime`` values only when that conversion is lossless. Other timestamps + remain ``msgpack.Timestamp``; use ``to_unix_nano()`` for exact epoch nanoseconds. Args: content: Raw MessagePack response body. @@ -42,8 +47,9 @@ def decode_msgpack_query_response(content: bytes) -> QueryResponse: ArcQueryError: If the body cannot be decoded or reports ``success=false``. """ try: - # timestamp=3 decodes native msgpack timestamp extensions to datetime. - envelope = msgpack.unpackb(content, timestamp=3, raw=False) + # timestamp=3 silently drops submicrosecond precision. Decode the native + # values first, then convert only timestamps datetime can represent exactly. + envelope = _decode_timestamps(msgpack.unpackb(content, timestamp=0, raw=False)) except Exception as e: # pragma: no cover - malformed body is rare raise ArcQueryError(f"Failed to decode msgpack query response: {e}") from e @@ -70,11 +76,32 @@ def decode_msgpack_query_response(content: bytes) -> QueryResponse: types=types, data=rows, row_count=row_count if row_count else len(rows), + truncated=envelope.get("truncated", False), + truncation_reason=envelope.get("truncation_reason"), + rows_capped=envelope.get("rows_capped", False), + row_cap=envelope.get("row_cap"), execution_time_ms=float(execution_time_ms), timestamp=str(timestamp) if timestamp is not None else None, ) +def _decode_timestamps(value: Any) -> Any: + """Preserve precision and range while retaining datetime where lossless.""" + if isinstance(value, msgpack.Timestamp): + if value.nanoseconds % 1000 == 0: + try: + return value.to_datetime() + except (OverflowError, ValueError): + # MessagePack supports a wider seconds range than datetime. + pass + return value + if isinstance(value, list): + return [_decode_timestamps(item) for item in value] + if isinstance(value, dict): + return {key: _decode_timestamps(item) for key, item in value.items()} + return value + + def _transpose_columns(column_data: list[list[Any]], row_count: int) -> list[list[Any]]: """Transpose an array-of-columns into row-major rows. diff --git a/tests/unit/http/test_delete_diagnostics.py b/tests/unit/http/test_delete_diagnostics.py new file mode 100644 index 0000000..fa9b7ce --- /dev/null +++ b/tests/unit/http/test_delete_diagnostics.py @@ -0,0 +1,162 @@ +"""DELETE diagnostics remain available without changing partial success semantics.""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest +from pytest_httpx import HTTPXMock + +from arc_client import ArcClient, AsyncArcClient +from arc_client.config import ClientConfig +from arc_client.exceptions import ArcError, ArcServerError +from arc_client.http.async_http import AsyncHTTPClient +from arc_client.http.sync_http import SyncHTTPClient + +ARRAY_FIELDS = ( + "files_processed", + "failed_files", + "unreachable_files", + "unflushed_nodes", + "cold_deleted_files", + "cold_files", +) +DIAGNOSTICS = { + "files_processed": ["rewritten.parquet"], + "failed_files": ["failed.parquet"], + "unreachable_files": ["missing.parquet"], + "unflushed_nodes": ["writer-2"], + "cold_deleted_files": ["cold-deleted.parquet"], + "cold_files": ["cold-partial.parquet"], + "cold_rows_not_deleted": 9, + "cold_not_examined": "bucket unavailable", +} +DELETE_ARGS = {"database": "db", "measurement": "cpu", "where": "value < 0"} + + +def add_delete_response(httpx_mock: HTTPXMock, status: int, body: dict[str, Any]) -> None: + httpx_mock.add_response( + method="POST", url="http://h:8000/api/v1/delete", status_code=status, json=body + ) + + +def assert_diagnostics(result: Any, body: dict[str, Any]) -> None: + for field, value in body.items(): + assert getattr(result, field) == value + + +@pytest.mark.parametrize("status", [200, 207]) +async def test_delete_diagnostics_public_clients(httpx_mock: HTTPXMock, status: int) -> None: + body = { + "success": status == 200, + "deleted_count": 4, + "affected_files": 3, + "rewritten_files": 1, + "execution_time_ms": 2.5, + "dry_run": False, + "error": None if status == 200 else "partial delete", + **DIAGNOSTICS, + } + add_delete_response(httpx_mock, status, body) + with ArcClient(host="h", max_retries=0) as client: + result = client.delete.delete(**DELETE_ARGS, dry_run=False, confirm=True) + assert_diagnostics(result, body) + + add_delete_response(httpx_mock, status, body) + async with AsyncArcClient(host="h", max_retries=0) as client: + result = await client.delete.delete(**DELETE_ARGS, dry_run=False, confirm=True) + assert_diagnostics(result, body) + requests = httpx_mock.get_requests() + assert len(requests) == 2 + for request in requests: + assert json.loads(request.content) == {**DELETE_ARGS, "dry_run": False, "confirm": True} + + +@pytest.mark.parametrize("array_value", ["omitted", None, []], ids=["omitted", "null", "empty"]) +async def test_delete_optional_arrays(httpx_mock: HTTPXMock, array_value: Any) -> None: + body = {"success": True} + if array_value != "omitted": + body.update(dict.fromkeys(ARRAY_FIELDS, array_value)) + add_delete_response(httpx_mock, 200, body) + with ArcClient(host="h", max_retries=0) as client: + result = client.delete.delete(**DELETE_ARGS) + for field in ARRAY_FIELDS: + assert getattr(result, field) == [] + assert result.cold_rows_not_deleted == 0 + assert result.cold_not_examined == "" + + add_delete_response(httpx_mock, 200, body) + async with AsyncArcClient(host="h", max_retries=0) as client: + result = await client.delete.delete(**DELETE_ARGS) + for field in ARRAY_FIELDS: + assert getattr(result, field) == [] + assert result.cold_rows_not_deleted == 0 + assert result.cold_not_examined == "" + + +@pytest.mark.parametrize( + "invalid", + [ + {"failed_files": "file.parquet"}, + {"unflushed_nodes": [1]}, + {"cold_rows_not_deleted": "many"}, + {"cold_not_examined": []}, + ], +) +async def test_delete_invalid_diagnostics_fail( + httpx_mock: HTTPXMock, invalid: dict[str, Any] +) -> None: + body = {"success": False, **invalid} + add_delete_response(httpx_mock, 207, body) + with ( + ArcClient(host="h", max_retries=0) as client, + pytest.raises(ArcError, match="Delete operation failed"), + ): + client.delete.delete(**DELETE_ARGS) + add_delete_response(httpx_mock, 207, body) + async with AsyncArcClient(host="h", max_retries=0) as client: + with pytest.raises(ArcError, match="Delete operation failed"): + await client.delete.delete(**DELETE_ARGS) + + +async def test_raw_delete_207_preserves_status_and_body(httpx_mock: HTTPXMock) -> None: + body = {"success": False, "error": "partial delete", **DIAGNOSTICS} + add_delete_response(httpx_mock, 207, body) + with SyncHTTPClient(ClientConfig(host="h", max_retries=0)) as http: + response = http.post("/api/v1/delete", json={**DELETE_ARGS, "dry_run": True}) + assert response.status_code == 207 + assert response.json() == body + add_delete_response(httpx_mock, 207, body) + async with AsyncHTTPClient(ClientConfig(host="h", max_retries=0)) as http: + response = await http.post("/api/v1/delete", json={**DELETE_ARGS, "dry_run": True}) + assert response.status_code == 207 + assert response.json() == body + + +@pytest.mark.httpx_mock(can_send_already_matched_responses=True) +async def test_delete_503_preserves_diagnostics_without_retry(httpx_mock: HTTPXMock) -> None: + body = { + "success": False, + "deleted_count": 4, + "affected_files": 3, + "rewritten_files": 1, + "execution_time_ms": 2.5, + "dry_run": False, + "error": "Delete partially completed", + **DIAGNOSTICS, + } + add_delete_response(httpx_mock, 503, body) + with ArcClient(host="h", retry_backoff=0) as client, pytest.raises(ArcServerError) as caught: + client.delete.delete(**DELETE_ARGS, dry_run=False, confirm=True) + assert caught.value.status_code == 503 + assert caught.value.response_body == body + assert len(httpx_mock.get_requests()) == 1 + + add_delete_response(httpx_mock, 503, body) + async with AsyncArcClient(host="h", retry_backoff=0) as client: + with pytest.raises(ArcServerError) as caught: + await client.delete.delete(**DELETE_ARGS, dry_run=False, confirm=True) + assert caught.value.status_code == 503 + assert caught.value.response_body == body + assert len(httpx_mock.get_requests()) == 2 diff --git a/tests/unit/http/test_query_metadata_contracts.py b/tests/unit/http/test_query_metadata_contracts.py new file mode 100644 index 0000000..5dc9e1f --- /dev/null +++ b/tests/unit/http/test_query_metadata_contracts.py @@ -0,0 +1,168 @@ +"""Query metadata survives actual JSON and MessagePack HTTP envelopes.""" + +from __future__ import annotations + +import json +from typing import Any + +import msgpack +import pytest +from pytest_httpx import HTTPXMock + +from arc_client import ArcClient, AsyncArcClient + +METADATA_CASES = [ + pytest.param({"rows_capped": True, "row_cap": 2}, id="capped"), + pytest.param({"rows_capped": False, "row_cap": 100}, id="uncapped"), + pytest.param({}, id="omitted"), + pytest.param({"truncated": True, "truncation_reason": "file disappeared"}, id="truncated"), +] + + +def add_query_response(httpx_mock: HTTPXMock, wire_format: str, metadata: dict[str, Any]) -> None: + body = { + "success": True, + "columns": ["host", "value"], + "types": ["utf8", "int64"], + "data": [["a", 3], ["b", None]], + "row_count": 2, + "execution_time_ms": 1.25, + "timestamp": "2026-10-27T00:00:00Z", + **metadata, + } + if wire_format == "msgpack": + body["data"] = [["a", "b"], [3, None]] + content = msgpack.packb(body, use_bin_type=True) + content_type = "application/msgpack" + path = "/api/v1/query/msgpack" + else: + content = json.dumps(body).encode() + content_type = "application/json" + path = "/api/v1/query" + httpx_mock.add_response( + method="POST", + url=f"http://h:8000{path}", + content=content, + headers={"Content-Type": content_type}, + ) + + +def assert_query_response(result: Any, metadata: dict[str, Any]) -> None: + assert result.success is True + assert result.columns == ["host", "value"] + assert result.types == ["utf8", "int64"] + assert result.data == [["a", 3], ["b", None]] + assert result.row_count == 2 + assert result.execution_time_ms == 1.25 + assert result.timestamp == "2026-10-27T00:00:00Z" + assert result.rows_capped is metadata.get("rows_capped", False) + assert result.row_cap == metadata.get("row_cap") + assert result.truncated is metadata.get("truncated", False) + assert result.truncation_reason == metadata.get("truncation_reason") + + +@pytest.mark.parametrize("wire_format", ["json", "msgpack"]) +@pytest.mark.parametrize("metadata", METADATA_CASES) +def test_query_metadata_sync( + httpx_mock: HTTPXMock, wire_format: str, metadata: dict[str, Any] +) -> None: + add_query_response(httpx_mock, wire_format, metadata) + with ArcClient(host="h", max_retries=0) as client: + query = client.query.query_msgpack if wire_format == "msgpack" else client.query.query + result = query("SELECT host, value FROM cpu") + assert_query_response(result, metadata) + assert json.loads(httpx_mock.get_request().content) == {"sql": "SELECT host, value FROM cpu"} + + +@pytest.mark.parametrize("wire_format", ["json", "msgpack"]) +@pytest.mark.parametrize("metadata", METADATA_CASES) +async def test_query_metadata_async( + httpx_mock: HTTPXMock, wire_format: str, metadata: dict[str, Any] +) -> None: + add_query_response(httpx_mock, wire_format, metadata) + async with AsyncArcClient(host="h", max_retries=0) as client: + query = client.query.query_msgpack if wire_format == "msgpack" else client.query.query + result = await query("SELECT host, value FROM cpu") + assert_query_response(result, metadata) + assert json.loads(httpx_mock.get_request().content) == {"sql": "SELECT host, value FROM cpu"} + + +TIMESTAMP_CASES = [ + pytest.param(1730000000, 123456789, id="nanoseconds"), + pytest.param(-1, 999999999, id="negative-one-nanosecond"), + pytest.param(0, 1, id="positive-one-nanosecond"), + pytest.param(1730000000, 123456000, id="microseconds"), + pytest.param(-1, 999999000, id="negative-one-microsecond"), + pytest.param(0, 0, id="epoch"), + pytest.param(253402300799, 999999000, id="datetime-maximum"), + pytest.param(2**63 - 1, 0, id="beyond-datetime-maximum"), + pytest.param(-(2**63), 0, id="below-datetime-minimum"), +] + + +def add_timestamp_response(httpx_mock: HTTPXMock, seconds: int, nanoseconds: int) -> None: + timestamp = msgpack.Timestamp(seconds, nanoseconds) + httpx_mock.add_response( + method="POST", + url="http://h:8000/api/v1/query/msgpack", + content=msgpack.packb( + { + "success": True, + "columns": ["time", "signed", "unsigned", "nested"], + "types": ["timestamp[ns]", "int64", "uint64", "struct"], + "data": [ + [timestamp, None], + [-(2**63), 2**63 - 1], + [2**64 - 1, 0], + [{"times": [timestamp, None], "count": 2**64 - 1}, None], + ], + "row_count": 2, + }, + use_bin_type=True, + ), + headers={"Content-Type": "application/msgpack"}, + ) + + +def assert_lossless_timestamp(result: Any, seconds: int, nanoseconds: int) -> None: + from datetime import datetime, timezone + + value = result.data[0][0] + if nanoseconds % 1000 or not (-62135596800 <= seconds <= 253402300799): + assert isinstance(value, msgpack.Timestamp) + assert value.seconds == seconds + assert value.nanoseconds == nanoseconds + assert value.to_unix_nano() == seconds * 1_000_000_000 + nanoseconds + else: + assert isinstance(value, datetime) + assert value.tzinfo == timezone.utc + delta = value - datetime(1970, 1, 1, tzinfo=timezone.utc) + actual_ns = ((delta.days * 86400 + delta.seconds) * 1_000_000 + delta.microseconds) * 1000 + assert actual_ns == seconds * 1_000_000_000 + nanoseconds + assert result.data[0][1:] == [ + -(2**63), + 2**64 - 1, + {"times": [value, None], "count": 2**64 - 1}, + ] + assert result.data[1] == [None, 2**63 - 1, 0, None] + assert result.types == ["timestamp[ns]", "int64", "uint64", "struct"] + + +@pytest.mark.parametrize("seconds,nanoseconds", TIMESTAMP_CASES) +def test_msgpack_timestamps_lossless_sync( + httpx_mock: HTTPXMock, seconds: int, nanoseconds: int +) -> None: + add_timestamp_response(httpx_mock, seconds, nanoseconds) + with ArcClient(host="h", max_retries=0) as client: + result = client.query.query_msgpack("SELECT * FROM cpu") + assert_lossless_timestamp(result, seconds, nanoseconds) + + +@pytest.mark.parametrize("seconds,nanoseconds", TIMESTAMP_CASES) +async def test_msgpack_timestamps_lossless_async( + httpx_mock: HTTPXMock, seconds: int, nanoseconds: int +) -> None: + add_timestamp_response(httpx_mock, seconds, nanoseconds) + async with AsyncArcClient(host="h", max_retries=0) as client: + result = await client.query.query_msgpack("SELECT * FROM cpu") + assert_lossless_timestamp(result, seconds, nanoseconds)