🗄️ 데이터베이스

인덱스

Index

데이터베이스 검색 속도를 높이기 위한 자료구조입니다. 책의 색인처럼 특정 데이터를 빠르게 찾을 수 있도록 별도의 구조를 만들어 쿼리 성능을 획기적으로 향상시킵니다.

📖 상세 설명

인덱스(Index)는 데이터베이스 테이블의 검색 속도를 향상시키기 위한 자료구조입니다. 책의 색인(Index)처럼 특정 키 값과 해당 데이터의 위치(포인터)를 매핑하여, 전체 테이블을 스캔하지 않고도 원하는 데이터를 빠르게 찾을 수 있게 합니다.

B-Tree 인덱스는 가장 일반적인 인덱스 유형입니다. 균형 잡힌 트리 구조로 O(log n) 시간 복잡도를 가지며, 범위 검색(BETWEEN, <, >)과 정렬(ORDER BY)에 효과적입니다. PostgreSQL, MySQL의 기본 인덱스 유형이며, 등호(=) 검색과 범위 검색 모두 지원합니다.

Hash 인덱스는 해시 함수를 사용하여 O(1) 시간 복잡도로 정확한 일치 검색(=)에 최적화되어 있습니다. 하지만 범위 검색이나 정렬에는 사용할 수 없으며, 메모리 기반 데이터베이스나 캐시에서 주로 활용됩니다.

복합 인덱스(Composite Index)는 두 개 이상의 컬럼을 조합한 인덱스입니다. 컬럼 순서가 중요하며, 왼쪽 컬럼부터 순서대로 사용될 때 효과적입니다. (A, B, C) 인덱스는 A 단독, (A, B), (A, B, C) 검색에 사용되지만, B나 C 단독 검색에는 사용되지 않습니다.

커버링 인덱스는 쿼리에 필요한 모든 컬럼이 인덱스에 포함되어 테이블 접근 없이 인덱스만으로 결과를 반환하는 경우입니다. 디스크 I/O를 크게 줄여 성능이 향상됩니다. INCLUDE 절이나 복합 인덱스로 구현할 수 있습니다.

💻 코드 예제

-- 인덱스 생성 및 활용 예제 (PostgreSQL/MySQL)

-- ============================================
-- 1. 기본 인덱스 생성
-- ============================================
-- 단일 컬럼 인덱스
CREATE INDEX idx_users_email ON users(email);

-- 유니크 인덱스 (중복 불허)
CREATE UNIQUE INDEX idx_users_email_unique ON users(email);

-- 부분 인덱스 (PostgreSQL) - 특정 조건의 행만 인덱싱
CREATE INDEX idx_orders_pending ON orders(created_at)
WHERE status = 'pending';

-- 함수 기반 인덱스 (표현식 인덱스)
CREATE INDEX idx_users_email_lower ON users(LOWER(email));


-- ============================================
-- 2. 복합 인덱스 (Composite Index)
-- ============================================
-- 컬럼 순서가 중요! 왼쪽부터 사용됨
CREATE INDEX idx_orders_user_date ON orders(user_id, created_at);

-- 이 인덱스가 사용되는 쿼리:
-- SELECT * FROM orders WHERE user_id = 1;  -- O
-- SELECT * FROM orders WHERE user_id = 1 AND created_at > '2024-01-01';  -- O
-- SELECT * FROM orders WHERE created_at > '2024-01-01';  -- X (user_id 없음)

-- 커버링 인덱스 (INCLUDE 사용)
CREATE INDEX idx_orders_covering ON orders(user_id, created_at)
INCLUDE (total_amount, status);


-- ============================================
-- 3. 특수 인덱스 유형 (PostgreSQL)
-- ============================================
-- GIN 인덱스 - 배열, JSON, 전문 검색
CREATE INDEX idx_products_tags ON products USING GIN(tags);
CREATE INDEX idx_docs_content ON documents USING GIN(to_tsvector('korean', content));

-- GiST 인덱스 - 지리 데이터, 범위
CREATE INDEX idx_places_location ON places USING GIST(location);

-- BRIN 인덱스 - 대용량 순차 데이터 (시계열)
CREATE INDEX idx_logs_timestamp ON logs USING BRIN(timestamp);


-- ============================================
-- 4. 인덱스 분석 및 실행 계획
-- ============================================
-- 실행 계획 확인
EXPLAIN ANALYZE
SELECT * FROM users WHERE email = 'test@example.com';

-- 예상 결과:
-- Index Scan using idx_users_email on users
--   Index Cond: (email = 'test@example.com')
--   Actual Time: 0.015..0.016 ms

-- 테이블 스캔 vs 인덱스 스캔 비교
EXPLAIN ANALYZE
SELECT * FROM orders WHERE user_id = 1;  -- Index Scan

EXPLAIN ANALYZE
SELECT * FROM orders WHERE total_amount > 50000;  -- Seq Scan (인덱스 없음)


-- ============================================
-- 5. 인덱스 관리
-- ============================================
-- 인덱스 목록 조회 (PostgreSQL)
SELECT indexname, indexdef
FROM pg_indexes
WHERE tablename = 'orders';

-- 인덱스 사용 통계
SELECT schemaname, relname, indexrelname, idx_scan, idx_tup_read
FROM pg_stat_user_indexes
WHERE relname = 'orders';

-- 사용되지 않는 인덱스 찾기
SELECT indexrelname, idx_scan
FROM pg_stat_user_indexes
WHERE idx_scan = 0 AND schemaname = 'public';

-- 인덱스 삭제
DROP INDEX idx_unused_index;

-- 인덱스 재구성 (단편화 해소)
REINDEX INDEX idx_orders_user_date;
-- 또는 동시 재생성 (락 최소화)
CREATE INDEX CONCURRENTLY idx_orders_new ON orders(user_id);


-- ============================================
-- 6. 인덱스 힌트 (MySQL)
-- ============================================
-- 특정 인덱스 강제 사용
SELECT * FROM orders USE INDEX (idx_orders_user_date)
WHERE user_id = 1;

-- 인덱스 사용 금지
SELECT * FROM orders IGNORE INDEX (idx_orders_user_date)
WHERE user_id = 1;
# 인덱스 분석 및 최적화 도구 - Python
import psycopg2
from dataclasses import dataclass
from typing import List, Dict

@dataclass
class IndexInfo:
    """인덱스 정보 클래스"""
    name: str
    table: str
    columns: List[str]
    size_mb: float
    scan_count: int
    is_unique: bool

class IndexAnalyzer:
    """인덱스 분석 도구"""

    def __init__(self, connection_string: str):
        self.conn_string = connection_string

    def get_all_indexes(self, schema: str = 'public') -> List[IndexInfo]:
        """모든 인덱스 정보 조회"""
        query = """
            SELECT
                i.relname as index_name,
                t.relname as table_name,
                array_agg(a.attname) as columns,
                pg_relation_size(i.oid) / 1024.0 / 1024.0 as size_mb,
                COALESCE(s.idx_scan, 0) as scan_count,
                ix.indisunique as is_unique
            FROM pg_index ix
            JOIN pg_class i ON i.oid = ix.indexrelid
            JOIN pg_class t ON t.oid = ix.indrelid
            JOIN pg_namespace n ON n.oid = t.relnamespace
            JOIN pg_attribute a ON a.attrelid = t.oid
                AND a.attnum = ANY(ix.indkey)
            LEFT JOIN pg_stat_user_indexes s ON s.indexrelid = i.oid
            WHERE n.nspname = %s
            GROUP BY i.relname, t.relname, i.oid, s.idx_scan, ix.indisunique
            ORDER BY size_mb DESC
        """

        with psycopg2.connect(self.conn_string) as conn:
            with conn.cursor() as cur:
                cur.execute(query, (schema,))
                return [
                    IndexInfo(
                        name=row[0],
                        table=row[1],
                        columns=row[2],
                        size_mb=float(row[3]),
                        scan_count=row[4],
                        is_unique=row[5]
                    )
                    for row in cur.fetchall()
                ]

    def find_unused_indexes(self, min_size_mb: float = 1.0) -> List[IndexInfo]:
        """사용되지 않는 인덱스 찾기"""
        all_indexes = self.get_all_indexes()
        return [
            idx for idx in all_indexes
            if idx.scan_count == 0
            and idx.size_mb >= min_size_mb
            and not idx.name.endswith('_pkey')  # PK 제외
        ]

    def find_duplicate_indexes(self) -> List[tuple]:
        """중복 인덱스 찾기 (같은 컬럼 조합)"""
        query = """
            SELECT
                array_agg(indexname) as index_names,
                tablename,
                array_agg(indexdef) as definitions
            FROM pg_indexes
            WHERE schemaname = 'public'
            GROUP BY tablename,
                regexp_replace(indexdef, '.* USING [^ ]+ \\((.*)\\)', '\\1')
            HAVING COUNT(*) > 1
        """

        with psycopg2.connect(self.conn_string) as conn:
            with conn.cursor() as cur:
                cur.execute(query)
                return cur.fetchall()

    def analyze_query_plan(self, query: str) -> Dict:
        """쿼리 실행 계획 분석"""
        explain_query = f"EXPLAIN (ANALYZE, FORMAT JSON) {query}"

        with psycopg2.connect(self.conn_string) as conn:
            with conn.cursor() as cur:
                cur.execute(explain_query)
                plan = cur.fetchone()[0][0]

                return {
                    'total_time_ms': plan['Execution Time'],
                    'planning_time_ms': plan['Planning Time'],
                    'node_type': plan['Plan']['Node Type'],
                    'uses_index': 'Index' in plan['Plan']['Node Type'],
                    'index_name': plan['Plan'].get('Index Name'),
                    'rows_returned': plan['Plan'].get('Actual Rows', 0)
                }

    def suggest_indexes(self, table: str) -> List[str]:
        """인덱스 제안 (자주 사용되는 WHERE 조건 기반)"""
        # pg_stat_statements 확장 필요
        query = """
            SELECT query
            FROM pg_stat_statements
            WHERE query ILIKE %s
            AND calls > 10
            ORDER BY total_time DESC
            LIMIT 20
        """

        suggestions = []
        # 실제 구현에서는 쿼리 파싱 후 WHERE 절 분석
        return suggestions


# 사용 예시
if __name__ == "__main__":
    analyzer = IndexAnalyzer("postgresql://localhost/mydb")

    # 사용되지 않는 인덱스 찾기
    unused = analyzer.find_unused_indexes(min_size_mb=10)
    print("🔍 사용되지 않는 인덱스 (10MB 이상):")
    for idx in unused:
        print(f"  - {idx.name}: {idx.size_mb:.2f}MB")

    # 중복 인덱스 찾기
    duplicates = analyzer.find_duplicate_indexes()
    print("\n⚠️ 중복 인덱스:")
    for dup in duplicates:
        print(f"  - 테이블 {dup[1]}: {dup[0]}")

    # 쿼리 분석
    result = analyzer.analyze_query_plan(
        "SELECT * FROM orders WHERE user_id = 1"
    )
    print(f"\n📊 쿼리 분석:")
    print(f"  - 인덱스 사용: {result['uses_index']}")
    print(f"  - 실행 시간: {result['total_time_ms']:.3f}ms")
// 인덱스 관리 예제 - Node.js (Prisma, Sequelize)
const { PrismaClient } = require('@prisma/client');

// ============================================
// 1. Prisma 스키마에서 인덱스 정의
// ============================================
/*
// schema.prisma
model User {
  id        Int      @id @default(autoincrement())
  email     String   @unique
  name      String
  createdAt DateTime @default(now())

  // 복합 인덱스
  @@index([name, createdAt])
}

model Order {
  id        Int      @id @default(autoincrement())
  userId    Int
  status    String
  amount    Decimal
  createdAt DateTime @default(now())

  // 복합 인덱스 + 부분 인덱스 (PostgreSQL)
  @@index([userId, createdAt])
  @@index([status], map: "idx_orders_pending")
}
*/

// ============================================
// 2. 인덱스 활용 쿼리 최적화
// ============================================
const prisma = new PrismaClient({
    log: ['query'],  // 쿼리 로깅
});

class OrderRepository {
    /**
     * 인덱스를 활용한 효율적인 조회
     * idx_orders_user_date(user_id, created_at) 활용
     */
    async getUserOrders(userId, startDate, endDate) {
        return prisma.order.findMany({
            where: {
                userId: userId,  // 인덱스 첫 번째 컬럼
                createdAt: {
                    gte: startDate,  // 인덱스 두 번째 컬럼
                    lte: endDate,
                },
            },
            orderBy: {
                createdAt: 'desc',  // 인덱스 정렬 활용
            },
            select: {
                id: true,
                amount: true,
                status: true,
                createdAt: true,
                // 필요한 컬럼만 선택 (커버링 인덱스 효과)
            },
        });
    }

    /**
     * 페이지네이션 - 커서 기반 (인덱스 효율적)
     */
    async getOrdersCursor(userId, cursor, limit = 20) {
        return prisma.order.findMany({
            where: {
                userId: userId,
                ...(cursor && {
                    createdAt: {
                        lt: new Date(cursor),
                    },
                }),
            },
            orderBy: {
                createdAt: 'desc',
            },
            take: limit,
        });
    }

    /**
     * 집계 쿼리 - 인덱스 활용
     */
    async getUserOrderStats(userId) {
        return prisma.order.aggregate({
            where: {
                userId: userId,
            },
            _count: true,
            _sum: {
                amount: true,
            },
            _avg: {
                amount: true,
            },
        });
    }
}

// ============================================
// 3. Sequelize에서 인덱스 정의
// ============================================
const { Sequelize, DataTypes, Model } = require('sequelize');

const sequelize = new Sequelize('database', 'user', 'password', {
    dialect: 'postgres',
    logging: console.log,
});

class Order extends Model {}

Order.init({
    id: {
        type: DataTypes.INTEGER,
        primaryKey: true,
        autoIncrement: true,
    },
    userId: {
        type: DataTypes.INTEGER,
        allowNull: false,
    },
    status: {
        type: DataTypes.STRING,
        allowNull: false,
    },
    amount: {
        type: DataTypes.DECIMAL(10, 2),
        allowNull: false,
    },
}, {
    sequelize,
    modelName: 'Order',
    indexes: [
        // 복합 인덱스
        {
            name: 'idx_orders_user_date',
            fields: ['userId', 'createdAt'],
        },
        // 부분 인덱스
        {
            name: 'idx_orders_pending',
            fields: ['createdAt'],
            where: {
                status: 'pending',
            },
        },
        // 유니크 인덱스
        {
            unique: true,
            fields: ['userId', 'externalId'],
        },
    ],
});

// ============================================
// 4. 인덱스 사용 확인 유틸리티
// ============================================
async function analyzeQuery(query, params) {
    const [result] = await sequelize.query(
        `EXPLAIN (ANALYZE, FORMAT JSON) ${query}`,
        { bind: params }
    );

    const plan = result[0]['QUERY PLAN'][0];

    return {
        executionTime: plan['Execution Time'],
        nodeType: plan['Plan']['Node Type'],
        usesIndex: plan['Plan']['Node Type'].includes('Index'),
        indexName: plan['Plan']['Index Name'] || null,
    };
}

// 사용 예시
async function main() {
    const repo = new OrderRepository();

    // 인덱스 활용 쿼리
    const orders = await repo.getUserOrders(
        123,
        new Date('2024-01-01'),
        new Date('2024-12-31')
    );
    console.log(`조회된 주문: ${orders.length}건`);

    // 쿼리 분석
    const analysis = await analyzeQuery(
        'SELECT * FROM "Orders" WHERE "userId" = $1',
        [123]
    );
    console.log('쿼리 분석:', analysis);
}

main().catch(console.error);

🗣️ 실무에서 이렇게 말하세요

💬 쿼리 성능 튜닝 회의에서
"이 쿼리가 3초나 걸리는 건 Full Table Scan 때문이에요. user_id와 created_at에 복합 인덱스를 걸면 인덱스 스캔으로 바뀌어서 밀리초 단위로 줄어들 거예요. EXPLAIN 결과 보여드릴게요."
💬 데이터베이스 설계 리뷰에서
"인덱스 순서가 중요합니다. (status, user_id)로 되어 있는데, user_id로만 검색하는 쿼리가 더 많으면 (user_id, status)로 바꿔야 해요. 왼쪽 컬럼부터 사용되거든요."
💬 운영 이슈 대응에서
"인덱스가 너무 많으면 INSERT/UPDATE가 느려져요. 지금 이 테이블에 인덱스가 12개인데, 사용되지 않는 인덱스 4개 삭제하고 중복되는 거 2개 합치면 쓰기 성능이 30% 정도 개선될 겁니다."

⚠️ 주의사항 & 베스트 프랙티스

과도한 인덱스 생성 금지

인덱스가 많으면 INSERT/UPDATE/DELETE 성능이 저하됩니다. 실제 쿼리 패턴을 분석하고 필요한 인덱스만 생성하세요.

복합 인덱스 순서 주의

복합 인덱스는 왼쪽 컬럼부터 순서대로 사용됩니다. 자주 단독으로 검색되는 컬럼을 앞에 배치하세요.

함수 적용 시 인덱스 무효화

WHERE LOWER(email) = 'test'처럼 컬럼에 함수를 적용하면 인덱스를 사용할 수 없습니다. 함수 기반 인덱스를 별도로 생성하세요.

인덱스 베스트 프랙티스

EXPLAIN으로 실행 계획 확인, 주기적인 미사용 인덱스 정리, 커버링 인덱스 활용, CONCURRENTLY 옵션으로 무중단 인덱스 생성.

🔗 관련 용어

📚 더 배우기