ZQYY.FQ-ETL/database/schema_verify_perf_indexes.sql

SET client_encoding TO "UTF8";

-- ============================================================================
-- 校验性能索引（ODS / DWD）
-- ----------------------------------------------------------------------------
-- 用途:
--   1) 加速校验查询（主键查找、窗口扫描、当前版本扫描）。
--   2) 保持数据语义不变（仅添加索引 + ANALYZE，不改写业务数据）。
--
-- 注意事项:
--   1) 本脚本具有幂等性（`CREATE INDEX IF NOT EXISTS`）。
--   2) 如有严格的在线 DDL 要求，请手动使用 `CREATE INDEX CONCURRENTLY`
--      在维护安全模式下执行（不可在事务块内运行）。
-- ============================================================================

DO $$
DECLARE
    rec RECORD;
    pk_cols TEXT[];
    pk_cols_sql TEXT;
    idx_name TEXT;
BEGIN
    FOR rec IN
        SELECT table_name
        FROM information_schema.tables
        WHERE table_schema = 'billiards_ods'
          AND table_type = 'BASE TABLE'
    LOOP
        IF EXISTS (
            SELECT 1
            FROM information_schema.columns
            WHERE table_schema = 'billiards_ods'
              AND table_name = rec.table_name
              AND column_name = 'fetched_at'
        ) THEN
            idx_name := left(format('idx_%s_vfy_fetched_at', rec.table_name), 50)
                        || '_' || substr(md5(rec.table_name || '_vfy_fetched_at'), 1, 8);
            EXECUTE format(
                'CREATE INDEX IF NOT EXISTS %I ON billiards_ods.%I (fetched_at)',
                idx_name, rec.table_name
            );

            SELECT array_agg(kcu.column_name ORDER BY kcu.ordinal_position)
            INTO pk_cols
            FROM information_schema.table_constraints tc
            JOIN information_schema.key_column_usage kcu
              ON tc.table_schema = kcu.table_schema
             AND tc.table_name = kcu.table_name
             AND tc.constraint_name = kcu.constraint_name
            WHERE tc.table_schema = 'billiards_ods'
              AND tc.table_name = rec.table_name
              AND tc.constraint_type = 'PRIMARY KEY';

            IF pk_cols IS NOT NULL AND coalesce(array_length(pk_cols, 1), 0) <= 3 THEN
                SELECT string_agg(format('%I', c), ', ')
                INTO pk_cols_sql
                FROM unnest(pk_cols) AS c;

                idx_name := left(format('idx_%s_vfy_fetched_pk', rec.table_name), 50)
                            || '_' || substr(md5(rec.table_name || '_vfy_fetched_pk'), 1, 8);
                EXECUTE format(
                    'CREATE INDEX IF NOT EXISTS %I ON billiards_ods.%I (fetched_at, %s)',
                    idx_name, rec.table_name, pk_cols_sql
                );
            END IF;
        END IF;
    END LOOP;
END
$$;

DO $$
DECLARE
    rec RECORD;
    tcol TEXT;
    pk_cols TEXT[];
    pk_cols_sql TEXT;
    idx_name TEXT;
    time_candidates TEXT[] := ARRAY[
        'pay_time',
        'create_time',
        'start_use_time',
        'scd2_start_time',
        'calc_time',
        'order_date',
        'fetched_at'
    ];
BEGIN
    FOR rec IN
        SELECT table_name
        FROM information_schema.tables
        WHERE table_schema = 'billiards_dwd'
          AND table_type = 'BASE TABLE'
    LOOP
        SELECT array_agg(kcu.column_name ORDER BY kcu.ordinal_position)
        INTO pk_cols
        FROM information_schema.table_constraints tc
        JOIN information_schema.key_column_usage kcu
          ON tc.table_schema = kcu.table_schema
         AND tc.table_name = kcu.table_name
         AND tc.constraint_name = kcu.constraint_name
        WHERE tc.table_schema = 'billiards_dwd'
          AND tc.table_name = rec.table_name
          AND tc.constraint_type = 'PRIMARY KEY';

        IF EXISTS (
            SELECT 1
            FROM information_schema.columns
            WHERE table_schema = 'billiards_dwd'
              AND table_name = rec.table_name
              AND column_name = 'scd2_is_current'
        ) AND pk_cols IS NOT NULL
          AND coalesce(array_length(pk_cols, 1), 0) BETWEEN 1 AND 4 THEN
            SELECT string_agg(format('%I', c), ', ')
            INTO pk_cols_sql
            FROM unnest(pk_cols) AS c;

            idx_name := left(format('idx_%s_vfy_pk_current', rec.table_name), 50)
                        || '_' || substr(md5(rec.table_name || '_vfy_pk_current'), 1, 8);
            EXECUTE format(
                'CREATE INDEX IF NOT EXISTS %I ON billiards_dwd.%I (%s, scd2_is_current)',
                idx_name, rec.table_name, pk_cols_sql
            );
        END IF;

        FOREACH tcol IN ARRAY time_candidates
        LOOP
            IF EXISTS (
                SELECT 1
                FROM information_schema.columns
                WHERE table_schema = 'billiards_dwd'
                  AND table_name = rec.table_name
                  AND column_name = tcol
            ) THEN
                idx_name := left(format('idx_%s_vfy_%s', rec.table_name, tcol), 50)
                            || '_' || substr(md5(rec.table_name || '_vfy_' || tcol), 1, 8);
                EXECUTE format(
                    'CREATE INDEX IF NOT EXISTS %I ON billiards_dwd.%I (%I)',
                    idx_name, rec.table_name, tcol
                );

                IF pk_cols IS NOT NULL AND coalesce(array_length(pk_cols, 1), 0) <= 3 THEN
                    SELECT string_agg(format('%I', c), ', ')
                    INTO pk_cols_sql
                    FROM unnest(pk_cols) AS c;

                    idx_name := left(format('idx_%s_vfy_%s_pk', rec.table_name, tcol), 50)
                                || '_' || substr(md5(rec.table_name || '_vfy_' || tcol || '_pk'), 1, 8);
                    EXECUTE format(
                        'CREATE INDEX IF NOT EXISTS %I ON billiards_dwd.%I (%I, %s)',
                        idx_name, rec.table_name, tcol, pk_cols_sql
                    );
                END IF;
            END IF;
        END LOOP;
    END LOOP;
END
$$;

DO $$
DECLARE
    rec RECORD;
BEGIN
    FOR rec IN
        SELECT table_schema, table_name
        FROM information_schema.tables
        WHERE table_schema IN ('billiards_ods', 'billiards_dwd')
          AND table_type = 'BASE TABLE'
    LOOP
        EXECUTE format('ANALYZE %I.%I', rec.table_schema, rec.table_name);
    END LOOP;
END
$$;