-- Tạo staging table — UNLOGGED để không ghi WAL, tăng tốc load
CREATE UNLOGGED TABLE orders_staging (
    order_id     BIGINT,
    customer_id  BIGINT,
    amount       NUMERIC(12,2),
    status       VARCHAR(50),
    updated_at   TIMESTAMPTZ
    -- Không có PRIMARY KEY, không có index, không có FK
);
 
-- Load file bằng COPY — nhanh nhất, không qua app layer
COPY orders_staging FROM '/data/orders_update.csv'
WITH (FORMAT csv, HEADER true, DELIMITER ',');
-- 1 triệu rows: ~30–60 giây tùy disk I/O

@Service
public class BulkLoader {
 
    private final DataSource dataSource;
 
    public long loadToStaging(Path csvFile) throws Exception {
        try (Connection conn = dataSource.getConnection();
             Reader reader = Files.newBufferedReader(csvFile)) {
 
            CopyManager copyManager = new CopyManager((BaseConnection) conn);
            return copyManager.copyIn(
                "COPY orders_staging FROM STDIN WITH (FORMAT csv, HEADER true)",
                reader
            );
        }
    }
}

-- Loại row không hợp lệ trước khi merge
DELETE FROM orders_staging
WHERE order_id IS NULL
   OR amount < 0
   OR status NOT IN ('PENDING', 'PAID', 'CANCELLED', 'REFUNDED');
 
-- Log vào dead-letter table nếu cần audit
INSERT INTO orders_staging_errors
SELECT *, NOW(), 'INVALID_STATUS'
FROM orders_staging
WHERE status NOT IN ('PENDING', 'PAID', 'CANCELLED', 'REFUNDED');

-- Merge một batch: rows từ offset đến offset + batch_size
INSERT INTO orders (order_id, customer_id, amount, status, updated_at)
SELECT order_id, customer_id, amount, status, updated_at
FROM orders_staging
ORDER BY order_id   -- consistent order tránh deadlock
LIMIT :batchSize OFFSET :offset
ON CONFLICT (order_id) DO UPDATE SET
    customer_id = EXCLUDED.customer_id,
    amount      = EXCLUDED.amount,
    status      = EXCLUDED.status,
    updated_at  = EXCLUDED.updated_at
WHERE orders.updated_at < EXCLUDED.updated_at;  -- chỉ update nếu staging mới hơn

@Service
@Slf4j
public class BulkMergeService {
 
    private static final int BATCH_SIZE = 10_000;
    private static final long BATCH_DELAY_MS = 50; // throttle để không bão hòa I/O
 
    private final JdbcTemplate jdbc;
 
    public void mergeFromStaging() throws Exception {
        long totalRows = jdbc.queryForObject(
            "SELECT COUNT(*) FROM orders_staging", Long.class);
 
        log.info("Starting merge: {} rows in staging", totalRows);
 
        long offset = 0;
        int batchNum = 0;
 
        while (offset < totalRows) {
            int merged = mergeBatch(offset, BATCH_SIZE);
            offset += BATCH_SIZE;
            batchNum++;
 
            if (batchNum % 10 == 0) {
                log.info("Progress: {}/{} rows processed ({}%)",
                    offset, totalRows,
                    String.format("%.1f", 100.0 * offset / totalRows));
            }
 
            Thread.sleep(BATCH_DELAY_MS); // throttle
        }
 
        log.info("Merge complete: {} batches, {} rows", batchNum, totalRows);
    }
 
    @Transactional  // mỗi batch = 1 transaction
    public int mergeBatch(long offset, int batchSize) {
        return jdbc.update("""
            INSERT INTO orders (order_id, customer_id, amount, status, updated_at)
            SELECT order_id, customer_id, amount, status, updated_at
            FROM orders_staging
            ORDER BY order_id
            LIMIT ? OFFSET ?
            ON CONFLICT (order_id) DO UPDATE SET
                customer_id = EXCLUDED.customer_id,
                amount      = EXCLUDED.amount,
                status      = EXCLUDED.status,
                updated_at  = EXCLUDED.updated_at
            WHERE orders.updated_at < EXCLUDED.updated_at
            """, batchSize, offset);
    }
}

-- Thêm: rate limit batch, monitor query time song song
SELECT query, state, wait_event, now() - query_start AS duration
FROM pg_stat_activity
WHERE state != 'idle'
ORDER BY duration DESC;

-- Drop secondary indexes trước khi merge
DROP INDEX CONCURRENTLY idx_orders_customer_id;
DROP INDEX CONCURRENTLY idx_orders_status;
DROP INDEX CONCURRENTLY idx_orders_updated_at;
 
-- Chạy merge (nhanh hơn 3–5x vì không update index mỗi row)
-- ...
 
-- Rebuild sau khi merge
CREATE INDEX CONCURRENTLY idx_orders_customer_id ON orders(customer_id);
CREATE INDEX CONCURRENTLY idx_orders_status ON orders(status);
CREATE INDEX CONCURRENTLY idx_orders_updated_at ON orders(updated_at);
-- CONCURRENTLY: build trong background, không lock bảng

-- 1. Tạo bảng mới với cấu trúc giống partition cũ
CREATE TABLE orders_2024_01_new (LIKE orders_2024_01 INCLUDING ALL);
 
-- 2. Copy toàn bộ partition cũ vào bảng mới
INSERT INTO orders_2024_01_new SELECT * FROM orders_2024_01;
 
-- 3. Merge staging vào bảng mới (không lock production)
INSERT INTO orders_2024_01_new
SELECT ... FROM orders_staging
ON CONFLICT (order_id) DO UPDATE SET ...;
 
-- 4. Atomic swap (cực nhanh, lock chỉ vài milliseconds)
BEGIN;
ALTER TABLE orders DETACH PARTITION orders_2024_01;
ALTER TABLE orders ATTACH PARTITION orders_2024_01_new
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
COMMIT;
 
-- 5. Drop partition cũ
DROP TABLE orders_2024_01;

-- Thêm cột tracking vào staging table
ALTER TABLE orders_staging ADD COLUMN merged BOOLEAN DEFAULT FALSE;
ALTER TABLE orders_staging ADD COLUMN merged_at TIMESTAMPTZ;
 
-- Chỉ merge các row chưa được merge
INSERT INTO orders (...)
SELECT ... FROM orders_staging
WHERE merged = FALSE
ORDER BY order_id
LIMIT :batchSize
ON CONFLICT ...;
 
-- Đánh dấu sau khi merge xong batch
UPDATE orders_staging SET merged = TRUE, merged_at = NOW()
WHERE order_id IN (... -- ids vừa merge);

-- Progress real-time
SELECT
    COUNT(*) FILTER (WHERE merged = TRUE)  AS done,
    COUNT(*) FILTER (WHERE merged = FALSE) AS remaining,
    COUNT(*)                               AS total,
    ROUND(100.0 * COUNT(*) FILTER (WHERE merged) / COUNT(*), 1) AS pct
FROM orders_staging;
 
-- Lock contention — xem ai đang bị block
SELECT
    blocked.pid,
    blocked.query,
    blocking.pid   AS blocking_pid,
    blocking.query AS blocking_query
FROM pg_stat_activity blocked
JOIN pg_stat_activity blocking
    ON blocking.pid = ANY(pg_blocking_pids(blocked.pid));
 
-- I/O load trong lúc merge
SELECT * FROM pg_stat_bgwriter;

# Prometheus alert — merge job stall
- alert: BulkMergeStalled
  expr: |
    increase(bulk_merge_rows_processed_total[5m]) == 0
    and bulk_merge_job_active == 1
  for: 5m
  annotations:
    summary: "Merge job không progress trong 5 phút — có thể bị deadlock hoặc lock wait"

CREATE TABLE orders_staging_dedup AS
SELECT DISTINCT ON (order_id) *
FROM orders_staging ORDER BY order_id, updated_at DESC;

INSERT INTO orders (...) SELECT ... FROM orders_staging
ON CONFLICT (order_id) DO UPDATE SET ...
RETURNING order_id,
    (xmax = 0) AS is_insert,
    (xmax != 0) AS is_update;