Fix Parquet IO-manager crash on non-scalar payload keys

The Dagster PayloadParquetIOManager crashed serializing source outputs:

  pyarrow.lib.ArrowInvalid: Could not convert <SiteRecord ...> ... for column
  "location"

A record's _payload can carry incidental non-scalar keys beyond its declared
schema — the FROST connectors leave the raw {location: SiteRecord, thing,
datastream, observation} in each observation's payload (ParameterTransformer
does rec.update(record)). The old pickle IO manager tolerated these; Parquet
can't. The combine only ever reads a record's declared keys anyway.

- orchestration/assets/products.py: the source asset now ships _schema_dict(r)
  = {k: r._payload.get(k) for k in r.keys} for records/sites/timeseries — only
  the declared scalar fields, dropping the junk.
- usgs/source.py: the USGS qualifier field is a list in the API; join it to a
  scalar string so it stays consistent with the other sources and serializes
  cleanly through both Parquet and the GeoJSON dump (it otherwise round-tripped
  as a numpy array).

Reproduced and fixed locally end-to-end: ebid (SiteRecord junk) and nwis
(qualifier list) both now round-trip records+sites+timeseries through the Parquet
handoff. Full suite green (233 passed); products.py compiles.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: jirhiker

backend/connectors/usgs/source.py

@@ -236,9 +236,11 @@ def _standardize_record(self, record: dict) -> dict:
             "datetime_measured": props["time"],
             "source_parameter_units": props["unit_of_measure"],
             # provisional vs approved, and provider qualifier flags — carried
-            # through to the timeseries product instead of dropped
+            # through to the timeseries product instead of dropped. qualifier is
+            # a list in the API; join to a scalar string so it stays consistent
+            # with the other sources and serializes cleanly (Parquet/GeoJSON).
             "approval_status": props.get("approval_status"),
-            "qualifier": props.get("qualifier"),
+            "qualifier": ", ".join(q) if isinstance((q := props.get("qualifier")), list) else q,
         }
 
     def _extract_site_records(self, records, site_record):

orchestration/assets/products.py

@@ -139,6 +139,13 @@
 SourceSpec = namedtuple("SourceSpec", "parameter scope source_key group")
 
 
+def _schema_dict(record) -> dict:
+    """A record's declared-schema fields only, as a plain scalar dict for the
+    Parquet IO-manager handoff. Drops incidental non-scalar payload keys that
+    would fail Arrow serialization (see the source asset)."""
+    return {k: record._payload.get(k) for k in record.keys}
+
+
 def _product_params(product: dict) -> list[str]:
     """The DIE parameter(s) a product unifies. Single-parameter products yield
     one; multi-analyte products yield their analyte list."""
@@ -298,11 +305,17 @@ def _source_asset(
                 summary_persister, timeseries_persister = unify_source_both(
                     config, spec.source_key
                 )
-                # Ship plain dicts across the IO manager; rebuild in combine.
-                records.extend(r._payload for r in summary_persister.records)
-                sites.extend(s._payload for s in timeseries_persister.sites)
+                # Ship plain scalar dicts across the IO manager; rebuild in
+                # combine. Emit only each record's declared schema keys — a
+                # record's _payload can also carry incidental non-scalar junk
+                # (e.g. the FROST connectors leave a SiteRecord in "location" and
+                # nested thing/datastream/observation dicts), which the pickle IO
+                # manager tolerated but the Parquet one cannot serialize. The
+                # combine only ever reads the declared keys anyway.
+                records.extend(_schema_dict(r) for r in summary_persister.records)
+                sites.extend(_schema_dict(s) for s in timeseries_persister.sites)
                 timeseries.extend(
-                    [o._payload for o in site_ts]
+                    [_schema_dict(o) for o in site_ts]
                     for site_ts in timeseries_persister.timeseries
                 )
         except Exception:

tests/test_usgs_metadata.py

@@ -19,9 +19,10 @@ def _raw_feature(approval="Provisional", qualifier=None):
 
 def test_standardize_record_keeps_approval_and_qualifier():
     s = NWISWaterLevelSource()
-    rec = s._standardize_record(_raw_feature("Approved", ["Static"]))
+    # API qualifier is a list; joined to a scalar string
+    rec = s._standardize_record(_raw_feature("Approved", ["Static", "Ice"]))
     assert rec["approval_status"] == "Approved"
-    assert rec["qualifier"] == ["Static"]
+    assert rec["qualifier"] == "Static, Ice"
 
 
 def test_standardize_record_missing_metadata_is_none():
@@ -46,7 +47,7 @@ def test_extract_parameter_record_populates_metadata():
     std = s._standardize_record(_raw_feature("Provisional", ["Ice"]))
     out = s._extract_parameter_record(std)
     assert out["approval_status"] == "Provisional"
-    assert out["qualifier"] == ["Ice"]
+    assert out["qualifier"] == "Ice"
 
 
 def test_parameter_record_schema_includes_metadata_keys():