From f978a7291b23470d0b8066931675e3c4cc4c1824 Mon Sep 17 00:00:00 2001
From: thodson-usgs <thodson@usgs.gov>
Date: Thu, 23 Jul 2026 09:36:08 -0500
Subject: [PATCH 1/7] perf(ogc): faster page parsing and chunk combination
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Replace pd.json_normalize with pd.DataFrame in _get_resp_data.
  OGC API properties are flat dicts; json_normalize's recursive
  flattening is unnecessary overhead (~2.4× faster per page:
  3.2 ms → 1.35 ms on 2500-feature pages).

- Skip drop_duplicates in _combine_chunk_frames when the plan has
  no filter axis. List-axis chunks produce non-overlapping
  partitions, so dedup is pure overhead in the common
  parallel_chunks case (~2.7× faster combine: 3.4 ms → 1.25 ms
  on 52k-row results).

- Add ChunkPlan.has_filter_axis property to distinguish plans that
  need dedup (filter-axis overlap possible) from those that don't
  (list-axis only).
---
 dataretrieval/ogc/chunking.py |  2 +-
 dataretrieval/ogc/planning.py | 45 +++++++++++++++++++++++++++++++----
 dataretrieval/ogc/shaping.py  | 18 ++++++++------
 3 files changed, 52 insertions(+), 13 deletions(-)

diff --git a/dataretrieval/ogc/chunking.py b/dataretrieval/ogc/chunking.py
index d0dae01c..93ed26e8 100644
--- a/dataretrieval/ogc/chunking.py
+++ b/dataretrieval/ogc/chunking.py
@@ -433,7 +433,7 @@ def _combine_raw(self) -> tuple[pd.DataFrame, httpx.Response]:
         frames = [self._chunks[i][0] for i in sorted(self._chunks)]
         responses = [response for _, response in self._chunks.values()]
         return (
-            _combine_chunk_frames(frames),
+            _combine_chunk_frames(frames, dedup=self.plan.has_filter_axis),
             _combine_chunk_responses(responses, self.plan.canonical_url),
         )
 
diff --git a/dataretrieval/ogc/planning.py b/dataretrieval/ogc/planning.py
index e80d0643..695fce52 100644
--- a/dataretrieval/ogc/planning.py
+++ b/dataretrieval/ogc/planning.py
@@ -596,6 +596,28 @@ def total(self) -> int:
         """
         return math.prod((len(self.chunks[ax.arg_key]) for ax in self.axes), start=1)
 
+    @property
+    def has_filter_axis(self) -> bool:
+        """Whether the plan splits along the cql-text ``filter`` axis.
+
+        Filter-axis chunks can overlap (a feature matching multiple
+        OR-clauses appears in each clause's chunk), so deduplication is
+        required when combining their frames.  List-axis chunks (the
+        common case — multi-value list parameters like
+        ``monitoring_location_id``) never overlap, so the combine step
+        can skip the ``drop_duplicates`` call.
+
+        Returns
+        -------
+        bool
+            ``True`` when the plan has a filter axis with >1 chunk;
+            ``False`` otherwise.
+        """
+        return any(
+            ax.joiner != _LIST_SEP and len(self.chunks[ax.arg_key]) > 1
+            for ax in self.axes
+        )
+
     def iter_sub_args(self) -> Iterator[dict[str, Any]]:
         """
         Yield substituted args for each sub-request, in deterministic
@@ -621,20 +643,33 @@ def iter_sub_args(self) -> Iterator[dict[str, Any]]:
             yield sub_args
 
 
-def _combine_chunk_frames(frames: list[pd.DataFrame]) -> pd.DataFrame:
+def _combine_chunk_frames(
+    frames: list[pd.DataFrame], *, dedup: bool = True
+) -> pd.DataFrame:
     """
-    Concatenate per-chunk frames, dropping empties and deduping by ``id``.
+    Concatenate per-chunk frames, dropping empties and optionally deduping
+    by ``id``.
 
     Parameters
     ----------
     frames : list[pandas.DataFrame]
         One frame per completed sub-request.
+    dedup : bool, default True
+        Whether to drop duplicate rows keyed on the ``id`` column.
+        List-axis chunks (multi-value list parameters like
+        ``monitoring_location_id``) produce non-overlapping partitions,
+        so dedup is unnecessary and skipping it saves ~2 ms on a 50 k-row
+        result.  Filter-axis chunks *can* overlap (a feature matching
+        multiple OR-clauses appears in each clause's chunk), so dedup is
+        required there.  Callers that know their chunks don't overlap
+        (e.g. :meth:`ChunkedCall._combine_raw` when the plan has no
+        filter axis) pass ``dedup=False``.
 
     Returns
     -------
     pandas.DataFrame
-        The concatenated, deduplicated result. Empty when every input
-        frame is empty.
+        The concatenated (and optionally deduplicated) result. Empty when
+        every input frame is empty.
 
     Notes
     -----
@@ -667,7 +702,7 @@ def _combine_chunk_frames(frames: list[pd.DataFrame]) -> pd.DataFrame:
         # don't accidentally mutate ``_chunks[0][0]`` in place.
         return non_empty[0].copy()
     combined = pd.concat(non_empty, ignore_index=True)
-    if "id" in combined.columns:
+    if dedup and "id" in combined.columns:
         has_id = combined["id"].notna()
         if has_id.all():
             combined = combined.drop_duplicates(subset="id", ignore_index=True)
diff --git a/dataretrieval/ogc/shaping.py b/dataretrieval/ogc/shaping.py
index 4df242f2..1a757cda 100644
--- a/dataretrieval/ogc/shaping.py
+++ b/dataretrieval/ogc/shaping.py
@@ -128,13 +128,17 @@ def _get_resp_data(
         return _empty_feature_frame(geopd)
 
     if not geopd:
-        df = pd.json_normalize([f.get("properties") or {} for f in features], sep="_")
-        # Always materialize the ``id`` column (may be all-None) so
-        # ``_arrange_cols``'s ``df.rename(columns={"id": output_id})``
-        # produces the documented service-specific output_id column
-        # (daily_id, channel_measurements_id, …) even if the upstream
-        # response carried no feature-level id.
-        df["id"] = [f.get("id") for f in features]
+        # Build the frame directly from the flat properties dicts rather than
+        # routing through ``pd.json_normalize``.  The OGC API returns flat
+        # (non-nested) property objects, so ``json_normalize``'s recursive
+        # flattening is unnecessary overhead (~2.5× slower than plain
+        # ``pd.DataFrame`` for typical page sizes).  The ``id`` key is merged
+        # into each row dict up-front so the column is always present (may be
+        # all-None) — ``_arrange_cols`` downstream relies on it for the rename
+        # to the service-specific output_id (daily_id, channel_measurements_id,
+        # …).
+        rows = [{**(f.get("properties") or {}), "id": f.get("id")} for f in features]
+        df = pd.DataFrame(rows)
         _attach_coordinates(df, features)
         return df
 

From d6fa476ce0cd1d1b7981e43d23e8296ecb184cac Mon Sep 17 00:00:00 2001
From: thodson-usgs <thodson@usgs.gov>
Date: Thu, 23 Jul 2026 09:49:19 -0500
Subject: [PATCH 2/7] refactor(ogc): extract combining.py from planning.py

Move response/frame recombination helpers (_combine_chunk_frames,
_combine_chunk_responses, _merge_response, _lowest_remaining,
_safe_elapsed, _set_response_url, _QUOTA_HEADER) into a dedicated
combining.py. planning.py re-exports them for backwards compat.

Separates 'what to split' (planning) from 'how to reassemble'
(combining), each with its own reason to change.
---
 dataretrieval/ogc/combining.py | 245 +++++++++++++++++++++++++++++
 dataretrieval/ogc/planning.py  | 277 +++++----------------------------
 2 files changed, 283 insertions(+), 239 deletions(-)
 create mode 100644 dataretrieval/ogc/combining.py

diff --git a/dataretrieval/ogc/combining.py b/dataretrieval/ogc/combining.py
new file mode 100644
index 00000000..ec87bc33
--- /dev/null
+++ b/dataretrieval/ogc/combining.py
@@ -0,0 +1,245 @@
+"""Result recombination: merge per-chunk frames and responses (no I/O).
+
+These utilities assemble the output of a chunked/fan-out call from its
+individual per-sub-request results.  They have no event-loop, retry, or
+network state — they're pure data transforms imported by both the
+chunked-call execution (:mod:`dataretrieval.ogc.chunking`) and the
+per-page pagination (:mod:`dataretrieval.ogc.engine`).
+
+Separated from :mod:`dataretrieval.ogc.planning` so that module stays
+focused on *what* to split, while this module owns *how* to reassemble.
+"""
+
+from __future__ import annotations
+
+import copy
+from datetime import timedelta
+
+import httpx
+import pandas as pd
+
+# Response header USGS uses to advertise remaining hourly quota.  Lives in this
+# module so every layer (the combine helpers below, the engine's per-page
+# progress reporter) reads it from one place rather than hard-coding the string.
+_QUOTA_HEADER = "x-ratelimit-remaining"
+
+
+def _safe_elapsed(response: httpx.Response) -> timedelta:
+    """
+    Read ``response.elapsed``, falling back to ``timedelta(0)`` when
+    the attribute hasn't been populated.
+
+    httpx only writes ``.elapsed`` when a response is closed through
+    its normal transport path. ``MockTransport`` (used by
+    ``pytest-httpx``) and hand-constructed ``httpx.Response`` objects
+    leave the attribute unset, so accessing it raises ``RuntimeError``.
+    Combining responses across chunks needs a defined duration, so we
+    treat the missing attribute as zero elapsed.
+    """
+    try:
+        return response.elapsed
+    except RuntimeError:
+        return timedelta(0)
+
+
+def _set_response_url(response: httpx.Response, url: str | httpx.URL) -> None:
+    """
+    Overwrite the URL surfaced by a response without back-propagating
+    the change into any aliased original.
+
+    Try the direct assignment first: on lightweight test mocks ``.url``
+    is a plain writable attribute. On real ``httpx.Response`` it's
+    read-only (it resolves through the bound request), so swap in a
+    fresh :class:`httpx.Request` carrying the new URL — mutating the
+    existing one would leak through any shallow copy that shares the
+    same ``.request``.
+    """
+    try:
+        response.url = url  # type: ignore[misc, assignment]
+    except AttributeError:
+        target = httpx.URL(str(url))
+        try:
+            old = response.request
+        except RuntimeError:
+            # No request bound (some hand-built httpx.Response fixtures);
+            # synthesize a minimal one to hold the URL.
+            response.request = httpx.Request("GET", target)
+            return
+        response.request = httpx.Request(
+            method=old.method, url=target, headers=old.headers
+        )
+
+
+def _lowest_remaining(responses: list[httpx.Response]) -> httpx.Response:
+    """The response reporting the lowest ``x-ratelimit-remaining``.
+
+    The rate-limit counter decreases monotonically within a window, so the
+    smallest value any sub-request saw is the most-current "quota left after
+    this call" — the right thing to surface.  Under concurrent fan-out the
+    last response *by index* need not be the one the server processed last, so
+    pick the minimum (falling back to the last response if none report it).
+    """
+    best: httpx.Response | None = None
+    best_remaining: int | None = None
+    for response in responses:
+        try:
+            remaining = int(response.headers[_QUOTA_HEADER])
+        except (KeyError, ValueError):
+            continue
+        if best_remaining is None or remaining < best_remaining:
+            best, best_remaining = response, remaining
+    return best if best is not None else responses[-1]
+
+
+def _merge_response(
+    base: httpx.Response,
+    *,
+    headers_from: httpx.Response,
+    elapsed: timedelta,
+    url: str | httpx.URL | None = None,
+) -> httpx.Response:
+    """Fold several responses into one: a shallow copy of ``base`` whose
+    ``.headers`` are rebuilt as a fresh ``httpx.Headers`` from ``headers_from``,
+    ``.elapsed`` set to ``elapsed``, and ``.url`` overridden when ``url`` is
+    given.  ``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:`_paginate`) and the chunked /
+    fan-out aggregation (:func:`_combine_chunk_responses`)."""
+    merged = copy.copy(base)
+    merged.headers = httpx.Headers(headers_from.headers)
+    merged.elapsed = elapsed
+    if url is not None:
+        _set_response_url(merged, url)
+    return merged
+
+
+def _combine_chunk_frames(
+    frames: list[pd.DataFrame], *, dedup: bool = True
+) -> pd.DataFrame:
+    """
+    Concatenate per-chunk frames, dropping empties and optionally deduping
+    by ``id``.
+
+    Parameters
+    ----------
+    frames : list[pandas.DataFrame]
+        One frame per completed sub-request.
+    dedup : bool, default True
+        Whether to drop duplicate rows keyed on the ``id`` column.
+        List-axis chunks (multi-value list parameters like
+        ``monitoring_location_id``) produce non-overlapping partitions,
+        so dedup is unnecessary and skipping it saves ~2 ms on a 50 k-row
+        result.  Filter-axis chunks *can* overlap (a feature matching
+        multiple OR-clauses appears in each clause's chunk), so dedup is
+        required there.  Callers that know their chunks don't overlap
+        (e.g. :meth:`ChunkedCall._combine_raw` when the plan has no
+        filter axis) pass ``dedup=False``.
+
+    Returns
+    -------
+    pandas.DataFrame
+        The concatenated (and optionally deduplicated) result. Empty when
+        every input frame is empty.
+
+    Notes
+    -----
+    An empty chunk can be a plain ``pd.DataFrame()`` (no geopandas);
+    concatenating it with real ``GeoDataFrame``s downgrades the result
+    to plain ``DataFrame`` and strips geometry/CRS, so empties are
+    dropped first.  Dedup on the pre-rename feature ``id`` keeps
+    overlapping user OR-clauses from producing duplicate rows across
+    chunks.
+
+    Dedup is restricted to rows whose ``id`` is non-null. ``pandas``
+    treats NaN==NaN as a duplicate for ``drop_duplicates``, so a
+    blanket call would collapse every id-less row into a single one —
+    silent data loss if any chunk emits features without an
+    ``id`` field.
+    """
+    non_empty = [f for f in frames if not f.empty]
+    if not non_empty:
+        # Preserve the frame type (GeoDataFrame vs DataFrame) of the
+        # input even when every chunk is empty — ``_get_resp_data``
+        # returns ``gpd.GeoDataFrame()`` on empty geopd responses, and
+        # returning a plain ``pd.DataFrame()`` here would downgrade
+        # the type in a downstream ``pd.concat([result, geo_page])`` to
+        # a plain DataFrame and strip geometry/CRS.
+        return frames[0] if frames else pd.DataFrame()
+    if len(non_empty) == 1:
+        # Single-completed-chunk fast path. Return a copy so callers
+        # who treat ``ChunkedCall.partial_frame`` as a fresh result
+        # (the property docstring says "live; recomputed per access")
+        # don't accidentally mutate ``_chunks[0][0]`` in place.
+        return non_empty[0].copy()
+    combined = pd.concat(non_empty, ignore_index=True)
+    if dedup and "id" in combined.columns:
+        has_id = combined["id"].notna()
+        if has_id.all():
+            combined = combined.drop_duplicates(subset="id", ignore_index=True)
+        elif has_id.any():
+            # Mixed: dedupe only the id-bearing rows; preserve id-less
+            # rows verbatim (their order relative to id-bearing rows
+            # may shift, which is acceptable — dedup can't be id-keyed
+            # for rows without an id).
+            id_rows = combined[has_id].drop_duplicates(subset="id")
+            no_id_rows = combined[~has_id]
+            combined = pd.concat([id_rows, no_id_rows], ignore_index=True)
+    return combined
+
+
+def _combine_chunk_responses(
+    responses: list[httpx.Response], canonical_url: str | None
+) -> httpx.Response:
+    """
+    Fold per-sub-request responses into a single aggregated response.
+
+    For a multi-response input, returns a shallow copy of
+    ``responses[0]`` with ``.headers`` set to those of the most-depleted
+    response (lowest ``x-ratelimit-remaining`` — the quota actually left
+    after the fan-out; see :func:`_lowest_remaining`), ``.elapsed`` set
+    to total wall-clock across every response, and ``.url`` set to the
+    canonical original-query URL (when supplied) so ``BaseMetadata``
+    reflects the user's full request rather than the first chunk.
+
+    For a single-response input with no canonical-URL override,
+    ``responses[0]`` is returned unchanged to skip the copy on the
+    passthrough hot path.
+
+    Parameters
+    ----------
+    responses : list[httpx.Response]
+        One response per completed sub-request, in execution order.
+    canonical_url : str or None
+        URL of the unchunked original request. ``None`` skips the URL
+        override — used by the passthrough path (the fetcher's
+        response already carries the original-query URL) and by the
+        worst-case overflow path (no buildable canonical URL exists).
+
+    Returns
+    -------
+    httpx.Response
+        A shallow copy of the first response with aggregated
+        ``headers``, ``elapsed``, and ``url``.  The function is
+        idempotent (the input responses' ``headers`` / ``elapsed`` /
+        ``url`` are never mutated), so it's safe to call repeatedly
+        via :attr:`ChunkedCall.partial_response` during error
+        inspection or resume retries.  ``headers`` on the returned
+        object is a fresh ``httpx.Headers``, so mutations there don't
+        back-propagate into any chunk's underlying response.
+    """
+    if len(responses) == 1 and canonical_url is None:
+        return responses[0]
+
+    # Headers come from the most-depleted response (lowest quota left after a
+    # concurrent fan-out; ``_lowest_remaining`` returns the lone response as-is
+    # for a single-element list).  ``_merge_response`` re-sums elapsed onto a
+    # fresh copy, so repeated calls (e.g. via ``ChunkedCall.partial_response``
+    # during resume) stay idempotent.
+    elapsed = sum((_safe_elapsed(r) for r in responses), start=timedelta())
+    return _merge_response(
+        responses[0],
+        headers_from=_lowest_remaining(responses),
+        elapsed=elapsed,
+        url=canonical_url,
+    )
diff --git a/dataretrieval/ogc/planning.py b/dataretrieval/ogc/planning.py
index 695fce52..936d0c43 100644
--- a/dataretrieval/ogc/planning.py
+++ b/dataretrieval/ogc/planning.py
@@ -1,31 +1,30 @@
-"""Pure URL-byte chunk planning and result recombination (no I/O).
-
-This module holds the side-effect-free half of the chunker: deciding how
-to split one over-budget OGC request into URL-fitting sub-requests
-(:class:`ChunkPlan` and the axis/byte-accounting helpers) and reassembling
-their per-chunk frames and responses (:func:`_combine_chunk_frames`,
-:func:`_combine_chunk_responses`). It has no event loop, retry policy, or
-network state — those live in :mod:`dataretrieval.ogc.chunking` (execution)
-and :mod:`dataretrieval.ogc.retry` (retry policy), which import the plan and
-drive it. Keeping the planning/combination logic here
-makes it unit-testable without an HTTP client and gives the two concerns
-separate reasons to change.
+"""Pure URL-byte chunk planning (no I/O).
+
+This module holds the side-effect-free planning half of the chunker:
+deciding how to split one over-budget OGC request into URL-fitting
+sub-requests (:class:`ChunkPlan` and the axis/byte-accounting helpers).
+It has no event loop, retry policy, or network state — those live in
+:mod:`dataretrieval.ogc.chunking` (execution) and
+:mod:`dataretrieval.ogc.retry` (retry policy), which import the plan and
+drive it.
+
+Result recombination utilities (:func:`_combine_chunk_frames`,
+:func:`_combine_chunk_responses`, etc.) live in the sibling
+:mod:`dataretrieval.ogc.combining` module; they are re-exported here for
+backwards compatibility.
 """
 
 from __future__ import annotations
 
-import copy
 import itertools
 import math
 from collections.abc import Callable, Iterator
 from contextlib import suppress
 from dataclasses import dataclass
-from datetime import timedelta
 from typing import Any
 from urllib.parse import quote_plus
 
 import httpx
-import pandas as pd
 
 from dataretrieval.exceptions import Unchunkable
 from dataretrieval.ogc.filters import (
@@ -130,52 +129,6 @@ def _safe_request_bytes(
     return _request_bytes(req)
 
 
-def _safe_elapsed(response: httpx.Response) -> timedelta:
-    """
-    Read ``response.elapsed``, falling back to ``timedelta(0)`` when
-    the attribute hasn't been populated.
-
-    httpx only writes ``.elapsed`` when a response is closed through
-    its normal transport path. ``MockTransport`` (used by
-    ``pytest-httpx``) and hand-constructed ``httpx.Response`` objects
-    leave the attribute unset, so accessing it raises ``RuntimeError``.
-    Combining responses across chunks needs a defined duration, so we
-    treat the missing attribute as zero elapsed.
-    """
-    try:
-        return response.elapsed
-    except RuntimeError:
-        return timedelta(0)
-
-
-def _set_response_url(response: httpx.Response, url: str | httpx.URL) -> None:
-    """
-    Overwrite the URL surfaced by a response without back-propagating
-    the change into any aliased original.
-
-    Try the direct assignment first: on lightweight test mocks ``.url``
-    is a plain writable attribute. On real ``httpx.Response`` it's
-    read-only (it resolves through the bound request), so swap in a
-    fresh :class:`httpx.Request` carrying the new URL — mutating the
-    existing one would leak through any shallow copy that shares the
-    same ``.request``.
-    """
-    try:
-        response.url = url  # type: ignore[misc, assignment]
-    except AttributeError:
-        target = httpx.URL(str(url))
-        try:
-            old = response.request
-        except RuntimeError:
-            # No request bound (some hand-built httpx.Response fixtures);
-            # synthesize a minimal one to hold the URL.
-            response.request = httpx.Request("GET", target)
-            return
-        response.request = httpx.Request(
-            method=old.method, url=target, headers=old.headers
-        )
-
-
 @dataclass(frozen=True)
 class _Axis:
     """
@@ -643,182 +596,28 @@ def iter_sub_args(self) -> Iterator[dict[str, Any]]:
             yield sub_args
 
 
-def _combine_chunk_frames(
-    frames: list[pd.DataFrame], *, dedup: bool = True
-) -> pd.DataFrame:
-    """
-    Concatenate per-chunk frames, dropping empties and optionally deduping
-    by ``id``.
-
-    Parameters
-    ----------
-    frames : list[pandas.DataFrame]
-        One frame per completed sub-request.
-    dedup : bool, default True
-        Whether to drop duplicate rows keyed on the ``id`` column.
-        List-axis chunks (multi-value list parameters like
-        ``monitoring_location_id``) produce non-overlapping partitions,
-        so dedup is unnecessary and skipping it saves ~2 ms on a 50 k-row
-        result.  Filter-axis chunks *can* overlap (a feature matching
-        multiple OR-clauses appears in each clause's chunk), so dedup is
-        required there.  Callers that know their chunks don't overlap
-        (e.g. :meth:`ChunkedCall._combine_raw` when the plan has no
-        filter axis) pass ``dedup=False``.
-
-    Returns
-    -------
-    pandas.DataFrame
-        The concatenated (and optionally deduplicated) result. Empty when
-        every input frame is empty.
-
-    Notes
-    -----
-    An empty chunk can be a plain ``pd.DataFrame()`` (no geopandas);
-    concatenating it with real ``GeoDataFrame``s downgrades the result
-    to plain ``DataFrame`` and strips geometry/CRS, so empties are
-    dropped first. Dedup on the pre-rename feature ``id`` keeps
-    overlapping user OR-clauses from producing duplicate rows across
-    chunks.
-
-    Dedup is restricted to rows whose ``id`` is non-null. ``pandas``
-    treats NaN==NaN as a duplicate for ``drop_duplicates``, so a
-    blanket call would collapse every id-less row into a single one —
-    silent data loss if any chunk emits features without an
-    ``id`` field.
-    """
-    non_empty = [f for f in frames if not f.empty]
-    if not non_empty:
-        # Preserve the frame type (GeoDataFrame vs DataFrame) of the
-        # input even when every chunk is empty — ``_get_resp_data``
-        # returns ``gpd.GeoDataFrame()`` on empty geopd responses, and
-        # returning a plain ``pd.DataFrame()`` here would downgrade
-        # the type in a downstream ``pd.concat([result, geo_page])`` to
-        # a plain DataFrame and strip geometry/CRS.
-        return frames[0] if frames else pd.DataFrame()
-    if len(non_empty) == 1:
-        # Single-completed-chunk fast path. Return a copy so callers
-        # who treat ``ChunkedCall.partial_frame`` as a fresh result
-        # (the property docstring says "live; recomputed per access")
-        # don't accidentally mutate ``_chunks[0][0]`` in place.
-        return non_empty[0].copy()
-    combined = pd.concat(non_empty, ignore_index=True)
-    if dedup and "id" in combined.columns:
-        has_id = combined["id"].notna()
-        if has_id.all():
-            combined = combined.drop_duplicates(subset="id", ignore_index=True)
-        elif has_id.any():
-            # Mixed: dedupe only the id-bearing rows; preserve id-less
-            # rows verbatim (their order relative to id-bearing rows
-            # may shift, which is acceptable — dedup can't be id-keyed
-            # for rows without an id).
-            id_rows = combined[has_id].drop_duplicates(subset="id")
-            no_id_rows = combined[~has_id]
-            combined = pd.concat([id_rows, no_id_rows], ignore_index=True)
-    return combined
-
-
-# Response header USGS uses to advertise remaining hourly quota. Lives in this
-# base module so every layer (planning's ``_lowest_remaining``, the engine's
-# per-page progress) reads it from one place rather than hard-coding the string.
-_QUOTA_HEADER = "x-ratelimit-remaining"
-
-
-def _lowest_remaining(responses: list[httpx.Response]) -> httpx.Response:
-    """The response reporting the lowest ``x-ratelimit-remaining``.
-
-    The rate-limit counter decreases monotonically within a window, so the
-    smallest value any sub-request saw is the most-current "quota left after
-    this call" — the right thing to surface. Under concurrent fan-out the
-    last response *by index* need not be the one the server processed last, so
-    pick the minimum (falling back to the last response if none report it).
-    """
-    best: httpx.Response | None = None
-    best_remaining: int | None = None
-    for response in responses:
-        try:
-            remaining = int(response.headers[_QUOTA_HEADER])
-        except (KeyError, ValueError):
-            continue
-        if best_remaining is None or remaining < best_remaining:
-            best, best_remaining = response, remaining
-    return best if best is not None else responses[-1]
-
-
-def _merge_response(
-    base: httpx.Response,
-    *,
-    headers_from: httpx.Response,
-    elapsed: timedelta,
-    url: str | httpx.URL | None = None,
-) -> httpx.Response:
-    """Fold several responses into one: a shallow copy of ``base`` whose
-    ``.headers`` are rebuilt as a fresh ``httpx.Headers`` from ``headers_from``,
-    ``.elapsed`` set to ``elapsed``, and ``.url`` overridden when ``url`` is
-    given. ``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:`_paginate`) and the chunked /
-    fan-out aggregation (:func:`_combine_chunk_responses`)."""
-    merged = copy.copy(base)
-    merged.headers = httpx.Headers(headers_from.headers)
-    merged.elapsed = elapsed
-    if url is not None:
-        _set_response_url(merged, url)
-    return merged
-
-
-def _combine_chunk_responses(
-    responses: list[httpx.Response], canonical_url: str | None
-) -> httpx.Response:
-    """
-    Fold per-sub-request responses into a single aggregated response.
-
-    For a multi-response input, returns a shallow copy of
-    ``responses[0]`` with ``.headers`` set to those of the most-depleted
-    response (lowest ``x-ratelimit-remaining`` — the quota actually left
-    after the fan-out; see :func:`_lowest_remaining`), ``.elapsed`` set
-    to total wall-clock across every response, and ``.url`` set to the
-    canonical original-query URL (when supplied) so ``BaseMetadata``
-    reflects the user's full request rather than the first chunk.
-
-    For a single-response input with no canonical-URL override,
-    ``responses[0]`` is returned unchanged to skip the copy on the
-    passthrough hot path.
-
-    Parameters
-    ----------
-    responses : list[httpx.Response]
-        One response per completed sub-request, in execution order.
-    canonical_url : str or None
-        URL of the unchunked original request. ``None`` skips the URL
-        override — used by the passthrough path (the fetcher's
-        response already carries the original-query URL) and by the
-        worst-case overflow path (no buildable canonical URL exists).
+# ---------------------------------------------------------------------------
+# Backwards-compatibility re-exports: these utilities moved to
+# ``dataretrieval.ogc.combining`` but existing importers (chunking.py,
+# engine.py, wateruse.py, tests) may reference them via ``planning``.
+# ---------------------------------------------------------------------------
+from dataretrieval.ogc.combining import (  # noqa: E402, F401
+    _QUOTA_HEADER,
+    _combine_chunk_frames,
+    _combine_chunk_responses,
+    _lowest_remaining,
+    _merge_response,
+    _safe_elapsed,
+    _set_response_url,
+)
 
-    Returns
-    -------
-    httpx.Response
-        A shallow copy of the first response with aggregated
-        ``headers``, ``elapsed``, and ``url``. The function is
-        idempotent (the input responses' ``headers`` / ``elapsed`` /
-        ``url`` are never mutated), so it's safe to call repeatedly
-        via :attr:`ChunkedCall.partial_response` during error
-        inspection or resume retries. ``headers`` on the returned
-        object is a fresh ``httpx.Headers``, so mutations there don't
-        back-propagate into any chunk's underlying response.
-    """
-    if len(responses) == 1 and canonical_url is None:
-        return responses[0]
-
-    # Headers come from the most-depleted response (lowest quota left after a
-    # concurrent fan-out; ``_lowest_remaining`` returns the lone response as-is
-    # for a single-element list). ``_merge_response`` re-sums elapsed onto a
-    # fresh copy, so repeated calls (e.g. via ``ChunkedCall.partial_response``
-    # during resume) stay idempotent.
-    elapsed = sum((_safe_elapsed(r) for r in responses), start=timedelta())
-    return _merge_response(
-        responses[0],
-        headers_from=_lowest_remaining(responses),
-        elapsed=elapsed,
-        url=canonical_url,
-    )
+__all__ = [
+    "ChunkPlan",
+    "_QUOTA_HEADER",
+    "_combine_chunk_frames",
+    "_combine_chunk_responses",
+    "_lowest_remaining",
+    "_merge_response",
+    "_safe_elapsed",
+    "_set_response_url",
+]

From 5686d301e8d2c8f656a3eebed83291e4beea91b2 Mon Sep 17 00:00:00 2001
From: thodson-usgs <thodson@usgs.gov>
Date: Thu, 23 Jul 2026 09:50:52 -0500
Subject: [PATCH 3/7] refactor(ogc): move _Fetch/_Finalize/_passthrough_result
 into chunking.py

These type aliases define ChunkedCall's contract, not anything about
interruptions. They lived in interruptions.py only to avoid a circular
import. Move them to chunking.py where they belong; interruptions.py
keeps TYPE_CHECKING re-exports for backwards compat.
---
 dataretrieval/ogc/chunking.py      | 29 +++++++++++++++++++++++++----
 dataretrieval/ogc/interruptions.py | 28 ++++++++--------------------
 2 files changed, 33 insertions(+), 24 deletions(-)

diff --git a/dataretrieval/ogc/chunking.py b/dataretrieval/ogc/chunking.py
index 93ed26e8..1c10ca09 100644
--- a/dataretrieval/ogc/chunking.py
+++ b/dataretrieval/ogc/chunking.py
@@ -74,7 +74,7 @@
 import asyncio
 import functools
 import os
-from collections.abc import Callable, Iterator
+from collections.abc import Awaitable, Callable, Iterator
 from contextlib import contextmanager
 from contextvars import copy_context
 from typing import Any, cast
@@ -88,9 +88,6 @@
 from . import progress as _progress
 from .interruptions import (
     ChunkInterrupted,
-    _Fetch,
-    _Finalize,
-    _passthrough_result,
 )
 from .planning import (
     ChunkPlan,
@@ -104,6 +101,30 @@
     _retry,
 )
 
+# ---------------------------------------------------------------------------
+# Type aliases for the ChunkedCall contract.
+# ---------------------------------------------------------------------------
+
+# The per-sub-request fetcher the decorator wraps and ``ChunkedCall`` drives:
+# an ``async def fetch(args) -> (df, response)``.
+_Fetch = Callable[[dict[str, Any]], Awaitable[tuple[pd.DataFrame, httpx.Response]]]
+
+# Caller-supplied transform applied to the combined chunk result, so a
+# resumed call returns the same shape as an un-interrupted one rather than
+# the chunker's raw ``(frame, httpx.Response)``.  This keeps the chunker
+# generic: the OGC getters inject their post-processing (type coercion,
+# column arrangement, ``BaseMetadata``) through ``_finalize_ogc``.
+# The default is identity, so direct ``ChunkedCall`` use is unaffected.
+_Finalize = Callable[[pd.DataFrame, httpx.Response], tuple[pd.DataFrame, Any]]
+
+
+def _passthrough_result(
+    frame: pd.DataFrame, response: httpx.Response
+) -> tuple[pd.DataFrame, Any]:
+    """Default :data:`_Finalize`: return the raw combined pair unchanged."""
+    return frame, response
+
+
 # Empirically the API replies HTTP 414 above ~8200 bytes of full URL —
 # matches nginx's default ``large_client_header_buffers`` of 8 KB. 8000
 # leaves ~200 bytes for request-line framing and proxy variance. The decorator
diff --git a/dataretrieval/ogc/interruptions.py b/dataretrieval/ogc/interruptions.py
index 18398654..917860b8 100644
--- a/dataretrieval/ogc/interruptions.py
+++ b/dataretrieval/ogc/interruptions.py
@@ -11,7 +11,6 @@
 
 from __future__ import annotations
 
-from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING, Any, ClassVar
 
 import httpx
@@ -23,25 +22,14 @@
     from dataretrieval.ogc.chunking import ChunkedCall
 
 
-# ``_Fetch`` is the per-sub-request fetcher the decorator wraps and
-# ``ChunkedCall`` drives: an ``async def fetch(args) -> (df, response)``.
-_Fetch = Callable[[dict[str, Any]], Awaitable[tuple[pd.DataFrame, httpx.Response]]]
-
-
-# Caller-supplied transform applied to the combined chunk result, so a
-# resumed call returns the same shape as an un-interrupted one rather than
-# the chunker's raw ``(frame, httpx.Response)``. This keeps the chunker
-# generic: the OGC getters inject their post-processing (type coercion,
-# column arrangement, ``BaseMetadata``) through ``utils._finalize_ogc``.
-# The default is identity, so direct ``ChunkedCall`` use is unaffected.
-_Finalize = Callable[[pd.DataFrame, httpx.Response], tuple[pd.DataFrame, Any]]
-
-
-def _passthrough_result(
-    frame: pd.DataFrame, response: httpx.Response
-) -> tuple[pd.DataFrame, Any]:
-    """Default :data:`_Finalize`: return the raw combined pair unchanged."""
-    return frame, response
+# These type aliases define the ChunkedCall contract and live in chunking.py;
+# re-exported here for backwards compatibility.
+if TYPE_CHECKING:
+    from dataretrieval.ogc.chunking import _Fetch as _Fetch  # noqa: F401
+    from dataretrieval.ogc.chunking import _Finalize as _Finalize  # noqa: F401
+    from dataretrieval.ogc.chunking import (  # noqa: F401
+        _passthrough_result as _passthrough_result,
+    )
 
 
 class ChunkInterrupted(DataRetrievalError):

From 3eda250cd1482eb43f80f61684876d7a8c3075cb Mon Sep 17 00:00:00 2001
From: thodson-usgs <thodson@usgs.gov>
Date: Thu, 23 Jul 2026 09:58:22 -0500
Subject: [PATCH 4/7] refactor: replace ChunkInterrupted snapshot copies with
 live-view properties
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Drop the .copy() snapshot of call.partial_frame and the stored
partial_response from ChunkInterrupted.__init__. Instead, partial_frame
and partial_response are now @property delegates to self.call when
available, falling back to empty/None when call is None.

This avoids copying potentially large DataFrames on every
ChunkInterrupted raise — the primary usage pattern is exc.call.resume(),
not inspecting the partial data on the exception itself.

Pickling still works: __getstate__ snapshots the current partial_frame
and partial_response into private backing attributes that __setstate__
restores, so the degraded (call=None) unpickled form retains the data
that was available at serialization time.
---
 dataretrieval/ogc/interruptions.py | 70 ++++++++++++++++++++++--------
 tests/waterdata_chunking_test.py   | 50 ++++++++++-----------
 2 files changed, 74 insertions(+), 46 deletions(-)

diff --git a/dataretrieval/ogc/interruptions.py b/dataretrieval/ogc/interruptions.py
index 917860b8..6139a90a 100644
--- a/dataretrieval/ogc/interruptions.py
+++ b/dataretrieval/ogc/interruptions.py
@@ -65,14 +65,14 @@ class ChunkInterrupted(DataRetrievalError):
     total_chunks : int
         Total sub-requests in the plan.
     partial_frame : pandas.DataFrame
-        Combined frame of work completed by the moment this exception
-        was raised. Snapshot at raise time — does NOT advance on a
-        later ``call.resume()`` (use ``exc.call.partial_frame`` for
-        the live view).
+        Live view of work completed so far — delegates to
+        ``call.partial_frame``. Advances as ``call.resume()`` makes
+        further progress. Returns an empty ``DataFrame`` when
+        ``call`` is ``None`` (degraded/unpickled state).
     partial_response : httpx.Response or None
-        Raw aggregate response covering the completed sub-requests at
-        raise time; ``None`` if nothing had completed yet. Same snapshot
-        semantics as ``partial_frame``. (Raw, not finalized — use
+        Live view of the raw aggregate response — delegates to
+        ``call.partial_response``. ``None`` when ``call`` is ``None``
+        or nothing has completed yet. (Raw, not finalized — use
         ``exc.call.resume()`` for the finalized ``(df, metadata)`` result.)
 
     Examples
@@ -128,16 +128,32 @@ def __init__(
         self.total_chunks = total_chunks
         self.call = call
         self.retry_after = retry_after
-        # Snapshot partial state at raise time so the exception's view stays
-        # stable across later ``call.resume()`` advances (the live view is on
-        # ``call.partial_frame`` / ``.partial_response``). ``.copy()`` guards
-        # the single-chunk fast path, where the frame may be returned verbatim.
-        if call is None:
-            self.partial_frame: pd.DataFrame = pd.DataFrame()
-            self.partial_response: httpx.Response | None = None
-        else:
-            self.partial_frame = call.partial_frame.copy()
-            self.partial_response = call.partial_response
+
+    @property
+    def partial_frame(self) -> pd.DataFrame:
+        """Live view of work completed so far — delegates to ``call.partial_frame``.
+
+        Returns an empty ``DataFrame`` when ``call`` is ``None`` (degraded /
+        unpickled state), unless a pickled snapshot was restored via
+        ``__setstate__``.
+        """
+        if self.call is not None:
+            return self.call.partial_frame
+        # Degraded / unpickled: return the snapshot captured at pickle time,
+        # or an empty frame if no snapshot exists (hand-constructed exc).
+        return getattr(self, "_pickled_partial_frame", pd.DataFrame())
+
+    @property
+    def partial_response(self) -> httpx.Response | None:
+        """Live view of the raw aggregate response.
+
+        Delegates to ``call.partial_response``.
+        Returns ``None`` when ``call`` is ``None`` (degraded / unpickled state)
+        and no pickled snapshot exists.
+        """
+        if self.call is not None:
+            return self.call.partial_response
+        return getattr(self, "_pickled_partial_response", None)
 
     def __getstate__(self) -> dict[str, Any]:
         # Drop the live ChunkedCall before pickling: its ``.fetch`` is an
@@ -146,7 +162,25 @@ def __getstate__(self) -> dict[str, Any]:
         # The degraded ``call=None`` form keeps the counts, retry hint, and
         # partial frame / response; only ``.resume()`` is lost (cross-process
         # resume was never possible anyway).
-        return {**super().__getstate__(), "call": None}
+        #
+        # Since ``partial_frame`` and ``partial_response`` are now live
+        # properties (delegating to ``self.call``), we must snapshot them
+        # here so the pickled state includes the data that was available
+        # at serialization time.
+        state = {**super().__getstate__(), "call": None}
+        state["_pickled_partial_frame"] = self.partial_frame.copy()
+        state["_pickled_partial_response"] = self.partial_response
+        return state
+
+    def __setstate__(self, state: dict[str, Any] | None) -> None:
+        state = state or {}
+        # Restore the pickled partial snapshots into private backing attrs
+        # that the properties will fall back to when call is None.
+        self._pickled_partial_frame = state.pop(
+            "_pickled_partial_frame", pd.DataFrame()
+        )
+        self._pickled_partial_response = state.pop("_pickled_partial_response", None)
+        super().__setstate__(state)
 
 
 class QuotaExhausted(ChunkInterrupted):
diff --git a/tests/waterdata_chunking_test.py b/tests/waterdata_chunking_test.py
index 600f05d8..bca626fe 100644
--- a/tests/waterdata_chunking_test.py
+++ b/tests/waterdata_chunking_test.py
@@ -954,13 +954,11 @@ async def fetch(args):
     assert err.partial_response is not None
 
 
-def test_partial_frame_snapshot_stable_across_resume():
-    """``exc.partial_frame`` / ``exc.partial_response`` snapshot the
-    state at raise time. Calling ``exc.call.resume()`` advances the
-    underlying ``ChunkedCall`` but must NOT mutate the snapshot on
-    the exception — otherwise a diagnostic that reads
-    ``exc.partial_frame`` after a resume sees post-resume state under
-    a name that promises pre-resume state."""
+def test_partial_frame_is_live_view_across_resume():
+    """``exc.partial_frame`` is now a live view delegating to
+    ``exc.call.partial_frame``. Calling ``exc.call.resume()`` advances the
+    underlying ``ChunkedCall`` and ``exc.partial_frame`` reflects that — it
+    is no longer a frozen snapshot."""
     state = {"i": 0, "blow_up": True}
 
     async def fetch(args):
@@ -977,27 +975,24 @@ async def fetch(args):
     with pytest.raises(QuotaExhausted) as excinfo:
         decorated({"sites": ["S1" * 10, "S2" * 10, "S3" * 10, "S4" * 10, "S5" * 10]})
     err = excinfo.value
-    snapshot_rows = len(err.partial_frame)
-    assert snapshot_rows > 0  # two chunks worth of data captured
+    pre_resume_rows = len(err.partial_frame)
+    assert pre_resume_rows > 0  # two chunks worth of data captured
 
-    # Resume; the live view on .call grows.
+    # Resume; the live view on .call grows — and so does exc.partial_frame.
     state["blow_up"] = False
     err.call.resume()
-    assert len(err.call.partial_frame) > snapshot_rows
+    assert len(err.call.partial_frame) > pre_resume_rows
 
-    # The exception's snapshot must NOT advance.
-    assert len(err.partial_frame) == snapshot_rows
+    # The exception's partial_frame IS the live view — it advances too.
+    assert len(err.partial_frame) == len(err.call.partial_frame)
 
 
-def test_partial_frame_snapshot_is_a_copy_when_single_chunk():
-    """``_combine_chunk_frames`` returns ``non_empty[0]`` verbatim on
-    its single-frame fast path. ``ChunkInterrupted.__init__`` must
-    therefore defensively ``.copy()`` so an in-place mutation of the
-    underlying chunk frame (e.g. user diagnostic code adding a
-    column on the live view) doesn't leak through the snapshot.
-    Companion to ``test_partial_frame_snapshot_stable_across_resume``,
-    which uses ≥2 completed chunks and so goes through
-    ``pd.concat`` (which already produces a fresh frame)."""
+def test_partial_frame_is_live_view_when_single_chunk():
+    """``partial_frame`` is a live view delegating to ``call.partial_frame``.
+    When a single chunk completes, ``partial_frame`` still reflects the
+    current state of ``call.partial_frame`` (which recomputes on each
+    access). This verifies the live-view semantics even on the single-
+    chunk fast path."""
     state = {"i": 0, "blow_up": True}
 
     async def fetch(args):
@@ -1020,12 +1015,11 @@ async def fetch(args):
     err = excinfo.value
     assert err.completed_chunks == 1
 
-    snapshot_cols = list(err.partial_frame.columns)
-    # Mutate the underlying chunk in place — the snapshot must NOT
-    # reflect the mutation.
-    err.call._chunks[0][0]["extra"] = 0
-    assert list(err.partial_frame.columns) == snapshot_cols
-    assert "extra" not in err.partial_frame.columns
+    # The live view delegates to call.partial_frame; verify it's
+    # accessible and reflects the call's current state.
+    assert not err.partial_frame.empty
+    assert list(err.partial_frame.columns) == ["sites"]
+    assert err.partial_frame.equals(err.call.partial_frame)
 
 
 def test_combine_chunk_responses_returns_independent_headers():

From 85346155fbcc627c2a7e38ed140644376d07b162 Mon Sep 17 00:00:00 2001
From: thodson-usgs <thodson@usgs.gov>
Date: Thu, 23 Jul 2026 09:59:23 -0500
Subject: [PATCH 5/7] refactor: unify error classification into shared
 _classify_transient helper

Extract the per-exception type matching from _classify_chunk_error into
a new _classify_transient(exc) helper that classifies a SINGLE exception
(no chain walk) into (interrupted_class, retry_after) | None.

_classify_chunk_error now calls _classify_transient on each step of its
__cause__ walk. _retryable continues to inspect only the top-level
exception (preserving the intentional divergence: initial-page raw
transients are cheap to retry; wrapped mid-pagination failures escalate
to a resumable ChunkInterrupted instead).

When a new transient error type needs recognition, it now goes in ONE
place (_classify_transient) and both the retry and escalation paths pick
it up automatically.
---
 dataretrieval/ogc/retry.py | 45 +++++++++++++++++++++++++++++++++-----
 1 file changed, 39 insertions(+), 6 deletions(-)

diff --git a/dataretrieval/ogc/retry.py b/dataretrieval/ogc/retry.py
index b81ce60e..affb4522 100644
--- a/dataretrieval/ogc/retry.py
+++ b/dataretrieval/ogc/retry.py
@@ -190,6 +190,38 @@ def backoff(self, attempt: int, retry_after: float | None) -> float:
 _NO_RETRY = RetryPolicy(max_retries=0)
 
 
+def _classify_transient(
+    exc: BaseException,
+) -> tuple[type[ChunkInterrupted], float | None] | None:
+    """
+    Classify a SINGLE exception as a known transient (resumable) failure.
+
+    Does NOT walk the ``__cause__`` chain — inspects only the exception
+    passed in. This is the shared taxonomy that both
+    :func:`_classify_chunk_error` (chain-walking) and :func:`_retryable`
+    (top-level only) delegate to, so a new error type need only be added
+    in one place.
+
+    Parameters
+    ----------
+    exc : BaseException
+        A single exception to classify.
+
+    Returns
+    -------
+    tuple[type[ChunkInterrupted], float or None] or None
+        ``(interrupted_class, retry_after)`` for a recognized transient
+        failure; ``None`` otherwise.
+    """
+    if isinstance(exc, RateLimited):
+        return QuotaExhausted, exc.retry_after
+    if isinstance(exc, ServiceUnavailable):
+        return ServiceInterrupted, exc.retry_after
+    if isinstance(exc, (httpx.HTTPError, httpx.InvalidURL)):
+        return ServiceInterrupted, None
+    return None
+
+
 def _classify_chunk_error(
     exc: BaseException,
 ) -> tuple[type[ChunkInterrupted], float | None] | None:
@@ -231,12 +263,9 @@ def _classify_chunk_error(
     """
     cur: BaseException | None = exc
     while cur is not None:
-        if isinstance(cur, RateLimited):
-            return QuotaExhausted, cur.retry_after
-        if isinstance(cur, ServiceUnavailable):
-            return ServiceInterrupted, cur.retry_after
-        if isinstance(cur, (httpx.HTTPError, httpx.InvalidURL)):
-            return ServiceInterrupted, None
+        result = _classify_transient(cur)
+        if result is not None:
+            return result
         cur = cur.__cause__
     return None
 
@@ -262,6 +291,10 @@ def _retryable(exc: BaseException) -> tuple[bool, float | None]:
         ``(retryable, retry_after)`` — the server ``Retry-After`` hint
         (seconds) when the transient carried one, else ``None``.
     """
+    # Only initial-page raw transients (TransientError, TransportError) are
+    # retryable — not wrapped mid-pagination DataRetrievalError. We use
+    # _classify_transient for the typed status errors and handle bare
+    # TransportError (which isn't a TransientError) separately.
     if isinstance(exc, TransientError):
         return True, exc.retry_after
     if isinstance(exc, httpx.TransportError):

From fda917003b90b6a66e13f4d6c331b6415a18fd70 Mon Sep 17 00:00:00 2001
From: thodson-usgs <thodson@usgs.gov>
Date: Thu, 23 Jul 2026 10:39:04 -0500
Subject: [PATCH 6/7] refactor(ogc): adopt combining module directly; drop dead
 re-export shims
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Follow-up cleanup on the module-boundary refactor so the split lands in
the import graph, not just the file layout.

- Import the recombination helpers straight from ogc.combining at every
  call site (chunking, engine, wateruse, tests) and delete planning.py's
  backwards-compat re-export block + __all__. The new module previously
  had zero direct importers; everything routed through planning.

- Drop the dead TYPE_CHECKING re-export of _Fetch/_Finalize/
  _passthrough_result from interruptions.py (no importers) and repoint the
  lone Sphinx xref in shaping.py to ogc.chunking._Finalize, their real home.

- Remove ChunkInterrupted.__setstate__: __getstate__ already writes the
  _pickled_* snapshot keys and the base DataRetrievalError.__setstate__
  restores them via __dict__.update, so the override was redundant (the
  property getattr fallbacks cover the hand-constructed case).

- Correct _classify_transient's docstring and _retryable's comment, which
  claimed a delegation that does not — and deliberately should not — exist
  (the two use intentionally different type sets).

No behavior change; mypy --strict and the affected test suites pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 dataretrieval/ogc/chunking.py      | 10 ++++-----
 dataretrieval/ogc/engine.py        |  2 +-
 dataretrieval/ogc/interruptions.py | 32 ++++++---------------------
 dataretrieval/ogc/planning.py      | 35 ++++--------------------------
 dataretrieval/ogc/retry.py         | 18 +++++++++------
 dataretrieval/ogc/shaping.py       |  2 +-
 dataretrieval/wateruse.py          |  2 +-
 tests/waterdata_chunking_test.py   |  8 ++++---
 tests/wateruse_test.py             |  2 +-
 9 files changed, 36 insertions(+), 75 deletions(-)

diff --git a/dataretrieval/ogc/chunking.py b/dataretrieval/ogc/chunking.py
index 1c10ca09..78efcfea 100644
--- a/dataretrieval/ogc/chunking.py
+++ b/dataretrieval/ogc/chunking.py
@@ -86,14 +86,14 @@
 from dataretrieval.utils import HTTPX_DEFAULTS, Ambient, _require_positive_int
 
 from . import progress as _progress
-from .interruptions import (
-    ChunkInterrupted,
-)
-from .planning import (
-    ChunkPlan,
+from .combining import (
     _combine_chunk_frames,
     _combine_chunk_responses,
 )
+from .interruptions import (
+    ChunkInterrupted,
+)
+from .planning import ChunkPlan
 from .retry import (
     _NO_RETRY,
     RetryPolicy,
diff --git a/dataretrieval/ogc/engine.py b/dataretrieval/ogc/engine.py
index 668fa7b7..3d22ac35 100644
--- a/dataretrieval/ogc/engine.py
+++ b/dataretrieval/ogc/engine.py
@@ -47,9 +47,9 @@
 from dataretrieval.ogc import chunking
 from dataretrieval.ogc import progress as _progress
 from dataretrieval.ogc.chunking import get_active_client
+from dataretrieval.ogc.combining import _QUOTA_HEADER, _merge_response, _safe_elapsed
 from dataretrieval.ogc.dates import _DATE_RANGE_PARAMS, _format_api_dates
 from dataretrieval.ogc.errors import _paginated_failure_message, _raise_for_non_200
-from dataretrieval.ogc.planning import _QUOTA_HEADER, _merge_response, _safe_elapsed
 from dataretrieval.ogc.shaping import GEOPANDAS, _finalize_ogc, _get_resp_data
 from dataretrieval.utils import (
     HTTPX_DEFAULTS,
diff --git a/dataretrieval/ogc/interruptions.py b/dataretrieval/ogc/interruptions.py
index 6139a90a..a40f1946 100644
--- a/dataretrieval/ogc/interruptions.py
+++ b/dataretrieval/ogc/interruptions.py
@@ -22,16 +22,6 @@
     from dataretrieval.ogc.chunking import ChunkedCall
 
 
-# These type aliases define the ChunkedCall contract and live in chunking.py;
-# re-exported here for backwards compatibility.
-if TYPE_CHECKING:
-    from dataretrieval.ogc.chunking import _Fetch as _Fetch  # noqa: F401
-    from dataretrieval.ogc.chunking import _Finalize as _Finalize  # noqa: F401
-    from dataretrieval.ogc.chunking import (  # noqa: F401
-        _passthrough_result as _passthrough_result,
-    )
-
-
 class ChunkInterrupted(DataRetrievalError):
     """
     Base class for mid-stream chunk failures whose completed work is
@@ -134,8 +124,8 @@ def partial_frame(self) -> pd.DataFrame:
         """Live view of work completed so far — delegates to ``call.partial_frame``.
 
         Returns an empty ``DataFrame`` when ``call`` is ``None`` (degraded /
-        unpickled state), unless a pickled snapshot was restored via
-        ``__setstate__``.
+        unpickled state), unless a pickled snapshot was restored during
+        unpickling.
         """
         if self.call is not None:
             return self.call.partial_frame
@@ -164,24 +154,16 @@ def __getstate__(self) -> dict[str, Any]:
         # resume was never possible anyway).
         #
         # Since ``partial_frame`` and ``partial_response`` are now live
-        # properties (delegating to ``self.call``), we must snapshot them
-        # here so the pickled state includes the data that was available
-        # at serialization time.
+        # properties (delegating to ``self.call``), snapshot them here into
+        # private ``_pickled_*`` keys. The base ``__setstate__`` restores
+        # those keys via ``self.__dict__.update``, and the properties fall
+        # back to them (through ``getattr``) once ``call`` is ``None`` — so
+        # no ``__setstate__`` override is needed.
         state = {**super().__getstate__(), "call": None}
         state["_pickled_partial_frame"] = self.partial_frame.copy()
         state["_pickled_partial_response"] = self.partial_response
         return state
 
-    def __setstate__(self, state: dict[str, Any] | None) -> None:
-        state = state or {}
-        # Restore the pickled partial snapshots into private backing attrs
-        # that the properties will fall back to when call is None.
-        self._pickled_partial_frame = state.pop(
-            "_pickled_partial_frame", pd.DataFrame()
-        )
-        self._pickled_partial_response = state.pop("_pickled_partial_response", None)
-        super().__setstate__(state)
-
 
 class QuotaExhausted(ChunkInterrupted):
     """
diff --git a/dataretrieval/ogc/planning.py b/dataretrieval/ogc/planning.py
index 936d0c43..45ae343e 100644
--- a/dataretrieval/ogc/planning.py
+++ b/dataretrieval/ogc/planning.py
@@ -8,10 +8,10 @@
 :mod:`dataretrieval.ogc.retry` (retry policy), which import the plan and
 drive it.
 
-Result recombination utilities (:func:`_combine_chunk_frames`,
-:func:`_combine_chunk_responses`, etc.) live in the sibling
-:mod:`dataretrieval.ogc.combining` module; they are re-exported here for
-backwards compatibility.
+Result recombination — reassembling the per-chunk frames and responses
+back into one result (:func:`_combine_chunk_frames`,
+:func:`_combine_chunk_responses`, etc.) — lives in the sibling
+:mod:`dataretrieval.ogc.combining` module, which callers import directly.
 """
 
 from __future__ import annotations
@@ -594,30 +594,3 @@ def iter_sub_args(self) -> Iterator[dict[str, Any]]:
             for axis, chunk in zip(self.axes, combo, strict=False):
                 sub_args[axis.arg_key] = axis.render(chunk)
             yield sub_args
-
-
-# ---------------------------------------------------------------------------
-# Backwards-compatibility re-exports: these utilities moved to
-# ``dataretrieval.ogc.combining`` but existing importers (chunking.py,
-# engine.py, wateruse.py, tests) may reference them via ``planning``.
-# ---------------------------------------------------------------------------
-from dataretrieval.ogc.combining import (  # noqa: E402, F401
-    _QUOTA_HEADER,
-    _combine_chunk_frames,
-    _combine_chunk_responses,
-    _lowest_remaining,
-    _merge_response,
-    _safe_elapsed,
-    _set_response_url,
-)
-
-__all__ = [
-    "ChunkPlan",
-    "_QUOTA_HEADER",
-    "_combine_chunk_frames",
-    "_combine_chunk_responses",
-    "_lowest_remaining",
-    "_merge_response",
-    "_safe_elapsed",
-    "_set_response_url",
-]
diff --git a/dataretrieval/ogc/retry.py b/dataretrieval/ogc/retry.py
index affb4522..8c3fab17 100644
--- a/dataretrieval/ogc/retry.py
+++ b/dataretrieval/ogc/retry.py
@@ -197,10 +197,11 @@ def _classify_transient(
     Classify a SINGLE exception as a known transient (resumable) failure.
 
     Does NOT walk the ``__cause__`` chain — inspects only the exception
-    passed in. This is the shared taxonomy that both
-    :func:`_classify_chunk_error` (chain-walking) and :func:`_retryable`
-    (top-level only) delegate to, so a new error type need only be added
-    in one place.
+    passed in. Factored out of :func:`_classify_chunk_error` (which walks
+    the ``__cause__`` chain and calls this on each link) so the transient
+    taxonomy lives in one place. :func:`_retryable` deliberately does *not*
+    delegate here — it uses a narrower set (``httpx.InvalidURL`` and a bare
+    ``httpx.HTTPError`` are resumable but not worth an automatic retry).
 
     Parameters
     ----------
@@ -292,9 +293,12 @@ def _retryable(exc: BaseException) -> tuple[bool, float | None]:
         (seconds) when the transient carried one, else ``None``.
     """
     # Only initial-page raw transients (TransientError, TransportError) are
-    # retryable — not wrapped mid-pagination DataRetrievalError. We use
-    # _classify_transient for the typed status errors and handle bare
-    # TransportError (which isn't a TransientError) separately.
+    # retryable — not a wrapped mid-pagination DataRetrievalError. This is a
+    # deliberately narrower taxonomy than _classify_transient's: RateLimited
+    # and ServiceUnavailable are caught via their TransientError base, but
+    # httpx.InvalidURL and bare httpx.HTTPError are intentionally excluded
+    # (a too-long cursor or non-transport HTTP error won't fix on retry), so
+    # this stays a separate check rather than delegating to _classify_transient.
     if isinstance(exc, TransientError):
         return True, exc.retry_after
     if isinstance(exc, httpx.TransportError):
diff --git a/dataretrieval/ogc/shaping.py b/dataretrieval/ogc/shaping.py
index 1a757cda..315be869 100644
--- a/dataretrieval/ogc/shaping.py
+++ b/dataretrieval/ogc/shaping.py
@@ -373,7 +373,7 @@ def _finalize_ogc(
     as :class:`~dataretrieval.utils.BaseMetadata`.
 
     Injected into the chunker as its ``finalize`` hook (see
-    :data:`~dataretrieval.ogc.interruptions._Finalize`) so the
+    :data:`~dataretrieval.ogc.chunking._Finalize`) so the
     un-interrupted return *and* a resumed ``ChunkInterrupted.call.resume()``
     produce the same post-processed ``(DataFrame, BaseMetadata)`` shape, not
     the chunker's raw frame and bare ``httpx.Response``.
diff --git a/dataretrieval/wateruse.py b/dataretrieval/wateruse.py
index 635a97fb..9d223f7c 100644
--- a/dataretrieval/wateruse.py
+++ b/dataretrieval/wateruse.py
@@ -53,8 +53,8 @@
 
 from dataretrieval.codes.states import to_state
 from dataretrieval.exceptions import DataRetrievalError
+from dataretrieval.ogc.combining import _combine_chunk_frames, _combine_chunk_responses
 from dataretrieval.ogc.engine import _paginate, _run_sync
-from dataretrieval.ogc.planning import _combine_chunk_frames, _combine_chunk_responses
 from dataretrieval.utils import (
     HTTPX_DEFAULTS,
     BaseMetadata,
diff --git a/tests/waterdata_chunking_test.py b/tests/waterdata_chunking_test.py
index bca626fe..868566e8 100644
--- a/tests/waterdata_chunking_test.py
+++ b/tests/waterdata_chunking_test.py
@@ -47,6 +47,11 @@
     multi_value_chunked,
     parallel_chunks,
 )
+from dataretrieval.ogc.combining import (
+    _QUOTA_HEADER,
+    _combine_chunk_frames,
+    _combine_chunk_responses,
+)
 from dataretrieval.ogc.interruptions import (
     ChunkInterrupted,
     QuotaExhausted,
@@ -56,10 +61,7 @@
     _LIST_SEP,
     _NEVER_CHUNK,
     _OR_SEP,
-    _QUOTA_HEADER,
     ChunkPlan,
-    _combine_chunk_frames,
-    _combine_chunk_responses,
     _extract_axes,
     _request_bytes,
     _safe_request_bytes,
diff --git a/tests/wateruse_test.py b/tests/wateruse_test.py
index 031aad74..00a843c8 100644
--- a/tests/wateruse_test.py
+++ b/tests/wateruse_test.py
@@ -306,7 +306,7 @@ def test_fan_out_surfaces_final_rate_limit_header(httpx_mock):
     assert md.header["x-ratelimit-remaining"] == "850"
 
 
-# (response aggregation now reuses ogc.planning._combine_chunk_responses; the
+# (response aggregation now reuses ogc.combining._combine_chunk_responses; the
 # integration test above pins the rate-limit-header behavior end-to-end.)
 
 

From 8cfac54dd59e6e93de634a9c851ca68f3b47c884 Mon Sep 17 00:00:00 2001
From: thodson-usgs <thodson@usgs.gov>
Date: Thu, 23 Jul 2026 10:55:36 -0500
Subject: [PATCH 7/7] refactor(ogc): keep snapshot semantics for
 ChunkInterrupted partial state
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Revert the live-view change (item 3 of #346): restore exc.partial_frame /
exc.partial_response as raise-time snapshots rather than properties that
delegate to exc.call.

Snapshot is the better contract for an exception — a record of the failure
moment, not a live handle:

- In a resume loop each interruption stays a faithful record of what it saw.
  Live-view aliased every exception to the one shared call, so after a second
  failure exc1.partial_frame and exc2.partial_frame returned the same (later)
  state.
- After a successful resume, live-view's .partial_frame returned the COMPLETE
  result — a field named "partial" that no longer was.
- Exception payloads shouldn't mutate after you catch them (cf.
  CalledProcessError.output).

Removing the delegation also drops the properties and the _pickled_* /
__getstate__ snapshot machinery it required: partial_frame/partial_response
are plain instance attributes again, pickled by the base __getstate__
(net -21 lines in interruptions.py).

No behavior change versus released main — this PR is unmerged, so live-view
never shipped. #346 is now a true no-behavioral-change refactor.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 dataretrieval/ogc/interruptions.py | 73 +++++++++++-------------------
 tests/waterdata_chunking_test.py   | 50 +++++++++++---------
 2 files changed, 54 insertions(+), 69 deletions(-)

diff --git a/dataretrieval/ogc/interruptions.py b/dataretrieval/ogc/interruptions.py
index a40f1946..8cb5723c 100644
--- a/dataretrieval/ogc/interruptions.py
+++ b/dataretrieval/ogc/interruptions.py
@@ -55,14 +55,14 @@ class ChunkInterrupted(DataRetrievalError):
     total_chunks : int
         Total sub-requests in the plan.
     partial_frame : pandas.DataFrame
-        Live view of work completed so far — delegates to
-        ``call.partial_frame``. Advances as ``call.resume()`` makes
-        further progress. Returns an empty ``DataFrame`` when
-        ``call`` is ``None`` (degraded/unpickled state).
+        Combined frame of work completed by the moment this exception
+        was raised. Snapshot at raise time — does NOT advance on a
+        later ``call.resume()`` (use ``exc.call.partial_frame`` for
+        the live view).
     partial_response : httpx.Response or None
-        Live view of the raw aggregate response — delegates to
-        ``call.partial_response``. ``None`` when ``call`` is ``None``
-        or nothing has completed yet. (Raw, not finalized — use
+        Raw aggregate response covering the completed sub-requests at
+        raise time; ``None`` if nothing had completed yet. Same snapshot
+        semantics as ``partial_frame``. (Raw, not finalized — use
         ``exc.call.resume()`` for the finalized ``(df, metadata)`` result.)
 
     Examples
@@ -118,51 +118,30 @@ def __init__(
         self.total_chunks = total_chunks
         self.call = call
         self.retry_after = retry_after
-
-    @property
-    def partial_frame(self) -> pd.DataFrame:
-        """Live view of work completed so far — delegates to ``call.partial_frame``.
-
-        Returns an empty ``DataFrame`` when ``call`` is ``None`` (degraded /
-        unpickled state), unless a pickled snapshot was restored during
-        unpickling.
-        """
-        if self.call is not None:
-            return self.call.partial_frame
-        # Degraded / unpickled: return the snapshot captured at pickle time,
-        # or an empty frame if no snapshot exists (hand-constructed exc).
-        return getattr(self, "_pickled_partial_frame", pd.DataFrame())
-
-    @property
-    def partial_response(self) -> httpx.Response | None:
-        """Live view of the raw aggregate response.
-
-        Delegates to ``call.partial_response``.
-        Returns ``None`` when ``call`` is ``None`` (degraded / unpickled state)
-        and no pickled snapshot exists.
-        """
-        if self.call is not None:
-            return self.call.partial_response
-        return getattr(self, "_pickled_partial_response", None)
+        # Snapshot partial state at raise time so the exception stays a stable
+        # record of the failure moment: ``exc.partial_frame`` /
+        # ``.partial_response`` do NOT advance on a later ``call.resume()``
+        # (that live view is on ``call.partial_frame`` / ``.partial_response``).
+        # This keeps each interruption in a resume loop a faithful record of
+        # what it saw, rather than every exception aliasing the shared call's
+        # advancing state. ``.copy()`` guards the single-chunk fast path, where
+        # the combined frame may be returned verbatim.
+        if call is None:
+            self.partial_frame: pd.DataFrame = pd.DataFrame()
+            self.partial_response: httpx.Response | None = None
+        else:
+            self.partial_frame = call.partial_frame.copy()
+            self.partial_response = call.partial_response
 
     def __getstate__(self) -> dict[str, Any]:
         # Drop the live ChunkedCall before pickling: its ``.fetch`` is an
         # undecorated module function pickle can't reference by name, so the
         # interruption can't cross a process boundary with ``.call`` attached.
-        # The degraded ``call=None`` form keeps the counts, retry hint, and
-        # partial frame / response; only ``.resume()`` is lost (cross-process
-        # resume was never possible anyway).
-        #
-        # Since ``partial_frame`` and ``partial_response`` are now live
-        # properties (delegating to ``self.call``), snapshot them here into
-        # private ``_pickled_*`` keys. The base ``__setstate__`` restores
-        # those keys via ``self.__dict__.update``, and the properties fall
-        # back to them (through ``getattr``) once ``call`` is ``None`` — so
-        # no ``__setstate__`` override is needed.
-        state = {**super().__getstate__(), "call": None}
-        state["_pickled_partial_frame"] = self.partial_frame.copy()
-        state["_pickled_partial_response"] = self.partial_response
-        return state
+        # The degraded ``call=None`` form keeps the counts, retry hint, and the
+        # snapshotted partial frame / response — plain instance attributes the
+        # base ``__getstate__`` already pickles; only ``.resume()`` is lost
+        # (cross-process resume was never possible anyway).
+        return {**super().__getstate__(), "call": None}
 
 
 class QuotaExhausted(ChunkInterrupted):
diff --git a/tests/waterdata_chunking_test.py b/tests/waterdata_chunking_test.py
index 868566e8..12278f29 100644
--- a/tests/waterdata_chunking_test.py
+++ b/tests/waterdata_chunking_test.py
@@ -956,11 +956,13 @@ async def fetch(args):
     assert err.partial_response is not None
 
 
-def test_partial_frame_is_live_view_across_resume():
-    """``exc.partial_frame`` is now a live view delegating to
-    ``exc.call.partial_frame``. Calling ``exc.call.resume()`` advances the
-    underlying ``ChunkedCall`` and ``exc.partial_frame`` reflects that — it
-    is no longer a frozen snapshot."""
+def test_partial_frame_snapshot_stable_across_resume():
+    """``exc.partial_frame`` / ``exc.partial_response`` snapshot the
+    state at raise time. Calling ``exc.call.resume()`` advances the
+    underlying ``ChunkedCall`` but must NOT mutate the snapshot on
+    the exception — otherwise a diagnostic that reads
+    ``exc.partial_frame`` after a resume sees post-resume state under
+    a name that promises pre-resume state."""
     state = {"i": 0, "blow_up": True}
 
     async def fetch(args):
@@ -977,24 +979,27 @@ async def fetch(args):
     with pytest.raises(QuotaExhausted) as excinfo:
         decorated({"sites": ["S1" * 10, "S2" * 10, "S3" * 10, "S4" * 10, "S5" * 10]})
     err = excinfo.value
-    pre_resume_rows = len(err.partial_frame)
-    assert pre_resume_rows > 0  # two chunks worth of data captured
+    snapshot_rows = len(err.partial_frame)
+    assert snapshot_rows > 0  # two chunks worth of data captured
 
-    # Resume; the live view on .call grows — and so does exc.partial_frame.
+    # Resume; the live view on .call grows.
     state["blow_up"] = False
     err.call.resume()
-    assert len(err.call.partial_frame) > pre_resume_rows
+    assert len(err.call.partial_frame) > snapshot_rows
 
-    # The exception's partial_frame IS the live view — it advances too.
-    assert len(err.partial_frame) == len(err.call.partial_frame)
+    # The exception's snapshot must NOT advance.
+    assert len(err.partial_frame) == snapshot_rows
 
 
-def test_partial_frame_is_live_view_when_single_chunk():
-    """``partial_frame`` is a live view delegating to ``call.partial_frame``.
-    When a single chunk completes, ``partial_frame`` still reflects the
-    current state of ``call.partial_frame`` (which recomputes on each
-    access). This verifies the live-view semantics even on the single-
-    chunk fast path."""
+def test_partial_frame_snapshot_is_a_copy_when_single_chunk():
+    """``_combine_chunk_frames`` returns ``non_empty[0]`` verbatim on
+    its single-frame fast path. ``ChunkInterrupted.__init__`` must
+    therefore defensively ``.copy()`` so an in-place mutation of the
+    underlying chunk frame (e.g. user diagnostic code adding a
+    column on the live view) doesn't leak through the snapshot.
+    Companion to ``test_partial_frame_snapshot_stable_across_resume``,
+    which uses ≥2 completed chunks and so goes through
+    ``pd.concat`` (which already produces a fresh frame)."""
     state = {"i": 0, "blow_up": True}
 
     async def fetch(args):
@@ -1017,11 +1022,12 @@ async def fetch(args):
     err = excinfo.value
     assert err.completed_chunks == 1
 
-    # The live view delegates to call.partial_frame; verify it's
-    # accessible and reflects the call's current state.
-    assert not err.partial_frame.empty
-    assert list(err.partial_frame.columns) == ["sites"]
-    assert err.partial_frame.equals(err.call.partial_frame)
+    snapshot_cols = list(err.partial_frame.columns)
+    # Mutate the underlying chunk in place — the snapshot must NOT
+    # reflect the mutation.
+    err.call._chunks[0][0]["extra"] = 0
+    assert list(err.partial_frame.columns) == snapshot_cols
+    assert "extra" not in err.partial_frame.columns
 
 
 def test_combine_chunk_responses_returns_independent_headers():