explain_query
Analyze SQL query execution plans to understand database performance and optimize queries with detailed cost estimates and hypothetical index simulation.
Instructions
Explains the execution plan for a SQL query, showing how the database will execute it and provides detailed cost estimates.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | Yes | SQL query to explain | |
| analyze | No | When True, actually runs the query to show real execution statistics instead of estimates. Takes longer but provides more accurate information. | |
| hypothetical_indexes | No | A list of hypothetical indexes to simulate. Each index must be a dictionary with these keys: - 'table': The table name to add the index to (e.g., 'users') - 'columns': List of column names to include in the index (e.g., ['email'] or ['last_name', 'first_name']) - 'using': Optional index method (default: 'btree', other options include 'hash', 'gist', etc.) Examples: [ {"table": "users", "columns": ["email"], "using": "btree"}, {"table": "orders", "columns": ["user_id", "created_at"]} ] If there is no hypothetical index, you can pass an empty list. |
Implementation Reference
- Primary handler function for the 'explain_query' tool. Decorated with @mcp.tool for registration. Defines input schema via Pydantic Field. Dispatches to ExplainPlanTool based on 'analyze' and 'hypothetical_indexes' parameters.
@mcp.tool(description="Explains the execution plan for a SQL query, showing how the database will execute it and provides detailed cost estimates.") async def explain_query( sql: str = Field(description="SQL query to explain"), analyze: bool = Field( description="When True, actually runs the query to show real execution statistics instead of estimates. " "Takes longer but provides more accurate information.", default=False, ), hypothetical_indexes: list[dict[str, Any]] = Field( description="""A list of hypothetical indexes to simulate. Each index must be a dictionary with these keys: - 'table': The table name to add the index to (e.g., 'users') - 'columns': List of column names to include in the index (e.g., ['email'] or ['last_name', 'first_name']) - 'using': Optional index method (default: 'btree', other options include 'hash', 'gist', etc.) Examples: [ {"table": "users", "columns": ["email"], "using": "btree"}, {"table": "orders", "columns": ["user_id", "created_at"]} ] If there is no hypothetical index, you can pass an empty list.""", default=[], ), ) -> ResponseType: """ Explains the execution plan for a SQL query. Args: sql: The SQL query to explain analyze: When True, actually runs the query for real statistics hypothetical_indexes: Optional list of indexes to simulate """ try: sql_driver = await get_sql_driver() explain_tool = ExplainPlanTool(sql_driver=sql_driver) result: ExplainPlanArtifact | ErrorResult | None = None # If hypothetical indexes are specified, check for HypoPG extension if hypothetical_indexes and len(hypothetical_indexes) > 0: if analyze: return format_error_response("Cannot use analyze and hypothetical indexes together") try: # Use the common utility function to check if hypopg is installed ( is_hypopg_installed, hypopg_message, ) = await check_hypopg_installation_status(sql_driver) # If hypopg is not installed, return the message if not is_hypopg_installed: return format_text_response(hypopg_message) # HypoPG is installed, proceed with explaining with hypothetical indexes result = await explain_tool.explain_with_hypothetical_indexes(sql, hypothetical_indexes) except Exception: raise # Re-raise the original exception elif analyze: try: # Use EXPLAIN ANALYZE result = await explain_tool.explain_analyze(sql) except Exception: raise # Re-raise the original exception else: try: # Use basic EXPLAIN result = await explain_tool.explain(sql) except Exception: raise # Re-raise the original exception if result and isinstance(result, ExplainPlanArtifact): return format_text_response(result.to_text()) else: error_message = "Error processing explain plan" if isinstance(result, ErrorResult): error_message = result.to_text() return format_error_response(error_message) except Exception as e: logger.error(f"Error explaining query: {e}") return format_error_response(str(e)) - Core helper method in ExplainPlanTool that constructs and executes the EXPLAIN (ANALYZE?) (GENERIC_PLAN?) query, parses JSON output, and returns ExplainPlanArtifact or ErrorResult.
async def _run_explain_query(self, query: str, analyze: bool = False, generic_plan: bool = False) -> ExplainPlanArtifact | ErrorResult: try: explain_options = ["FORMAT JSON"] if analyze: explain_options.append("ANALYZE") if generic_plan: explain_options.append("GENERIC_PLAN") explain_q = f"EXPLAIN ({', '.join(explain_options)}) {query}" logger.debug(f"RUNNING EXPLAIN QUERY: {explain_q}") rows = await self.sql_driver.execute_query(explain_q) # type: ignore if rows is None: return ErrorResult("No results returned from EXPLAIN") query_plan_data = rows[0].cells["QUERY PLAN"] if not isinstance(query_plan_data, list): return ErrorResult(f"Expected list from EXPLAIN, got {type(query_plan_data)}") if len(query_plan_data) == 0: return ErrorResult("No results returned from EXPLAIN") plan_dict = query_plan_data[0] if not isinstance(plan_dict, dict): return ErrorResult(f"Expected dict in EXPLAIN result list, got {type(plan_dict)} with value {plan_dict}") try: return ExplainPlanArtifact.from_json_data(plan_dict) except Exception as e: return ErrorResult(f"Internal error converting explain plan - do not retry: {e}") except Exception as e: return ErrorResult(f"Error executing explain plan: {e}") - Helper method in ExplainPlanTool for handling hypothetical indexes using hypopg extension: validates input, creates hypopg indexes, runs explain, processes result.
async def explain_with_hypothetical_indexes( self, sql_query: str, hypothetical_indexes: list[dict[str, Any]] ) -> ExplainPlanArtifact | ErrorResult: """ Generate an explain plan for a query as if certain indexes existed. Args: sql_query: The SQL query to explain hypothetical_indexes: List of index definitions as dictionaries Returns: ExplainPlanArtifact or ErrorResult """ try: # Validate index definitions format if not isinstance(hypothetical_indexes, list): return ErrorResult(f"Expected list of index definitions, got {type(hypothetical_indexes)}") for idx in hypothetical_indexes: if not isinstance(idx, dict): return ErrorResult(f"Expected dictionary for index definition, got {type(idx)}") if "table" not in idx: return ErrorResult("Missing 'table' in index definition") if "columns" not in idx: return ErrorResult("Missing 'columns' in index definition") if not isinstance(idx["columns"], list): # Try to convert to list if it's not already try: idx["columns"] = list(idx["columns"]) if hasattr(idx["columns"], "__iter__") else [idx["columns"]] except Exception as e: return ErrorResult(f"Expected list for 'columns', got {type(idx['columns'])}: {e}") # Convert the index definitions to IndexConfig objects indexes = frozenset( IndexDefinition( table=idx["table"], columns=tuple(idx["columns"]), using=idx.get("using", "btree"), ) for idx in hypothetical_indexes ) # Check if the query contains bind variables modified_sql_query, use_generic_plan = await self.replace_query_parameters_if_needed(sql_query) # Generate the explain plan using the static method plan_data = await self.generate_explain_plan_with_hypothetical_indexes(modified_sql_query, indexes, use_generic_plan) # Check if we got a valid plan if not plan_data or not isinstance(plan_data, dict) or "Plan" not in plan_data: return ErrorResult("Failed to generate a valid explain plan with the hypothetical indexes") try: # Convert the plan data to an ExplainPlanArtifact return ExplainPlanArtifact.from_json_data(plan_data) except Exception as e: return ErrorResult(f"Error converting explain plan: {e}") except Exception as e: logger.error(f"Error in explain_with_hypothetical_indexes: {e}", exc_info=True) return ErrorResult(f"Error generating explain plan with hypothetical indexes: {e}") - Output schema: ExplainPlanArtifact dataclass parses PostgreSQL EXPLAIN JSON into structured PlanNode tree with costs, actual times, etc. Includes from_json_data factory and to_text formatter.
@define class ExplainPlanArtifact: value: str plan_tree: PlanNode planning_time: float | None = field(default=None) execution_time: float | None = field(default=None) def __init__( self, value: str, plan_tree: PlanNode, planning_time: float | None = None, execution_time: float | None = None, ): self.value = value self.plan_tree = plan_tree self.planning_time = planning_time self.execution_time = execution_time def to_text(self) -> str: """Convert the explain plan to a compact text representation.""" lines: list[str] = [] # Compact timing info if available timing_parts: list[str] = [] if self.planning_time is not None: timing_parts.append(f"plan={self.planning_time:.3f}ms") if self.execution_time is not None: timing_parts.append(f"exec={self.execution_time:.3f}ms") if timing_parts: lines.append("Timing: " + " ".join(timing_parts)) # Plan tree lines.append(self._format_plan_node(self.plan_tree)) return "\n".join(lines) @staticmethod def _format_plan_node(node: PlanNode, level: int = 0) -> str: """Recursively format a plan node and its children. Args: node: The plan node to format level: The current indentation level Returns: str: A formatted string representation of the node and its children """ indent = " " * level line = f"{indent}{node.node_type} cost={node.startup_cost:.2f}..{node.total_cost:.2f}" # Relation name if present if node.relation_name: line += f" rel={node.relation_name}" # Planned rows line += f" rows={node.plan_rows}" # Actual metrics if available if node.actual_total_time is not None and node.actual_startup_time is not None: line += f" act={node.actual_startup_time:.2f}..{node.actual_total_time:.2f}ms" if node.actual_rows is not None: line += f" act_rows={node.actual_rows}" if node.actual_loops is not None: line += f" loops={node.actual_loops}" # Filter (truncate long) if node.filter: filt = node.filter if len(filt) > 100: filt = filt[:97] + "..." line += f" | filter={filt}" # Buffer info if node.shared_hit_blocks is not None: line += f" | buf hit={node.shared_hit_blocks} read={node.shared_read_blocks} written={node.shared_written_blocks}" # Children if node.children: for child in node.children: line += "\n" + ExplainPlanArtifact._format_plan_node(child, level + 1) return line @classmethod def from_json_data(cls, plan_data: dict[str, Any]) -> "ExplainPlanArtifact": if "Plan" not in plan_data: raise ValueError("Missing 'Plan' field in explain plan data") # Create plan tree from the "Plan" field plan_tree = PlanNode.from_json_data(plan_data["Plan"]) # Extract optional timing information planning_time = plan_data.get("Planning Time") execution_time = plan_data.get("Execution Time") return cls( value=json.dumps(plan_data, indent=2), plan_tree=plan_tree, planning_time=planning_time, execution_time=execution_time, ) @staticmethod - Main public helper method in ExplainPlanTool: handles bind parameter replacement for generic plans (PG16+), then calls _run_explain_query.
async def explain(self, sql_query: str, do_analyze: bool = False) -> ExplainPlanArtifact | ErrorResult: """ Generate an EXPLAIN plan for a SQL query. Args: sql_query: The SQL query to explain Returns: ExplainPlanArtifact or ErrorResult """ modified_sql_query, use_generic_plan = await self.replace_query_parameters_if_needed(sql_query) return await self._run_explain_query(modified_sql_query, analyze=do_analyze, generic_plan=use_generic_plan)