fix: restore type coercion + DDL fallback for insert_rows

Prior commit broke coercion by moving the if/elif into wrong scope —
varchar truncation and numeric coercion only ran in the elif branch,
leaving the normal column_info path doing nothing. Restructured so
both column_info and ddl_types paths resolve info/data_type first,
then shared coercion code runs for both.

Also adds is_fact_table name-based detection so schemas without FK
constraints still resolve correctly (SALES, PERFORMANCE, JOURNEYS, etc).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

legitdata_bridge.py

@@ -233,7 +233,9 @@ class KeyPairSnowflakeWriter:
             if isinstance(value, date):
                 return value.strftime('%Y-%m-%d')
             
-            # Apply constraints based on column metadata
+            # Apply constraints based on column metadata.
+            # Resolve info/data_type from column_info (DESCRIBE TABLE) first,
+            # then fall back to ddl_types (parsed DDL) so coercion always runs.
             col_upper = col_name.upper()
             if col_upper in column_info:
                 info = column_info[col_upper]
@@ -242,38 +244,40 @@ class KeyPairSnowflakeWriter:
                 # Fallback: use DDL-declared type when DESCRIBE TABLE didn't return this column
                 info = {}
                 data_type = ddl_types[col_upper].upper()
-                
-                # Truncate strings to fit VARCHAR length
-                if 'VARCHAR' in data_type or 'TEXT' in data_type:
-                    max_len = info.get('length', 255)
-                    if isinstance(value, str) and len(value) > max_len:
-                        value = value[:max_len]
-                
-                # Clamp numbers to fit DECIMAL precision.
-                # Includes Snowflake's internal DESCRIBE TABLE names: FIXED (integers), REAL (floats).
-                elif any(t in data_type for t in (
-                    'NUMBER', 'DECIMAL', 'NUMERIC', 'INT', 'FLOAT',
-                    'FIXED', 'REAL', 'DOUBLE', 'MONEY',
-                )):
-                    # If a string landed in a numeric column, strip formatting and coerce
-                    if isinstance(value, str):
-                        import re as _re
-                        cleaned = _re.sub(r'[^\d.\-+eE]', '', value.replace(',', ''))
-                        try:
-                            value = float(cleaned) if cleaned else None
-                        except (ValueError, TypeError):
-                            return None  # Can't coerce — use NULL rather than crash
-                    precision = info.get('precision', 38)
-                    scale = info.get('scale', 0)
-                    if isinstance(value, (int, float)):
-                        # Max value for given precision/scale
-                        max_val = 10 ** (precision - scale) - (10 ** -scale)
-                        min_val = -max_val
-                        value = max(min_val, min(max_val, value))
-                        # Round to scale
-                        if scale > 0:
-                            value = round(value, scale)
-            
+            else:
+                return value  # No type info — pass through unchanged
+
+            # Truncate strings to fit VARCHAR length
+            if 'VARCHAR' in data_type or 'TEXT' in data_type:
+                max_len = info.get('length', 255)
+                if isinstance(value, str) and len(value) > max_len:
+                    value = value[:max_len]
+
+            # Clamp numbers to fit DECIMAL precision.
+            # Includes Snowflake's internal DESCRIBE TABLE names: FIXED (integers), REAL (floats).
+            elif any(t in data_type for t in (
+                'NUMBER', 'DECIMAL', 'NUMERIC', 'INT', 'FLOAT',
+                'FIXED', 'REAL', 'DOUBLE', 'MONEY',
+            )):
+                # If a string landed in a numeric column, strip formatting and coerce
+                if isinstance(value, str):
+                    import re as _re
+                    cleaned = _re.sub(r'[^\d.\-+eE]', '', value.replace(',', ''))
+                    try:
+                        value = float(cleaned) if cleaned else None
+                    except (ValueError, TypeError):
+                        return None  # Can't coerce — use NULL rather than crash
+                precision = info.get('precision', 38)
+                scale = info.get('scale', 0)
+                if isinstance(value, (int, float)):
+                    # Max value for given precision/scale
+                    max_val = 10 ** (precision - scale) - (10 ** -scale)
+                    min_val = -max_val
+                    value = max(min_val, min(max_val, value))
+                    # Round to scale
+                    if scale > 0:
+                        value = round(value, scale)
+
             return value
         
         total_inserted = 0

legitdata_project/legitdata/ddl/models.py

@@ -4,6 +4,28 @@ from dataclasses import dataclass, field
 from enum import Enum
 from typing import Optional
 
+# Name fragments that strongly suggest a fact/transaction table.
+# Checked against individual words in the table name (split on underscore).
+_FACT_TABLE_KEYWORDS = {
+    'SALES', 'FACT', 'TRANSACTION', 'TRANSACTIONS',
+    'ORDER', 'ORDERS', 'EVENT', 'EVENTS',
+    'METRIC', 'METRICS', 'PERFORMANCE',
+    'ACTIVITY', 'ACTIVITIES',
+    'BOOKING', 'BOOKINGS', 'RESERVATION', 'RESERVATIONS',
+    'REVENUE', 'INVOICE', 'INVOICES',
+    'CLAIM', 'CLAIMS', 'TRADE', 'TRADES',
+    'DEAL', 'DEALS', 'PIPELINE',
+    'JOURNEY', 'JOURNEYS',
+    'INTERACTION', 'INTERACTIONS',
+    'PURCHASE', 'PURCHASES',
+    'PAYMENT', 'PAYMENTS',
+    'SESSION', 'SESSIONS',
+    'CONVERSION', 'CONVERSIONS',
+    'IMPRESSION', 'IMPRESSIONS',
+    'SHIPMENT', 'SHIPMENTS',
+    'RETURN', 'RETURNS',
+}
+
 
 class ColumnClassification(Enum):
     """How a column's data should be sourced."""
@@ -51,13 +73,22 @@ class Table:
     
     @property
     def is_fact_table(self) -> bool:
-        """Heuristic: fact tables have multiple FKs."""
-        return len(self.foreign_keys) >= 2
-    
+        """Identify fact tables by FK count and/or name patterns.
+
+        Primary signal: 2+ foreign keys (schema has explicit FK constraints).
+        Secondary signal: table name contains a known fact-table keyword.
+        Both paths are checked so AI-generated DDL without FK constraints
+        still resolves correctly for standard fact-table naming conventions.
+        """
+        if len(self.foreign_keys) >= 2:
+            return True
+        words = set(self.name.upper().split('_'))
+        return bool(words & _FACT_TABLE_KEYWORDS)
+
     @property
     def is_dimension_table(self) -> bool:
-        """Heuristic: dimension tables have 0-1 FKs."""
-        return len(self.foreign_keys) < 2
+        """Dimension tables are tables that are not fact tables."""
+        return not self.is_fact_table
     
     def get_column(self, name: str) -> Optional[Column]:
         """Get a column by name (case-insensitive)."""