fix(query): retain authoritative Arrow completion metadata

.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

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/query/arrow_response.py

@@ -0,0 +1,89 @@
+"""Decode native Arrow data and authoritative completion metadata."""
+
+from __future__ import annotations
+
+import math
+import re
+
+import httpx
+import pyarrow as pa
+import pyarrow.ipc as ipc
+
+from arc_client.exceptions import ArcQueryError
+from arc_client.http.arrow_trailers import TRAILERS_EXTENSION, ArrowTrailers
+
+_CAP = b"arc-rows-capped"
+_TRUNCATED = b"arc-stream-truncated"
+_TIME = b"arc-execution-time-ms"
+_FIELDS = {_CAP, _TRUNCATED, _TIME}
+_RESERVED = {b"arc.rows_capped", b"arc.row_cap", b"arc.execution_time_ms"}
+
+
+def decode_arrow_response(response: httpx.Response) -> pa.Table:
+    """Keep the Table API and record server metadata in its schema.
+
+    Missing legacy metadata is unknown. An announced Arc trailer requires a
+    completed capture; malformed or truncated responses never become tables.
+    """
+    if "application/json" in response.headers.get("content-type", ""):
+        raise ArcQueryError(response.json().get("error", "Query failed with unknown error"))
+
+    announced = {
+        name.strip().lower().encode("ascii")
+        for value in response.headers.get_list("trailer")
+        for name in value.split(",")
+    }
+    sink = response.extensions.get(TRAILERS_EXTENSION)
+    complete = isinstance(sink, ArrowTrailers) and sink.complete
+    if announced & _FIELDS and not complete:
+        raise ArcQueryError(
+            "Arrow completion metadata could not be captured by this HTTP transport"
+        )
+
+    values: dict[bytes, bytes] = {}
+    fields = list(response.headers.raw)
+    final_timing = False
+    if isinstance(sink, ArrowTrailers) and sink.complete:
+        fields.extend(sink.headers)
+        final_timing = any(name.lower() == _TIME and value.strip() for name, value in sink.headers)
+    for name, value in fields:
+        name, value = name.lower(), value.strip()
+        if name not in _FIELDS or not value:
+            continue
+        if name in values and values[name] != value:
+            raise ArcQueryError("Conflicting Arrow completion metadata")
+        values[name] = value
+
+    if _TRUNCATED in values:
+        reason = values[_TRUNCATED].decode("utf-8", errors="replace")
+        raise ArcQueryError("Arrow stream truncated: " + reason)
+
+    metadata: dict[bytes, bytes] = {}
+    if _CAP in values:
+        cap = values[_CAP]
+        if not re.fullmatch(rb"[0-9]{1,19}", cap) or not 0 < int(cap) <= 2**63 - 1:
+            raise ArcQueryError("Invalid Arrow row-cap metadata")
+        metadata[b"arc.rows_capped"] = b"true"
+        metadata[b"arc.row_cap"] = str(int(cap)).encode("ascii")
+    elif complete and _CAP in announced and final_timing:
+        # Both supported Arc versions announce this trailer on every result;
+        # an empty (release) or absent (main) cap means uncapped. Require the
+        # always-sent execution marker in FINAL trailers too: a gateway that
+        # promotes timing but drops trailers cannot prove uncapped completion.
+        metadata[b"arc.rows_capped"] = b"false"
+    if _TIME in values:
+        try:
+            elapsed = float(values[_TIME])
+        except ValueError as error:
+            raise ArcQueryError("Invalid Arrow execution-time metadata") from error
+        if not math.isfinite(elapsed) or elapsed < 0:
+            raise ArcQueryError("Invalid Arrow execution-time metadata")
+        metadata[b"arc.execution_time_ms"] = values[_TIME]
+
+    with ipc.open_stream(response.content) as reader:
+        table = reader.read_all()
+    # These names are SDK-reserved. Wire completion metadata takes precedence
+    # over schema content, including when the server supplied no cap signal.
+    schema_metadata = {k: v for k, v in (table.schema.metadata or {}).items() if k not in _RESERVED}
+    schema_metadata.update(metadata)
+    return table.replace_schema_metadata(schema_metadata or None)

src/arc_client/query/async_executor.py

@@ -5,12 +5,12 @@
 from typing import TYPE_CHECKING, Any, Optional
 
 import pyarrow as pa
-import pyarrow.ipc as ipc
 
 from arc_client.config import ClientConfig
 from arc_client.exceptions import ArcNotSupportedError, ArcQueryError, ArcValidationError
 from arc_client.http.async_http import AsyncHTTPClient
 from arc_client.models.query import EstimateResponse, MeasurementInfo, QueryResponse
+from arc_client.query.arrow_response import decode_arrow_response
 from arc_client.query.msgpack_response import decode_msgpack_query_response
 from arc_client.validation import validate_name
 
@@ -196,7 +196,9 @@ async def query_arrow(self, sql: str, database: Optional[str] = None) -> pa.Tabl
             database: Target database. Uses client default if not specified.
 
         Returns:
-            pyarrow.Table with query results.
+            pyarrow.Table with query results and completion metadata in
+            its schema: arc.rows_capped, arc.row_cap, arc.execution_time_ms.
+            Missing cap metadata means unknown. Truncation raises ArcQueryError.
 
         Raises:
             ArcQueryError: If the query fails.
@@ -218,15 +220,7 @@ async def query_arrow(self, sql: str, database: Optional[str] = None) -> pa.Tabl
                 headers=headers,
             )
 
-            # Check if we got JSON error response instead of Arrow
-            content_type = response.headers.get("content-type", "")
-            if "application/json" in content_type:
-                error_data = response.json()
-                raise ArcQueryError(error_data.get("error", "Query failed with unknown error"))
-
-            # Parse Arrow IPC stream
-            reader = ipc.open_stream(response.content)
-            return reader.read_all()
+            return decode_arrow_response(response)
 
         except ArcQueryError:
             raise