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

NEWS.md

@@ -1,3 +1,18 @@
+**09/15/2026:** Water Data and NGWMN OGC responses build frames faster
+for flat properties and point geometries, with unchanged results,
+and a fan-out no longer holds each chunk's page body in memory
+for the life of the call.
+**Behavior change:** the raw aggregate response reached through `partial_response`
+-- on `ChunkInterrupted`, `FanOutInterrupted`, and the `ChunkedCall` they carry --
+now has an empty body rather than one arbitrary page's bytes.
+Status, headers, URL, and elapsed are unchanged; the data is the returned frame.
+Response metadata is unaffected: it has never retained the response object.
+The page-body lifetime contract is documented in
+[ADR 0006](docs/source/architecture/decisions/0006-service-neutral-transport.rst).
+**Deprecation:** `utils.format_datetime` now emits a `DeprecationWarning`
+and will be removed on or after 2027-09-15.
+Combine the columns with `pandas.to_datetime` directly.
+
 **09/09/2026:** **Bug fix:** code and identifier columns keep their leading zeros. A bare `pandas.read_csv` infers a zero-padded code as a number, so `waterdata.get_samples()` returned parameter code `00060` as `60` and HUC12 `070700050502` as `70700050502`, and `nwis.get_info()` returned `huc_cd` `02060005` as `2060005`. One rule now decides what a code column is — a name ending in `code`, the RDB abbreviation `_cd`, or a name containing `identifier`, `huc`, or `fips` — and every delimited response is parsed through it: the Samples and WQP CSV readers, `rdb.read_rdb` (which reads the names from the RDB header rather than the caller listing them), and the Water Use CSV pages. **Behavior change:** these columns now hold strings. `waterdata.get_samples()`: `USGSpcode`, `Location_HUCEightDigitCode`, `Location_HUCTwelveDigitCode`, `SampleCollectionMethod_Identifier` (`get_samples_summary()` shares the parse; no column in its current profile was affected). `nwis.get_info()`, `nwis.what_sites()`, and `nwis.get_record(service="site")`: `huc_cd`, `state_cd`, `county_cd`, `district_cd`. A comparison against a number — `df["USGSpcode"] == 60` — or a merge onto a numeric key now matches nothing instead of raising, so compare against the padded string (`== "00060"`) or call `.astype(int)` where the number is what you want. **Behavior change:** a count whose name reads as an identifier is numeric again. WQP's `AlternateLocation_IdentifierCount` has been read as text since 05/31/2026 because "Identifier" appears in its name; a name ending in `count` is now excluded from the rule, so the same column has one dtype in every service that reports it. Measurement columns are unchanged, and the `waterdata` OGC getters and `ngwmn` were never affected: their JSON responses deliver codes as strings and numeric coercion there is limited to a fixed list of measurement columns. **Correction to the 1.2.0 notes:** the same fix was applied to the nine `wqp` getters on 05/31/2026 and never recorded here — `wqp.get_results()` and the `what_*` getters have returned HUCs, parameter codes, and FIPS codes as strings since that release.
 
 **09/01/2026:** **Announcement:** We at USGS Water Data for the Nation want your feedback! Tell us how we're doing by taking our quick [survey](https://usgswaterresources.gov1.qualtrics.com/jfe/form/SV_07gX8G1DeOtVrH8), available through September 2026.

dataretrieval/_deprecation.py

@@ -19,6 +19,7 @@
     "waterdata.get_cql(service=)": "2027-08-09",
     "wateruse": "2027-08-11",
     "ogc.interruptions": "2027-08-25",
+    "utils.format_datetime": "2027-09-15",
 }
 
 

dataretrieval/combining.py

@@ -70,6 +70,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``.
 
@@ -102,13 +112,15 @@ def _merge_response(
 
     The copy's ``.headers`` are rebuilt as a new ``httpx.Headers`` from
     ``headers_from``, ``.elapsed`` is set to ``elapsed``, and ``.url`` is
-    overridden when ``url`` is given.  ``base`` and ``headers_from`` are never
-    mutated, and the new ``httpx.Headers`` means downstream mutations don't
-    back-propagate into any underlying response — so callers may re-merge
-    idempotently.  This is the one low-level merge used by both pagination
+    overridden when ``url`` is given, and its body is emptied (ADR 0006).
+    ``base`` and ``headers_from`` are never mutated, and the new
+    ``httpx.Headers`` means downstream mutations don't back-propagate into any
+    underlying response — so callers may re-merge idempotently.  This is the
+    one low-level merge used by 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:
@@ -98,23 +104,108 @@ def _geo_feature_frame(features: list[dict[str, Any]]) -> pd.DataFrame:
     )
 
 
+def _properties_frame(features: list[dict[str, Any]]) -> pd.DataFrame:
+    """Build plain-frame properties, flattening nested dicts with underscores."""
+    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="_")
+    return frame
+
+
+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])
+    if len(missing) == len(features):
+        return gpd.array.from_shapely([None] * len(features))
+    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 without nulls."""
+    return (
+        isinstance(value, (list, tuple))
+        and len(value) == 2
+        and value[0] is not None
+        and value[1] is not None
+    )
+
+
+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`.
+
+    Keeps nested properties as raw dicts, matching ``from_features``.
+    Declines pages with a ``geometry`` property to preserve the fallback's
+    collision handling.
+    """
+    points = _point_geometries(features)
+    if points is None:
+        return None
+    frame = pd.DataFrame([feature.get("properties") or {} for feature in features])
+    if "geometry" in frame.columns:
+        return None
+    return gpd.GeoDataFrame(frame, geometry=points, crs=_CRS, copy=False)
+
+
 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,
 )
@@ -79,6 +80,22 @@ def paginated_failure_message(
     )
 
 
+def _finish_page(
+    page: httpx.Response, frame: pd.DataFrame, *, release_body: bool
+) -> None:
+    """Report a parsed page and release its body when transport owns it."""
+    note_progress()
+    reporter = _progress.current()
+    if reporter is not None:
+        reporter.set_rate_remaining(
+            page.headers.get(_QUOTA_HEADER),
+            limit=page.headers.get("x-ratelimit-limit"),
+        )
+        reporter.add_page(rows=len(frame))
+    if release_body:
+        _drop_body(page)
+
+
 async def paginate(
     initial_req: httpx.Request,
     *,
@@ -96,17 +113,6 @@ async def paginate(
     metadata aggregation.
     """
     logger.debug("Requesting: %s", initial_req.url)
-    reporter = _progress.current()
-
-    def report_page(page: httpx.Response, frame: pd.DataFrame) -> None:
-        note_progress()  # a walk still receiving pages is not stalled
-        if reporter is not None:
-            reporter.set_rate_remaining(
-                page.headers.get(_QUOTA_HEADER),
-                limit=page.headers.get("x-ratelimit-limit"),
-            )
-            reporter.add_page(rows=len(frame))
-
     async with _client_for(client) as session:
         response = await session.send(initial_req)
         raise_for_status(response)
@@ -124,7 +130,7 @@ def report_page(page: httpx.Response, frame: pd.DataFrame) -> None:
         frames = [frame]
         nrows = len(frame)
         seen: set[Any] = set()
-        report_page(response, frame)
+        _finish_page(response, frame, release_body=client is None)
 
         while (
             cursor is not None
@@ -139,7 +145,7 @@ def report_page(page: httpx.Response, frame: pd.DataFrame) -> None:
                 frames.append(frame)
                 nrows += len(frame)
                 total_elapsed += _safe_elapsed(response)
-                report_page(response, frame)
+                _finish_page(response, frame, release_body=client is None)
             except Exception as exc:
                 logger.warning(
                     "Request failed at cursor %r. Data download interrupted.", cursor

dataretrieval/utils.py

@@ -16,6 +16,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
@@ -58,8 +59,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

@@ -53,6 +53,22 @@ 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 request, not the query's combined data;
+retaining page bodies also increases memory use during a fan-out.
+The aggregate is always a copy with its body emptied,
+regardless of who owns the HTTP client.
+
+Page-body lifetime depends on client ownership.
+When no client is explicitly supplied (``client is None``),
+transport releases each page body after parsing and progress reporting.
+This includes clients obtained from the ambient fan-out context,
+which are internally owned.
+An adapter needing bytes must consume them in its page parser.
+When a caller explicitly supplies a client,
+transport leaves its per-page response bodies unchanged.
+
 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
@@ -132,6 +148,8 @@ Consequences
   advice to obtain one is not printed for a service that cannot use it.
 - The transport package is internal infrastructure, not a new public API
   contract.
+- Reading ``.content`` or ``.text`` off an aggregated response returns empty.
+  Callers wanting the payload use the returned frame.
 - Keeping presentation and frame assembly out means transport is roughly 570
   lines across five modules, each recognizably HTTP execution policy. Retry is
   the one complex module, because two independent bounds are what make retry
@@ -150,18 +168,22 @@ which failures are re-sent, cancellation, no-partial fan-out behavior, and
 credential host scoping. The exemptions above are covered by the liveness and
 retry tests over excluded waits and the first-attempt case. Next-page link
 validation is covered by the shared link-policy tests over foreign hosts and
-embedded userinfo.
+embedded userinfo. ``test_merge_response_empties_the_body_but_keeps_the_rest``
+pins the empty-body contract and the metadata that must survive it.
 
 Notes
 -----
 
-The waiting-time, next-page-link, and credentials-leaf clauses were added
-after the original decision, consolidating under ADR 0000 the rules the code
-was stating in prose -- the budget exemptions were argued in five places
+The waiting-time, next-page-link, credentials-leaf, and empty-body clauses were
+added after the original decision, consolidating under ADR 0000 the rules the
+code was stating in prose -- the budget exemptions were argued in five places
 across ``transport/retry.py``, ``transport/liveness.py``, and
 ``transport/fanout.py``.
 
 One sentence of the original Decision was also corrected rather than added to:
 it scoped automatic retry to "gateway 5xx", which was never true of a fanned-out
 call -- those re-send any 5xx, and only the single-shot adapters are limited to
 the gateway statuses. The decision is unchanged; the sentence now describes it.
+
+The page-body lifetime clause was corrected to distinguish internally owned
+clients from explicitly caller-supplied clients.