存算分离 + 开放表格式，是现在最稳的数据架构选择

-- 在 Athena 中创建 Iceberg 表（Silver 层订单明细）
CREATE TABLE silver.orders (
    order_id        STRING,
    user_id         STRING,
    order_status    STRING,
    total_amount    DECIMAL(18, 2),
    channel         STRING,
    created_at      TIMESTAMP,
    updated_at      TIMESTAMP,
    dt              DATE
)
PARTITIONED BY (dt)
LOCATION 's3://mushan-lakehouse/silver/orders/'
TBLPROPERTIES (
    'table_type'                        = 'ICEBERG',
    'format'                            = 'PARQUET',
    'write_compression'                 = 'ZSTD',
    'optimize_rewrite_delete_file_threshold' = '10'
);

-- CDC 增量合并：来自 Bronze 层的变更记录合入 Silver
-- Iceberg 的 MERGE 是原子操作，不会出现读到半成品的情况
MERGE INTO silver.orders AS t
USING (
    SELECT
        order_id,
        user_id,
        order_status,
        CAST(total_amount AS DECIMAL(18, 2))  AS total_amount,
        channel,
        created_at,
        updated_at,
        CAST(created_at AS DATE)              AS dt,
        op                                     -- I / U / D
    FROM (
        SELECT *,
               ROW_NUMBER() OVER (
                   PARTITION BY order_id ORDER BY updated_at DESC
               ) AS rn
        FROM bronze.orders_cdc
        WHERE ingest_date = CURRENT_DATE
    )
    WHERE rn = 1                               -- 同一主键只取最新变更
) AS s
ON t.order_id = s.order_id

WHEN MATCHED AND s.op = 'D' THEN DELETE

WHEN MATCHED AND s.op IN ('I', 'U') AND s.updated_at > t.updated_at THEN
    UPDATE SET
        order_status = s.order_status,
        total_amount = s.total_amount,
        channel      = s.channel,
        updated_at   = s.updated_at

WHEN NOT MATCHED AND s.op IN ('I', 'U') THEN
    INSERT (order_id, user_id, order_status, total_amount, channel, created_at, updated_at, dt)
    VALUES (s.order_id, s.user_id, s.order_status, s.total_amount, s.channel, s.created_at, s.updated_at, s.dt);

-- 定期维护：合并小文件、清理过期快照
OPTIMIZE silver.orders REWRITE DATA USING BIN_PACK;
VACUUM silver.orders;