Sharding

// MongoDB 샤딩 설정 및 사용 - Node.js
const { MongoClient } = require('mongodb');

/**
 * MongoDB 샤딩 클러스터 연결 및 설정
 * mongos 라우터를 통해 자동으로 샤드 라우팅
 */
class MongoShardingExample {
    constructor(mongosUri) {
        // mongos 라우터에 연결 (샤딩 클러스터의 진입점)
        this.uri = mongosUri; // e.g., 'mongodb://mongos1:27017,mongos2:27017'
        this.client = null;
        this.db = null;
    }

    async connect() {
        this.client = new MongoClient(this.uri);
        await this.client.connect();
        this.db = this.client.db('ecommerce');
        console.log('✅ MongoDB 샤딩 클러스터 연결 완료');
    }

    /**
     * 컬렉션 샤딩 활성화
     * mongos에서 실행 (보통 초기 설정 시 한 번)
     */
    async enableSharding() {
        const admin = this.client.db('admin');

        // 1. 데이터베이스 샤딩 활성화
        await admin.command({ enableSharding: 'ecommerce' });

        // 2. 컬렉션 샤딩 - Hash 기반
        await admin.command({
            shardCollection: 'ecommerce.orders',
            key: { user_id: 'hashed' }  // Hash 샤딩
        });

        // 3. 컬렉션 샤딩 - Range 기반
        await admin.command({
            shardCollection: 'ecommerce.logs',
            key: { timestamp: 1 }  // Range 샤딩 (시간순)
        });

        // 4. 복합 샤드 키
        await admin.command({
            shardCollection: 'ecommerce.products',
            key: { category: 1, product_id: 1 }  // 카테고리 + ID
        });

        console.log('✅ 샤딩 설정 완료');
    }

    /**
     * 주문 생성 - 자동으로 적절한 샤드로 라우팅
     */
    async createOrder(userId, items, totalAmount) {
        const orders = this.db.collection('orders');

        const order = {
            user_id: userId,  // 샤드 키
            items: items,
            total_amount: totalAmount,
            status: 'pending',
            created_at: new Date()
        };

        // MongoDB가 user_id 해시값으로 자동 라우팅
        const result = await orders.insertOne(order);
        console.log(`✅ 주문 생성: ${result.insertedId}`);
        return result.insertedId;
    }

    /**
     * 단일 샤드 쿼리 - 샤드 키 포함 (빠름)
     */
    async getUserOrders(userId) {
        const orders = this.db.collection('orders');

        // user_id(샤드 키)가 쿼리에 포함되어 단일 샤드만 조회
        const cursor = orders.find({ user_id: userId })
            .sort({ created_at: -1 })
            .limit(10);

        return await cursor.toArray();
    }

    /**
     * Scatter-Gather 쿼리 - 모든 샤드 조회 (느림)
     */
    async getRecentOrders(status, limit = 100) {
        const orders = this.db.collection('orders');

        // 샤드 키 없이 조회 → 모든 샤드에 쿼리 전송
        // 가능하면 피하거나, 별도 집계 컬렉션 활용
        const cursor = orders.find({ status: status })
            .sort({ created_at: -1 })
            .limit(limit);

        return await cursor.toArray();
    }

    /**
     * 샤딩 통계 확인
     */
    async getShardingStats() {
        const admin = this.client.db('admin');

        // 샤드 분포 확인
        const stats = await admin.command({
            shardDistribution: 'ecommerce.orders'
        });

        // 각 샤드별 청크 수 확인
        const config = this.client.db('config');
        const chunks = await config.collection('chunks')
            .aggregate([
                { $match: { ns: 'ecommerce.orders' } },
                { $group: { _id: '$shard', count: { $sum: 1 } } }
            ]).toArray();

        console.log('📊 샤드별 청크 분포:', chunks);
        return { stats, chunks };
    }

    /**
     * 태그 기반 샤딩 (Zone Sharding)
     * 지역별 데이터 격리에 유용
     */
    async setupZoneSharding() {
        const admin = this.client.db('admin');

        // 샤드에 태그(zone) 할당
        await admin.command({
            addShardToZone: 'shard0',
            zone: 'ASIA'
        });
        await admin.command({
            addShardToZone: 'shard1',
            zone: 'EUROPE'
        });

        // zone 범위 설정
        await admin.command({
            updateZoneKeyRange: 'ecommerce.users',
            min: { region: 'asia', _id: MinKey },
            max: { region: 'asia', _id: MaxKey },
            zone: 'ASIA'
        });
    }

    async close() {
        await this.client.close();
    }
}

// 사용 예시
async function main() {
    const mongo = new MongoShardingExample(
        'mongodb://mongos1:27017,mongos2:27017'
    );

    await mongo.connect();

    // 주문 생성 (자동 샤드 분배)
    await mongo.createOrder('user_001', [
        { product: 'iPhone', qty: 1 }
    ], 1200000);

    // 단일 사용자 주문 조회 (단일 샤드)
    const orders = await mongo.getUserOrders('user_001');
    console.log('사용자 주문:', orders);

    // 샤딩 통계
    await mongo.getShardingStats();

    await mongo.close();
}

main().catch(console.error);

-- ============================================
-- Citus (PostgreSQL 분산 확장) 샤딩 설정
-- ============================================

-- 1. 분산 테이블 생성 (Hash 샤딩)
CREATE TABLE orders (
    order_id BIGSERIAL,
    user_id BIGINT NOT NULL,
    product_id INT,
    amount DECIMAL(10,2),
    status VARCHAR(20),
    created_at TIMESTAMP DEFAULT NOW(),
    PRIMARY KEY (user_id, order_id)  -- 샤드 키 포함 필수
);

-- user_id로 샤딩 (32개 샤드로 분산)
SELECT create_distributed_table('orders', 'user_id');

-- 2. Reference 테이블 (모든 노드에 복제)
CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name VARCHAR(200),
    price DECIMAL(10,2),
    category VARCHAR(50)
);

-- 모든 워커 노드에 복제 (조인 최적화)
SELECT create_reference_table('products');


-- 3. 같은 샤드 키로 Co-location (조인 최적화)
CREATE TABLE order_items (
    id BIGSERIAL,
    order_id BIGINT,
    user_id BIGINT NOT NULL,  -- 같은 샤드 키
    product_id INT,
    quantity INT,
    PRIMARY KEY (user_id, id)
);

SELECT create_distributed_table('order_items', 'user_id',
    colocate_with => 'orders');

-- Co-located 조인 (단일 샤드 내에서 실행, 빠름)
SELECT o.order_id, o.amount, oi.quantity, p.name
FROM orders o
JOIN order_items oi ON o.order_id = oi.order_id
    AND o.user_id = oi.user_id  -- 샤드 키 조인 조건 필수!
JOIN products p ON oi.product_id = p.id
WHERE o.user_id = 12345;


-- ============================================
-- Vitess (MySQL 샤딩) VSchema 예시
-- ============================================
/*
{
  "sharded": true,
  "vindexes": {
    "hash": {
      "type": "hash"
    },
    "user_id_idx": {
      "type": "hash",
      "owner": "orders"
    }
  },
  "tables": {
    "orders": {
      "column_vindexes": [
        {
          "column": "user_id",
          "name": "hash"
        }
      ]
    },
    "users": {
      "column_vindexes": [
        {
          "column": "id",
          "name": "hash"
        }
      ]
    }
  }
}
*/


-- ============================================
-- 샤딩 모니터링 쿼리
-- ============================================

-- Citus: 샤드 분포 확인
SELECT
    nodename,
    count(*) as shard_count,
    pg_size_pretty(sum(shard_size)) as total_size
FROM citus_shards
JOIN citus_dist_stat ON shard_id = shardid
GROUP BY nodename;

-- 특정 데이터가 어느 샤드에 있는지 확인
SELECT get_shard_id_for_distribution_column('orders', 12345);

-- 샤드 리밸런싱 (데이터 재분배)
SELECT rebalance_table_shards('orders');


-- ============================================
-- 크로스 샤드 쿼리 최적화
-- ============================================

-- BAD: 모든 샤드 스캔 (느림)
SELECT COUNT(*) FROM orders WHERE status = 'pending';

-- GOOD: 샤드 키 필터 포함 (빠름)
SELECT COUNT(*) FROM orders
WHERE user_id = 12345 AND status = 'pending';

-- 집계가 필요하면 별도 요약 테이블 활용
CREATE TABLE daily_order_stats (
    stat_date DATE PRIMARY KEY,
    total_orders BIGINT,
    total_revenue DECIMAL(15,2)
);
SELECT create_reference_table('daily_order_stats');


-- ============================================
-- PostgreSQL 수동 샤딩 (FDW 활용)
-- ============================================

-- 원격 서버 정의
CREATE SERVER shard0 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (host 'shard0.db.com', dbname 'users');

CREATE SERVER shard1 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (host 'shard1.db.com', dbname 'users');

-- 외부 테이블 정의
CREATE FOREIGN TABLE users_shard0 (
    id BIGINT, name TEXT, email TEXT
) SERVER shard0 OPTIONS (table_name 'users');

-- 파티션 테이블로 통합
CREATE TABLE users (
    id BIGINT, name TEXT, email TEXT
) PARTITION BY HASH (id);

ALTER TABLE users ATTACH PARTITION users_shard0
FOR VALUES WITH (modulus 2, remainder 0);