explain_with_indexes
Analyze PostgreSQL query performance with proposed indexes before implementation. Compare execution plans to determine if new indexes would improve performance without creating them.
Instructions
Run EXPLAIN on a query, optionally with hypothetical indexes.
This tool allows you to see how a query would perform with proposed indexes WITHOUT actually creating them. Requires HypoPG extension for hypothetical testing.
Use this to:
Compare execution plans with and without specific indexes
Test if a proposed index would be used
Estimate the performance impact of new indexes
Returns both the original and hypothetical execution plans for comparison.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The SQL query to explain | |
| hypothetical_indexes | No | List of hypothetical indexes to test | |
| analyze | No | Whether to use EXPLAIN ANALYZE (executes the query) |
Implementation Reference
- The main handler function that executes the tool. It validates input, runs EXPLAIN on the query, optionally creates hypothetical indexes using HypoPG, compares execution plans, calculates estimated improvement, identifies used indexes, and cleans up hypothetical indexes.async def run_tool(self, arguments: dict[str, Any]) -> Sequence[TextContent]: try: self.validate_required_args(arguments, ["query"]) query = arguments["query"] hypothetical_indexes = arguments.get("hypothetical_indexes", []) analyze = arguments.get("analyze", False) # Get the original execution plan explain_opts = "ANALYZE, " if analyze else "" original_explain = f"EXPLAIN ({explain_opts}FORMAT JSON) {query}" original_result = await self.sql_driver.execute_query(original_explain) original_plan = self._extract_plan(original_result) output = { "query": query, "original_plan": original_plan, "original_cost": self._extract_cost(original_plan) } # Test with hypothetical indexes if provided if hypothetical_indexes: hypopg_status = await self.hypopg_service.check_status() if not hypopg_status.is_installed: output["error"] = ( "HypoPG extension is not available. " "Install it with: CREATE EXTENSION hypopg;" ) return self.format_json_result(output) # Create hypothetical indexes created_indexes = [] try: for idx_spec in hypothetical_indexes: hypo_index = await self.hypopg_service.create_index( table=idx_spec["table"], columns=idx_spec["columns"], using=idx_spec.get("index_type", "btree"), ) created_indexes.append({ **idx_spec, "index_name": hypo_index.index_name, "index_oid": hypo_index.indexrelid, "estimated_size": hypo_index.estimated_size }) # Get the plan with hypothetical indexes hypo_result = await self.sql_driver.execute_query(original_explain) hypo_plan = self._extract_plan(hypo_result) output["hypothetical_indexes"] = created_indexes output["plan_with_indexes"] = hypo_plan output["cost_with_indexes"] = self._extract_cost(hypo_plan) # Calculate improvement original_cost = output["original_cost"] new_cost = output["cost_with_indexes"] if original_cost > 0: improvement = ((original_cost - new_cost) / original_cost) * 100 output["estimated_improvement_percent"] = round(improvement, 2) # Check which hypothetical indexes were used output["indexes_used"] = self._find_used_indexes(hypo_plan, created_indexes) finally: # Clean up hypothetical indexes await self.hypopg_service.reset() return self.format_json_result(output) except Exception as e: return self.format_error(e)
- Defines the tool's input schema, including the required 'query' parameter, optional 'hypothetical_indexes' array specifying table, columns, index_type, unique, and 'analyze' boolean.def get_tool_definition(self) -> Tool: return Tool( name=self.name, description=self.description, inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to explain" }, "hypothetical_indexes": { "type": "array", "items": { "type": "object", "properties": { "table": { "type": "string", "description": "Table name" }, "columns": { "type": "array", "items": {"type": "string"}, "description": "Columns for the index" }, "index_type": { "type": "string", "enum": ["btree", "hash", "gin", "gist", "brin"], "default": "btree" }, "unique": { "type": "boolean", "default": False } }, "required": ["table", "columns"] }, "description": "List of hypothetical indexes to test" }, "analyze": { "type": "boolean", "description": "Whether to use EXPLAIN ANALYZE (executes the query)", "default": False } }, "required": ["query"] }, annotations=self.get_annotations() )
- src/pgtuner_mcp/server.py:154-154 (registration)Registers the ExplainQueryToolHandler instance with the global tool_handlers dictionary via add_tool_handler in the register_all_tools function.add_tool_handler(ExplainQueryToolHandler(driver, hypopg_service))
- Helper method to extract the execution plan JSON from the EXPLAIN query result.def _extract_plan(self, result: list[dict]) -> dict: """Extract the execution plan from EXPLAIN result.""" if not result: return {} plan_data = result[0].get("QUERY PLAN", result[0]) if isinstance(plan_data, str): plan_data = json.loads(plan_data) if isinstance(plan_data, list) and len(plan_data) > 0: return plan_data[0] return plan_data
- Helper method to traverse the execution plan and identify which hypothetical indexes were actually used in the query plan.def _find_used_indexes( self, plan: dict, created_indexes: list[dict] ) -> list[dict]: """Find which hypothetical indexes were used in the plan.""" used = [] index_names = {idx["index_name"] for idx in created_indexes if "index_name" in idx} def check_node(node: dict): if not isinstance(node, dict): return node_type = node.get("Node Type", "") if "Index" in node_type: idx_name = node.get("Index Name", "") if any(name in idx_name for name in index_names): used.append({ "index_name": idx_name, "scan_type": node_type, "startup_cost": node.get("Startup Cost"), "total_cost": node.get("Total Cost") }) for child in node.get("Plans", []): check_node(child) check_node(plan.get("Plan", plan)) return used