perf: fast feature-frame builds and response-body release in the chunked OGC path

NEWS.md

@@ -1,5 +1,7 @@
 **08/25/2026:** Removed `dataretrieval.ogc.retry`, which only re-exported private helpers. Deprecated `dataretrieval.ogc.interruptions`; import exceptions from `dataretrieval` or `dataretrieval.interruptions` instead. The old path will be removed in a future major release, no earlier than 2027-08-25.
 
+**08/22/2026:** The OGC feature-frame builders take vectorized fast paths, halving the client-side CPU of a chunked call. Flat feature properties — every Water Data / NGWMN collection — build through a plain `DataFrame` constructor instead of `pd.json_normalize` (~2.6x), and all-point geometry pages build their `GeoDataFrame` through one vectorized `geopandas.points_from_xy` call instead of the per-feature Python walk in `GeoDataFrame.from_features` (~4.8x). Results are identical — a page with nested properties, or any non-point or malformed geometry, takes the previous path — and this CPU runs inside the fan-out's event loop, so trimming it also tightens chunk overlap (measured: a 93k-row, 12-chunk `get_daily` re-pull fell from 1.33 s to 0.95 s median wall; spatial frame construction alone is 4.7–4.9x faster at 10k–500k rows). The gain is larger under pandas 2, where every string column is object dtype: the nested-value check that decides the fast path is gated on `infer_dtype`, so it walks only a genuinely mixed column rather than all of them. Aggregated responses also no longer retain a page body: the response behind a getter's metadata and each per-chunk aggregate held for resume each carried one full JSON page's bytes, which on a many-chunk fan-out kept roughly the whole download resident until the call finished (~19% lower peak Python-heap in a 12-chunk replay, growing with chunk count and page size). **Behavior change:** the aggregated `httpx.Response` a completed call or `FanOutInterrupted.partial_response` carries now has an empty body — it was previously one arbitrary page's bytes, not the query's data; status, headers, URL, and elapsed are unchanged, and per-page responses are untouched. **Deprecation:** `utils.format_datetime` — orphaned since the qw services it shaped responses for were retired — now emits a `DeprecationWarning`; it will be removed on or after 2027-08-22. Combine the columns with `pandas.to_datetime` directly.
+
 **08/20/2026:** The `state` filter now accepts the five US territories. `dataretrieval.codes.states` held the 50 states and DC, so `ngwmn.get_sites(state='Puerto Rico')`, `waterdata.get_monitoring_locations(state_name='Puerto Rico')` via the unified `state`, and `nwdc.get_wateruse(state='PR')` were refused locally -- while all three services carry the data (NGWMN answers with 36 Puerto Rico monitoring locations, the Water Data monitoring-locations collection returns Puerto Rico sites, and legacy NWIS lists 1,148 stream sites for `stateCd=PR`). American Samoa, Guam, the Northern Mariana Islands, Puerto Rico and the US Virgin Islands are now in both code tables under their real ANSI/FIPS codes, so every encoding resolves: `'Puerto Rico'`, `'PR'`, `'72'` and `'US:72'` all normalize alike. **Behavior change:** a territory that used to raise `ValueError` now produces a request. A value the table genuinely does not hold still fails fast.
 
 **08/20/2026:** Argument checks now share one vocabulary, and every rejection explains how to correct the call. `dataretrieval._validation` owns the message shape for each check that recurs across the adapters -- a value outside a closed vocabulary (`require_one_of`), a missing argument (`require_argument`, `require_together`), a query with no filter at all (`require_any_of`), and arguments that conflict (`require_exactly_one`, `reject_together`) -- so a new check cannot invent its own phrasing, which is how `get_reference_table` came to tell callers who passed a bad `collection` that their *code service* was invalid. Each check takes the caller's own spelling of the parameter and a remedy for the move it cannot derive, and every check raises `ValueError` -- one class for a bad argument value. **Behavior change:** the text of those rejections moves with them -- `"Unrecognized service: 'x'. get_record serves …"` is now `"Invalid service: 'x'. Valid options are: …"`, and the major-filter and bounding-box complaints in `query_waterdata` / `query_waterservices` are rendered in the shared form. **Behavior change:** the deprecated `nwis` query entry points (`query_waterdata`, `query_waterservices`, `get_record`) now answer a missing major filter, an incomplete bounding box, or an unknown service with `ValueError` rather than their historic `TypeError` -- `TypeError` remains for a genuinely mistyped argument, such as a non-string `sites`. Code catching `TypeError` there, or matching on the old strings, must update. **Bug fix:** a `None` passed as a major filter (`query_waterservices(service='dv', sites=None)`) counted as a filter and reached the service as an empty `sites=`; `None` now means not supplied, and the call is refused with the filters that would have served. **Bug fix:** four messages told callers to do something that raised again or named a parameter their getter does not accept -- `nldi.get_features(comid=…, feature_id=…)` said to supply the missing half of the feature pair, and the corrected call then failed on the conflict with `comid`; `nwdc` answered `state=[]` by saying exactly one of `state`, `county` or `huc` must be given, when exactly one was; `codes.states.apply_state` offered NGWMN callers a `state_name` / `state_code` parameter no NGWMN getter accepts; and `BaseMetadata` pointed NGWMN and NWDC callers at a Water Data getter. **Bug fix:** `nldi.get_features(navigation_mode=…)` without a `data_source` spelled `None` into the URL path and returned an empty frame from a 200; it now raises and names the source to pass. **Behavior change:** `waterdata.get_nearest_continuous` appends `time` and `monitoring_location_id` to an explicit `properties` list rather than honoring it verbatim -- omitting either silently collapsed every monitoring location into one row per target, a wrong answer rather than an error -- so a caller who passed `properties=['time', 'value']` now gets a third column. **Behavior change:** `nwis.query_waterdata` serves `'peaks'` only; the `'ratings'` URL it used to build was never an NwisWeb program and answered with an HTML error page. `nwis.get_record(service='ratings')` is unaffected -- it routes to `get_ratings`, which is served from a different endpoint. New: `waterdata.get_cql(..., max_rows=N)` caps the total rows a CQL query returns.

dataretrieval/_deprecation.py

@@ -27,6 +27,7 @@
     "nwis": "2027-05-06",
     "waterdata.get_cql(service=)": "2027-08-09",
     "wateruse": "2027-08-11",
+    "utils.format_datetime": "2027-08-22",
     "ogc.interruptions": "2027-08-25",
 }
 

dataretrieval/combining.py

@@ -8,9 +8,10 @@
 Separated from :mod:`dataretrieval.ogc.planning` so that module stays
 focused on *what* to split, while this module owns *how* to reassemble.
 
-A top-level leaf rather than part of :mod:`dataretrieval.transport`: these are
-pandas transforms over already-fetched results, with no HTTP or event-loop
-concern, consumed by chunk planning and service fan-out as well as by pagination.
+A top-level leaf rather than part of :mod:`dataretrieval.transport`, holding
+the response adjusters (url, elapsed, headers, body release) alongside the
+frame merges. See ADR 0003 for the dependency direction and ADR 0006 for the
+transport boundary and the aggregated-response contract.
 """
 
 from __future__ import annotations
@@ -72,6 +73,16 @@ def _set_response_url(response: httpx.Response, url: str | httpx.URL) -> None:
     response.request = httpx.Request(method=old.method, url=target, headers=old.headers)
 
 
+def _drop_body(response: httpx.Response) -> None:
+    """Free a response's fetched body, keeping status/headers/URL readable.
+
+    ``_content`` and ``_text`` are the slots httpx caches a read and a decoded
+    body in; both are emptied so the two accessors stay valid and agree.
+    """
+    response._content = b""
+    response.__dict__.pop("_text", None)
+
+
 def _lowest_remaining(responses: list[httpx.Response]) -> httpx.Response:
     """The response reporting the lowest ``x-ratelimit-remaining``.
 
@@ -104,13 +115,15 @@ def _merge_response(
 
     The copy's ``.headers`` are rebuilt as a fresh ``httpx.Headers`` from
     ``headers_from``, ``.elapsed`` is set to ``elapsed``, and ``.url`` is
-    overridden when ``url`` is given.  ``base`` and ``headers_from`` are never
+    overridden when ``url`` is given, and its body is emptied (ADR 0006).
+    ``base`` and ``headers_from`` are never
     mutated, and the fresh ``httpx.Headers`` means downstream mutations don't
     back-propagate into any underlying response — so callers may re-fold
     idempotently.  This is the one low-level merge behind both pagination
     (:func:`~dataretrieval.transport.pagination.paginate`) and the chunked /
     fan-out aggregation (:func:`_combine_chunk_responses`)."""
     merged = copy.copy(base)
+    _drop_body(merged)
     merged.headers = httpx.Headers(headers_from.headers)
     merged.elapsed = elapsed
     if url is not None:

dataretrieval/ogc/shaping.py

@@ -11,11 +11,13 @@
 from __future__ import annotations
 
 import logging
+import math
 import re
 from typing import Any
 
 import httpx
 import pandas as pd
+from pandas.api.types import infer_dtype
 
 from dataretrieval._response_metadata import BaseMetadata
 from dataretrieval.ogc.policy import DEFAULT_DIALECT, OgcDialect
@@ -35,6 +37,10 @@
 # (EPSG:4269).
 _CRS = "EPSG:4326"
 
+# What ``infer_dtype`` reports for an object column that cannot hide a dict.
+# Named this way round so an unfamiliar label falls through to the scan.
+_FLAT_DTYPES = frozenset({"string", "empty"})
+
 # Whether geopandas is present is a static, environment-level fact, so warn
 # once here at import time rather than per query/chunk.
 if not GEOPANDAS:
@@ -99,23 +105,112 @@ def _geo_feature_frame(features: list[dict[str, Any]]) -> pd.DataFrame:
     )
 
 
+def _properties_frame(features: list[dict[str, Any]]) -> tuple[pd.DataFrame, bool]:
+    """Build the frame of feature properties, and say whether values nested.
+
+    Flat properties build with the plain ``DataFrame`` constructor; a nested
+    value in *any* feature routes the whole page through ``json_normalize``
+    so no row keeps a raw dict.
+
+    Only the plain builder may act on a normalized frame: ``from_features``
+    keeps raw dicts, so a spatial page that normalized would disagree with
+    its own fallback about the column names.
+    """
+    properties = [feature.get("properties") or {} for feature in features]
+    frame = pd.DataFrame(properties)
+    if any(_holds_nested_value(column) for _, column in frame.items()):
+        return pd.json_normalize(properties, sep="_"), True
+    return frame, False
+
+
+def _holds_nested_value(column: pd.Series) -> bool:
+    """Whether a built column carries a raw dict that needs flattening."""
+    if column.dtype != object:
+        return False
+    values = column.to_numpy()
+    if infer_dtype(values, skipna=True) in _FLAT_DTYPES:
+        return False
+    return any(isinstance(value, dict) for value in values)
+
+
 def _plain_feature_frame(
     features: list[dict[str, Any]], *, include_geometry: bool
 ) -> pd.DataFrame:
     """Build a plain DataFrame from GeoJSON features."""
-    properties = [feature.get("properties") or {} for feature in features]
-    df = pd.json_normalize(properties, sep="_")
+    df, _ = _properties_frame(features)
     df["id"] = [feature.get("id") for feature in features]
     if include_geometry:
         _attach_coordinates(df, features)
     return df
 
 
+def _point_geometries(features: list[dict[str, Any]]) -> Any:
+    """Build the geometry array for an all-2D-point page in one vectorized
+    :func:`geopandas.points_from_xy` call.
+
+    Returns ``None`` when any feature carries a non-point or malformed
+    geometry, so the caller falls back to :func:`_geo_feature_frame`. A
+    feature with no geometry stays ``None``, matching that fallback. Two
+    flat x/y lists rather than coordinate pairs: the paired form is slower.
+    """
+    xs: list[Any] = []
+    ys: list[Any] = []
+    missing: list[int] = []
+    for index, feature in enumerate(features):
+        geometry = feature.get("geometry") or {}
+        if not geometry:
+            missing.append(index)
+            xs.append(math.nan)
+            ys.append(math.nan)
+            continue
+        xy: Any = geometry.get("coordinates")
+        if geometry.get("type") != "Point" or not _is_pair(xy):
+            return None
+        xs.append(xy[0])
+        ys.append(xy[1])
+    try:
+        points = gpd.points_from_xy(xs, ys)
+    except (TypeError, ValueError):
+        return None
+    points[missing] = None
+    return points
+
+
+def _is_pair(value: Any) -> bool:
+    """Whether ``value`` is a two-element coordinate sequence."""
+    return isinstance(value, (list, tuple)) and len(value) == 2
+
+
+def _point_feature_frame(features: list[dict[str, Any]]) -> pd.DataFrame | None:
+    """Fast-path GeoDataFrame for an all-2D-point page, or ``None`` to fall
+    back to :func:`_geo_feature_frame`.
+
+    Declines a page whose properties needed flattening, so a chunked call's
+    column names cannot depend on which pages happened to be all points, and
+    one naming a ``geometry`` property, which collides with the geometry
+    column and which ``from_features`` cannot shape either. Properties are
+    built first because both bail-outs read that frame.
+    """
+    frame, normalized = _properties_frame(features)
+    if normalized or "geometry" in frame.columns:
+        return None
+    points = _point_geometries(features)
+    if points is None:
+        return None
+    return gpd.GeoDataFrame(frame, geometry=points, crs=_CRS)
+
+
 def _spatial_feature_frame(features: list[dict[str, Any]]) -> pd.DataFrame:
     """Build a GeoDataFrame from GeoJSON features with ``id`` first."""
-    df = _geo_feature_frame(features)
+    df = _point_feature_frame(features)
+    if df is None:
+        df = _geo_feature_frame(features)
     df["id"] = [f.get("id") for f in features]
-    return df[["id"] + [col for col in df.columns if col != "id"]]
+    # Pin both names: the fast path appends ``geometry`` last and
+    # ``from_features`` emits it first, so only naming them keeps the two
+    # paths' column order identical across a chunked concat.
+    ordered = ["id", "geometry"]
+    return df[ordered + [col for col in df.columns if col not in ordered]]
 
 
 def _get_resp_data(

dataretrieval/transport/pagination.py

@@ -18,6 +18,7 @@
 from dataretrieval import progress as _progress
 from dataretrieval.combining import (
     _QUOTA_HEADER,
+    _drop_body,
     _merge_response,
     _safe_elapsed,
 )
@@ -126,6 +127,10 @@ def report_page(page: httpx.Response, frame: pd.DataFrame) -> None:
         nrows = len(frame)
         seen: set[Any] = set()
         report_page(response, frame)
+        # Parsed and never read again. httpx responses sit in a reference
+        # cycle, so a superseded page waits on the cyclic GC with its body
+        # still charged -- free every page as it is parsed, not just this one.
+        _drop_body(response)
 
         while (
             cursor is not None
@@ -141,6 +146,7 @@ def report_page(page: httpx.Response, frame: pd.DataFrame) -> None:
                 nrows += len(frame)
                 total_elapsed += _safe_elapsed(response)
                 report_page(response, frame)
+                _drop_body(response)
             except Exception as exc:  # noqa: BLE001
                 logger.warning(
                     "Request failed at cursor %r. Data download interrupted.", cursor

dataretrieval/utils.py

@@ -17,6 +17,7 @@
 
 import pandas as pd
 
+import dataretrieval._deprecation as _deprecation
 import dataretrieval._querying as _querying
 import dataretrieval.transport.http as _transport_http
 from dataretrieval._ambient import Ambient  # noqa: F401 - compatibility re-export
@@ -59,8 +60,16 @@ def format_datetime(
     df: ``pandas.DataFrame``
         The data frame with a formatted 'datetime' column.
 
+    Deprecated: the qw services this shaped responses for are retired and
+    nothing in the package calls it. Combine the columns with
+    :func:`pandas.to_datetime` directly. See
+    :data:`dataretrieval._deprecation.REMOVALS` for the removal horizon.
     """
-    # create a datetime index from the columns in qwdata response
+    _deprecation.warn_deprecated(
+        "`utils.format_datetime`",
+        replacement="a direct `pandas.to_datetime` over the combined columns",
+        removal=_deprecation.REMOVALS["utils.format_datetime"],
+    )
     df[tz_field] = df[tz_field].map(tz)
 
     df["datetime"] = pd.to_datetime(

docs/source/architecture/decisions/0006-service-neutral-transport.rst

@@ -51,6 +51,13 @@ through transport:
 - ``dataretrieval.combining`` -- pandas frame and response assembly. Transport
   returns results *through* it.
 
+An aggregated response describes the call, not a page: its status, headers,
+URL, and elapsed are meaningful, and its body is empty. A page body belongs to
+one arbitrary request of many, so carrying it would invite callers to read it
+as the query's data, and holding it keeps the whole download resident for the
+life of a fan-out. Pages are freed as they are parsed, and the aggregate is
+built without one. Per-page responses handed to an adapter are untouched.
+
 Transport depends only on stable package leaves and third-party infrastructure.
 It must not import OGC modules or service adapters. Service adapters inject
 request construction, response parsing, cursor extraction, and API-specific
@@ -95,6 +102,9 @@ Consequences
   service that cannot use an API key is not told to obtain one.
 - The transport package is internal infrastructure, not a new public API
   promise.
+- Reading ``.content`` or ``.text`` off an aggregated response returns empty.
+  Callers wanting the payload use the returned frame; an adapter wanting bytes
+  reads them from the per-page response before it is aggregated.
 - Keeping presentation and frame assembly out means transport is roughly 570
   lines across five modules, each recognizably HTTP execution policy. Retry is
   the one intricate module, and it is intricate because two independent bounds
@@ -110,4 +120,15 @@ transport, and that only ``dataretrieval.credentials`` names the API-key host.
 Component and adapter tests cover cursor termination, row caps, response
 aggregation, retry exhaustion, ``Retry-After`` limits, the no-progress budget,
 which failures are re-sent, cancellation, no-partial fan-out behavior, and
-credential host scoping.
+credential host scoping. ``test_merge_response_empties_the_body_but_keeps_the_rest``
+pins the empty-body contract and the metadata that must survive it.
+
+Notes
+-----
+
+The empty-body clause was added after the original decision; the rest of the
+ADR is unchanged. Freeing pages as they are parsed measured 434 MB -> 306 MB
+peak heap on a live seven-page, 303k-row Water Data walk (2026-08-26). Note
+when re-measuring that only network-built responses are freed: an
+``httpx.Response(content=...)`` test double keeps its bytes in
+``response.stream``, so a ``MockTransport`` benchmark shows no gain.

tests/utils_test.py

@@ -8,6 +8,7 @@
 import pytest
 
 from dataretrieval import _querying, _wqx, exceptions, nwis, utils
+from dataretrieval._deprecation import REMOVALS
 
 
 class Test_Ambient:
@@ -607,7 +608,13 @@ def test_retrying_get_maps_invalid_url(monkeypatch):
 class TestFormatDatetime:
     """``format_datetime`` joins the three columns NWIS RDB splits a
     timestamp across, and is the only place the package parses a local time
-    with a named zone."""
+    with a named zone. It is deprecated -- the qw services it shaped
+    responses for are retired -- so every call here also warns."""
+
+    def test_is_deprecated_with_the_published_horizon(self):
+        df = pd.DataFrame({"d": ["2020-01-01"], "t": ["12:00"], "z": ["EST"]})
+        with pytest.warns(DeprecationWarning, match=REMOVALS["utils.format_datetime"]):
+            utils.format_datetime(df, "d", "t", "z")
 
     def test_joins_date_time_and_zone_into_utc(self):
         df = pd.DataFrame(