From bd26e7ab43159b948ff4328174b4571918f4f992 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Thu, 30 Apr 2026 21:52:40 -0700 Subject: [PATCH] Fix incremental snapshot merge: coerce new row types to match existing schema event_volume (and potentially other columns) can arrive as int64 from the pipeline while the stored snapshot uses double, causing pa.concat_tables to fail with "incompatible types" every run and silently fall back to a full rebuild. _coerce_schema() casts new rows to the existing snapshot's types before concatenation so incremental works without a full rebuild. Co-Authored-By: Claude Sonnet 4.6 --- libs/export/canonical_snapshots.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/libs/export/canonical_snapshots.py b/libs/export/canonical_snapshots.py index d3a9358..6a9b249 100644 --- a/libs/export/canonical_snapshots.py +++ b/libs/export/canonical_snapshots.py @@ -29,6 +29,23 @@ from libs.export.snapshot_export import export_dataset_snapshot logger = get_logger(__name__) + +def _coerce_schema(table: pa.Table, target_schema: pa.Schema) -> pa.Table: + """Cast columns in table to match target_schema types where they differ numerically.""" + for i in range(len(target_schema)): + field = target_schema.field(i) + if field.name not in table.schema.names: + continue + col_idx = table.schema.get_field_index(field.name) + existing_type = table.schema.field(col_idx).type + if existing_type == field.type: + continue + try: + table = table.set_column(col_idx, field, table.column(field.name).cast(field.type, safe=False)) + except Exception: + pass # leave as-is; concat_tables will promote or raise with clear error + return table + _ENRICHMENT_SCRIPT_BY_STEP = { "earnings_history_enrich": Path("scripts/enrich_earnings_history_features.py"), "peer_surprise_enrich": Path("scripts/enrich_peer_surprise_features.py"), @@ -382,6 +399,7 @@ async def incremental_update_canonical_snapshot( existing_test = pq.read_table(str(target_dir / "test.parquet")) # 7. Append new rows to test split; align schemas (new columns get null in old rows) + new_test = _coerce_schema(new_test, existing_test.schema) merged_test = pa.concat_tables([existing_test, new_test], promote_options="default") if "event_date" in merged_test.column_names: sort_idx = pc.sort_indices(merged_test, sort_keys=[("event_date", "ascending")])