get_hg_query_plan
Analyze SQL query execution plans in Hologres databases to optimize performance and troubleshoot issues.
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 logic for get_hg_query_plan tool: extracts the query argument, validates it, and prefixes the SQL with 'EXPLAIN ' before delegating to handle_call_tool.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)Registers the 'get_hg_query_plan' tool in the MCP server's list_tools() function, including its name, description, and input schema.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"] } ),
- Pydantic input schema definition for the get_hg_query_plan tool, requiring a 'query' string.inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to analyze in Hologres database" } }, "required": ["query"] }
- Helper function that executes the prepared SQL query (EXPLAIN <user_query>) on the Hologres database, formats the result as CSV-like text, and handles errors.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)}"
- Helper function to establish a connection to the Hologres database with retry logic, used by handle_call_tool.def connect_with_retry(retries=3): attempt = 0 err_msg = "" while attempt <= retries: try: config = get_db_config() conn = psycopg.connect(**config) conn.autocommit = True with conn.cursor() as cursor: cursor.execute("SELECT 1;") cursor.fetchone() return conn except psycopg.Error as e: err_msg = f"Connection failed: {e}" attempt += 1 if attempt <= retries: print(f"Retrying connection (attempt {attempt + 1} of {retries + 1})...") time.sleep(5) # 等待 2 秒后再次尝试连接 raise psycopg.Error(f"Failed to connect to Hologres database after retrying: {err_msg}")