fix(query): preserve query metadata and partial-delete diagnostics

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.

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.

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)
 
 

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``.
 

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.
 

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