explain_query
Analyze SQL query execution plans with detailed cost estimates, simulate hypothetical indexes, and get real execution statistics for optimized database performance on Postgres MCP.
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 |
|---|---|---|---|
| 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. | |
| sql | Yes | SQL query to explain |
Implementation Reference
- src/postgres_mcp/server.py:310-387 (handler)Primary handler and registration for the 'explain_query' tool via @mcp.tool decorator. Processes inputs and delegates to ExplainPlanTool methods.@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))
- src/postgres_mcp/server.py:312-331 (schema)Input schema definitions using Pydantic Field for the explain_query tool parameters: sql, analyze, hypothetical_indexes.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:
- Core helper method in ExplainPlanTool that constructs and executes the EXPLAIN query with options (ANALYZE, GENERIC_PLAN), parses JSON output, and creates ExplainPlanArtifact.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 that prepares the query (handles bind params) and calls _run_explain_query. Called by the main handler.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)