Idea Codex

1"""Migration refinement:0013 — per-pull-run provenance on entity tables (ADR-0029).23Adds write-once `pull_run_id` (the pull run that produced the row, from the raw4doc) + `first_refined_at` (net-new marker; backfilled = refined_at for existing5rows) to the 6 single-PK entity tables. Composite-PK/snapshot/trends tables are6out of scope (Phase 1b-ii). steam_apps already has first_refined_at (m_0010).7"""89from __future__ import annotations1011import duckdb1213ID = "0013_curated_provenance"1415# (table, needs_first_refined_at) — steam_apps already has it (m_0010);16# steam_reviews does NOT have it (confirmed: m_0004_steam has no first_refined_at17# on steam_reviews, and no subsequent migration adds it before m_0013).18_TABLES = [19    ("reddit_submissions", True),20    ("reddit_comments", True),21    ("hn_stories", True),22    ("hn_comments", True),23    ("steam_apps", False),24    ("steam_reviews", True),25]262728def up(con: duckdb.DuckDBPyConnection) -> None:29    # Ordered in three passes within the migration transaction. DuckDB rejects30    # CREATE INDEX while there are outstanding row UPDATEs in the same31    # transaction, so ALL index creation must happen BEFORE the backfill UPDATEs32    # (ADD COLUMN is DDL and does not count as an outstanding update). On a33    # populated DB the naive add->update->index ordering raises34    # "Cannot create index with outstanding updates".35    # Pass 1 — add columns (DDL).36    for table, needs_fra in _TABLES:37        con.execute(f"ALTER TABLE {table} ADD COLUMN pull_run_id VARCHAR")38        if needs_fra:39            con.execute(f"ALTER TABLE {table} ADD COLUMN first_refined_at TIMESTAMPTZ")40    # Pass 2 — create indexes (before any UPDATE; built on the fresh NULL columns).41    for table, _ in _TABLES:42        con.execute(43            f"CREATE INDEX IF NOT EXISTS {table}_first_refined_idx ON {table}(first_refined_at)"44        )45        con.execute(f"CREATE INDEX IF NOT EXISTS {table}_pull_run_idx ON {table}(pull_run_id)")46    # Pass 3 — backfill first_refined_at for existing rows (UPDATEs last).47    for table, needs_fra in _TABLES:48        if needs_fra:49            con.execute(50                f"UPDATE {table} SET first_refined_at = refined_at WHERE first_refined_at IS NULL"51            )

idea-codex430b9e9src/market_research/contexts/refinement/adapters/migrations/m_0013_curated_provenance.pySatır 1 – 5151 satır

"""Migration refinement:0013 — per-pull-run provenance on entity tables (ADR-0029).

Adds write-once `pull_run_id` (the pull run that produced the row, from the raw
doc) + `first_refined_at` (net-new marker; backfilled = refined_at for existing
rows) to the 6 single-PK entity tables. Composite-PK/snapshot/trends tables are
out of scope (Phase 1b-ii). steam_apps already has first_refined_at (m_0010).
"""

from __future__ import annotations

import duckdb

ID = "0013_curated_provenance"

# (table, needs_first_refined_at) — steam_apps already has it (m_0010);
# steam_reviews does NOT have it (confirmed: m_0004_steam has no first_refined_at
# on steam_reviews, and no subsequent migration adds it before m_0013).
_TABLES = [
    ("reddit_submissions", True),
    ("reddit_comments", True),
    ("hn_stories", True),
    ("hn_comments", True),
    ("steam_apps", False),
    ("steam_reviews", True),
]


def up(con: duckdb.DuckDBPyConnection) -> None:
    # Ordered in three passes within the migration transaction. DuckDB rejects
    # CREATE INDEX while there are outstanding row UPDATEs in the same
    # transaction, so ALL index creation must happen BEFORE the backfill UPDATEs
    # (ADD COLUMN is DDL and does not count as an outstanding update). On a
    # populated DB the naive add->update->index ordering raises
    # "Cannot create index with outstanding updates".
    # Pass 1 — add columns (DDL).
    for table, needs_fra in _TABLES:
        con.execute(f"ALTER TABLE {table} ADD COLUMN pull_run_id VARCHAR")
        if needs_fra:
            con.execute(f"ALTER TABLE {table} ADD COLUMN first_refined_at TIMESTAMPTZ")
    # Pass 2 — create indexes (before any UPDATE; built on the fresh NULL columns).
    for table, _ in _TABLES:
        con.execute(
            f"CREATE INDEX IF NOT EXISTS {table}_first_refined_idx ON {table}(first_refined_at)"
        )
        con.execute(f"CREATE INDEX IF NOT EXISTS {table}_pull_run_idx ON {table}(pull_run_id)")
    # Pass 3 — backfill first_refined_at for existing rows (UPDATEs last).
    for table, needs_fra in _TABLES:
        if needs_fra:
            con.execute(
                f"UPDATE {table} SET first_refined_at = refined_at WHERE first_refined_at IS NULL"
            )