""" Plan parser for analyzing TiDB execution plans. """ import re import json from typing import List, Optional, Dict, Any class TableFullScanInfo: """Information about a TableFullScan operator.""" def __init__(self, table_name: str, act_rows: int, operator_info: str = "", database: str = "", task: str = ""): self.table_name = table_name # Table name without database prefix self.database = database # Database name self.act_rows = act_rows self.operator_info = operator_info self.task = task # Task type (e.g., "mpp[tiflash]", "cop[tikv]") def is_tiflash(self) -> bool: """Check if this TableFullScan is executed on TiFlash.""" return 'tiflash' in self.task.lower() or 'mpp' in self.task.lower() def __repr__(self): return f"TableFullScan(db={self.database}, table={self.table_name}, actRows={self.act_rows}, task={self.task})" class PlanParser: """Parser for TiDB execution plans.""" # Pattern to match TableFullScan operators # Example: TableFullScan_123 actRows:12345 TABLE_FULL_SCAN_PATTERN = re.compile( r'TableFullScan[_\w]*.*?actRows:\s*(\d+)', re.IGNORECASE | re.DOTALL ) # Alternative pattern for different plan formats TABLE_FULL_SCAN_PATTERN_ALT = re.compile( r'TableFullScan[_\w]*.*?act_rows[:\s]+(\d+)', re.IGNORECASE | re.DOTALL ) # Pattern to extract table name TABLE_NAME_PATTERN = re.compile( r'table:\s*(\w+)', re.IGNORECASE ) @classmethod def parse_plan(cls, decoded_plan: str) -> List[TableFullScanInfo]: """ Parse an execution plan and extract TableFullScan information. Args: decoded_plan: The decoded execution plan as a string Returns: List of TableFullScanInfo objects found in the plan """ if not decoded_plan: return [] table_full_scans = [] # Try to parse as JSON first (some plans are in JSON format) try: plan_json = json.loads(decoded_plan) table_full_scans.extend(cls._parse_json_plan(plan_json)) except (json.JSONDecodeError, TypeError): # If not JSON, parse as text table_full_scans.extend(cls._parse_text_plan(decoded_plan)) return table_full_scans @classmethod def _parse_json_plan(cls, plan_json: Any) -> List[TableFullScanInfo]: """Parse JSON format plan.""" table_full_scans = [] def traverse(node): if isinstance(node, dict): # Check if this node is a TableFullScan # Check multiple fields: id, operator, type, name node_id = node.get('id', '') operator = node.get('operator', node.get('type', node.get('name', ''))) # Check if this is a TableFullScan operator is_table_full_scan = ( 'TableFullScan' in node_id or 'TableFullScan' in operator or 'table_full_scan' in str(node_id).lower() or 'table_full_scan' in str(operator).lower() ) if is_table_full_scan: # Try different field names for actRows - ONLY accept actual rows, not estimates act_rows = ( node.get('actRows') or node.get('act_rows') or node.get('actual_rows') or None # Don't fallback to estRows ) # Only process if we have actual rows available if act_rows is not None and act_rows != '': # Extract table name and database from accessObjects (new format) table_name = 'unknown' database = '' # Try accessObjects array first (new format) access_objects = node.get('accessObjects', []) if isinstance(access_objects, list) and len(access_objects) > 0: scan_object = access_objects[0].get('scanObject', {}) if isinstance(scan_object, dict): database = scan_object.get('database', '') table_name = scan_object.get('table', 'unknown') # Fallback: Try accessObject string (old format) if table_name == 'unknown': access_object = node.get('accessObject', '') operator_info = node.get('operatorInfo', '') # Try to extract from accessObject (e.g., "table:users, partition:p0") if access_object and 'table:' in access_object: table_name = access_object.split('table:')[1].split(',')[0].strip() # Try to extract from operatorInfo elif operator_info and 'table:' in operator_info: table_name = operator_info.split('table:')[1].split(',')[0].strip() # Fallback to other fields else: table_name = ( node.get('table') or node.get('table_name') or 'unknown' ) if isinstance(act_rows, (int, float, str)) and act_rows: try: act_rows_int = int(float(str(act_rows))) # Extract task information task = node.get('task', node.get('taskType', '')) table_full_scans.append( TableFullScanInfo( table_name=str(table_name), act_rows=act_rows_int, operator_info=str(node), database=str(database), task=str(task) ) ) except (ValueError, TypeError): pass # Recursively traverse children for key, value in node.items(): if isinstance(value, (dict, list)): traverse(value) elif isinstance(node, list): for item in node: traverse(item) traverse(plan_json) return table_full_scans @classmethod def _parse_text_plan(cls, plan_text: str) -> List[TableFullScanInfo]: """Parse text format plan.""" table_full_scans = [] # Split plan into lines for better parsing lines = plan_text.split('\n') # Find all TableFullScan operators for i, line in enumerate(lines): if 'TableFullScan' in line: # Extract actRows from current or nearby lines act_rows = cls._extract_act_rows(lines, i) if act_rows is not None: # Try to extract table name table_name = cls._extract_table_name(lines, i) table_full_scans.append( TableFullScanInfo( table_name=table_name, act_rows=act_rows, operator_info=line.strip() ) ) # If line-based parsing didn't work, try regex patterns if not table_full_scans: # Try primary pattern matches = cls.TABLE_FULL_SCAN_PATTERN.findall(plan_text) for match in matches: try: act_rows = int(match) table_full_scans.append( TableFullScanInfo( table_name='unknown', act_rows=act_rows ) ) except ValueError: pass # Try alternative pattern if primary didn't match if not table_full_scans: matches = cls.TABLE_FULL_SCAN_PATTERN_ALT.findall(plan_text) for match in matches: try: act_rows = int(match) table_full_scans.append( TableFullScanInfo( table_name='unknown', act_rows=act_rows ) ) except ValueError: pass return table_full_scans @classmethod def _extract_act_rows(cls, lines: List[str], start_index: int) -> Optional[int]: """Extract actRows value from the line or nearby lines.""" # Check current line and next few lines for i in range(start_index, min(start_index + 5, len(lines))): line = lines[i] # Try pattern 1: "actRows: 12345" or "actRows 12345" match = re.search(r'actRows[:\s]+(\d+)', line, re.IGNORECASE) if not match: match = re.search(r'act_rows[:\s]+(\d+)', line, re.IGNORECASE) if not match: match = re.search(r'actual[_\s]rows[:\s]+(\d+)', line, re.IGNORECASE) if match: try: return int(match.group(1)) except ValueError: pass # Try pattern 2: Table-style layout (columns separated by whitespace) # Example: "TableFullScan_4 cop[tikv] 10000 25678 table:orders" # where columns are: id, task, estRows, actRows, operatorInfo if 'TableFullScan' in line: parts = line.split() # Try to find numeric columns (usually estRows and actRows are consecutive) for j in range(len(parts) - 1): try: # If we find two consecutive numbers, the second one is likely actRows int(parts[j]) act_rows = int(parts[j + 1]) return act_rows except (ValueError, IndexError): continue return None @classmethod def _extract_table_name(cls, lines: List[str], start_index: int) -> str: """Extract table name from the line or nearby lines.""" # Check current line and nearby lines for i in range(max(0, start_index - 2), min(start_index + 3, len(lines))): line = lines[i] match = cls.TABLE_NAME_PATTERN.search(line) if match: return match.group(1) return 'unknown' @classmethod def get_max_table_full_scan_rows(cls, decoded_plan: str) -> Optional[int]: """ Get the maximum actRows from all TableFullScan operators in the plan. Args: decoded_plan: The decoded execution plan Returns: Maximum actRows value, or None if no TableFullScan found """ table_full_scans = cls.parse_plan(decoded_plan) if not table_full_scans: return None return max(scan.act_rows for scan in table_full_scans) @classmethod def has_table_full_scan(cls, decoded_plan: str) -> bool: """ Check if the plan contains any TableFullScan operator. Args: decoded_plan: The decoded execution plan Returns: True if plan contains TableFullScan, False otherwise """ return len(cls.parse_plan(decoded_plan)) > 0