added legacy schema versioning

app/db.py

@@ -108,21 +108,13 @@ def ensure_raw_table(conn: duckdb.DuckDBPyConnection):
 def ensure_schema_versions_table(conn: duckdb.DuckDBPyConnection):
     """
     Tracks schema evolution for each entity table.
-    Enables rollback, audit, and cross-org schema drift monitoring.
-    
-    Table: main.schema_versions
-        - version_id: Auto-incrementing schema version
-        - table_name: Target entity table (e.g., 'sales_canonical')
-        - schema_json: {"column": "dtype", ...}
-        - created_at: When version was detected
-        - applied_at: When version was successfully applied
-        - status: pending | applied | failed
-        - rows_at_migration: Row count snapshot for migration tracking
+    Compatible with DuckDB 0.10.3 constraint limitations.
     """
     conn.execute("CREATE SCHEMA IF NOT EXISTS main")
+    # Use legacy SERIAL syntax instead of IDENTITY
     conn.execute("""
         CREATE TABLE IF NOT EXISTS main.schema_versions (
-            version_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+            version_id BIGINT PRIMARY KEY,
             table_name VARCHAR NOT NULL,
             schema_json JSON NOT NULL,
             created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
@@ -131,7 +123,13 @@ def ensure_schema_versions_table(conn: duckdb.DuckDBPyConnection):
             rows_at_migration BIGINT
         )
     """)
-
+    
+    # Create sequence if it doesn't exist (for manual auto-increment)
+    conn.execute("""
+        CREATE SEQUENCE IF NOT EXISTS schema_version_seq 
+        START WITH 1 
+        INCREMENT BY 1
+    """)
 
 def infer_duckdb_type(value: Any) -> str:
     """

app/mapper.py

@@ -417,14 +417,7 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
     industry_confidence = industry_info["confidence"]
     print(f"[canonify] 🎯 Entity: {entity_type}, Industry: {industry} ({industry_confidence:.2%})")
 
-    # 7️⃣ ENFORCE SCHEMA CONTRACT
-    try:
-        enforce_schema_contract(df, org_id)
-    except ValueError as e:
-        print(f"[canonify] ❌ Schema contract violated: {e}")
-        # For now, log and continue; in strict mode, raise HTTPException
-
-    # 8️⃣ SCHEMA VERSIONING & TRANSACTIONAL INSERT
+        # 8️⃣ SCHEMA VERSIONING & TRANSACTIONAL INSERT
     os.makedirs("./db", exist_ok=True)
     
     with transactional_conn(org_id) as duck:
@@ -445,11 +438,14 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
         
         version_id = None
         if is_new_schema:
+            # Manual auto-increment for DuckDB 0.10.3 compatibility
             version_id = duck.execute("""
-                INSERT INTO main.schema_versions (table_name, schema_json, status) 
-                VALUES (?, ?, 'pending')
-            """, (f"{entity_type}_canonical", json.dumps(current_schema)))
-            version_id = duck.execute("SELECT last_insert_rowid()").fetchone()[0]
+                INSERT INTO main.schema_versions 
+                (version_id, table_name, schema_json, status) 
+                VALUES (nextval('schema_version_seq'), ?, ?, 'pending') 
+                RETURNING version_id
+            """, (f"{entity_type}_canonical", json.dumps(current_schema))).fetchone()[0]
+            print(f"[canonify] 📝 Created schema v{version_id} for {entity_type}_canonical")
         
         # 8b) Ensure table exists with current schema
         table_name = ensure_canonical_table(duck, df, entity_type)
@@ -472,10 +468,8 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
                 print(f"[canonify] 💾 Inserted {len(df_to_insert)} rows into {table_name}")
             
             # 8d) Mark schema as applied post-insert
-                        # 8d) Mark schema as applied post-insert (DuckDB-safe)
             if is_new_schema and version_id:
                 try:
-                    # Use simple UPDATE without complex constraints
                     duck.execute("""
                         UPDATE main.schema_versions 
                         SET applied_at = CURRENT_TIMESTAMP, status = 'applied'
@@ -484,7 +478,7 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
                     print(f"[canonify] ✅ Schema v{version_id} marked as applied")
                 except Exception as e:
                     print(f"[canonify] ⚠️ Schema update warning (non-critical): {e}")
-                    # Continue even if update fails—data is already inserted
+
     duration_ms = (datetime.now() - start_time).total_seconds() * 1000
     print(f"[canonify] ✅ Pipeline complete in {duration_ms:.2f}ms for {org_id}")