gather_hg_table_statistics
Collect table statistics in Hologres databases to improve query optimization and generate better query execution plans.
Instructions
Execute the ANALYZE TABLE command to have Hologres collect table statistics, enabling QO to generate better query plans
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| schema | Yes | Schema name in Hologres database | |
| table | Yes | Table name in Hologres database |
Implementation Reference
- Dispatches the tool call by validating schema and table arguments, then constructs the ANALYZE query for gathering table statistics.
elif name == "gather_hg_table_statistics": schema = arguments.get("schema") table = arguments.get("table") if not all([schema, table]): raise ValueError("Schema and table are required") query = f"ANALYZE {schema}.{table}" elif name == "get_hg_query_plan": - Defines the input schema requiring 'schema' and 'table' strings.
inputSchema={ "type": "object", "properties": { "schema": { "type": "string", "description": "Schema name in Hologres database" }, "table": { "type": "string", "description": "Table name in Hologres database" } }, "required": ["schema", "table"] } - src/hologres_mcp_server/server.py:404-421 (registration)Registers the tool in the list_tools() function with name, description, and schema.
Tool( name="gather_hg_table_statistics", description="Execute the ANALYZE TABLE command to have Hologres collect table statistics, enabling QO to generate better query plans", inputSchema={ "type": "object", "properties": { "schema": { "type": "string", "description": "Schema name in Hologres database" }, "table": { "type": "string", "description": "Table name in Hologres database" } }, "required": ["schema", "table"] } ), - Executes the constructed query on the database connection. For this tool, specially returns a success message after execution since ANALYZE does not return query results.
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)}" - Special handling for ANALYZE command: returns success message without fetching results.
# 特殊处理 ANALYZE 命令 if tool_name == "gather_hg_table_statistics": return f"Successfully {query}"