get_table_stats
Retrieve table statistics including row count and size from CockroachDB to analyze database performance and storage usage.
Instructions
Get statistics for a table.
Args:
table: Table name (schema.table or just table).
Returns:
Table statistics including row count and size.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| table | Yes |
Implementation Reference
- Core handler function that executes the tool logic: parses table name, validates schema, queries CockroachDB for row count and table size using pg_total_relation_size, returns formatted statistics.async def get_table_stats(table: str) -> dict[str, Any]: """Get statistics for a table. Args: table: Table name (schema.table or just table). Returns: Table statistics. """ conn = await connection_manager.ensure_connected() # Parse schema and table name if "." in table: schema, table_name = table.rsplit(".", 1) else: schema = "public" table_name = table # Check if schema is allowed if not _is_allowed_schema(schema): return {"status": "error", "error": f"Schema '{schema}' is not allowed"} try: # Get table size and row count async with conn.cursor() as cur: # Row count await cur.execute(f"SELECT COUNT(*) as count FROM {schema}.{table_name}") count_row = await cur.fetchone() row_count = count_row["count"] if count_row else 0 # Table size using CockroachDB specific query await cur.execute( """ SELECT pg_size_pretty(pg_total_relation_size(%s::regclass)) as total_size, pg_total_relation_size(%s::regclass) as total_bytes """, (f"{schema}.{table_name}", f"{schema}.{table_name}"), ) size_row = await cur.fetchone() return { "schema": schema, "table": table_name, "full_name": f"{schema}.{table_name}", "row_count": row_count, "total_size": size_row.get("total_size") if size_row else None, "total_bytes": size_row.get("total_bytes") if size_row else None, } except Exception as e: return {"status": "error", "error": str(e)}
- src/cockroachdb_mcp/server.py:189-202 (registration)MCP tool registration via @mcp.tool() decorator. Thin wrapper handler that delegates to the core implementation in tables.get_table_stats with error handling.@mcp.tool() async def get_table_stats(table: str) -> dict[str, Any]: """Get statistics for a table. Args: table: Table name (schema.table or just table). Returns: Table statistics including row count and size. """ try: return await tables.get_table_stats(table) except Exception as e: return {"status": "error", "error": str(e)}