fix: align Snowflake loader identifiers with DDL compiler (fixes 'invalid identifier')

CREATE TABLE normalized column/table names (upper + non-alnum->_) but the row
loader quoted them raw, so a non-uppercase-safe name (e.g. a Marketing funnel
measure 'Checkout Initiated') was created as CHECKOUT_INITIATED yet INSERTed as
"Checkout Initiated" -> Snowflake 'invalid identifier'. Extract one shared
safe_identifier() and use it in both paths. Names already UPPER_SNAKE are
unaffected (no regression).

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

demoprep_app/ddl/from_dataset.py

@@ -5,6 +5,22 @@ from __future__ import annotations
 from demoprep_app.dataset.contracts import DatasetBundle, DatasetColumn, DatasetTable
 
 
+def safe_identifier(identifier: str) -> str:
+    """Normalize a name into the Snowflake identifier this pipeline creates.
+
+    Single source of truth: CREATE TABLE (this module) and the row loader
+    (integrations/snowflake/dataset_writer.py) MUST agree on this, or the
+    INSERT references a column name the table doesn't have and Snowflake
+    raises "invalid identifier". Uppercases and replaces any non-alphanumeric
+    character with '_' (so 'Checkout Initiated' -> 'CHECKOUT_INITIATED')."""
+    safe = "".join(ch if ch.isalnum() or ch == "_" else "_" for ch in identifier.strip().upper())
+    if not safe:
+        raise ValueError("Identifier cannot be empty")
+    if safe[0].isdigit():
+        safe = f"_{safe}"
+    return safe
+
+
 class DatasetDdlCompiler:
     """Derive simple Snowflake DDL from canonical dataset metadata."""
 
@@ -66,12 +82,7 @@ class DatasetDdlCompiler:
 
     @staticmethod
     def _safe_ident(identifier: str) -> str:
-        safe = "".join(ch if ch.isalnum() or ch == "_" else "_" for ch in identifier.strip().upper())
-        if not safe:
-            raise ValueError("Identifier cannot be empty")
-        if safe[0].isdigit():
-            safe = f"_{safe}"
-        return safe
+        return safe_identifier(identifier)
 
     @staticmethod
     def _snowflake_type(data_type: str) -> str:

demoprep_app/integrations/snowflake/dataset_writer.py

@@ -51,5 +51,12 @@ def populate_dataset_bundle(
 
 
 def _quote_ident(identifier: str) -> str:
-    escaped = identifier.replace('"', '""')
+    # Normalize the SAME way the DDL compiler does before quoting, so the
+    # INSERT targets the exact columns CREATE TABLE produced. A raw name like
+    # 'Checkout Initiated' is created as CHECKOUT_INITIATED by the DDL; quoting
+    # the raw name here would look for "Checkout Initiated" and Snowflake would
+    # raise "invalid identifier".
+    from demoprep_app.ddl.from_dataset import safe_identifier
+
+    escaped = safe_identifier(identifier).replace('"', '""')
     return f'"{escaped}"'