guard: catch CREATE/INSERT column mismatches before and during Snowflake load

The Gap e2e failure: FACT_CAMPAIGN_PERFORMANCE's row loader referenced
CHECKOUT_INITIATED, a column CREATE TABLE never declared -> Snowflake
"invalid identifier", ~10 minutes into the run, no indication of which
table/column. Root mechanism wasn't pinned with certainty (DDL and the
loader both derive columns from the same DatasetBundle object, so they
can't diverge in the code as read) — these guards catch the failure
class immediately and name the exact column, regardless of cause.

- validator.py: _check_schema_consistency() — in-memory, at
validate_bundle() time. Flags (a) two column names that collide after
SQL-identifier normalization (duplicate column in CREATE TABLE), and
(b) a generated row carrying a key no declared column covers (INSERT
would reference an identifier CREATE TABLE never defined). Runs before
any DDL is derived or Snowflake is touched.

- dataset_writer.py: _verify_table_columns() — right before each table's
INSERT, DESCRIBE TABLE and diff against the columns we're about to
insert. Catches the mismatch even if its cause is outside this code
(stale worker, schema predating this run, etc.) by checking reality
instead of assuming the DDL that ran matches the object being loaded.

Tests: 4 new validator cases (clean bundle, row-key mismatch, normalized
duplicate names, empty-table no-op) + 2 new dataset_writer cases (missing
column raises before any INSERT; DESCRIBE runs per non-empty table).
Extended test_dataset_writer's FakeCursor with DESCRIBE TABLE support.

Verified locally: new tests pass, full suite unchanged (2 pre-existing
unrelated failures confirmed present on HEAD before this change), and
the blueprint smoke script (real engine, no mocks) passes clean — no
false positives on real generated data.

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

demoprep_app/dataset/validator.py

@@ -63,6 +63,8 @@ def validate_bundle(blueprint: DemoBlueprint, bundle: DatasetBundle) -> Validati
     report = ValidationReport()
     tables = bundle.table_map()
 
+    _check_schema_consistency(bundle, report)
+
     fact_rows_by_name: dict[str, list[dict[str, Any]]] = {}
     for fact in blueprint.facts:
         table = tables.get(fact.name)
@@ -89,6 +91,54 @@ def validate_bundle(blueprint: DemoBlueprint, bundle: DatasetBundle) -> Validati
 # ---------------------------------------------------------------------------
 
 
+def _check_schema_consistency(bundle: DatasetBundle, report: ValidationReport) -> None:
+    """Guard against CREATE TABLE / INSERT column mismatches before either is ever
+    sent to Snowflake. Without this, a mismatch surfaces ~10 minutes later as a
+    cryptic 'invalid identifier' SQL-compilation error deep inside an executemany —
+    this catches it here, in-memory, in milliseconds, and names the exact column.
+
+    Two failure modes guarded:
+      1. Two distinct column names collide under SQL-identifier normalization
+         (e.g. 'Checkout-Initiated' and 'Checkout Initiated' both ->
+         CHECKOUT_INITIATED) — CREATE TABLE would get a duplicate column.
+      2. A generated row carries a key that normalization doesn't map onto any
+         declared column — the INSERT would reference an identifier CREATE
+         TABLE never defined.
+
+    Both directions derive from the SAME table.columns/table.rows the DDL
+    compiler and the row loader each read independently, so if this check
+    passes, CREATE and INSERT are provably looking at the same column set.
+    """
+    from demoprep_app.ddl.from_dataset import safe_identifier
+
+    for table in bundle.tables:
+        seen: dict[str, str] = {}
+        for column in table.columns:
+            safe = safe_identifier(column.name)
+            if safe in seen and seen[safe] != column.name:
+                report.integrity_errors.append(
+                    f"Table {table.name}: columns '{seen[safe]}' and '{column.name}' both "
+                    f"normalize to SQL identifier '{safe}' — CREATE TABLE would get a "
+                    "duplicate column."
+                )
+                continue
+            seen[safe] = column.name
+
+        if not table.rows:
+            continue
+        declared = set(seen.keys())
+        # A generation bug is systemic (every row has it the same way), so the
+        # first row is as informative as scanning all of them and far cheaper.
+        row_keys = {safe_identifier(k) for k in table.rows[0].keys()}
+        extra = row_keys - declared
+        if extra:
+            report.integrity_errors.append(
+                f"Table {table.name}: generated rows contain column(s) {sorted(extra)} "
+                "that CREATE TABLE would not declare for this table — INSERT would fail "
+                f"with 'invalid identifier'. Declared columns: {sorted(declared)}."
+            )
+
+
 def _check_integrity(blueprint: DemoBlueprint, fact, tables: dict, fact_rows: list[dict[str, Any]], report: ValidationReport) -> None:
     for dim in blueprint.dimensions:
         if dim.name not in fact.dimension_names:

demoprep_app/integrations/snowflake/dataset_writer.py

@@ -29,6 +29,7 @@ def populate_dataset_bundle(
             quoted_columns = ", ".join(_quote_ident(column) for column in columns)
             placeholders = ", ".join(["%s"] * len(columns))
             values = [tuple(row.get(column) for column in columns) for row in table.rows]
+            _verify_table_columns(cursor, table.name, columns)
             if progress_callback:
                 progress_callback(f"Loading {table.name}: {len(values):,} rows")
             cursor.execute(f"TRUNCATE TABLE {_quote_ident(table.name)}")
@@ -50,6 +51,33 @@ def populate_dataset_bundle(
         cursor.close()
 
 
+def _verify_table_columns(cursor: Any, table_name: str, expected_columns: list[str]) -> None:
+    """Fail fast, with a clear diagnosis, if the table Snowflake actually has
+    doesn't carry every column we're about to INSERT into.
+
+    Without this, a mismatch (whatever its cause — a stale worker running old
+    DDL, a schema that predates this run, a generation bug validate_bundle
+    missed) surfaces as a bare 'invalid identifier' SQL-compilation error deep
+    inside executemany, with no indication of which table or column. This
+    names both, before a single row is sent, by checking reality (what
+    Snowflake actually created) against the dataset object's expectations —
+    the two are independently derived, so this is the one place they're
+    provably reconciled.
+    """
+    from demoprep_app.ddl.from_dataset import safe_identifier
+
+    cursor.execute(f"DESCRIBE TABLE {_quote_ident(table_name)}")
+    actual = {row[0].upper() for row in cursor.fetchall()}
+    expected = {safe_identifier(c) for c in expected_columns}
+    missing = expected - actual
+    if missing:
+        raise RuntimeError(
+            f"Table {table_name} is missing column(s) {sorted(missing)} that the generated "
+            "dataset expects to insert. CREATE TABLE and the row loader disagree on this "
+            f"table's schema. Actual columns in Snowflake: {sorted(actual)}."
+        )
+
+
 def _quote_ident(identifier: str) -> str:
     # Normalize the SAME way the DDL compiler does before quoting, so the
     # INSERT targets the exact columns CREATE TABLE produced. A raw name like

tests/test_dataset_writer.py

@@ -6,14 +6,25 @@ from demoprep_app.scenario.contract import ScenarioContract
 
 
 class FakeCursor:
-    def __init__(self, fail_on_executemany=False):
+    def __init__(self, fail_on_executemany=False, describe_columns=None):
         self.fail_on_executemany = fail_on_executemany
+        # table_name -> list of column names DESCRIBE TABLE should report for it.
+        # Tests that don't care about the column-verification guard can omit this
+        # and get an empty result, which is why they must opt in to match reality.
+        self.describe_columns = describe_columns or {}
         self.executed = []
         self.executemany_calls = []
         self.closed = False
+        self._last_describe_table = None
 
     def execute(self, sql):
         self.executed.append(sql)
+        if sql.upper().startswith("DESCRIBE TABLE"):
+            self._last_describe_table = sql.split('"')[1]
+
+    def fetchall(self):
+        cols = self.describe_columns.get(self._last_describe_table, [])
+        return [(c,) for c in cols]
 
     def executemany(self, sql, values):
         if self.fail_on_executemany:
@@ -74,7 +85,7 @@ def _bundle():
 
 
 def test_populate_dataset_bundle_inserts_rows_and_returns_counts():
-    cursor = FakeCursor()
+    cursor = FakeCursor(describe_columns={"ACCOUNTS": ["ACCOUNT_ID", "ACCOUNT_NAME"]})
     connection = FakeConnection(cursor)
     progress = []
 
@@ -99,7 +110,10 @@ def test_populate_dataset_bundle_inserts_rows_and_returns_counts():
 
 
 def test_populate_dataset_bundle_rolls_back_and_closes_cursor_on_error():
-    cursor = FakeCursor(fail_on_executemany=True)
+    cursor = FakeCursor(
+        fail_on_executemany=True,
+        describe_columns={"ACCOUNTS": ["ACCOUNT_ID", "ACCOUNT_NAME"]},
+    )
     connection = FakeConnection(cursor)
 
     with pytest.raises(RuntimeError, match="insert failed"):
@@ -108,3 +122,31 @@ def test_populate_dataset_bundle_rolls_back_and_closes_cursor_on_error():
     assert connection.committed is False
     assert connection.rolled_back is True
     assert cursor.closed is True
+
+
+def test_populate_dataset_bundle_raises_clear_error_when_column_missing():
+    # Snowflake's table is missing ACCOUNT_NAME — e.g. the CREATE TABLE that ran
+    # didn't match the dataset object being loaded. This must fail BEFORE any
+    # INSERT is attempted, naming the exact missing column, instead of letting
+    # Snowflake raise a bare "invalid identifier" deep inside executemany.
+    cursor = FakeCursor(describe_columns={"ACCOUNTS": ["ACCOUNT_ID"]})
+    connection = FakeConnection(cursor)
+
+    with pytest.raises(RuntimeError, match="ACCOUNTS.*ACCOUNT_NAME"):
+        populate_dataset_bundle(connection, "DEMO_SCHEMA", _bundle())
+
+    assert cursor.executemany_calls == []
+    assert connection.committed is False
+    assert connection.rolled_back is True
+    assert cursor.closed is True
+
+
+def test_populate_dataset_bundle_verifies_columns_before_each_table_insert():
+    cursor = FakeCursor(describe_columns={"ACCOUNTS": ["ACCOUNT_ID", "ACCOUNT_NAME"]})
+    connection = FakeConnection(cursor)
+
+    populate_dataset_bundle(connection, "DEMO_SCHEMA", _bundle())
+
+    assert 'DESCRIBE TABLE "ACCOUNTS"' in cursor.executed
+    # EMPTY_DIM has no rows, so it's skipped entirely (no DESCRIBE, no INSERT).
+    assert 'DESCRIBE TABLE "EMPTY_DIM"' not in cursor.executed

tests/test_validator_schema_consistency.py

@@ -0,0 +1,127 @@
+"""Guard against CREATE TABLE / INSERT column mismatches — catch them in-memory,
+at validate_bundle() time, instead of ~10 minutes later as a cryptic Snowflake
+'invalid identifier' error deep inside an executemany (see the Gap e2e failure:
+FACT_CAMPAIGN_PERFORMANCE's row loader referenced a CHECKOUT_INITIATED column
+CREATE TABLE never declared).
+"""
+
+from __future__ import annotations
+
+from demoprep_app.dataset.contracts import DatasetBundle, DatasetColumn, DatasetTable
+from demoprep_app.dataset.validator import ValidationReport, _check_schema_consistency
+from demoprep_app.scenario.contract import ScenarioContract
+
+
+def _scenario() -> ScenarioContract:
+    return ScenarioContract(
+        company_name="Acme",
+        company_url="https://example.com",
+        use_case="Marketing Funnel",
+        scenario_type="campaign_performance",
+        fact_grain="campaign-day",
+    )
+
+
+def test_schema_consistency_passes_for_a_clean_bundle():
+    bundle = DatasetBundle(
+        scenario=_scenario(),
+        tables=[
+            DatasetTable(
+                name="FACT_CAMPAIGN_PERFORMANCE",
+                grain="campaign-day",
+                columns=[
+                    DatasetColumn("FACT_CAMPAIGN_PERFORMANCE_KEY", "fact_key", "NUMBER", nullable=False),
+                    DatasetColumn("IMPRESSIONS", "impressions", "NUMBER", nullable=False),
+                    DatasetColumn("CLICKS", "clicks", "NUMBER", nullable=False),
+                ],
+                rows=[{"FACT_CAMPAIGN_PERFORMANCE_KEY": 1, "IMPRESSIONS": 100, "CLICKS": 5}],
+                is_fact=True,
+            )
+        ],
+    )
+    report = ValidationReport()
+
+    _check_schema_consistency(bundle, report)
+
+    assert report.integrity_errors == []
+
+
+def test_schema_consistency_catches_a_row_key_with_no_declared_column():
+    # Mirrors the Gap failure: the row loader would INSERT a column
+    # (CHECKOUT_INITIATED) that CREATE TABLE never declared for this table.
+    bundle = DatasetBundle(
+        scenario=_scenario(),
+        tables=[
+            DatasetTable(
+                name="FACT_CAMPAIGN_PERFORMANCE",
+                grain="campaign-day",
+                columns=[
+                    DatasetColumn("FACT_CAMPAIGN_PERFORMANCE_KEY", "fact_key", "NUMBER", nullable=False),
+                    DatasetColumn("ADD_TO_CART", "add_to_cart", "NUMBER", nullable=False),
+                    DatasetColumn("CONVERSIONS", "conversions", "NUMBER", nullable=False),
+                ],
+                rows=[
+                    {
+                        "FACT_CAMPAIGN_PERFORMANCE_KEY": 1,
+                        "ADD_TO_CART": 10,
+                        "CONVERSIONS": 2,
+                        "CHECKOUT_INITIATED": 6,  # not a declared column
+                    }
+                ],
+                is_fact=True,
+            )
+        ],
+    )
+    report = ValidationReport()
+
+    _check_schema_consistency(bundle, report)
+
+    assert len(report.integrity_errors) == 1
+    assert "FACT_CAMPAIGN_PERFORMANCE" in report.integrity_errors[0]
+    assert "CHECKOUT_INITIATED" in report.integrity_errors[0]
+
+
+def test_schema_consistency_catches_column_names_that_collide_after_normalization():
+    # Two distinct Python-level names that normalize to the same SQL identifier
+    # would make CREATE TABLE emit a duplicate column.
+    bundle = DatasetBundle(
+        scenario=_scenario(),
+        tables=[
+            DatasetTable(
+                name="FACT_CAMPAIGN_PERFORMANCE",
+                grain="campaign-day",
+                columns=[
+                    DatasetColumn("Checkout Initiated", "checkout_initiated", "NUMBER", nullable=False),
+                    DatasetColumn("Checkout-Initiated", "checkout_initiated_2", "NUMBER", nullable=False),
+                ],
+                rows=[{"Checkout Initiated": 1, "Checkout-Initiated": 2}],
+                is_fact=True,
+            )
+        ],
+    )
+    report = ValidationReport()
+
+    _check_schema_consistency(bundle, report)
+
+    assert len(report.integrity_errors) == 1
+    assert "CHECKOUT_INITIATED" in report.integrity_errors[0]
+    assert "duplicate column" in report.integrity_errors[0]
+
+
+def test_schema_consistency_ignores_empty_tables():
+    bundle = DatasetBundle(
+        scenario=_scenario(),
+        tables=[
+            DatasetTable(
+                name="EMPTY_DIM",
+                grain="empty",
+                columns=[DatasetColumn("EMPTY_DIM_KEY", "dimension_key", "NUMBER", nullable=False)],
+                rows=[],
+            )
+        ],
+    )
+    report = ValidationReport()
+
+    _check_schema_consistency(bundle, report)
+
+    assert report.integrity_errors == []