add lock to descover schema

app/mapper.py

@@ -474,27 +474,7 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
                 # In app/mapper.py, after `rows_inserted = len(df_to_insert)`
                 print(f"[canonify] 💾 Inserted {rows_inserted} rows into {table_name}")
 
-                # ==================== DIAGNOSTIC QUERIES (DISABLE IN PROD) ====================
-                print(f"[canonify] 🔬 DIAGNOSTIC: Querying table immediately post-insert...")
-                try:
-                    diagnostic_conn = get_conn(org_id)
-    
-                    # Check total row count
-                    total_rows = diagnostic_conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
-                    print(f"[canonify] 🔬 DIAGNOSTIC: Table {table_name} has {total_rows} total rows")
-    
-                    # Sample the data
-                    sample = diagnostic_conn.execute(f"SELECT * FROM {table_name} LIMIT 3").df()
-                    if not sample.empty:
-                        print(f"[canonify] 🔬 DIAGNOSTIC: Sample data:\n{sample.to_string()}")
-                    else:
-                        print(f"[canonify] 🔬 DIAGNOSTIC: Table exists but SELECT returned ZERO rows")
-    
-                        diagnostic_conn.close()
-                except Exception as diag_e:
-                    print(f"[canonify] 🔬 DIAGNOSTIC ERROR: {diag_e}")
-# ============================================================================
-            
+                
             # Mark schema as applied
             if is_new_schema and version_id:
                 try:

app/tasks/analytics_worker.py

@@ -44,7 +44,7 @@ class AnalyticsWorker:
         self.hours_window = hours_window
         
         # Core engines
-        self.schema = OrgSchema(org_id)
+        
         self.col_embedder = ColumnEmbeddingService()
         self.txn_embedder = EmbeddingService()
         self.vector_service = VectorService(org_id)
@@ -364,6 +364,9 @@ class AnalyticsWorker:
 
     def _discover_schema(self):
         """Schema discovery with proper caching and error handling"""
+        if not self._entity_type:
+            logger.error("[SCHEMA] ❌ Cannot discover schema: entity_type not set")
+            raise ValueError("entity_type must be set before schema discovery")
         try:
             logger.info("[SCHEMA] 🧠 Cache miss, discovering...")