From d2754226e6fabdf8ee8b46c09f8286f7bbd9dd11 Mon Sep 17 00:00:00 2001 From: jross Date: Fri, 10 Jul 2026 15:33:54 -0600 Subject: [PATCH] Harden Parquet IO against mixed-type parameter_value columns WQP non-detects arrive as ResultMeasureValue='ND'. In _apply_unit_conversion numeric results convert to float while float('ND') raises and the raw string is kept ("Keeping ... for time series data"). The result is a parameter_value column mixing floats and qualitative strings, which crashed the Parquet handoff: pyarrow.lib.ArrowInvalid: Could not convert 'ND' with type str: tried to convert to double ... for column parameter_value pyarrow infers one logical type per column and can't hold both. Rather than drop the qualitative markers (they are intentional time-series data), stringify any object column that mixes numeric values with non-numeric strings before writing. - backend/persisters/geodataframe.py: new _frame_to_parquet_bytes helper scans columns and casts mixed numeric/str columns to str, preserving nulls. Both dicts_to_parquet_bytes (records/sites) and timeseries_to_parquet_bytes (observations) route through it, since the records path can carry the same mix. Qualitative markers ('ND', '<0.5') now survive the round-trip; all-numeric batches keep their float64 dtype. Every downstream consumer already float()-coerces and skips non-numeric values (backend.trend_stats.daily_series), so string storage in ND-bearing batches is safe. Note: parameter_value dtype now varies by batch (float64 when clean, str when any qualitative value present). Verified against pyarrow: the mixed column round-trips with markers intact and all-numeric batches stay float64; local persister suite green. Co-Authored-By: Claude Opus 4.8 --- backend/persisters/geodataframe.py | 35 +++++++++++++++++++++++++----- 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/backend/persisters/geodataframe.py b/backend/persisters/geodataframe.py index eb220cc..c39f523 100644 --- a/backend/persisters/geodataframe.py +++ b/backend/persisters/geodataframe.py @@ -196,13 +196,38 @@ def _clean_nans(records: list[dict]) -> list[dict]: return [{k: (None if pd.isna(v) else v) for k, v in row.items()} for row in records] +def _frame_to_parquet_bytes(df: pd.DataFrame) -> bytes: + """Serialize *df* to Parquet, stringifying any object column that mixes + numeric values with non-numeric strings. + + pyarrow infers one logical type per column and raises ArrowInvalid if an + object column holds both — e.g. a WQP analyte's parameter_value column + carrying floats *and* the non-detect marker 'ND' (numeric results convert to + float in _apply_unit_conversion, qualitative ones stay strings). Casting such + a column to str lets the qualitative markers survive the round-trip; every + downstream consumer already float()-coerces and skips non-numeric values + (see backend.trend_stats.daily_series).""" + for col in df.columns: + vals = [v for v in df[col] if not _is_null(v)] + has_num = any(isinstance(v, (int, float)) and not isinstance(v, bool) for v in vals) + has_str = any(isinstance(v, str) for v in vals) + if has_num and has_str: + df[col] = [None if _is_null(v) else str(v) for v in df[col]] + buf = io.BytesIO() + df.to_parquet(buf, index=False) + return buf.getvalue() + + +def _is_null(v) -> bool: + """None or a float NaN. Scalar-only — payload columns are flat scalars.""" + return v is None or (isinstance(v, float) and v != v) + + def dicts_to_parquet_bytes(dicts: list[dict]) -> bytes: """Serialize a flat list of payload dicts (records or sites) to Parquet. object dtype keeps exact types; a uniform column set is fine — the record classes tolerate extra null-valued keys on rebuild.""" - buf = io.BytesIO() - pd.DataFrame(dicts, dtype=object).to_parquet(buf, index=False) - return buf.getvalue() + return _frame_to_parquet_bytes(pd.DataFrame(dicts, dtype=object)) def parquet_bytes_to_dicts(data: bytes) -> list[dict]: @@ -218,9 +243,7 @@ def timeseries_to_parquet_bytes(timeseries: list[list[dict]]) -> bytes: for site_idx, site_ts in enumerate(timeseries): for obs in site_ts: rows.append({**obs, _SITE_IDX: site_idx}) - buf = io.BytesIO() - pd.DataFrame(rows, dtype=object).to_parquet(buf, index=False) - return buf.getvalue() + return _frame_to_parquet_bytes(pd.DataFrame(rows, dtype=object)) def parquet_bytes_to_timeseries(data: bytes) -> list[list[dict]]: