PostgreSQL-Performance-Tuner-Mcp

Instructions

Implementation Reference

• src/pgtuner_mcp/tools/tools_index.py:670-791 (handler)

  The main handler function that executes the tool logic: queries pg_stat_user_indexes for unused indexes (0 scans), detects duplicate/overlapping indexes, generates DROP recommendations, and returns JSON summary.

  async def run_tool(self, arguments: dict[str, Any]) -> Sequence[TextContent]:
      try:
          schema_name = arguments.get("schema_name", "public")
          min_size_mb = arguments.get("min_size_mb", 0)
          include_duplicates = arguments.get("include_duplicates", True)
  
          # Query only user indexes, excluding system schemas
          unused_query = """
              SELECT
                  s.schemaname,
                  s.relname as table_name,
                  s.indexrelname as index_name,
                  s.idx_scan as scans,
                  s.idx_tup_read as tuples_read,
                  s.idx_tup_fetch as tuples_fetched,
                  pg_size_pretty(pg_relation_size(s.indexrelid)) as size,
                  pg_relation_size(s.indexrelid) as size_bytes,
                  pg_get_indexdef(s.indexrelid) as definition,
                  t.n_live_tup as table_rows
              FROM pg_stat_user_indexes s
              JOIN pg_stat_user_tables t ON s.relid = t.relid
              WHERE s.schemaname = %s
                AND s.schemaname NOT IN ('pg_catalog', 'information_schema', 'pg_toast')
                AND pg_relation_size(s.indexrelid) >= %s * 1024 * 1024
                AND s.indexrelname NOT LIKE '%%_pkey'
                AND s.idx_scan = 0
              ORDER BY pg_relation_size(s.indexrelid) DESC
          """
  
          unused_results = await self.sql_driver.execute_query(
              unused_query,
              [schema_name, min_size_mb]
          )
  
          output = {
              "schema": schema_name,
              "unused_indexes": unused_results,
              "unused_count": len(unused_results),
              "potential_savings_bytes": sum(
                  r.get("size_bytes", 0) for r in unused_results
              )
          }
  
          # Find duplicate/overlapping indexes (user tables only)
          if include_duplicates:
              duplicate_query = """
                  WITH index_cols AS (
                      SELECT
                          n.nspname as schema_name,
                          t.relname as table_name,
                          i.relname as index_name,
                          pg_get_indexdef(i.oid) as definition,
                          array_agg(a.attname ORDER BY k.n) as columns,
                          pg_relation_size(i.oid) as size_bytes
                      FROM pg_index x
                      JOIN pg_class t ON t.oid = x.indrelid
                      JOIN pg_class i ON i.oid = x.indexrelid
                      JOIN pg_namespace n ON n.oid = t.relnamespace
                      CROSS JOIN unnest(x.indkey) WITH ORDINALITY AS k(attnum, n)
                      JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = k.attnum
                      WHERE n.nspname = %s
                        AND n.nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast')
                      GROUP BY n.nspname, t.relname, i.relname, i.oid
                  )
                  SELECT
                      a.table_name,
                      a.index_name as index1,
                      a.columns as columns1,
                      a.definition as definition1,
                      a.size_bytes as size1,
                      b.index_name as index2,
                      b.columns as columns2,
                      b.definition as definition2,
                      b.size_bytes as size2,
                      CASE
                          WHEN a.columns = b.columns THEN 'duplicate'
                          WHEN a.columns[1:array_length(b.columns, 1)] = b.columns THEN 'overlapping'
                          ELSE 'related'
                      END as relationship
                  FROM index_cols a
                  JOIN index_cols b ON a.table_name = b.table_name
                      AND a.index_name < b.index_name
                  WHERE a.columns = b.columns
                     OR a.columns[1:array_length(b.columns, 1)] = b.columns
              """
  
              duplicate_results = await self.sql_driver.execute_query(
                  duplicate_query,
                  [schema_name]
              )
  
              output["duplicate_indexes"] = duplicate_results
              output["duplicate_count"] = len(duplicate_results)
  
          # Generate recommendations
          recommendations = []
  
          for idx in unused_results[:5]:
              size = idx.get("size", "unknown")
              recommendations.append(
                  f"DROP INDEX {schema_name}.{idx['index_name']}; -- {size}, 0 scans"
              )
  
          if include_duplicates:
              for dup in duplicate_results[:5]:
                  if dup["relationship"] == "duplicate":
                      # Recommend dropping the larger one
                      if dup["size1"] > dup["size2"]:
                          recommendations.append(
                              f"DROP INDEX {schema_name}.{dup['index1']}; -- duplicate of {dup['index2']}"
                          )
                      else:
                          recommendations.append(
                              f"DROP INDEX {schema_name}.{dup['index2']}; -- duplicate of {dup['index1']}"
                          )
  
          output["recommendations"] = recommendations
  
          return self.format_json_result(output)
  
      except Exception as e:
          return self.format_error(e)

• src/pgtuner_mcp/tools/tools_index.py:642-668 (schema)

  Defines the tool schema including input parameters: schema_name, min_size_mb, include_duplicates.

  def get_tool_definition(self) -> Tool:
      return Tool(
          name=self.name,
          description=self.description,
          inputSchema={
              "type": "object",
              "properties": {
                  "schema_name": {
                      "type": "string",
                      "description": "Schema to analyze (default: public)",
                      "default": "public"
                  },
                  "min_size_mb": {
                      "type": "number",
                      "description": "Minimum index size in MB to include",
                      "default": 0
                  },
                  "include_duplicates": {
                      "type": "boolean",
                      "description": "Include analysis of duplicate/overlapping indexes",
                      "default": True
                  }
              },
              "required": []
          },
          annotations=self.get_annotations()
      )

• src/pgtuner_mcp/server.py:136-169 (registration)

  Registers all tool handlers including UnusedIndexesToolHandler(driver) via add_tool_handler in the central register_all_tools function.

  def register_all_tools() -> None:
      """
      Register all available tool handlers.
  
      This function serves as the central registry for all tools.
      New tool handlers should be added here for automatic registration.
      """
      driver = get_sql_driver()
      hypopg_service = HypoPGService(driver)
      index_advisor = IndexAdvisor(driver)
  
      # Performance analysis tools
      add_tool_handler(GetSlowQueriesToolHandler(driver))
      add_tool_handler(AnalyzeQueryToolHandler(driver))
      add_tool_handler(TableStatsToolHandler(driver))
  
      # Index tuning tools
      add_tool_handler(IndexAdvisorToolHandler(index_advisor))
      add_tool_handler(ExplainQueryToolHandler(driver, hypopg_service))
      add_tool_handler(HypoPGToolHandler(hypopg_service))
      add_tool_handler(UnusedIndexesToolHandler(driver))
  
      # Database health tools
      add_tool_handler(DatabaseHealthToolHandler(driver))
      add_tool_handler(ActiveQueriesToolHandler(driver))
      add_tool_handler(WaitEventsToolHandler(driver))
      add_tool_handler(DatabaseSettingsToolHandler(driver))
  
      # Bloat detection tools (using pgstattuple extension)
      add_tool_handler(TableBloatToolHandler(driver))
      add_tool_handler(IndexBloatToolHandler(driver))
      add_tool_handler(DatabaseBloatSummaryToolHandler(driver))
  
      logger.info(f"Registered {len(tool_handlers)} tool handlers")

• src/pgtuner_mcp/server.py:156-156 (registration)

  Specific registration line for the UnusedIndexesToolHandler.

  add_tool_handler(UnusedIndexesToolHandler(driver))