自动化PostgreSQL数据库索引创建与删除方案问询

我正在尝试实现PostgreSQL数据库的自动化分析与索引管理流程——利用pg_stat_statements采集查询数据，通过Rust代码按周期自动分析并执行索引的创建/删除操作，目标是精准优化数据库性能。目前还没测试这个方案，希望大家给点反馈或改进建议，以下是我写的Rust代码：

use sqlx::postgres::{PgPool, PgPoolOptions};
use sqlx::{Error, Row};
use std::collections::{HashMap, HashSet};
use tokio::time::{self, Duration};
use log::{info, error};

const INDEX_THRESHOLD_CALLS: i64 = 100; // 考虑创建索引的最小调用次数
const INDEX_DELETE_THRESHOLD: f64 = 0.2; // 删除索引的百分比阈值

#[tokio::main]
async fn main() -> Result<(), Error> {
    env_logger::init();

    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect("postgres://username:password@localhost/dbname")
        .await?;

    schedule_index_management(pool.clone()).await;

    Ok(())
}

async fn schedule_index_management(pool: PgPool) {
    let mut interval = time::interval(Duration::from_secs(24 * 60 * 60)); // 24小时执行一次

    loop {
        interval.tick().await;

        info!("Running index management task...");

        if let Err(e) = manage_indexes(&pool).await {
            error!("Failed to manage indexes: {:?}", e);
        }

        info!("Index management task completed.");
    }
}

async fn manage_indexes(pool: &PgPool) -> Result<(), Error> {
    let (indexes_to_create, indexes_to_delete) = analyze_queries(pool).await?;

    create_indexes(pool, indexes_to_create).await?;
    delete_indexes(pool, indexes_to_delete).await?;

    Ok(())
}

async fn analyze_queries(pool: &PgPool) -> Result<(Vec<String>, Vec<String>), Error> {
    let rows = sqlx::query(
        "SELECT query, calls, mean_exec_time 
         FROM pg_stat_statements 
         ORDER BY calls DESC"
    )
    .fetch_all(pool)
    .await?;

    let mut indexes_to_create = Vec::new();
    let mut indexes_to_delete = Vec::new();
    let mut existing_indexes = HashSet::new();
    let mut query_usage = HashMap::new();

    let existing_indexes_rows = sqlx::query("SELECT indexname FROM pg_indexes WHERE schemaname = 'public'")
        .fetch_all(pool)
        .await?;

    for row in existing_indexes_rows {
        let index_name: String = row.get("indexname");
        existing_indexes.insert(index_name);
    }

    for row in rows {
        let query: String = row.get("query");
        let calls: i64 = row.get("calls");

        analyze_query(&query, calls, &mut indexes_to_create, &mut query_usage);
    }

    identify_indexes_to_delete(&query_usage, &existing_indexes, &mut indexes_to_delete);

    Ok((indexes_to_create, indexes_to_delete))
}

fn analyze_query(
    query: &str,
    calls: i64,
    indexes_to_create: &mut Vec<String>,
    query_usage: &mut HashMap<String, i64>,
) {
    if calls >= INDEX_THRESHOLD_CALLS {
        // 示例：实现更复杂的逻辑从查询中提取表和列
        if let Some((table, column)) = extract_table_and_column(query) {
            // 示例：根据查询类型和使用模式确定合适的索引类型
            let index_type = determine_index_type(query);

            let index_name = format!("idx_{}_{}", table, column);
            let index_query = match index_type {
                IndexType::BTree => format!("CREATE INDEX IF NOT EXISTS {} ON {} USING btree ({})", index_name, table, column),
                IndexType::Hash => format!("CREATE INDEX IF NOT EXISTS {} ON {} USING hash ({})", index_name, table, column),
                IndexType::GiST => format!("CREATE INDEX IF NOT EXISTS {} ON {} USING gist ({})", index_name, table, column),
                IndexType::GIN => format!("CREATE INDEX IF NOT EXISTS {} ON {} USING gin ({})", index_name, table, column),
                IndexType::BRIN => format!("CREATE INDEX IF NOT EXISTS {} ON {} USING brin ({})", index_name, table, column),
            };

            indexes_to_create.push(index_query.clone());
            query_usage.insert(index_name.clone(), calls);
        }
    }
}

fn extract_table_and_column(query: &str) -> Option<(String, String)> {
    // 从查询中提取表和列的占位逻辑
    if query.contains("WHERE") {
        // 示例：解析简单的WHERE条件
        let parts: Vec<&str> = query.split_whitespace().collect();
        if parts.len() > 4 {
            let table = parts[3].to_string();
            let column = parts[5].split('=').next()?.to_string();
            return Some((table, column));
        }
    }
    None
}

enum IndexType {
    BTree,
    Hash,
    GiST,
    GIN,
    BRIN,
}

fn determine_index_type(query: &str) -> IndexType {
    // 示例：根据查询类型和使用模式确定合适的索引类型
    if query.contains("JOIN") {
        IndexType::Hash // 示例：对JOIN使用哈希索引
    } else if query.contains("WHERE") {
        IndexType::BTree // 示例：对WHERE子句使用B-tree索引
    } else {
        IndexType::BTree // 默认使用B-tree索引
    }
}

fn identify_indexes_to_delete(
    query_usage: &HashMap<String, i64>,
    existing_indexes: &HashSet<String>,
    indexes_to_delete: &mut Vec<String>,
) {
    for index in existing_indexes {
        let usage = query_usage.get(index).copied().unwrap_or(0);
        if usage < (INDEX_THRESHOLD_CALLS as f64 * INDEX_DELETE_THRESHOLD) as i64 {
            indexes_to_delete.push(format!("DROP INDEX IF EXISTS {}", index));
        }
    }
}

async fn create_indexes(pool: &PgPool, indexes: Vec<String>) -> Result<(), Error> {
    for index in indexes {
        if let Err(e) = sqlx::query(&index).execute(pool).await {
            error!("Failed to create index: {}", e);
        } else {
            info!("Created index: {}", index);
        }
    }
    Ok(())
}

async fn delete_indexes(pool: &PgPool, indexes: Vec<String>) -> Result<(), Error> {
    for index in indexes {
        if let Err(e) = sqlx::query(&index).execute(pool).await {
            error!("Failed to delete index: {}", e);
        } else {
            info!("Deleted index: {}", index);
        }
    }
    Ok(())
}

改进建议与反馈 #

• 查询解析逻辑强化：当前extract_table_and_column只能处理极简的WHERE语句，实际生产中查询会有嵌套、表别名、多表关联、子查询等复杂场景。建议：

  • 用Rust的sqlparser库做专业SQL解析，精准提取涉及的表、列和操作类型；
  • 结合pg_stat_statements的queryid，调用pg_get_query(queryid)获取标准化后的查询语句，避免因参数不同导致的重复解析。

• 索引类型选择逻辑修正：

  • Hash索引在PostgreSQL中仅支持等值查询，且不支持排序，JOIN场景更适合用B-tree（通用场景）或GiST（特殊数据类型如几何类型），不要默认给JOIN分配Hash索引；
  • GiST/GIN/BRIN的选择要结合数据类型：全文搜索用GIN，几何/地理类型用GiST，超大分区表用BRIN，不能仅靠关键字判断。

• 索引删除判断逻辑优化：

  • 当前仅依赖本次采集的查询调用数判断索引是否有用，会遗漏低频但关键的查询（如周/月报表）。建议结合pg_stat_user_indexes中的idx_scan字段（索引被扫描的次数）来判断真实使用情况；
  • 加入保护规则：排除主键、唯一索引，以及通过自定义元数据表标记的"保留索引"，避免误删核心索引。

• 性能与稳定性提升：

  • 创建索引时添加CONCURRENTLY关键字，避免锁表影响业务，但注意CREATE INDEX CONCURRENTLY不能在事务中执行，代码要单独处理这类语句；
  • 把阈值、执行周期、目标Schema等参数改成可配置项（从环境变量或配置文件读取），不要硬编码；
  • 增加操作日志：将每次创建/删除的索引信息记录到数据库表中，方便回溯和审计；
  • 调整连接池配置：根据数据库实际负载设置合理的最大连接数，避免占用过多连接资源。

• 数据采集准确性优化：

  • 确保pg_stat_statements正确配置：track_activity_query_size要足够大以容纳长查询，pg_stat_statements.track设为all以采集所有类型的查询；
  • 分析时过滤掉系统内部查询（如pg_catalog相关的语句），只处理用户业务查询。

• 测试与验证机制：

  • 增加dry-run模式：只输出拟执行的索引操作，不实际执行，方便验证逻辑正确性；
  • 加入效果评估：记录操作前后的查询平均耗时、索引扫描次数等指标，验证优化效果。

内容的提问来源于stack exchange，提问作者Awm