1"""Compaction: merge session_date delta files older than `min_age_days` into base files per (symbol, year).2Asserts the (instrument_id, timestamp_utc) uniqueness invariant before publishing."""3from __future__ import annotations45from datetime import UTC, datetime, timedelta6from typing import Any78import duckdb9import pyarrow as pa10from findata_core.db import connection11from findata_schema import BARS_SCHEMA1213from findata_ingest.lake import Lake141516def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:17 p = job.get("params") or {}18 dataset_id = job["dataset_id"]19 min_age = int(p.get("min_age_days", 35))20 max_files = int(p.get("max_delta_files", 40))21 cutoff = (datetime.now(UTC) - timedelta(days=min_age)).date()22 with connection() as conn:23 deltas = conn.execute("SELECT path, date_min FROM lake_files WHERE dataset_id=%s AND layout='delta' AND version_removed IS NULL AND date_max <= %s ORDER BY date_min LIMIT %s",24 (dataset_id, cutoff, max_files)).fetchall()25 if not deltas:26 return {"compacted_files": 0}27 paths = [str(lake.abs(d["path"])) for d in deltas]28 con = duckdb.connect()29 con.execute("SET threads TO 8")30 syms = [r[0] for r in con.sql("SELECT DISTINCT symbol_as_reported FROM read_parquet($p, hive_partitioning=false) ORDER BY 1", params={"p": paths}).fetchall()]31 years = sorted({d["date_min"].year for d in deltas})32 written, retire = [], [d["path"] for d in deltas]33 rows = 034 for s in syms:35 for y in years:36 new = con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false) WHERE symbol_as_reported=$s AND year(session_date)=$y", params={"p": paths, "s": s, "y": y}).to_arrow_table()37 if new.num_rows == 0:38 continue39 with connection() as conn:40 old = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='base' AND version_removed IS NULL AND symbol_min=%s AND symbol_max=%s AND date_min>=%s AND date_max<=%s",41 (dataset_id, s, s, f"{y}-01-01", f"{y}-12-31")).fetchall()42 parts = [new.cast(BARS_SCHEMA)]43 for o in old:44 t = con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false)", params={"p": [str(lake.abs(o["path"]))]}).to_arrow_table()45 parts.append(t.cast(BARS_SCHEMA))46 retire.append(o["path"])47 merged = pa.concat_tables(parts)48 con.register("m", merged)49 dup = con.sql("SELECT count(*) AS n FROM (SELECT instrument_id, timestamp_utc FROM m GROUP BY 1,2 HAVING count(*)>1)").fetchall()[0][0]50 con.unregister("m")51 if dup:52 raise RuntimeError(f"compaction invariant violated: {dup} duplicate (instrument_id, timestamp_utc) for {s} {y}")53 written.append(lake.write_table(dataset_id, merged, "base", {"symbol": s, "year": str(y)}))54 rows += merged.num_rows55 pub = lake.publish(dataset_id, written, sorted(set(retire)), job.get("run_id"), note=f"compaction {len(deltas)} delta files")56 log.info("compaction.done", deltas=len(deltas), base_files=len(written), rows=rows, **pub)57 return {"compacted_files": len(deltas), "base_files_written": len(written), "rows": rows, **pub}58