fix(sdk): integrate query diagnostics and Arrow metadata into main

.github/workflows/ci.yml

@@ -134,3 +134,20 @@ jobs:
         with:
           files: ./coverage.xml
           fail_ci_if_error: false
+
+  arrow-httpx-027:
+    name: Arrow transport (HTTPX 0.27)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: astral-sh/setup-uv@v4
+        with:
+          version: "latest"
+      - run: uv python install 3.11
+      - run: uv sync --extra dev --extra all
+      - name: Install compatible legacy transport test dependencies
+        run: uv pip install --no-deps httpx==0.27.2 pytest-httpx==0.30.0 sniffio==1.3.1
+      - name: Verify real HTTPX 0.27 wire compatibility
+        run: |
+          uv run --no-sync python -c "import httpx; print(httpx.__version__)"
+          uv run --no-sync pytest tests/unit/http/test_arrow_metadata.py --no-cov -v --tb=short

CHANGELOG.md

@@ -12,6 +12,15 @@ plus bug fixes and the first real HTTP-layer test coverage.
 
 ### Added
 
+- Arrow completion metadata in the existing `pyarrow.Table` schema for sync and
+  async queries: policy cap and execution time are retained; failed, malformed,
+  conflicting or unobservable announced completion metadata is rejected.
+  Native column types and unrelated schema metadata remain intact. Trailer
+  capture uses explicitly bounded HTTPX/httpcore/h11 compatibility versions.
+  Promoted timing headers cannot prove uncapped completion when final trailers
+  are missing.
+
+
 - **MessagePack query** (`query.query_msgpack`, experimental): decodes the
   columnar MessagePack response with native timestamp handling and returns the
   same `QueryResponse` shape as `query()` (with per-column Arrow `types`). Raises
@@ -66,6 +75,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.

README.md

@@ -242,9 +242,27 @@ pl_df = client.query.query_polars("SELECT * FROM default.cpu LIMIT 1000")
 ### PyArrow Table (Zero-Copy)
 
 ```python
-table = client.query.query_arrow("SELECT * FROM default.cpu LIMIT 1000")
+table = client.query.query_arrow("SELECT * FROM cpu LIMIT 1000")
+metadata = table.schema.metadata or {}
+if metadata.get(b"arc.rows_capped") == b"true":
+    print("Policy capped the result at", int(metadata[b"arc.row_cap"]))
 ```
 
+`query_arrow()` still returns a `pyarrow.Table`, with authoritative completion
+metadata under reserved schema keys `arc.rows_capped`, `arc.row_cap` and
+`arc.execution_time_ms` (byte keys and values). A missing cap flag means unknown;
+`false` means the supported server completed its uncapped trailer contract.
+Timing promoted into initial headers cannot prove uncapped completion after
+final trailers disappear.
+Native column types and unrelated schema metadata are preserved. Truncated,
+malformed, conflicting or unobservable announced metadata raises `ArcQueryError`.
+DataFrame conversions can drop schema metadata; inspect the Arrow table first
+when completeness matters.
+
+Trailer capture is tested with HTTPX 0.27/0.28, httpcore 1.0.9 and h11 0.16.
+Those dependency bounds are deliberate: the adapter observes HTTP/1.1 parser
+completion without replacing HTTPX's TLS, proxy, timeout or pooling behavior.
+
 ### MessagePack Response (experimental)
 
 A compact columnar MessagePack response with native timestamp decoding. Returns
@@ -498,3 +516,41 @@ See the [full documentation](https://docs.basekick.net/arc/sdks/python]) for det
 ## License
 
 MIT License - see [LICENSE](LICENSE) for details.
+
+
+## Server compatibility and safe management updates
+
+The SDK distinguishes partial updates from complete replacements:
+
+- `update()` sends only supplied non-`None` fields, without fetching a definition.
+  Use this on servers with partial PUT semantics. Existing `None` arguments remain omitted;
+  use `from arc_client import CLEAR` to send JSON null for nullable update fields.
+- `replace()` sends every mutable field explicitly in one PUT. Arc 26.09.3 uses
+  full-definition PUT semantics for retention policies and continuous queries, so use
+  `replace()` on that release. The method also works on newer servers. It intentionally
+  overwrites the whole definition; coordinate concurrent edits before calling it.
+
+There is no server-capability guessing or retry with a different update payload.
+Both sync and async clients provide the same methods. CQ replacements include
+`tag_columns`; retention replacements include `database`, `measurement`, and `buffer_days`.
+
+Query results retain `truncated`, `truncation_reason`, `rows_capped`, and `row_cap`
+when their wire format supplies them. Experimental `query_msgpack()` returns UTC
+`datetime` values when a timestamp is exactly representable at microsecond resolution.
+Submicrosecond or out-of-range timestamps remain `msgpack.Timestamp`, preserving
+`.seconds`, `.nanoseconds`, and `.to_unix_nano()`. Such cells require an explicit
+serialization choice when exporting JSON. Arrow completion metadata is available
+in the returned table schema as described above.
+
+DELETE results expose failed, unreachable, unflushed, and cold-tier omission details.
+HTTP 207 remains a result that callers must inspect. Non-2xx typed exceptions retain
+structured details in `error.response_body`. Data DELETE requests are never automatically
+replayed after a server or transport failure, because work may already have committed.
+
+Automatic 429/503 and ambiguous transport retries apply only to read-only HTTP methods.
+Other mutations retry only connection-establishment failures; unknown write outcomes
+are surfaced for the caller to reconcile, rather than replayed automatically.
+
+Query requests already carry the configured database in `x-arc-database`. Use
+unqualified measurement names in those SQL requests, for example
+`client.query.query("SELECT * FROM cpu LIMIT 10")`.

pyproject.toml

@@ -27,7 +27,9 @@ classifiers = [
 ]
 
 dependencies = [
-    "httpx>=0.27.0",
+    "httpx>=0.27.0,<0.29",
+    "httpcore==1.0.9",
+    "h11>=0.16.0,<0.17",
     "msgpack>=1.0.7",
     "pyarrow>=15.0.0",
     "pydantic>=2.6.0",

src/arc_client/http/arrow_trailers.py

@@ -0,0 +1,84 @@
+"""Bounded HTTPX/httpcore HTTP/1.1 compatibility adapter for Arrow trailers.
+
+HTTPX 0.27/0.28 and httpcore 1.0.9 discard h11 EndOfMessage headers. The
+response hook runs before body consumption, even if the body is already in
+the parser's receive buffer. Only the owning connection's parser is observed;
+TLS, proxies, timeouts, framing and pool ownership remain with HTTPX/httpcore.
+An unsupported stream shape is detected by the Arrow decoder, never guessed.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Optional, cast
+
+import h11
+import httpcore
+import httpx
+
+TRAILERS_EXTENSION = "arc.arrow_trailers"
+
+
+@dataclass
+class ArrowTrailers:
+    headers: list[tuple[bytes, bytes]] = field(default_factory=list)
+    complete: bool = False
+
+
+class _ObservedState:
+    """Delegate protocol state unchanged, retaining only the current sink."""
+
+    def __init__(self, state: Any) -> None:
+        self._state = state
+        self.sink: Optional[ArrowTrailers] = None
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._state, name)
+
+    def next_event(self) -> Any:
+        event = self._state.next_event()
+        if isinstance(event, h11.EndOfMessage):
+            sink, self.sink = self.sink, None
+            if sink is not None:
+                sink.headers = list(event.headers.raw_items())
+                sink.complete = True
+        return event
+
+    def start_next_cycle(self) -> None:
+        self.sink = None
+        self._state.start_next_cycle()
+
+
+def capture_arrow_trailers(response: httpx.Response) -> None:
+    """Attach a fresh per-response sink to the supported HTTP/1.1 parser."""
+    if (
+        response.request.method != "POST"
+        or response.request.url.path.rstrip("/") != "/api/v1/query/arrow"
+    ):
+        return
+    sink = ArrowTrailers()
+    response.extensions[TRAILERS_EXTENSION] = sink
+    # Bound{Sync,Async}Stream -> HTTPX response stream -> PoolByteStream ->
+    # HTTP11ConnectionByteStream -> owning HTTP11 connection. Deliberately
+    # bounded to the dependency versions declared in pyproject.toml.
+    owner: Any = response.stream
+    for attribute in ("_stream", "_httpcore_stream", "_stream", "_connection"):
+        owner = getattr(owner, attribute, None)
+        if owner is None:
+            return
+    if not isinstance(owner, (httpcore.HTTP11Connection, httpcore.AsyncHTTP11Connection)):
+        return
+    state: Any = owner._h11_state
+    if isinstance(state, _ObservedState):
+        observer = state
+    elif isinstance(state, h11.Connection):
+        observer = _ObservedState(state)
+        cast(Any, owner)._h11_state = observer
+    else:
+        return
+    observer.sink = sink
+
+
+async def capture_async_arrow_trailers(response: httpx.Response) -> None:
+    """Async HTTPX hook; observing parser events itself performs no I/O."""
+    capture_arrow_trailers(response)

src/arc_client/http/async_http.py

@@ -8,6 +8,7 @@
 import httpx
 
 from arc_client.config import ClientConfig
+from arc_client.http.arrow_trailers import capture_async_arrow_trailers
 from arc_client.http.base import (
     RETRYABLE_TRANSPORT_ERRORS,
     HTTPClientBase,
@@ -33,6 +34,7 @@ def _get_client(self) -> httpx.AsyncClient:
                 base_url=self._base_url,
                 timeout=self.config.timeout,
                 verify=self.config.verify_ssl,
+                event_hooks={"response": [capture_async_arrow_trailers]},
             )
         return self._client
 

src/arc_client/http/sync_http.py

@@ -8,6 +8,7 @@
 import httpx
 
 from arc_client.config import ClientConfig
+from arc_client.http.arrow_trailers import capture_arrow_trailers
 from arc_client.http.base import (
     RETRYABLE_TRANSPORT_ERRORS,
     HTTPClientBase,
@@ -33,6 +34,7 @@ def _get_client(self) -> httpx.Client:
                 base_url=self._base_url,
                 timeout=self.config.timeout,
                 verify=self.config.verify_ssl,
+                event_hooks={"response": [capture_arrow_trailers]},
             )
         return self._client
 

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)