get_hg_query_plan
Retrieve the execution plan for a SQL query in the Hologres database to analyze and optimize query performance using the Hologres MCP Server.
Instructions
Get query plan for a SQL query in Hologres database
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The SQL query to analyze in Hologres database |
Implementation Reference
- Specific handler dispatch for 'get_hg_query_plan' tool in the main call_tool function: extracts query argument, validates it, and prepends 'EXPLAIN' to generate the explain query.elif name == "get_hg_query_plan": query = arguments.get("query") if not query: raise ValueError("Query is required") query = f"EXPLAIN {query}"
- src/hologres_mcp_server/server.py:422-435 (registration)Tool registration in list_tools(): defines name, description, and input schema (requiring 'query' string).Tool( name="get_hg_query_plan", description="Get query plan for a SQL query in Hologres database", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to analyze in Hologres database" } }, "required": ["query"] } ),
- Input schema definition for the 'get_hg_query_plan' tool: object with required 'query' property of type string.inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to analyze in Hologres database" } }, "required": ["query"] }
- Shared utility function handle_call_tool that connects to Hologres DB, optionally sets serverless mode, executes the prepared EXPLAIN query, fetches results, and formats as CSV (columns then rows). Called after query preparation in server.py.def handle_call_tool(tool_name, query, serverless = False): """Handle callTool method.""" config = get_db_config() try: with connect_with_retry() as conn: with conn.cursor() as cursor: # 特殊处理 serverless computing 查询 if serverless: cursor.execute("set hg_computing_resource='serverless'") # Execute the query cursor.execute(query) # 特殊处理 ANALYZE 命令 if tool_name == "gather_hg_table_statistics": return f"Successfully {query}" # 处理其他有返回结果的查询 if cursor.description: # SELECT query columns = [desc[0] for desc in cursor.description] rows = cursor.fetchall() result = [",".join(map(str, row)) for row in rows] return "\n".join([",".join(columns)] + result) elif tool_name == "execute_dml_sql": # Non-SELECT query row_count = cursor.rowcount return f"Query executed successfully. {row_count} rows affected." else: return "Query executed successfully" except Exception as e: return f"Error executing query: {str(e)}"