import sys import re import json import os import spacy import networkx as nx import torch import torch.nn as nn from transformers import AutoTokenizer, AutoModel from sentence_transformers.cross_encoder import CrossEncoder # ========================================================= # SKELETON CONSTANTS # ========================================================= SIMPLE = 0 PURE_AGG = 1 GROUPED_AGG = 2 SUPERLATIVE = 3 EXCEPT_QUERY = 4 LABEL_NAMES = {0:'SIMPLE', 1:'PURE_AGG', 2:'GROUPED_AGG', 3:'SUPERLATIVE', 4:'EXCEPT_QUERY'} # ========================================================= # SKELETON CLASSIFIER # ========================================================= class _SkeletonModel(nn.Module): def __init__(self, backbone, n_classes): super().__init__() self.backbone = backbone self.classifier = nn.Linear(384, n_classes) self.dropout = nn.Dropout(0.1) def forward(self, input_ids, attention_mask): out = self.backbone(input_ids=input_ids, attention_mask=attention_mask) pooled = out.last_hidden_state[:, 0, :] return self.classifier(self.dropout(pooled)) class SkeletonClassifier: def __init__(self, model_path): with open(f"{model_path}/labels.json") as f: self.label_names = json.load(f) n_classes = len(self.label_names) self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') self.tokenizer = AutoTokenizer.from_pretrained(model_path) backbone = AutoModel.from_pretrained('cross-encoder/ms-marco-MiniLM-L-6-v2') self.model = _SkeletonModel(backbone, n_classes).to(self.device) self.model.load_state_dict( torch.load(f"{model_path}/model.pt", map_location=self.device) ) self.model.eval() def predict(self, question): enc = self.tokenizer([question], padding=True, truncation=True, max_length=128, return_tensors='pt') with torch.no_grad(): logits = self.model( enc['input_ids'].to(self.device), enc['attention_mask'].to(self.device) ) label = int(logits.argmax(dim=1).item()) return label, self.label_names[str(label)] #============================================================ # DB Class #============================================================ from sentence_transformers import SentenceTransformer, util import sqlite3 class DBValueLookup: def __init__(self, db_dir, model_name='all-MiniLM-L6-v2'): self.db_dir = db_dir self._cache = {} # (db_id, table, col) -> [values] self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') print(f"Loading Semantic DB Lookup model ({model_name})...") self.model = SentenceTransformer(model_name).to(self.device) def get_values(self, db_id, table, col): key = (db_id, table, col) if key not in self._cache: db_path = f"{self.db_dir}/{db_id}/{db_id}.sqlite" try: conn = sqlite3.connect(db_path) # Fetch distinct values, ignore nulls rows = conn.execute( f"SELECT DISTINCT {col} FROM {table} WHERE {col} IS NOT NULL LIMIT 200" ).fetchall() conn.close() self._cache[key] = [str(r[0]) for r in rows if r[0]] except Exception: self._cache[key] = [] return self._cache[key] def semantic_match(self, value_text, db_id, table, col, threshold=0.5): candidates = self.get_values(db_id, table, col) if not candidates: return value_text # Fast-path: Exact match saves us running the neural model v_lower = value_text.lower() for c in candidates: if v_lower == c.lower(): return c # Semantic matching query_embedding = self.model.encode(value_text, convert_to_tensor=True, show_progress_bar=False, device=self.device) db_embeddings = self.model.encode(candidates, convert_to_tensor=True, show_progress_bar=False, device=self.device) cosine_scores = util.cos_sim(query_embedding, db_embeddings) best_score_val, best_idx = torch.max(cosine_scores, dim=1) best_score = best_score_val.item() if best_score >= threshold: return candidates[best_idx] return value_text # Fallback to original text if no good match #========================================================== # OPERATORCLISSIFIER CLASSS # ========================================================= class OperatorClassifier: def __init__(self, model_path): import torch import torch.nn as nn from transformers import AutoTokenizer, AutoModel self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') self.tokenizer = AutoTokenizer.from_pretrained(model_path) with open(f"{model_path}/operators.json") as f: self.operators = json.load(f) backbone = AutoModel.from_pretrained('cross-encoder/ms-marco-MiniLM-L-6-v2') class OperatorModel(nn.Module): def __init__(self, backbone, n_classes): super().__init__() self.backbone = backbone self.classifier = nn.Linear(384, n_classes) self.dropout = nn.Dropout(0.1) def forward(self, input_ids, attention_mask): out = self.backbone(input_ids=input_ids, attention_mask=attention_mask) pooled = out.last_hidden_state[:, 0, :] return self.classifier(self.dropout(pooled)) self.model = OperatorModel(backbone, len(self.operators)).to(self.device) self.model.load_state_dict( torch.load(f"{model_path}/model.pt", map_location=self.device) ) self.model.eval() def predict(self, context): import torch enc = self.tokenizer([context], padding=True, truncation=True, max_length=64, return_tensors='pt') with torch.no_grad(): logits = self.model( enc['input_ids'].to(self.device), enc['attention_mask'].to(self.device) ) idx = int(logits.argmax(dim=1).item()) return self.operators[idx] #========================================================= # 1. SCHEMA GRAPH # ========================================================= class SchemaGraph: def __init__(self, schema_text=None, spider_schema=None): self.graph = nx.Graph() self.tables = {} self.primary_keys = {} if spider_schema: self._parse_spider(spider_schema) elif schema_text: self._parse_text(schema_text) self._build_graph() def _parse_text(self, text): table_pattern = re.compile(r'CREATE\s+TABLE\s+(\w+)\s*\((.*?)\);', re.I | re.S) for match in table_pattern.finditer(text): table = match.group(1).lower() body = match.group(2) cols = {} for line in body.split(','): line = line.strip() if not line: continue parts = line.split() col_name = parts[0].lower() col_type = "TEXT" if len(parts) > 1 and any(t in parts[1].upper() for t in ["INT", "FLOAT", "NUMBER"]): col_type = "NUM" cols[col_name] = col_type # if "PRIMARY KEY" in line.upper() or col_name in ('id', f"{table}_id"): if "PRIMARY KEY" in line.upper() or col_name in ('id', f"{table}_id") or col_name.endswith('id'): self.primary_keys[table] = col_name self.tables[table] = cols self.graph.add_node(table) def _parse_spider(self, schema_json): table_names_orig = schema_json['table_names_original'] table_names_norm = schema_json['table_names'] # The natural language table names col_names_orig = schema_json['column_names_original'] col_names_norm = schema_json['column_names'] # The natural language column names col_types = schema_json['column_types'] # New dictionaries to hold the translation mapping self.norm_table_names = {} self.norm_column_names = {} for i, tbl in enumerate(table_names_orig): t_name = tbl.lower() self.tables[t_name] = {} self.graph.add_node(t_name) # Map original table name -> normalized table name self.norm_table_names[t_name] = table_names_norm[i].lower() for i, (tbl_idx, col_name) in enumerate(col_names_orig): if tbl_idx == -1: continue t_name = table_names_orig[tbl_idx].lower() c_name = col_name.lower() # Spider stores the normalized name at index 1 of the inner list norm_c_name = col_names_norm[i][1].lower() c_type = "NUM" if col_types[i] == "number" else "TEXT" self.tables[t_name][c_name] = c_type # Map (table, original_column) -> normalized column self.norm_column_names[(t_name, c_name)] = norm_c_name if c_name in ('id', f"{t_name}_id"): self.primary_keys[t_name] = c_name for (col_idx_1, col_idx_2) in schema_json['foreign_keys']: t1_idx, c1_name = col_names_orig[col_idx_1] t2_idx, c2_name = col_names_orig[col_idx_2] t1 = table_names_orig[t1_idx].lower() t2 = table_names_orig[t2_idx].lower() self.graph.add_edge(t1, t2, on=f"{t1}.{c1_name.lower()} = {t2}.{c2_name.lower()}") def _build_graph(self): tables = list(self.tables.keys()) for t1 in tables: for col in self.tables[t1]: for t2 in tables: if col in (f"{t2}_id", f"{t2.rstrip('s')}_id"): if t1 == t2: continue pk = self.primary_keys.get(t2) if pk and not self.graph.has_edge(t1, t2): self.graph.add_edge(t1, t2, on=f"{t1}.{col} = {t2}.{pk}") # ========================================================= # 2. LINGUISTIC ENGINE # ========================================================= class LinguisticEngine: def __init__(self, schema_graph, table_model_path, column_model_path, value_model_path, skeleton_model_path, db_id=None): self.schema = schema_graph self.db_id = db_id print("Loading spaCy...") self.nlp = spacy.load("en_core_web_sm") print("Loading DB value lookup...") self.db_lookup = DBValueLookup('/kaggle/working/spider_data/database') print("Loading table linker...") self.table_linker = CrossEncoder(table_model_path) print("Loading column linker...") self.column_linker = CrossEncoder(column_model_path) print("Loading value linker...") self.value_linker = CrossEncoder(value_model_path) print("Loading skeleton classifier...") self.skeleton = SkeletonClassifier(skeleton_model_path) print("Ready.") print("Loading operator classifier...") self.operator_clf = OperatorClassifier('schema_linking_data/model_operator') self.boolean_cols = self._build_boolean_column_registry() def _serialize_table(self, table_name): cols = list(self.schema.tables.get(table_name, {}).keys()) # Use mapped table name and mapped column names norm_t_name = getattr(self.schema, 'norm_table_names', {}).get(table_name, table_name) norm_cols = [ getattr(self.schema, 'norm_column_names', {}).get((table_name, c), c) for c in cols ] return f"{norm_t_name} | {', '.join(norm_cols)}" def _serialize_column(self, table_name, col_name): all_cols = list(self.schema.tables.get(table_name, {}).keys()) col_type = self.schema.tables[table_name].get(col_name, 'text') # Safely fetch normalized names (falling back to raw names if not a Spider schema) norm_t_name = getattr(self.schema, 'norm_table_names', {}).get(table_name, table_name) norm_c_name = getattr(self.schema, 'norm_column_names', {}).get((table_name, col_name), col_name) # Normalize the context columns as well norm_context_list = [ getattr(self.schema, 'norm_column_names', {}).get((table_name, c), c) for c in all_cols if c != col_name ] context = ', '.join(norm_context_list) return f"{norm_t_name} | {norm_c_name} | {col_type} | context: {context}" def resolve_ordering(self, question, active_tables): q_lower = question.lower() doc = self.nlp(q_lower) target_spans = [] direction = "DESC" dir_map = { 'highest': 'DESC', 'largest': 'DESC', 'maximum': 'DESC', 'best': 'DESC', 'oldest': 'DESC', 'most': 'DESC', 'lowest': 'ASC', 'smallest': 'ASC', 'minimum': 'ASC', 'worst': 'ASC', 'youngest': 'ASC', 'least': 'ASC' } for token in doc: if token.tag_ in ['JJS', 'RBS'] or token.text in dir_map: # Safely set direction based on the first superlative encountered direction = dir_map.get(token.text, "ASC" if any(x in token.text for x in ['least','fewest','smallest','lowest','worst']) else "DESC") # Hard-target known abstract concepts to prevent noun pollution if token.text in ['youngest', 'oldest']: target_spans.append('age') elif token.text in ['most', 'least']: return 'COUNT(*)', direction else: head = token.head if head.pos_ in ['NOUN', 'PROPN'] and head.text != token.text: modifiers = [child.text for child in head.children if child.dep_ in ['amod', 'compound'] and child.text != token.text] modifiers.append(head.text) target_spans.append(" ".join(modifiers)) else: target_spans.append(token.text) break if not target_spans: if any(k in q_lower for k in ["descending order","ascending order","order by","sorted by","ordered by"]): chunks = list(doc.noun_chunks) target_span = chunks[-1].text if chunks else question.split()[-1] target_spans.append(target_span) if "asc" in q_lower: direction = "ASC" if not target_spans: return None, None # --- Structural Domain Override for Time-Series Superlatives --- # The prompt explicitly defines anomalies as being ranked by data_value_double # if getattr(self, 'db_id', None) == 'influx_system' and 'sys_target_alerts' in active_tables: # if any(w in question.lower() for w in ['anomaly', 'bottleneck', 'spike']): # return 'data_value_double', direction # --- Structural Domain Override for Time-Series Superlatives --- if getattr(self, 'db_id', None) == 'influx_system' and 'sys_target_alerts' in active_tables: # Protect aggregate queries! If they ask for a count, do NOT force the raw double value is_aggregate = any(w in question.lower() for w in ['count', 'total', 'how many', 'average', 'sum']) if not is_aggregate: if any(w in question.lower() for w in ['anomaly', 'bottleneck', 'spike']): return 'data_value_double', direction candidates = [(t, c) for t in active_tables for c in self.schema.tables[t] if c not in ('id', f"{t}_id")] # candidates = [(t, c) for t in active_tables for c in self.schema.tables[t] if c not in ('id', f"{t}_id")] if not candidates: return None, None best_overall_score = -float('inf') best_overall_match = None for span in target_spans: pairs = [(span, self._serialize_column(t, c)) for t, c in candidates] scores = self.column_linker.predict(pairs, show_progress_bar=False) best_idx = int(scores.argmax()) best_score = float(scores[best_idx]) if best_score > best_overall_score: best_overall_score = best_score best_t, best_c = candidates[best_idx] best_overall_match = f"{best_t}.{best_c}" return best_overall_match, direction def detect_aggregation(self, question): q = question.lower() found_aggs = [] # Aggregation should only trigger on mathematical intent checks = [ (["average", "mean", "avg"], "AVG"), (["maximum", "max", "highest"], "MAX"), (["minimum", "min", "lowest"], "MIN"), (["how many", "count the number"], "COUNT"), # Removed "all" and generic "count" (["total sum", "sum of"], "SUM") ] # Robust mathematical intent detection for "count" if "count" in q and any(w in q for w in ["per", "total", "of", "for"]): found_aggs.append((q.find("count"), "COUNT")) for keywords, agg_type in checks: for kw in keywords: idx = q.find(kw) if idx != -1: if agg_type == "SUM" and "total number" in q: continue found_aggs.append((idx, agg_type)) break found_aggs.sort(key=lambda x: x[0]) if 'SUM' in [a for _, a in found_aggs] and 'COUNT' in [a for _, a in found_aggs]: if 'total number' in q or 'total count' in q or 'count per' in q: found_aggs = [(i, a) for i, a in found_aggs if a == 'COUNT'] return [agg for _, agg in found_aggs] def bind_values(self, question, active_tables, window_size=5, debug=False): question = question.replace('\u2018', "'").replace('\u2019', "'").replace('\u201c', '"').replace('\u201d', '"') quoted = [(m.group(1), False, True) for m in re.finditer(r"['\"](.+?)['\"]", question)] top_n_positions = {m.start(1) for m in re.finditer( r'\b(?:top|bottom|worst|best)\s+(\d+)\b', question.lower())} numbered = [] for m in re.finditer(r'\b(\d+(?:\.\d+)?)\b', question): if m.start() in top_n_positions: continue num_str = m.group(1) if not any(num_str in q[0] for q in quoted): numbered.append((num_str, True, False)) doc = self.nlp(question) entity_texts = set(q[0].lower() for q in quoted) valid_ent_types = {'PERSON', 'ORG', 'GPE', 'LOC', 'FAC', 'PRODUCT', 'NORP', 'EVENT', 'WORK_OF_ART'} entities = [] for ent in doc.ents: if ent.text.lower() in entity_texts or re.search(r'\d+', ent.text) or ent.label_ not in valid_ent_types: continue entities.append((ent.text, False, False)) # ---> CHATBOT ENTITY SAFETY NET <--- # Users won't use quotes in a chatbot. We must explicitly scan the text # for known target names or critical identifiers. if self.db_id == 'derby_system' or self.db_id == 'influx_system': # Fast-fetch all known target names from Derby known_targets = self.db_lookup.get_values('derby_system', 'target', 'name') for kt in known_targets: if str(kt).lower() in question.lower() and len(str(kt)) > 3: # If the unquoted string exists in the DB, force it into the entity list! # The boolean flags are (is_numeric=False, is_quoted=False) entities.append((str(kt), False, False)) search_tables = set(active_tables) for t in active_tables: search_tables.update(self.schema.graph.neighbors(t)) if self.db_id: seen_texts = entity_texts | set(v[0].lower() for v in numbered) # ---> SURGICAL FIX: FORCE EXTRACTION OF TARGET NAMES <--- # Look for words containing underscores and numbers (e.g. MySQL_QUICK_1711_1) # complex_names = re.findall(r'\b[A-Za-z0-9]+_[A-Za-z0-9_]+\b', question) # for name in complex_names: # if name.lower() not in seen_texts: # ---> SURGICAL FIX: FORCE EXTRACTION OF TARGET NAMES <--- complex_names = re.findall(r'\b[A-Za-z0-9]+(?:_[A-Za-z0-9_]+|-[\w\-]+)+\b', question) for name in complex_names: # Prevent extracting substrings of already-quoted values! if name.lower() not in seen_texts and not any(name.lower() in st for st in seen_texts): # ALWAYS trust complex names as entities, skip the DB existence check here! # It will be evaluated by the structural trust rule downstream. entities.append((name, False, False)) seen_texts.add(name.lower()) # Standard spaCy noun extraction for token in doc: if token.pos_ not in ['NOUN', 'PROPN', 'X']: continue for txt in [token.text, token.lemma_]: if txt.lower() in seen_texts or len(txt) < 3: continue for table in search_tables: for col, col_type in self.schema.tables.get(table, {}).items(): if col_type != 'TEXT': continue db_vals = self.db_lookup.get_values(self.db_id, table, col) if any(txt.lower() == str(v).lower() for v in db_vals): entities.append((txt, False, False)) seen_texts.add(txt.lower()) seen_texts.add(token.text.lower()) break else: continue break # STRUCTURAL FIX 1: Deduplicate all extracted values unique_vals = [] seen = set() for v in quoted + numbered + entities: if v[0].lower() not in seen: seen.add(v[0].lower()) unique_vals.append(v) all_values = unique_vals if not all_values: return [] candidates = [(table, col, col_type) for table in search_tables for col, col_type in self.schema.tables.get(table, {}).items() if col != 'id' and not col.endswith('_id')] if not candidates: return [] filters = [] skip_vals = set() for val_text, is_numeric, is_quoted in all_values: if val_text in skip_vals: continue val_pos = question.lower().find(val_text.lower()) if val_pos == -1: continue before = question[:val_pos].split()[-window_size:] after = question[val_pos + len(val_text):].split()[:window_size] context = ' '.join(before + [val_text] + after) valid_candidates = candidates # --- THE FIX: STRUCTURAL TRUST --- # If the word contains underscores, hyphens mixed with numbers, or looks like a system ID, # we trust it is a valid entity even if the DB lookup fails. is_complex_identifier = bool(re.match(r'^[A-Za-z0-9]+(?:_[A-Za-z0-9_]+|-[\w\-]+)+$', val_text)) if self.db_id and not is_numeric and not is_quoted and not is_complex_identifier: exact_matches = [] for t, c, ct in candidates: if ct == 'TEXT': db_vals = self.db_lookup.get_values(self.db_id, t, c) if val_text.lower() in [str(v).lower() for v in db_vals]: exact_matches.append((t, c, ct)) # Discard unquoted, simple conversational nouns that do not physically exist in the DB! if not exact_matches: continue valid_candidates = exact_matches pairs = [(context, self._serialize_column(t, c)) for t, c, ct in valid_candidates] scores = self.value_linker.predict(pairs, show_progress_bar=False) sorted_indices = scores.argsort()[::-1] for idx in sorted_indices: best_score = float(scores[idx]) t, c, ct = valid_candidates[idx] # ---> BULLETPROOF LEXICAL OVERRIDE <--- # Check if the column name exists in the question, ignoring spaces and underscores col_clean = c.lower().replace("_", "") q_clean = question.lower().replace("_", "").replace(" ", "") is_lexical = col_clean in q_clean threshold = -2.0 if is_numeric else -5.0 # If the neural net hates it, but the column name is literally in the sentence, trust the text! if best_score <= threshold and not is_lexical: continue # Keep looking, don't break entirely if ct == 'NUM' and not is_numeric: continue op = self.operator_clf.predict(context) like_triggers = ['have', 'having', 'contain', 'containing', 'start', 'end', 'like', 'starts with', 'ends with'] if is_quoted and any(k in context.lower() for k in like_triggers): op = 'LIKE' val = f"'%{val_text}%'" else: if not is_numeric and not is_quoted and ct == 'TEXT' and self.db_id: grounded_val = self.db_lookup.semantic_match(val_text, self.db_id, t, c) val = f"'{grounded_val}'" else: val = val_text if is_numeric else f"'{val_text}'" if op == 'BETWEEN': remaining = question[question.lower().find(val_text.lower()) + len(val_text):] next_num = re.search(r'\b(\d+(?:\.\d+)?)\b', remaining) if next_num: next_val = next_num.group(1) val = f"{val_text} AND {next_val}" skip_vals.add(next_val) else: op = '=' intent = 'EXCLUDE' if op == '!=' and ct == 'TEXT' else ('INCLUDE' if op in ('=', 'LIKE') and ct == 'TEXT' else 'COMPARE') filters.append((t, c, op, val, intent)) break return filters def _build_boolean_column_registry(self): boolean_cols = set() for table_name, columns in self.schema.tables.items(): for col_name, col_info in columns.items(): col_type = col_info.get('type', '').upper() if isinstance(col_info, dict) else '' # Strategy 1: explicit BOOLEAN type if 'BOOLEAN' in col_type or 'BOOL' in col_type: boolean_cols.add((table_name, col_name)) # Strategy 2: naming convention patterns elif re.match(r'^is_|^has_|^hlc_is_', col_name): boolean_cols.add((table_name, col_name)) # Strategy 3: enum-like columns — low cardinality status/type/platform cols # These should filter, not display elif re.match(r'.+_(status|type|platform|period|provider)$', col_name): boolean_cols.add((table_name, col_name)) # Strategy 4: day-of-week, enabled flag elif col_name in ('enabled', 'status', 'platform'): boolean_cols.add((table_name, col_name)) return boolean_cols def extract_intent(self, question, top_k_tables=3, top_k_cols=6, table_margin=2.0, col_margin=2.0, debug=False): # PATCH: Derby execution-plan queries must stay in performance_schema + sql_plan only. # With 12 tables in derby_system, the report/template tables score within margin # for phrasing like "get the execution plan for digest X" because words like # "execution", "plan", "time" match scheduler/report column descriptions. # We short-circuit the neural table linker entirely for this intent. q_lower_pre = question.lower() # _force_tables = None # if getattr(self, 'db_id', None) == 'derby_system': # plan_kws = ['execution plan', 'sql plan', 'explain plan', 'query plan', # 'plain text plan', 'plan for digest', 'plan for the digest'] # has_plan = any(kw in q_lower_pre for kw in plan_kws) # has_digest_ref = bool(re.search(r'\bdigest\b', q_lower_pre)) and 'plan' in q_lower_pre # if has_plan or has_digest_ref: # _force_tables = ['performance_schema', 'sql_plan'] _force_tables = None _injected_digest_filter = None if getattr(self, 'db_id', None) == 'derby_system': plan_kws = ['execution plan', 'sql plan', 'explain plan', 'query plan', 'plain text plan', 'plan for digest', 'plan for the digest'] has_plan = any(kw in q_lower_pre for kw in plan_kws) has_digest_ref = bool(re.search(r'\bdigest\b', q_lower_pre)) and 'plan' in q_lower_pre if has_plan or has_digest_ref: _force_tables = ['performance_schema', 'sql_plan'] # Also extract digest value if present but unquoted, so value binder # doesn't miss it (bind_values only catches quoted or numeric tokens). digest_val_match = re.search( r"\bdigest\s+['\"]?([\w\-]+--[\w\-]+)['\"]?", q_lower_pre ) if digest_val_match: _injected_digest_filter = digest_val_match.group(1) all_tables = list(self.schema.tables.keys()) table_pairs = [(question, self._serialize_table(t)) for t in all_tables] all_tables = [t for t in self.schema.tables.keys()] if self.db_id == 'influx_system': # If we are in Influx, the ONLY valid tables are metrics valid_tables = ['sys_target_alerts', 'target_based_time_series_data'] else: # If we are in Derby, exclude the Influx-specific tables valid_tables = [t for t in self.schema.tables.keys() if t not in ['sys_target_alerts', 'target_based_time_series_data']] # Filter the neural network's search space table_pairs = [(question, self._serialize_table(t)) for t in valid_tables] table_scores = self.table_linker.predict(table_pairs, show_progress_bar=False) sorted_table_indices = table_scores.argsort()[::-1] best_table_score = table_scores[sorted_table_indices[0]] active_tables = [] for idx in sorted_table_indices[:top_k_tables]: score = table_scores[idx] if score >= (best_table_score - table_margin): # active_tables.append(all_tables[idx]) active_tables.append(valid_tables[idx]) # Apply force-lock after neural scoring (preserves scores for col linker) # if _force_tables is not None: # active_tables = _force_tables # Apply force-lock after neural scoring (preserves scores for col linker) if _force_tables is not None: active_tables = _force_tables # Store injected digest for generate_sql to consume via engine attribute self._injected_digest_filter = getattr(self, '_injected_digest_filter', None) \ if not hasattr(self, '_injected_digest_filter') else _injected_digest_filter self._injected_digest_filter = _injected_digest_filter # ── 1. The Smart NLP Lexical Anchor Extraction ── doc = self.nlp(question.lower()) # Extract base lemmas ONLY for meaningful parts of speech. meaningful_lemmas = { token.lemma_ for token in doc if token.pos_ in ['NOUN', 'PROPN', 'ADJ'] } # Extract noun chunks for multi-word concepts noun_chunks = {chunk.text for chunk in doc.noun_chunks} # ── 2. Smart Table Rescue ── # Force-include tables if their clean lemma is explicitly a noun in the question. for t in all_tables: # Check against the raw table name AND Spider's normalized English name norm_t = getattr(self.schema, 'norm_table_names', {}).get(t, t).lower() if t in meaningful_lemmas or norm_t in meaningful_lemmas: if t not in active_tables: active_tables.append(t) if debug: print(f"[extract_intent] active_tables={active_tables}") col_hits = [] for table in active_tables: cols = [c for c in self.schema.tables[table] if c not in ('id', f"{table}_id")] if not cols: continue col_pairs = [(question, self._serialize_column(table, c)) for c in cols] col_scores = self.column_linker.predict(col_pairs, show_progress_bar=False) sorted_col_indices = col_scores.argsort()[::-1] best_col_score = col_scores[sorted_col_indices[0]] for idx in sorted_col_indices: score = col_scores[idx] if score >= (best_col_score - col_margin): col_hits.append((table, cols[idx], float(score))) # ── 3. Smart Column Rescue ── already_included = {(t, c) for t, c, s in col_hits} for table in active_tables: for col in self.schema.tables[table]: if col in ('id',) or col.endswith('_id'): continue # Fetch Spider's normalized name (e.g., "fname" -> "first name") norm_c = getattr(self.schema, 'norm_column_names', {}).get((table, col), col).lower() # We rescue the column if its normalized name is a direct lemma # OR if it's explicitly contained inside a noun chunk. is_lemma_match = norm_c in meaningful_lemmas is_chunk_match = any(norm_c in chunk for chunk in noun_chunks) if is_lemma_match or is_chunk_match: if (table, col) not in already_included: # Append with a score of 0.0 so it survives pruning but doesn't override neural top-picks col_hits.append((table, col, 0.0)) rescue_triggers = { 'enabled': ['disabled', 'inactive', 'off', 'disable'], 'status': ['running', 'stopped', 'failed', 'error', 'active', 'inactive'] } for col_name, triggers in rescue_triggers.items(): if any(w in question.lower() for w in triggers): for t in active_tables: if col_name in self.schema.tables.get(t, {}): # Check if the column is already in the list of tuples if not any(c == col_name for _, c, _ in col_hits): col_hits.append((t, col_name, 0.0)) if debug: print(f"[extract_intent] col_hits={col_hits}") filters = self.bind_values(question, active_tables, debug=debug) if filters: filter_tables = set(t for t, c, op, val, intent in filters) if len(filter_tables) == 1: sole = list(filter_tables)[0] # Only keep tables reachable from sole_table reachable = set(nx.single_source_shortest_path(self.schema.graph, sole).keys()) active_tables = [t for t in active_tables if t in reachable] # Rescue any tables discovered by the DB value matcher back into active_tables for t, c, op, val, intent in filters: if t not in active_tables: active_tables.append(t) if debug: print(f"[extract_intent] filters={filters}") return active_tables, col_hits, filters # ========================================================= # 3. SQL COMPOSER # ========================================================= def generate_sql(question, engine, injected_tables=None, injected_col_hits=None, injected_filters=None, debug=True): # ── MEMORY INJECTION ── if injected_tables is not None and injected_filters is not None: active_tables = injected_tables col_hits = injected_col_hits or [] filters = injected_filters else: active_tables, col_hits, filters = engine.extract_intent(question, debug=debug) injected = getattr(engine, '_injected_digest_filter', None) if injected and not any('digest' in w for _, c, _, _, _ in filters for w in [c]): filters = list(filters) + [('performance_schema', 'digest', '=', f"'{injected}'", 'INCLUDE')] engine._injected_digest_filter = None for t, c, op, val, intent in filters: if t not in active_tables: active_tables.append(t) # ── 1. BOOLEAN COLUMN PROMOTION (POLARITY FIX) ── promoted_filter_cols = set() new_filters = list(filters) q_lower = question.lower() # FATAL FLAW FIX: We must only promote booleans belonging to the absolute core tables. core_table = active_tables[0] if active_tables else None for t, c, score in col_hits: # Strict enforcement: If this column doesn't belong to the main subject table, ignore it. if t != core_table: continue col_type_raw = engine.schema.tables.get(t, {}).get(c, 'TEXT') is_boolean_col = ( 'BOOL' in str(col_type_raw).upper() or c.startswith('is_') or c.startswith('has_') or c.startswith('hlc_is_') or c in ('enabled', 'status', 'platform') or c.endswith('_status') or c.endswith('_period') or c.endswith('_type') or c.endswith('_provider') or c.endswith('_day_of_week') # FIX 2: Add day_of_week to the enum list ) if not is_boolean_col: continue if any(fc == c and ft == t for ft, fc, _, _, _ in filters): continue # 1. THE NEURAL GUARDRAIL (Relaxed for antonyms) col_words = [w for w in c.lower().split('_') if w not in ('is', 'has', 'hlc')] lexical_extensions = set(col_words) if 'enabled' in col_words: lexical_extensions.update(['disabled', 'active', 'inactive']) if 'status' in col_words: lexical_extensions.update(['running', 'stopped', 'failed', 'error', 'active', 'inactive']) if 'dynamic' in col_words: lexical_extensions.update(['static']) has_lexical_match = any(w in q_lower for w in lexical_extensions) # if score < -5.0 and not has_lexical_match: # Relaxed slightly more to let template_type through # continue # THE FIX: Require score >= 0.0 unless there is explicit lexical proof in the prompt if score < 0.0 and not has_lexical_match: continue candidate_values = engine.db_lookup.get_values(engine.db_id, t, c) val_set = set(str(v).lower() for v in candidate_values) if candidate_values else set() universal_negation = [ 'not', 'no', 'false', 'without', 'excluding', 'exclude', 'disabled', 'inactive', 'stopped', 'off', 'never', 'fails', 'local', 'on-prem', 'on-premise' ] has_negation = any(re.search(rf'\b{w}\b', q_lower) for w in universal_negation) best_value = None # 2. BULLETPROOF POLARITY ROUTING # FIX 1: We no longer rely on val_set for columns starting with is/has. We force the structure. is_true_false_col = c.startswith('is_') or c.startswith('has_') or c.startswith('hlc_is_') if is_true_false_col: if 'demo' in c.lower(): if has_negation: best_value = 'false' elif 'demo' in q_lower: best_value = 'true' else: best_value = 'false' elif has_negation: best_value = 'false' else: best_value = 'true' # # 2. STATUS ENUM ROUTING # elif 'status' in c.lower() or c == 'enabled': # if has_negation or 'inactive' in q_lower or 'disabled' in q_lower: # best_value = 'Inactive' if 'active' in val_set or 'Inactive' in val_set else 'false' # elif any(w in q_lower for w in ['active', 'running', 'enabled', 'working']): # best_value = 'Active' if 'active' in val_set or 'Active' in val_set else 'true' # else: # continue # # 3. DERBY TARGET STATUS ROUTING # elif 'running' in val_set or 'stopped' in val_set: # if any(w in q_lower for w in ['error', 'failed', 'failing']): # best_value = 'Error' # elif any(w in q_lower for w in ['stopped', 'down', 'inactive', 'not running']): # best_value = 'Stopped' # elif any(w in q_lower for w in ['running', 'active', 'up', 'working']): # best_value = 'Running' # else: # continue # 2. DERBY TARGET STATUS ROUTING (Check this FIRST!) elif 'running' in val_set or 'stopped' in val_set: if any(w in q_lower for w in ['error', 'failed', 'failing']): best_value = 'Error' elif any(w in q_lower for w in ['stopped', 'down', 'inactive', 'not running']): best_value = 'Stopped' elif any(w in q_lower for w in ['running', 'active', 'up', 'working']): best_value = 'Running' else: continue # 3. GENERIC STATUS ENUM ROUTING (Fallback) elif 'status' in c.lower() or c == 'enabled': if has_negation or 'inactive' in q_lower or 'disabled' in q_lower: best_value = 'Inactive' if 'active' in val_set or 'Inactive' in val_set else 'false' elif any(w in q_lower for w in ['active', 'running', 'enabled', 'working']): best_value = 'Active' if 'active' in val_set or 'Active' in val_set else 'true' else: continue # 4. FALLBACK FOR OTHER ENUMS (e.g. DAILY, Dashboard,R ENUMS MONDAY) if best_value is None and candidate_values: # BYPASS NEURAL NET: If the exact DB value is in the question, just use it! for val in candidate_values: # We check lower() to match, but we use the original case 'val' for the SQL if str(val).lower() in q_lower: best_value = val break # If the pure string match failed, ONLY THEN ask the neural net to guess if best_value is None: pairs = [(question, f"{engine._serialize_column(t, c)} | value: {val}") for val in candidate_values] scores = engine.value_linker.predict(pairs, show_progress_bar=False) best_idx = int(scores.argmax()) best_score = float(scores[best_idx]) # Require a positive score to prevent wild hallucinations like "DASHBOARD" for "FinOps" if best_score > 0.0: best_value = candidate_values[best_idx] if best_value is not None: new_filters.append((t, c, '=', f"'{best_value}'", 'INCLUDE')) promoted_filter_cols.add((t, c)) # 3. THE "CLOUD" DOUBLE-BIND CLEANUP cleaned_filters = [] for ft, fc, fop, fval, fintent in new_filters: if (ft, fc) in promoted_filter_cols: cleaned_filters.append((ft, fc, fop, fval, fintent)) continue val_clean = fval.replace("'", "").replace("%", "").lower() is_redundant = False for pt, pc in promoted_filter_cols: if pt == ft and val_clean in pc.lower() and len(val_clean) > 3: is_redundant = True break if not is_redundant: cleaned_filters.append((ft, fc, fop, fval, fintent)) filters = cleaned_filters col_hits = [(t, c, s) for t, c, s in col_hits if (t, c) not in promoted_filter_cols] sorted_hits = sorted(col_hits, key=lambda x: -x[2]) # ── 2. PREPARE THE FROM SLOT ── # ── 2. PREPARE THE FROM SLOT (GENERALIZED GRAPH PRUNING) ── path_nodes = set(active_tables) if len(active_tables) > 1: root = active_tables[0] valid_tables = [root] path_nodes = {root} for tgt in active_tables[1:]: try: path = nx.shortest_path(engine.schema.graph, root, tgt) path_nodes.update(path) valid_tables.append(tgt) except nx.NetworkXNoPath: # Silently drop hallucinated tables that can't be joined pass active_tables = valid_tables ordered_nodes = list(path_nodes) else: ordered_nodes = active_tables start_node = ordered_nodes[0] if ordered_nodes else active_tables[0] table_aliases = {tbl: f"T{i+1}" for i, tbl in enumerate(ordered_nodes)} subgraph = engine.schema.graph.subgraph(ordered_nodes) use_aliases = len(ordered_nodes) > 1 if use_aliases: join_clauses = [f"{start_node} AS {table_aliases[start_node]}"] try: for u, v in nx.bfs_edges(subgraph, source=start_node): edge_data = engine.schema.graph.get_edge_data(u, v) if edge_data: on_cond = edge_data['on'] left_part, right_part = on_cond.split(' = ') t1, c1 = left_part.strip().split('.') t2, c2 = right_part.strip().split('.') u_col, v_col = (c1, c2) if t1 == u else (c2, c1) join_clauses.append(f"JOIN {v} AS {table_aliases[v]} ON {table_aliases[u]}.{u_col} = {table_aliases[v]}.{v_col}") except Exception: pass joins_sql = ' ' + ' '.join(join_clauses) else: joins_sql = f" {start_node}" def apply_alias(col_str): if "." in col_str: t_name, c_name = col_str.split('.') if not use_aliases: return c_name if t_name in table_aliases: return f"{table_aliases[t_name]}.{c_name}" return col_str # ── 3. AST SLOTS INITIALIZATION ── slots = {"DISTINCT": False, "SELECT": [], "FROM": joins_sql, "WHERE": [], "GROUP BY": [], "HAVING": "", "ORDER BY": "", "LIMIT": ""} skeleton_label, skeleton_name = engine.skeleton.predict(question) aggs = engine.detect_aggregation(question) if re.search(r'\bper\s+\w+\b', q_lower) and aggs: skeleton_label = GROUPED_AGG if skeleton_label == EXCEPT_QUERY and aggs: skeleton_label = PURE_AGG t_main = next((t for t in active_tables), None) t_sub = next((t for t in active_tables if t != t_main), None) if t_main and t_sub: main_pk = engine.schema.primary_keys.get(t_main) or next((c for c in engine.schema.tables.get(t_main, {}) if 'id' in c.lower()), None) fk_in_sub = next((c for c in engine.schema.tables.get(t_sub, {}) if c == main_pk or (t_main.rstrip('s') in c and 'id' in c)), main_pk) if main_pk and fk_in_sub: sub_filters = [f"{c} {op} {val}" for t, c, op, val, intent in filters if t == t_sub] sub_where = f" WHERE {' AND '.join(sub_filters)}" if sub_filters else "" slots["WHERE"].append(f"{main_pk} NOT IN (SELECT {fk_in_sub} FROM {t_sub}{sub_where})") slots["FROM"] = f" {t_main}" use_aliases = False table_aliases = {t_main: t_main} # ── 4. FILL THE WHERE SLOT ── filter_dict = {} for t, c, op, val, intent in filters: col_str = apply_alias(f"{t}.{c}") if col_str not in filter_dict: filter_dict[col_str] = [] filter_dict[col_str].append(f"{col_str} {op} {val}") intersect_vals = [] except_outer_table = next((t for t in active_tables), None) if skeleton_label == EXCEPT_QUERY else None for col_str, conditions in filter_dict.items(): if except_outer_table is not None: filter_table = next((t for t, c, op, val, intent in filters if apply_alias(f"{t}.{c}") == col_str), None) if filter_table and filter_table != except_outer_table: continue if len(conditions) > 1 and ("both" in q_lower or "and" in q_lower): intersect_vals = conditions elif len(conditions) > 1: slots["WHERE"].append("(" + " OR ".join(conditions) + ")") else: slots["WHERE"].append(conditions[0]) if "average" in q_lower and any(kw in q_lower for kw in ["above", "older", "greater", "more", "higher"]): num_col_raw = next((f"{t}.{c}" for t, c, s in sorted_hits if engine.schema.tables[t].get(c) == 'NUM'), None) if num_col_raw: tbl, col = num_col_raw.split('.') slots["WHERE"].append(f"{apply_alias(num_col_raw)} > (SELECT avg({col}) FROM {tbl})") skeleton_label = SIMPLE elif "average" in q_lower and any(kw in q_lower for kw in ["below", "smaller", "less", "under", "lower"]): num_col_raw = next((f"{t}.{c}" for t, c, s in sorted_hits if engine.schema.tables[t].get(c) == 'NUM'), None) if num_col_raw: tbl, col = num_col_raw.split('.') slots["WHERE"].append(f"{apply_alias(num_col_raw)} < (SELECT avg({col}) FROM {tbl})") skeleton_label = SIMPLE # ── 5. PREPARE COLUMNS & SHOW ALL FIX ── structural_cols = set() for u, v, data in engine.schema.graph.edges(data=True): for part in data.get('on', '').split(' = '): if '.' in part.strip(): structural_cols.add(tuple(part.strip().split('.'))) for tbl, pk in engine.schema.primary_keys.items(): structural_cols.add((tbl, pk)) q_tokens = set(re.findall(r'\w+', q_lower)) candidates_with_scores = [(t, c, s) for t, c, s in sorted_hits if (t, c) not in structural_cols or len(active_tables) == 1 or c.lower() in q_tokens] if candidates_with_scores: best_score = candidates_with_scores[0][2] if best_score < 0: candidates_with_scores = [candidates_with_scores[0]] else: safe_candidates = [] for t, c, s in candidates_with_scores: if s >= best_score - 2.0: safe_candidates.append((t, c, s)) elif s == 0.0 and (c.lower() in q_tokens or c.lower() + 's' in q_tokens or c.lower().rstrip('s') in q_tokens): safe_candidates.append((t, c, s)) candidates_with_scores = safe_candidates else: candidates_with_scores = sorted_hits[:1] # candidates = [(t, c) for t, c, s in candidates_with_scores] # ORPHANED COLUMN KILL SWITCH: Only allow columns if their table survived the JOIN graph candidates = [(t, c) for t, c, s in candidates_with_scores if t in ordered_nodes] # Fallback if the killer switch wipes everything out if not candidates: candidates = [(start_node, 'name')] if 'name' in engine.schema.tables.get(start_node, {}) else [(start_node, list(engine.schema.tables.get(start_node, {}).keys())[0])] math_words = {'average', 'max', 'min', 'sum', 'highest', 'lowest', 'maximum', 'minimum'} for t, c in list(candidates): if c.lower() in math_words: other_nums = [col for col, ctype in engine.schema.tables.get(t, {}).items() if ctype == 'NUM' and col.lower() not in math_words] if other_nums and aggs: candidates.remove((t, c)) has_name_col = any('name' in c.lower() for t, c in candidates) for t in active_tables: for c in engine.schema.tables[t]: if (t, c) in candidates or c == 'id' or c.endswith('_id'): continue c_clean = c.lower() if c_clean in ['name', 'pettype', 'type', 'country'] and c_clean in q_tokens: if c_clean == 'name' and has_name_col: continue candidates.append((t, c)) filter_cols = set((t, c) for t, c, op, val, intent in filters) if len(candidates) > 1: pure = [x for x in candidates if x not in filter_cols] if pure: candidates = pure else: for t in active_tables: for fallback in ['name', 'title', 'integration_name']: if fallback in engine.schema.tables.get(t, {}): candidates = [(t, fallback)] break else: continue break elif len(candidates) == 1 and candidates[0] in filter_cols: t = candidates[0][0] display_fallback = next((c for c in engine.schema.tables.get(t, {}) if c not in ('id',) and not c.endswith('_id') and c not in {fc for _, fc, _, _, _ in filters}), None) if display_fallback: candidates = [(t, display_fallback)] if re.search(r'\bid\b', q_lower): for t in active_tables: pk = engine.schema.primary_keys.get(t) if pk and (t, pk) not in filter_cols and t not in [ft for ft, fc, fo, fv, fi in filters]: candidates = [(t, pk)] + [x for x in candidates if x != (t, pk)] break # --- 🤖 CHATBOT CONVERSATIONAL BYPASS --- # Phrases your coworker uses that imply they want the "whole row" chatbot_phrases = [ "give me", "show me", "tell me about", "get me", "profile for", "information on", "info for", "details of", "all target", "complete profile" ] # Standard list phrases list_phrases = ["list all", "show all", "get all", "list demo", "show demo", "list reports", "list targets"] is_asking_for_everything = any(p in q_lower for p in chatbot_phrases + list_phrases) # Keywords that suggest the user wants a SPECIFIC calculation (which would break SELECT *) # We add "status" and "platform" here so "What is the status" doesn't trigger SELECT * explicit_math_or_col = any(re.search(rf'\b{w}\b', q_lower) for w in [ "names", "ids", "types", "platforms", "status", "text", "plan", "average", "count", "how many", "total", "avg", "sum" ]) # If they use chatbot phrasing and didn't ask for a specific math/column, give them SELECT * if is_asking_for_everything and not explicit_math_or_col and not aggs: top_cols = ["*"] display_col = "*" else: top_cols = [apply_alias(f"{t}.{c}") for t, c in candidates] top_cols.sort(key=lambda x: 0 if 'name' in x.lower() else 1) display_col = top_cols[0] if top_cols else "*" def get_group_col(disp_col): if disp_col == "*": return disp_col if "." in disp_col: alias = disp_col.split('.')[0] t_name = next((k for k, v in table_aliases.items() if v == alias), None) if use_aliases else alias else: t_name = start_node alias = start_node if not use_aliases else table_aliases[start_node] if t_name and 'name' in disp_col.lower(): pk = f"{t_name}_id" if f"{t_name}_id" in engine.schema.tables[t_name] else "id" if pk in engine.schema.tables[t_name]: return f"{alias}.{pk}" if use_aliases else pk return disp_col sort_col_raw, sort_dir = engine.resolve_ordering(question, active_tables) sort_col = apply_alias(sort_col_raw) if sort_col_raw else None is_explicit_order = any(k in q_lower for k in ["ordered by", "order by", "sorted by", "descending", "ascending"]) top_n_match = re.search(r'\b(?:top|bottom|worst|best|highest|lowest)\s+(\d+)\b', q_lower) forced_limit = int(top_n_match.group(1)) if top_n_match else None # ---> ROBUST SKELETON DOWNGRADE <--- # If the AI predicts SUPERLATIVE but there is nothing to sort by, no limit requested, # and no explicit count requested, then the classification is a false positive. if skeleton_label == SUPERLATIVE and not sort_col_raw and not top_n_match: is_count_select = any(q_lower.startswith(x) for x in ['how many', 'find the number', 'what is the number', 'number of']) if not is_count_select: skeleton_label = SIMPLE if sort_col_raw and skeleton_label not in [GROUPED_AGG, SUPERLATIVE, EXCEPT_QUERY] and not top_n_match: if not aggs and not is_explicit_order: if any(w in q_lower for w in ['youngest', 'oldest', 'most', 'least', 'highest', 'lowest', 'maximum', 'minimum', 'best', 'worst', 'largest', 'smallest']): skeleton_label = SUPERLATIVE # Down-grade false-positive PURE_AGG intents if skeleton_label == PURE_AGG and not aggs: is_count_word = any(w in q_lower for w in ['how many', 'count', 'total', 'number of']) if not is_count_word: skeleton_label = SIMPLE # ── 6. POPULATE AST BY NEURAL SKELETON ── if skeleton_label == EXCEPT_QUERY or (any(kw in q_lower.split() for kw in ["without"]) or "did not" in q_lower): t_main = active_tables[0] for col_str in top_cols: if col_str == "*": continue prefix = col_str.split('.')[0] if '.' in col_str else start_node cand_t = next((k for k, v in table_aliases.items() if v == prefix), prefix) if cand_t in active_tables: t_main = cand_t break t_sub = next((t for t in active_tables if t != t_main and engine.schema.primary_keys.get(t_main) in engine.schema.tables.get(t, {})), next((t for t in active_tables if t != t_main), None)) slots["FROM"] = f" {t_main} AS {table_aliases[t_main]}" if use_aliases else f" {t_main}" main_pk = engine.schema.primary_keys.get(t_main) or next((c for c in engine.schema.tables.get(t_main, {}) if 'id' in c.lower()), list(engine.schema.tables.get(t_main, {}).keys())[0]) if t_sub: fk_in_sub = next((c for c in engine.schema.tables.get(t_sub, {}) if c == main_pk or t_main in c), main_pk) target_alias = table_aliases[t_main] if use_aliases else t_main inner_wheres = [] outer_wheres = [] bridge_filter_tables = set(t for t, c, op, val, intent in filters if t != t_main and t != t_sub) has_explicit_exclude = any(i == 'EXCLUDE' for _, _, _, _, i in filters) for t, c, op, val, intent in filters: if t == t_sub or t in bridge_filter_tables: if intent == 'EXCLUDE': inner_wheres.append(f"{c} = {val}") elif intent == 'INCLUDE' and (has_explicit_exclude or t == t_main): outer_wheres.append((t, c, op, val)) else: inner_wheres.append(f"{c} {op} {val}") if outer_wheres: slots["FROM"] = joins_sql use_aliases = len(ordered_nodes) > 1 table_aliases = {tbl: f"T{i+1}" for i, tbl in enumerate(ordered_nodes)} for t, c, op, val in outer_wheres: slots["WHERE"].append(f"{apply_alias(f'{t}.{c}')} {op} {val}") inner_where_str = f" WHERE {' AND '.join(inner_wheres)}" if inner_wheres else "" if bridge_filter_tables: bridge_t = list(bridge_filter_tables)[0] edge = engine.schema.graph.get_edge_data(t_sub, bridge_t) if edge: sub_join = f"{t_sub} JOIN {bridge_t} ON {edge['on']}" slots["WHERE"].append(f"{target_alias}.{main_pk} NOT IN (SELECT {fk_in_sub} FROM {sub_join}{inner_where_str})") else: slots["WHERE"].append(f"{target_alias}.{main_pk} NOT IN (SELECT {fk_in_sub} FROM {t_sub}{inner_where_str})") else: slots["WHERE"].append(f"{target_alias}.{main_pk} NOT IN (SELECT {fk_in_sub} FROM {t_sub}{inner_where_str})") slots["SELECT"] = top_cols if top_cols else ["*"] elif skeleton_label == SUPERLATIVE: order_dir_sql = " DESC" if sort_dir == 'DESC' else "" is_count_select = any(q_lower.startswith(x) for x in ['how many', 'find the number', 'what is the number', 'number of']) if is_count_select and len(active_tables) >= 2 and sort_col_raw: sort_t = sort_col_raw.split('.')[0] if '.' in sort_col_raw else start_node t_inner = next((k for k, v in table_aliases.items() if v == sort_t), sort_t) if use_aliases else sort_t t_outer = next((t for t in active_tables if t != t_inner), active_tables[0]) sort_fk = next((c for c in engine.schema.tables.get(t_outer, {}) if t_inner in c or c == f"{t_inner}_id"), None) inner_pk = next((c for c in engine.schema.tables.get(t_inner, {}) if c == 'id' or c.endswith('_id')), 'id') if sort_fk: slots["SELECT"] = ["count(*)"] slots["FROM"] = f" {t_outer}" if not use_aliases else f" {t_outer} AS {table_aliases[t_outer]}" target_alias = table_aliases[t_outer] if use_aliases else t_outer inner_col = sort_col_raw.split('.')[-1] slots["WHERE"].append(f"{target_alias}.{sort_fk} = (SELECT {inner_pk} FROM {t_inner} ORDER BY {inner_col}{order_dir_sql} LIMIT 1)") else: slots["ORDER BY"] = f"ORDER BY {sort_col}{order_dir_sql}" if not is_explicit_order: slots["LIMIT"] = "LIMIT 1" slots["SELECT"].append("count(*)") else: slots["ORDER BY"] = f"ORDER BY {sort_col}{order_dir_sql}" if sort_col else "ORDER BY count(*) DESC" if not is_explicit_order: slots["LIMIT"] = "LIMIT 1" if is_count_select: select_cols = ["count(*)"] elif top_cols: select_cols = [c for c in top_cols if c != sort_col or is_explicit_order] if not select_cols: select_cols = top_cols else: select_cols = ["*"] for sc in select_cols: slots["SELECT"].append(sc) d_col = select_cols[0] if select_cols else "*" if "count(*)" in slots["ORDER BY"].lower() and d_col != "count(*)": slots["GROUP BY"].append(get_group_col(d_col if d_col != "*" and "count" not in d_col else top_cols[0])) elif skeleton_label == GROUPED_AGG: group_candidates = [col for col in top_cols if 'id' not in col.lower() and col != '*'] if not group_candidates: group_candidates = [col for col in top_cols if col != '*'] if not group_candidates and display_col != '*': group_candidates = [display_col] elif not group_candidates: group_candidates = [apply_alias(f"{start_node}.name")] def get_col_type(c): q_lower = question.lower() prefix = c.split('.')[0] if '.' in c else start_node t = next((k for k, v in table_aliases.items() if v == prefix), prefix) return engine.schema.tables.get(t, {}).get(c.split('.')[-1], 'TEXT') numeric_cols = [c for c in group_candidates if get_col_type(c) == 'NUM' and c.split('.')[-1] not in [x[1] for x in structural_cols]] categ_cols = [c for c in group_candidates if c not in numeric_cols] if aggs and numeric_cols: for agg in aggs: for nc in numeric_cols: slots["SELECT"].append(f"{agg.lower()}({nc})") group_by_col = categ_cols[0] if categ_cols else group_candidates[0] slots["SELECT"].extend(categ_cols if categ_cols else group_candidates) else: slots["SELECT"].extend(group_candidates) group_by_col = group_candidates[0] slots["GROUP BY"].append(get_group_col(group_by_col)) if any(k in q_lower for k in ['how many', 'number of', 'count']): slots["SELECT"].append("count(*)") elif skeleton_label == PURE_AGG: use_count_distinct = any(k in q_lower for k in ["distinct", "different", "unique", "various"]) count_str = f"count(DISTINCT {display_col})" if use_count_distinct and display_col != '*' else "count(*)" if not aggs: slots["SELECT"].append(count_str) else: target_col = sort_col if sort_col and sort_col != 'count(*)' else display_col has_literal_avg = any('average' in engine.schema.tables[t] for t in active_tables) for agg in aggs: if agg == "COUNT": slots["SELECT"].append(count_str) else: if agg == "AVG" and has_literal_avg and "average of" not in q_lower: avg_raw = next(f"{t}.average" for t in active_tables if 'average' in engine.schema.tables[t]) slots["SELECT"].append(apply_alias(avg_raw)) else: slots["SELECT"].append(f"{agg.lower()}({target_col})") for col in top_cols: if col != target_col and col not in slots["SELECT"] and col != '*' and col.split('.')[-1] in q_lower: slots["SELECT"].append(col) else: if any(k in q_lower for k in ["distinct", "unique", "different"]): slots["DISTINCT"] = True for c in top_cols: slots["SELECT"].append(c) if sort_col and any(k in q_lower for k in ["order by", "sorted by", "descending", "ascending"]): order_dir_sql = " DESC" if sort_dir == 'DESC' else "" slots["ORDER BY"] = f"ORDER BY {sort_col}{order_dir_sql}" if top_n_match: slots["ORDER BY"] = f"ORDER BY {sort_col if sort_col else 'data_value_double'} DESC" slots["LIMIT"] = f"LIMIT {forced_limit}" having_match = re.search(r'(more than|at least|fewer than|less than|greater than|over|under)\s+(\d+)', q_lower) if having_match and slots["GROUP BY"]: op_map = {'more than': '>', 'over': '>', 'greater than': '>', 'at least': '>=', 'fewer than': '<', 'less than': '<', 'under': '<'} op = op_map.get(having_match.group(1), '>') num = having_match.group(2) slots["HAVING"] = f"HAVING count(*) {op} {num}" # ── 7. DIALECT ENFORCEMENT ENGINE ── if getattr(engine, 'db_id', None) == 'influx_system': active_tables = ['sys_target_alerts'] slots["FROM"] = " sys_target_alerts" if not any('is_demo_alert' in w for w in slots["WHERE"]): slots["WHERE"].append("is_demo_alert != 'true'") is_bottom = any(w in q_lower for w in ['bottom', 'least', 'smallest', 'lowest']) order_dir = "ASC" if is_bottom else "DESC" if aggs: # If it's just asking for top/bottom anomalies (Superlative), DO NOT group. # We must return the raw 4 columns so the orchestrator can extract the digest! if skeleton_label == SUPERLATIVE or top_n_match or any(w in q_lower for w in ['top', 'bottom', 'worst']): slots["SELECT"] = ['target', 'metric', 'data_value_double', 'time'] slots["GROUP BY"] = [] slots["ORDER BY"] = f"ORDER BY data_value_double {order_dir}" slots["LIMIT"] = f"LIMIT {forced_limit}" if forced_limit else "LIMIT 5" else: # True aggregations (like "average by target") that don't feed into a multi-step digest lookup new_select = ['target'] for agg in aggs: if agg in ('MAX', 'MIN', 'AVG', 'SUM'): new_select.append(f"{agg.lower()}(data_value_double)") elif agg == 'COUNT': new_select.append("count(*)") slots["SELECT"] = new_select slots["GROUP BY"] = ['target'] if top_n_match: slots["ORDER BY"] = f"ORDER BY {new_select[1]} {order_dir}" slots["LIMIT"] = f"LIMIT {forced_limit}" if forced_limit else "LIMIT 5" else: slots["SELECT"] = [re.sub(r'\bT\d+\.', '', c) for c in slots["SELECT"]] if "*" in slots["SELECT"] and len(slots["SELECT"]) > 1: slots["SELECT"].remove("*") INFLUX_FIXED_COLS = {'target', 'metric', 'data_value_double', 'time', 'is_demo_alert', 'owner_id', 'templatename', 'confidencescore', 'measurement_value', 'data_value_long'} neural_pick = [c for c in slots["SELECT"] if c.lower() not in INFLUX_FIXED_COLS and c != '*'] where_cols = [c.split('.')[-1] for c in filter_dict.keys()] # ---> ORCHESTRATOR SAFETY LOCK <--- # The python orchestrator hardcodes indices 0, 1, 2, 3. We MUST guarantee they exist in this exact order. final_select = ['target', 'metric', 'data_value_double', 'time'] # Append anything else the user explicitly asked for to the end for col in neural_pick + where_cols: if col not in final_select and col != '*': final_select.append(col) slots["SELECT"] = final_select slots["GROUP BY"] = [] if top_n_match or skeleton_label == SUPERLATIVE or any(w in q_lower for w in ['top', 'bottom', 'worst']): slots["ORDER BY"] = f"ORDER BY data_value_double {order_dir}" slots["LIMIT"] = f"LIMIT {forced_limit}" if forced_limit else "LIMIT 5" elif getattr(engine, 'db_id', None) == 'derby_system': is_plan = any('plan' in c.lower() for c in slots["SELECT"]) has_digest = any('digest' in w.lower() for w in slots["WHERE"]) or 'digest' in q_lower if is_plan and has_digest: slots["FROM"] = " performance_schema AS T1 JOIN sql_plan AS T2 ON T1.sql_plan_id = T2.id" plan_cols = {'sql_plan', 'sql_plain_text_plan'} new_select = [] for c in slots["SELECT"]: if c == '*': new_select.append('*') continue bare = re.sub(r'\bT\d+\.', '', c) if bare in plan_cols: new_select.append(f"T2.{bare}") else: new_select.append(f"T1.{bare}") slots["SELECT"] = new_select # new_where = [] # for w in slots["WHERE"]: # parts = w.split(' ', 2) # if len(parts) == 3 and not w.startswith('('): new_where.append(f"T1.digest {parts[1]} {parts[2]}") # else: new_where.append(w) # slots["WHERE"] = new_where new_where = [] for w in slots["WHERE"]: # Wipe out any hallucinated table prefixes and strictly enforce T1.digest w_clean = re.sub(r'\b(?:T\d+|performance_schema|sql_plan)\.digest\b', 'T1.digest', w) w_clean = re.sub(r'(? 0;" }, # ── CLOUD PRICING ── "CLOUD_PRICING_INFO": { "anchors": [ "How does cloud pricing work?", "Show me the cost of cloud databases.", "What are the pricing lookup codes?", "Show me billing and credit costs for AWS.", "cloud pricing table", "show pricing info", "what does cloud monitoring cost", "AWS pricing", "Azure pricing", "GCP pricing", "credit cost table", "show me the pricing chart", "cloud cost breakdown" ], "clarification": "Did you mean to view the cloud database pricing chart?", "target_db": "derby_system", "sql": "SELECT DISTINCT cloud_database_provider, region, credit_cost FROM cloud_database_pricing;" }, "CLOUD_PRICING_BY_PROVIDER": { "anchors": [ "show AWS prices", "what does Azure cost", "GCP credit cost", "pricing for Amazon RDS", "show me costs by cloud provider" ], "clarification": "Did you mean to see pricing broken down by cloud provider?", "target_db": "derby_system", "sql": "SELECT cloud_database_provider, region, credit_cost, lookup_code FROM cloud_database_pricing ORDER BY cloud_database_provider;" }, # ── COLLECTORS (DBACTDC) ── "COLLECTOR_INFO": { "anchors": [ "What are collector agents?", "Show me the dbactdc instances.", "List all the monitoring guards.", "Where are the collectors installed?", "What are the SSH hosts for the collectors?", "show me collectors", "list my collectors", "which collectors are running", "are my collectors active", "collector status", "dbactdc status", "is the collector up", "show collector agents", "list monitoring agents", "show dbactdc", "get collector list", "collector health" ], "clarification": "Did you mean to list the active collector agents (DBACTDC instances)?", "target_db": "derby_system", "sql": "SELECT name, status, platform, ssh_host FROM dbactdc_instance;" }, # "COLLECTOR_STATUS": { # "anchors": [ # "which collectors are inactive", "show stopped collectors", # "any collectors with errors", "collector availability", # "are all collectors up", "which collectors are down", # "show me failed collectors", "active collectors only" # ], # "clarification": "Did you mean to check which collectors are currently active or inactive?", # "target_db": "derby_system", # "sql": "SELECT name, status, platform, ssh_host FROM dbactdc_instance;" # }, "COLLECTOR_PLATFORM": { "anchors": [ "which collectors are on windows", "linux collectors", "show collector platforms", "what OS are collectors on", "remote collectors", "local collectors" ], "clarification": "Did you mean to see what platform each collector runs on?", "target_db": "derby_system", "sql": "SELECT name, platform, is_remote, ssh_host FROM dbactdc_instance;" }, # ── INTEGRATIONS ── "EXTERNAL_INTEGRATIONS": { "anchors": [ "What external integrations are supported?", "Do you connect to Datadog or AppDynamics?", "Show me the third party tools.", "What alerts can be sent to New Relic?", "show integrations", "list integrations", "what tools are connected", "show third party connections", "what monitoring tools are linked", "do you integrate with anything", "show me all integrations", "which integrations are enabled", "external connections" ], "clarification": "Did you mean to view the configured external integrations?", "target_db": "derby_system", "sql": "SELECT integration_name, title, enabled FROM integration;" }, # "ACTIVE_INTEGRATIONS": { # "anchors": [ # "which integrations are active", "show enabled integrations", # "what tools are currently active", "active third party connections", # "which integrations are turned on" # ], # "clarification": "Did you mean to list only the currently enabled integrations?", # "target_db": "derby_system", # "sql": "SELECT integration_name, title FROM integration WHERE enabled = 'true';" # }, # ── REPORTS ── "REPORTING_CAPABILITIES": { "anchors": [ "What types of reports exist?", "Show me the report scheduling options.", "Can I get PDF emails?", "List the automated reports.", "show me reports", "list reports", "what reports do I have", "show scheduled reports", "report list", "what is being reported", "show me all reports", "get report list", "reporting schedule" ], "clarification": "Did you mean to ask about the automated report schedules and formats?", "target_db": "derby_system", "sql": "SELECT name, title, scheduler_params_period, is_email_pdf_report_enabled FROM report;" }, # "SCHEDULED_REPORTS": { # "anchors": [ # "which reports run automatically", "show daily reports", "show weekly reports", # "show monthly reports", "scheduled report list", "reports that run on schedule", # "what reports are automated", "show hourly reports" # ], # "clarification": "Did you mean to see which reports run on an automated schedule?", # "target_db": "derby_system", # "sql": "SELECT name, title, scheduler_params_period, scheduler_params_day_of_week FROM report WHERE is_scheduled = 'true';" # }, "PDF_REPORTS": { "anchors": [ "which reports send PDF emails", "show PDF report list", "what reports email PDFs", "PDF email reports" ], "clarification": "Did you mean to list reports that send PDF emails?", "target_db": "derby_system", "sql": "SELECT name, title FROM report WHERE is_email_pdf_report_enabled = 'true';" }, # ── TEMPLATES & DASHBOARDS ── "DASHBOARD_TEMPLATES": { "anchors": [ "What dashboard templates do you have?", "Show me the UI dashlets.", "What is a FinOps template?", "How are charts configured?", "show templates", "list templates", "what templates exist", "show me dashboards", "chart templates", "list dashboards", "what dashboards are available", "show all templates", "show me FinOps templates", "show dynamic templates" ], "clarification": "Did you mean to list the available dashboard and chart templates?", "target_db": "derby_system", "sql": "SELECT name, template_type, is_dynamic FROM template;" }, # "FINOPS_TEMPLATES": { # "anchors": [ # "show finops templates", "list cost templates", "FinOps dashboards", # "which templates are for cost", "cloud cost templates", "finance templates" # ], # "clarification": "Did you mean to list FinOps cost-analysis templates?", # "target_db": "derby_system", # "sql": "SELECT name, template_type, finance_targets_count FROM template WHERE template_type = 'FinOps';" # }, "DYNAMIC_TEMPLATES": { "anchors": [ "which templates are dynamic", "show dynamic dashboards", "list adaptive templates", "templates that auto update" ], "clarification": "Did you mean to list templates that dynamically adapt to available metrics?", "target_db": "derby_system", "sql": "SELECT name, template_type FROM template WHERE is_dynamic = 'true';" }, "SYSTEM_DEFAULT_TEMPLATES": { "anchors": [ "which templates are system defaults", "show default templates", "built in templates", "out of the box templates" ], "clarification": "Did you mean to list system-provided default templates?", "target_db": "derby_system", "sql": "SELECT name, template_type FROM template WHERE is_system_default = 'true';" }, # ── PERFORMANCE SCHEMA / SQL DIGESTS ── "PERFORMANCE_SCHEMA_INFO": { "anchors": [ "Where are the execution plans stored?", "How do you track SQL digests?", "Show me how SQL text is mapped.", "What is the performance schema?", "show sql digests", "list recent sql queries", "show me captured sql", "what sql has been captured", "show performance schema", "show digest table", "list sql digests", "recent query digests" ], "clarification": "Did you mean to ask how SQL queries and execution plans are stored?", "target_db": "derby_system", "sql": "SELECT digest, substr(digest_text, 1, 80) as sql_preview FROM performance_schema LIMIT 10;" }, "SQL_PLANS": { "anchors": [ "show sql execution plans", "list query plans", "show me query plans", "what execution plans are stored", "show all plans", "sql plan table" ], "clarification": "Did you mean to view stored SQL execution plans?", "target_db": "derby_system", "sql": "SELECT id, substr(sql_plain_text_plan, 1, 100) as plan_preview FROM sql_plan LIMIT 10;" }, # ── ANOMALIES / INFLUX ── "RECENT_ANOMALIES": { "anchors": [ "show recent anomalies", "what anomalies were detected", "list alerts", "show me spikes", "any bottlenecks detected", "what performance issues exist", "show me recent alerts", "latest anomalies", "current performance problems", "what is wrong with my databases", "show performance issues", "any current problems", "show spikes", "performance alerts", "what bottlenecks exist", "show me the worst issues" ], "clarification": "Did you mean to view recent performance anomalies and alerts?", "target_db": "influx_system", "sql": "SELECT target, metric, data_value_double, time FROM sys_target_alerts WHERE is_demo_alert != 'true' ORDER BY data_value_double DESC LIMIT 20;" }, # "ANOMALIES_BY_TARGET": { # "anchors": [ # "which target has the most anomalies", "show anomalies per target", # "target anomaly summary", "which database has the most issues", # "anomaly count by target", "show me problems by target" # ], # "clarification": "Did you mean to see anomaly counts grouped by target?", # "target_db": "influx_system", # "sql": "SELECT target, count(*) FROM sys_target_alerts WHERE is_demo_alert != 'true' GROUP BY target ORDER BY count(*) DESC;" # }, # ── OUT OF SCOPE / REJECT ── "OUT_OF_SCOPE": { "anchors": [ "what is the weather today", "tell me a joke", "who won the football game", "write me a poem", "what is 2 plus 2", "how do I cook pasta", "what is the capital of France", "who is the president", "recommend a movie", "what time is it", "tell me something funny", "what is the meaning of life", "help me write an email", "translate this to spanish", "what stocks should I buy" ], "clarification": "I can only answer questions about your monitored database targets, performance metrics, collectors, reports, and SQL analysis. Could you rephrase?", "target_db": None, "sql": None } } # --------------------------------------------------------- # EXHAUSTIVE TARGET DIALECT ROUTING & EXTRACTION # --------------------------------------------------------- TARGET_DIALECTS = { "SHOW_TABLES": { "mysql": "SELECT table_name FROM information_schema.tables WHERE table_schema = DATABASE();", "oracle": "SELECT table_name FROM all_tables;", "postgres": "SELECT tablename FROM pg_tables WHERE schemaname NOT IN ('pg_catalog', 'information_schema');", "sqlserver": "SELECT name AS table_name FROM sys.tables;", "snowflake": "SELECT table_name FROM INFORMATION_SCHEMA.TABLES;", "mongodb": "db.getCollectionNames();", "cassandra": "SELECT keyspace_name, table_name FROM system_schema.tables;", "redshift": "SELECT tablename FROM pg_catalog.svv_table_info;", "db2": "SELECT tabname FROM SYSCAT.TABLES WHERE tabschema NOT LIKE 'SYS%';", "sybase": "SELECT name FROM sysobjects WHERE type = 'U';", "clickhouse": "SELECT name FROM system.tables WHERE database != 'system';" }, "SHOW_COLUMNS": { "mysql": "SELECT table_name, column_name, data_type FROM information_schema.columns WHERE table_schema = DATABASE();", "oracle": "SELECT table_name, column_name, data_type FROM all_tab_columns;", "postgres": "SELECT relname AS table_name, relkind FROM pg_class WHERE relkind = 'r';", "sqlserver": "SELECT object_name(object_id) AS table_name, name AS column_name FROM sys.columns;", "snowflake": "SELECT table_name, column_name, data_type FROM INFORMATION_SCHEMA.COLUMNS;", "mongodb": "db.collection.findOne();", "cassandra": "SELECT keyspace_name, table_name, column_name, type FROM system_schema.columns;", "redshift": "SELECT tablename, column, type FROM pg_catalog.pg_table_def;", "db2": "SELECT tabname, colname, typename FROM SYSCAT.COLUMNS;", "sybase": "SELECT object_name(id) AS table_name, name AS column_name FROM syscolumns;", "clickhouse": "SELECT table, name, type FROM system.columns WHERE database != 'system';" }, "SHOW_INDEXES": { "mysql": "SELECT index_name, table_name FROM information_schema.statistics;", "oracle": "SELECT index_name, table_name FROM all_indexes;", "postgres": "SELECT indexname, tablename FROM pg_indexes;", "sqlserver": "SELECT object_name(object_id) AS table_name, name AS index_name FROM sys.indexes;", "snowflake": "SHOW INDEXES;", "mongodb": "db.collection.getIndexes();", "cassandra": "SELECT keyspace_name, table_name, index_name FROM system_schema.indexes;", "redshift": "SELECT tablename, sortkey1 FROM pg_catalog.svv_table_info;", "db2": "SELECT indname, tabname FROM SYSCAT.INDEXES;", "sybase": "SELECT name FROM sysindexes;", "clickhouse": "SELECT name, table FROM system.parts;" }, "SHOW_DIAGNOSTICS": { "mysql": "SELECT * FROM information_schema.processlist WHERE command != 'Sleep';", "oracle": "SELECT sid, serial#, status, osuser, machine FROM V$SESSION WHERE type != 'BACKGROUND';", "postgres": "SELECT name, setting, short_desc FROM pg_settings WHERE category LIKE 'Resource%';", "sqlserver": "SELECT session_id, status, cpu_time, memory_usage FROM sys.dm_exec_sessions WHERE is_user_process = 1;", "snowflake": "SELECT * FROM TABLE(INFORMATION_SCHEMA.WAREHOUSE_LOAD_HISTORY(DATEADD('hour',-1,CURRENT_TIMESTAMP())));", "mongodb": "db.serverStatus().mem;", "cassandra": "SELECT * FROM system.local;", "redshift": "SELECT * FROM pg_catalog.svv_query_metrics ORDER BY query_execution_time DESC LIMIT 10;", "db2": "SELECT * FROM SYSIBMADM.HEALTH_MON;", "sybase": "SELECT * FROM monEngine;", "clickhouse": "SELECT metric, value FROM system.metrics LIMIT 10;" }, "EXPLAIN_TEMPLATE": { "mysql": "EXPLAIN ANALYZE {query};", "oracle": "EXPLAIN PLAN FOR {query};", "postgres": "EXPLAIN (ANALYZE, BUFFERS) {query};", "sqlserver": "SET SHOWPLAN_TEXT ON; {query};", "snowflake": "EXPLAIN USING TABULAR {query};", "mongodb": "db.collection.find({query}).explain();", "clickhouse": "EXPLAIN {query};", "redshift": "EXPLAIN {query};", "db2": "EXPLAIN PLAN SELECTION FOR {query};", "sybase": "SET SHOWPLAN ON; {query};" } } dba_intents = { "TARGET_SHOW_TABLES": { "anchors": ["show me the tables for", "list tables in target", "what tables exist in", "show all tables", "database structure for", "what are the tables in"], "intent_key": "SHOW_TABLES" }, "TARGET_SHOW_COLUMNS": { "anchors": ["show columns for", "what are the columns in", "list data types for", "show table schema in", "what fields are in"], "intent_key": "SHOW_COLUMNS" }, "TARGET_SHOW_INDEXES": { "anchors": ["show indexes for", "list the indexes in", "what indexes are on", "show index statistics"], "intent_key": "SHOW_INDEXES" }, "TARGET_DIAGNOSTICS": { "anchors": ["check health of", "show active sessions in", "what are the system settings for", "is the database busy", "show metrics for", "diagnostic report for"], "intent_key": "SHOW_DIAGNOSTICS" }, "TARGET_EXPLAIN": { "anchors": ["explain why this is slow", "get execution plan for", "run explain analyze on", "show plan for", "how is this query running", "explain why the query with digest is slow", "get execution plan for digest", "explain digest"], "intent_key": "EXPLAIN_TEMPLATE" } } def parse_composite_digest(digest_string): """ Implements the 'schema_name--original_digest' rule from the training data. Returns (schema_name, original_digest) """ if not digest_string or "--" not in digest_string: return "public", digest_string # Default fallback parts = digest_string.split("--", 1) schema = parts[0] if parts[0] != "NULL" else "public" original_digest = parts[1] return schema, original_digest def normalize_db_type(raw_type): """Maps Derby's hlc_database_type to our dictionary keys.""" if not raw_type: return None raw = raw_type.lower() if 'mysql' in raw: return 'mysql' if 'postgres' in raw or 'psql' in raw: return 'postgres' if 'oracle' in raw: return 'oracle' if 'sqlserver' in raw or 'mssql' in raw: return 'sqlserver' if 'snowflake' in raw: return 'snowflake' if 'mongo' in raw: return 'mongodb' if 'cassandra' in raw: return 'cassandra' if 'redshift' in raw: return 'redshift' if 'db2' in raw: return 'db2' return raw def extract_target_context(question, derby_db_path="./spider_data/database/derby_system/derby_system.sqlite"): try: conn = sqlite3.connect(derby_db_path) # Fetching hlc_database_type (engine), not hlc_server_type (cloud/on-prem) targets = conn.execute("SELECT name, hlc_database_type FROM target WHERE name IS NOT NULL").fetchall() conn.close() except Exception as e: print(f"[WARNING] Could not connect to Derby to resolve targets: {e}") return None, None q_lower = question.lower() quoted_matches = re.findall(r"['\"]([^'\"]+)['\"]", question) for q_match in quoted_matches: for t_name, t_type in targets: if q_match.lower() == t_name.lower(): return t_name, normalize_db_type(t_type) for t_name, t_type in targets: if re.search(rf'\b{re.escape(t_name.lower())}\b', q_lower): return t_name, normalize_db_type(t_type) return None, None def extract_platform_context(question): """Deterministically extracts known OS platforms from the text.""" # Define the exact casing your database expects platforms = ['Linux', 'Windows', 'AIX', 'Solaris', 'HP-UX', 'Ubuntu', 'RHEL'] q_lower = question.lower() for p in platforms: # Use regex word boundaries (\b) so "Linux" doesn't trigger inside another word if re.search(rf'\b{p.lower()}\b', q_lower): return p return None # ---------------------------------------------------------# --------------------------------------------------------- anchor_texts = [] anchor_intents = [] # Load standard FAQ intents for intent_id, data in faq_intents.items(): for anchor in data["anchors"]: anchor_texts.append(anchor) anchor_intents.append(intent_id) # NEW: Load the DBA target intents for intent_id, data in dba_intents.items(): for anchor in data["anchors"]: anchor_texts.append(anchor) anchor_intents.append(intent_id) anchor_embeddings = router_model.encode(anchor_texts, convert_to_tensor=True) def semantic_metadata_router(user_question, embedding_model, anchor_embeddings, threshold_high=0.85, threshold_mid=0.75): q_emb = embedding_model.encode(user_question, convert_to_tensor=True) cosine_scores = util.cos_sim(q_emb, anchor_embeddings)[0] best_score_val, best_idx = torch.max(cosine_scores, dim=0) best_score = best_score_val.item() winning_intent_id = anchor_intents[best_idx] print(f" [ROUTER] Intent={winning_intent_id} Score={best_score:.2f}") # --- FIX: Prevent KeyError --- if winning_intent_id not in faq_intents: return {"status": "PASS", "sql": None, "target_db": None, "message": None} intent_data = faq_intents[winning_intent_id] if winning_intent_id == "OUT_OF_SCOPE" and best_score >= threshold_mid: return {"status": "REJECT", "message": intent_data["clarification"], "sql": None, "target_db": None} if best_score >= threshold_high: return {"status": "MATCH", "sql": intent_data["sql"], "target_db": intent_data["target_db"], "message": "Direct Match"} elif best_score >= threshold_mid: return {"status": "CLARIFY", "message": intent_data["clarification"], "sql": intent_data["sql"], "target_db": intent_data["target_db"]} else: return {"status": "PASS", "sql": None, "target_db": None, "message": None} def route_query(question): q_lower = question.lower() if any(k in q_lower for k in ['sql query text', 'actual sql', 'execution plan']): return 'derby_system' influx_triggers = [ 'anomaly', 'anomalies', 'spike', 'bottleneck', 'bottlenecks', 'alert', 'alerts', 'time series', 'worst performing', 'top sql issue', 'top sql issues', 'sql problem', 'sql problems', 'capture source', 'capture method', 'performance issue', 'performance bottleneck', 'detected for' ] if any(trigger in q_lower for trigger in influx_triggers): return 'influx_system' return 'derby_system' def word_to_num(word): word_map = { 'single': 1, 'one': 1, 'two': 2, 'three': 3, 'four': 4, 'five': 5, 'six': 6, 'seven': 7, 'eight': 8, 'nine': 9, 'ten': 10, 'dozen': 12 } return word_map.get(word.lower(), None) # ========================================================= # DIALOGUE STATE TRACKER (DST) # ========================================================= class DialogueStateTracker: def __init__(self): self.state = { "active_target": None, "active_db_type": None, "last_intent": None, "neural_tables": [], "neural_filters": [], "target_schema_cache": {} } def update_state(self, target_name=None, db_type=None, intent=None): if target_name is not None: self.state["active_target"] = target_name if db_type is not None: self.state["active_db_type"] = db_type if intent is not None: self.state["last_intent"] = intent def clear_target_context(self): # The dedicated kill switch self.state["active_target"] = None self.state["active_db_type"] = None # THE FIX: Wipe neural memory so old filters don't haunt new targets self.state["neural_tables"] = [] self.state["neural_filters"] = [] def update_neural_context(self, tables, filters): self.state["neural_tables"] = tables self.state["neural_filters"] = filters def cache_schema(self, target_name, schema_graph): self.state["target_schema_cache"][target_name] = schema_graph def get_state(self): return self.state # Initialize this in your main app execution block (e.g., Gradio/Streamlit state) chat_memory = DialogueStateTracker() import json import sqlite3 import re def format_results_as_md_table(cursor, results): if not results: return "_No data found for this query._" col_names = [desc[0] for desc in cursor.description] if len(results) == 1 and len(col_names) > 5: detail_view = "### 📋 Record Detail View\n\n| Property | Value |\n| :--- | :--- |\n" row = results[0] for col, val in zip(col_names, row): detail_view += f"| **{col}** | {str(val).replace('|', '|')} |\n" return detail_view header = "| " + " | ".join(col_names) + " |" separator = "|" + "|".join(["---" for _ in col_names]) + "|" rows = ["| " + " | ".join([str(i).replace('|', '|') for i in r]) + " |" for r in results] return f'
\n\n' + "\n".join([header, separator] + rows) + "\n\n
" def parse_composite_digest(digest_string): """ Implements the 'schema_name--original_digest' rule from the training data. Returns (schema_name, original_digest) """ if not digest_string or "--" not in digest_string: return "public", digest_string # Default fallback parts = digest_string.split("--", 1) schema = parts[0] if parts[0] != "NULL" else "public" original_digest = parts[1] return schema, original_digest import os import re import json import sqlite3 import torch from sentence_transformers import util def run_multi_step_workflow(question, engine_derby, engine_influx, chat_memory, debug=True): q_lower = question.lower() current_state = chat_memory.get_state() # 1. ALWAYS extract context from the CURRENT question first extracted_name, extracted_type = extract_target_context(question) # 2. SEMANTIC CONTEXT ESCAPE HATCH (System Mode Check) q_emb = router_model.encode(question, convert_to_tensor=True) scores = util.cos_sim(q_emb, anchor_embeddings)[0] best_idx = int(torch.argmax(scores).item()) best_intent = anchor_intents[best_idx] best_score = float(scores[best_idx]) if debug: print(f"[DEBUG ROUTER] Question: '{question}' | Picked Intent: {best_intent} | Score: {best_score:.4f}") # Context Wipe Hook: If they specifically ask for "all", clear the memory so we don't accidentally filter is_global_query = any(w in q_lower for w in [ "all target", "every target", "all the target", "list all", "across all", "all of them", "in influxdb", "in influx", "in derby", "system wide", "all databases", "overall" ]) # --- DETERMINISTIC CONTEXT SWITCHING --- if extracted_name: # User explicitly mentioned a database (e.g. clickhouse_prod) if extracted_name != current_state["active_target"]: if debug: print(f"[MEMORY] New target detected: {extracted_name}. Switching context.") chat_memory.clear_target_context() chat_memory.update_state(target_name=extracted_name, db_type=extracted_type) target_name, db_type = extracted_name, extracted_type elif is_global_query: if debug: print(f"[MEMORY] Global query detected ('all targets'). Wiping target memory to prevent accidental filtering.") chat_memory.clear_target_context() target_name, db_type = None, None else: # Fallback to RECALL memory target_name = current_state["active_target"] db_type = current_state["active_db_type"] if debug and target_name: print(f"[MEMORY] Recalled target from memory: {target_name}") # ========================================================================= # THE GATEKEEPER: INTELLIGENT MODE DETECTION # ========================================================================= # Only go "inside" the target database (CASE A) if it's a DBA intent, # a raw execution command, or explicitly asking for internal schema data. is_dba_intent = best_intent in dba_intents and best_score > 0.30 is_raw_execute = bool(re.search(r'\b(execute|run)\b', q_lower) and re.search(r'\b(select|update|insert|delete|with|call)\s', q_lower)) is_schema_introspection = any(w in q_lower for w in ['tables in', 'columns in', 'indexes in', 'schema for']) # This boolean safely gates CASE A is_in_target_mode = target_name and (is_dba_intent or is_raw_execute or is_schema_introspection) # ========================================================================= # EXECUTION PIPELINE # ========================================================================= # CASE A: Native DBA / Target Database Query (Inside the Target DB) if is_in_target_mode and target_name and db_type: if debug: print(f"[ORCHESTRATOR] 🎯 Mode: IN-TARGET Context ({target_name})") # ── PRIORITY 0.5: RAW SQL EXECUTION ── if is_raw_execute: if debug: print(f"[ORCHESTRATOR] ⚡ Action: RAW_EXECUTE triggered.") chat_memory.update_state(intent="RAW_EXECUTE") # Strategy A: Look for SQL safely wrapped in quotes or backticks sql_match = re.search(r"['\`\"](SELECT|UPDATE|INSERT|DELETE|WITH|CALL)[\s\S]*?['\`\"]", question, re.IGNORECASE) if sql_match: raw_sql = sql_match.group(0).strip("'`\"") else: # Strategy B: Grab everything from the SQL keyword to the end of the sentence sql_match = re.search(r"\b(SELECT|UPDATE|INSERT|DELETE|WITH|CALL)\s[\s\S]*", question, re.IGNORECASE) raw_sql = sql_match.group(0) # Clean up "pollution": Strip "in [Target_Name]" if they put it at the end raw_sql = re.sub(rf"(?i)\s+in\s+['\"]?{re.escape(target_name)}['\"]?\s*$", "", raw_sql).strip() return (f"### 🛠️ Generated Execution Plan\n" f"* **Target System:** `{target_name}`\n" f"* **Engine:** `{db_type.upper()}`\n" f"* **Action:** `RAW_EXECUTE`\n" f" ```sql\n {raw_sql}\n ```\n\n*Note: Direct query payload ready for dispatch.*") # ── PRIORITY 1: EXPLAIN LOGIC ── if best_intent == "TARGET_EXPLAIN" and best_score > 0.60: if debug: print(f"[ORCHESTRATOR] ⚡ Action: EXPLAIN triggered.") digest_match = re.search(r"['\"](.+?--.+?)['\"]", question) if digest_match: digest = digest_match.group(1) schema_name, _ = parse_composite_digest(digest) # Fetch text from performance_schema conn = sqlite3.connect("./spider_data/database/derby_system/derby_system.sqlite") row = conn.execute("SELECT digest_text FROM performance_schema WHERE digest = ?", (digest,)).fetchone() conn.close() if row: template = TARGET_DIALECTS["EXPLAIN_TEMPLATE"].get(db_type, "{query}") prefix = "" if db_type in ['oracle', 'db2']: prefix = f"SET SCHEMA {schema_name}; " elif db_type == 'mysql': prefix = f"USE {schema_name}; " elif db_type == 'postgres': prefix = f"SET search_path TO {schema_name}; " final_explain_sql = prefix + template.format(query=row[0]) return (f"### 🛠️ Generated Execution Plan\n* **Target:** `{target_name}`\n" f"* **Action:** `EXPLAIN` (Schema: `{schema_name}`)\n" f" ```sql\n {final_explain_sql}\n ```\n*Note: Query ready for dispatch.*") else: return f"⚠️ Intent recognized as `EXPLAIN`, but digest `{digest}` was not found in the Derby performance_schema." else: return "⚠️ Intent recognized as `EXPLAIN`, but no valid digest ID (e.g., 'SCHEMA--id') was found in your prompt." # ── PRIORITY 2: READYMADE DIALECTS ── if best_intent in dba_intents and best_score > 0.30: action = dba_intents[best_intent]["intent_key"] if debug: print(f"[ORCHESTRATOR] ⚡ Action: DIALECT matched ({action}).") dialect_sql = TARGET_DIALECTS.get(action, {}).get(db_type) if dialect_sql: # --- SMART TABLE FILTER INJECTION --- if action in ["SHOW_COLUMNS", "SHOW_INDEXES"]: table_match = re.search(r"\btable\s+['\"]?([a-zA-Z0-9_]+)['\"]?", question, re.IGNORECASE) if table_match: tbl = table_match.group(1) # Map the correct dialect-specific column name for "table" if db_type == "postgres": t_col = "relname" if action == "SHOW_COLUMNS" else "tablename" elif db_type == "db2": t_col = "tabname" elif db_type == "redshift": t_col = "tablename" elif db_type == "sqlserver": t_col = "object_name(object_id)" elif db_type == "sybase": t_col = "object_name(id)" elif db_type == "clickhouse": t_col = "table" else: t_col = "table_name" # MySQL, Oracle, Cassandra, Snowflake fuzzy_filter = f" LOWER({t_col}) LIKE LOWER('%{tbl}%')" if "WHERE" in dialect_sql.upper(): dialect_sql = dialect_sql.replace(";", f" AND{fuzzy_filter};") else: dialect_sql = dialect_sql.replace(";", f" WHERE{fuzzy_filter};") return (f"### 🛠️ Generated Execution Plan\n* **Target:** `{target_name}` ({db_type.upper()})\n" f"* **Action:** `{action}`\n" f" ```sql\n {dialect_sql}\n ```\n\n*Note: Using native system views.*") # ── PRIORITY 3: DYNAMIC DATA QUERY (Hijack Engine) ── if debug: print(f"[ORCHESTRATOR] 🧠 Action: DYNAMIC TARGET QUERY for {target_name}") chat_memory.update_state(intent="DYNAMIC_TARGET_QUERY") # Fetch or build SchemaGraph dynamically from mock SQLite if target_name in current_state["target_schema_cache"]: target_schema = current_state["target_schema_cache"][target_name] else: db_path = f"./spider_data/database/{target_name}/{target_name}.sqlite" if not os.path.exists(db_path): return f"⚠️ Mock database file for `{target_name}` not found in demo environment. Cannot build schema." conn = sqlite3.connect(db_path) schema_text = "" for row in conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND sql IS NOT NULL"): schema_text += row[0] + ";\n" conn.close() target_schema = SchemaGraph(schema_text=schema_text) chat_memory.cache_schema(target_name, target_schema) # Optimization: Temporarily hijack engine_derby instead of loading a new ML pipeline original_schema = engine_derby.schema original_db_id = engine_derby.db_id engine_derby.schema = target_schema engine_derby.db_id = None # Disable value lookup for remote targets try: target_sql = generate_sql(question, engine_derby, debug=debug) finally: # Always restore the engine back to Derby! engine_derby.schema = original_schema engine_derby.db_id = original_db_id return (f"### 🛠️ Generated Execution Plan\n* **Target System:** `{target_name}`\n" f"* **Engine:** `{db_type.upper()}`\n* **Action:** Dynamic Data Query\n" f" ```sql\n {target_sql}\n ```\n\n*Note: Query ready for dispatch.*") # CASE B: Standard Monitoring / System Query (Derby/Influx) else: if debug: print(f"[ORCHESTRATOR] 🌐 Mode: SYSTEM Context (Derby/Influx metadata)") # ── PRIORITY 1: MULTI-STEP PERFORMANCE ANALYSIS (The JSON Builder) ── needs_sql_text = bool(re.search(r'\b(sql quer(?:y|ies)|sql texts?|actual sql|actual quer(?:y|ies)|execution plans?|sql plans?|quer(?:y|ies) (?:behind|causing|responsible|running)|what sql|what quer(?:y|ies)|show.{0,20}quer(?:y|ies)|look like|what do they look)\b', q_lower)) needs_anomalies = bool(re.search(r'\b(worst|bottlenecks?|anomal(?:y|ies)|spikes?|issues?|problems?|most anomalous|performing|severe)\b', q_lower)) # Trigger if asking for anomalies AND a target is in memory, OR if asking for plans + anomalies, OR asking for plans globally trigger_multistep = (needs_sql_text and needs_anomalies) or (needs_sql_text and is_global_query) or (needs_anomalies and target_name) if trigger_multistep: if debug: print(f"[ORCHESTRATOR] ⚡ Action: MULTI-STEP QUERY Detected!") chat_memory.update_state(intent="MULTI_STEP_PERFORMANCE") execution_steps = [] influx_limit = 5 digit_match = re.search(r'\b(?:top|worst|highest|bottom|least)\s+(\d+)\b', q_lower) word_match = re.search(r'\b(?:top|worst|highest|bottom|least)\s+(one|two|three|four|five|six|seven|eight|nine|ten)\b', q_lower) if digit_match: influx_limit = int(digit_match.group(1)) elif word_match: parsed_num = word_to_num(word_match.group(1)) if parsed_num: influx_limit = parsed_num elif any(w in q_lower for w in ['single', 'most anomalous', 'worst target', 'worst query']): influx_limit = 1 # influx_prompt = f"Find the top {influx_limit} worst performing queries." # # THE FIX: If context holds a target, explicitly feed it to the multi-step prompt # if target_name and not is_global_query: # if debug: print(f"[ORCHESTRATOR] 💉 Injecting target '{target_name}' into Influx Multi-Step Prompt.") # influx_prompt += f" for target {target_name}" # influx_sql = generate_sql(influx_prompt, engine_influx, debug=debug) influx_prompt = f"Find the top {influx_limit} worst performing queries." # Extract via neural engine, but DO NOT generate SQL yet inf_tables, inf_cols, inf_filters = engine_influx.extract_intent(influx_prompt, debug=debug) # ---> THE SCRUBBER <--- if target_name and not is_global_query: if debug: print(f"[ORCHESTRATOR] 💉 Scrubbing Influx hallucinations & forcing target: '{target_name}'.") # Drop any filter the neural engine tried to guess for 'target' inf_filters = [f for f in inf_filters if f[1] != 'target'] # Inject the absolute truth inf_filters.append(('sys_target_alerts', 'target', '=', f"'{target_name}'", 'INCLUDE')) if 'sys_target_alerts' not in inf_tables: inf_tables.append('sys_target_alerts') # Pass the forcefully injected arrays to bypass the neural linker influx_sql = generate_sql(influx_prompt, engine_influx, injected_tables=inf_tables, injected_col_hits=inf_cols, injected_filters=inf_filters, debug=debug) execution_steps.append(f"* **Step 1** (Target: `influx_system`):\n ```sql\n {influx_sql}\n ```") try: conn_in = sqlite3.connect("./spider_data/database/influx_system/influx_system.sqlite") influx_results = conn_in.execute(influx_sql).fetchall() conn_in.close() except Exception as e: return f"🚨 InfluxDB Execution Error: {e}" digests = [row[1] for row in influx_results if row[1]] if not digests: return "\n".join(execution_steps) + "\n\n**Result:** No performance bottlenecks found in InfluxDB." digest_list_str = ", ".join([f"'{d}'" for d in digests]) derby_text_sql = f"SELECT digest, digest_text, sql_plan_id FROM performance_schema WHERE digest IN ({digest_list_str})" execution_steps.append(f"* **Step 2** (Target: `derby_system`):\n *(Extracts digests from Step 1)*\n ```sql\n {derby_text_sql}\n ```") try: conn_derby = sqlite3.connect("./spider_data/database/derby_system/derby_system.sqlite") derby_text_results = conn_derby.execute(derby_text_sql).fetchall() except Exception as e: return f"🚨 Derby Query Error: {e}" sql_metadata = {} plan_ids_to_fetch = set() for row in derby_text_results: d_id, d_text, p_id = row[0], row[1], row[2] sql_metadata[d_id] = {"text": d_text, "plan_id": p_id} if p_id is not None: plan_ids_to_fetch.add(str(p_id)) plan_metadata = {} if plan_ids_to_fetch: plan_id_str = ", ".join(plan_ids_to_fetch) derby_plan_sql = f"SELECT id, sql_plan, sql_plain_text_plan FROM sql_plan WHERE id IN ({plan_id_str})" execution_steps.append(f"* **Step 3** (Target: `derby_system`):\n *(Extracts plan IDs from Step 2)*\n ```sql\n {derby_plan_sql}\n ```") try: derby_plan_results = conn_derby.execute(derby_plan_sql).fetchall() for row in derby_plan_results: p_id, json_plan, plain_plan = row[0], row[1], row[2] plan_metadata[p_id] = json_plan if json_plan else plain_plan except Exception as e: pass conn_derby.close() llm_payload = [] for row in influx_results: target, metric_digest, severity, timestamp = row[0], row[1], row[2], row[3] meta = sql_metadata.get(metric_digest, {}) actual_sql = meta.get("text", "-- Metadata Not Found in Derby --") plan_id = meta.get("plan_id") execution_plan = plan_metadata.get(str(plan_id), "-- No Plan Linked --") if plan_id else "-- No Plan Linked --" llm_payload.append({ "target_server": target, "digest": metric_digest, "anomaly_score": severity, "timestamp": timestamp, "sql_query": actual_sql, "execution_plan": execution_plan }) formatted_payload = json.dumps(llm_payload, indent=2) exec_plan_md = "### 🛠️ Generated Execution Plan\n" + "\n".join(execution_steps) return f"{exec_plan_md}\n\n### 📦 Final JSON Payload (For LLM)\n```json\n{formatted_payload}\n```" # ── PRIORITY 2: SEMANTIC ROUTER (FAQ & Metadata Queries) ── router_result = semantic_metadata_router(question, router_model, anchor_embeddings) if router_result["status"] in ["MATCH", "CLARIFY"]: if debug: print(f"[ORCHESTRATOR] ⚡ Action: SEMANTIC ROUTER MATCH ({best_intent})") chat_memory.update_state(intent=router_result.get("sql")) # Track FAQ intent sql_query = router_result["sql"] target_db = router_result.get("target_db", "derby_system") # ---> THE SURGICAL FIX FOR TARGET DETAILS <--- if best_intent == "TARGET_DETAILS" and target_name: if debug: print(f"[ORCHESTRATOR] 💉 Intercepting TARGET_DETAILS to inject target: {target_name}") sql_query = f"SELECT * FROM target WHERE name = '{target_name}';" # --------------------------------------------- exec_plan = f"### 🛠️ Generated Execution Plan\n* **Step 1** (Target: `{target_db}`):\n ```sql\n {sql_query}\n ```\n" db_path = f"./spider_data/database/{target_db}/{target_db}.sqlite" try: conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(sql_query) results = cursor.fetchall() if "sqlite_master" in sql_query: formatted_results = "\n".join([f"- **{row[0]}**" for row in results if row[0] != 'sqlite_sequence']) else: formatted_results = format_results_as_md_table(cursor, results) conn.close() except Exception as e: formatted_results = f"🚨 **SQL EXECUTION ERROR:**\n```\n{e}\n```" final_output = f"{exec_plan}\n### 📊 Execution Results:\n{formatted_results}" if router_result["status"] == "CLARIFY": return f"🤖 **SYSTEM QUESTION:**\n{router_result['message']}\n*(Assuming YES...)*\n\n{final_output}" return final_output # ── PRIORITY 3: STANDARD NEURAL DB QUERY (With Memory Injection) ── if debug: print(f"[ORCHESTRATOR] ➡️ Action: NEURAL DB QUERY Fallback.") target_db = route_query(question) engine = engine_derby if target_db == 'derby_system' else engine_influx # 1. Run extractor on current question active_tables, col_hits, new_filters = engine.extract_intent(question, debug=debug) # # 2. THE SCRUBBER & TARGET INJECTION (Turn 4 fix) # if target_name and not is_global_query: # target_col = 'target' if target_db == 'influx_system' else 'name' # target_tbl = 'sys_target_alerts' if target_db == 'influx_system' else 'target' # cleaned_filters = [] # for ft, fc, fop, fval, fintent in new_filters: # val_clean = fval.replace("'", "").lower() # target_clean = target_name.lower() # # Rule 1: Neural engine hallucinated the value for our target column # is_hallucinated_val = (fc == target_col and val_clean != target_clean) # # Rule 2: Neural engine applied our exact target name to the WRONG column (e.g. db_target_status) # is_wrong_col = (target_clean in val_clean and fc != target_col) # if is_hallucinated_val or is_wrong_col: # if debug: print(f"[ORCHESTRATOR] 🧹 Scrubbing hallucinated target binding: {fc} = {fval}") # continue # cleaned_filters.append((ft, fc, fop, fval, fintent)) # new_filters = cleaned_filters # # Ensure we don't duplicate the filter, then inject the pure deterministic truth # if not any(fc == target_col and target_name.lower() in fval.lower() for _, fc, _, fval, _ in new_filters): # if debug: print(f"[ORCHESTRATOR] 💉 Injecting pure target filter for: '{target_name}'") # new_filters.append((target_tbl, target_col, '=', f"'{target_name}'", 'INCLUDE')) # if target_tbl not in active_tables: # active_tables.append(target_tbl) # ========================================================= # 2. THE EXPANDED SCRUBBER (Table Names & Hallucinations) # ========================================================= cleaned_filters = [] for ft, fc, fop, fval, fintent in new_filters: val_clean = fval.replace("'", "").lower() # RULE 1: The Conversational Noun Killer # If the user used a table name conversationally (e.g. "target"), drop it. if val_clean in [t.lower() for t in active_tables] or val_clean == 'target': if debug: print(f"[ORCHESTRATOR] 🧹 Scrubbing conversational filler: {fc} = {fval}") continue # RULE 2: Target Name Hallucination Checks if target_name and not is_global_query: target_col = 'target' if target_db == 'influx_system' else 'name' target_clean = target_name.lower() is_hallucinated_val = (fc == target_col and val_clean != target_clean) is_wrong_col = (target_clean in val_clean and fc != target_col) if is_hallucinated_val or is_wrong_col: if debug: print(f"[ORCHESTRATOR] 🧹 Scrubbing hallucinated target binding: {fc} = {fval}") continue cleaned_filters.append((ft, fc, fop, fval, fintent)) new_filters = cleaned_filters # 2B. Force Inject Target Filter if target_name and not is_global_query: target_col = 'target' if target_db == 'influx_system' else 'name' target_tbl = 'sys_target_alerts' if target_db == 'influx_system' else 'target' if not any(fc == target_col and target_name.lower() in fval.lower() for _, fc, _, fval, _ in new_filters): if debug: print(f"[ORCHESTRATOR] 💉 Injecting pure target filter for: '{target_name}'") new_filters.append((target_tbl, target_col, '=', f"'{target_name}'", 'INCLUDE')) if target_tbl not in active_tables: active_tables.append(target_tbl) # ========================================================= # 2B. THE NEW FEATURE: Deterministic Platform Injection # ========================================================= # platform_name = extract_platform_context(question) # if platform_name and target_db == 'derby_system': # if debug: print(f"[ORCHESTRATOR] 💉 Injecting OS platform filter: '{platform_name}'") # # Step A: Scrub any bad guesses the neural net made about the platform column # new_filters = [f for f in new_filters if f[1] != 'hlc_server_platform'] # # Step B: Inject the absolute truth # new_filters.append(('target', 'hlc_server_platform', '=', f"'{platform_name}'", 'INCLUDE')) # # Step C: Ensure the target table is active so the query doesn't crash # if 'target' not in active_tables: # active_tables.append('target') # ========================================================= # 2B. THE NEW FEATURE: Deterministic Platform Injection # ========================================================= platform_name = extract_platform_context(question) if platform_name and target_db == 'derby_system': # NEW: Look for negation words right before the platform name negation_pattern = r'\b(not|excluding|except|other than|without|non)\b(?:(?!\b(?:and|or)\b).){0,20}?\b' + platform_name.lower() + r'\b' is_negated = bool(re.search(negation_pattern, q_lower)) op = '!=' if is_negated else '=' intent_val = 'EXCLUDE' if is_negated else 'INCLUDE' if debug: print(f"[ORCHESTRATOR] 💉 Injecting OS platform filter: '{platform_name}' (Negated: {is_negated})") # Step A: Scrub any bad guesses the neural net made about the platform column new_filters = [f for f in new_filters if f[1] != 'hlc_server_platform'] # Step B: Inject the absolute truth WITH correct polarity new_filters.append(('target', 'hlc_server_platform', op, f"'{platform_name}'", intent_val)) if 'target' not in active_tables: active_tables.append('target') # ========================================================= # 3. ── MEMORY INJECTION ── Check if it's a follow-up follow_up_pattern = r'\b(those|them|these|what about|how about|which of|out of)\b' is_follow_up = bool(re.search(follow_up_pattern, q_lower)) if is_follow_up and current_state["neural_filters"]: if debug: print("[MEMORY] Follow-up detected. Injecting previous neural context.") # THE FIX: Use a composite key (Table, Column) to prevent cross-table overwriting merged_filters = {} # 1. Load the old historical filters first for ft, fc, fop, fval, fint in current_state["neural_filters"]: merged_filters[(ft, fc)] = (ft, fc, fop, fval, fint) # 2. Overwrite with the fresh filters from the current turn for ft, fc, fop, fval, fint in new_filters: merged_filters[(ft, fc)] = (ft, fc, fop, fval, fint) # 3. Convert back to list new_filters = list(merged_filters.values()) for t in current_state["neural_tables"]: if t not in active_tables: active_tables.append(t) # Save new context for the next turn chat_memory.update_neural_context(active_tables, new_filters) # Generate using injected arrays sql_query = generate_sql(question, engine, injected_tables=active_tables, injected_col_hits=col_hits, injected_filters=new_filters, debug=debug) exec_plan = f"### 🛠️ Generated Execution Plan\n* **Step 1** (Target: `{target_db}`):\n ```sql\n {sql_query}\n ```\n" db_path = f"./spider_data/database/{target_db}/{target_db}.sqlite" try: conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute(sql_query) results = cursor.fetchall() formatted_results = format_results_as_md_table(cursor, results) conn.close() except Exception as e: formatted_results = f"🚨 **SQL EXECUTION ERROR:**\n```\n{e}\n```" return f"{exec_plan}\n### 📊 Execution Results:\n{formatted_results}" #AastikATripathiFinal