DuckDB MERGE INTO 与 Dynamic Filters：ETL 性能翻倍秘籍

MERGE INTO dim_products d
USING staging_products s
ON d.product_id = s.product_id
WHEN MATCHED THEN UPDATE SET
    product_name = s.product_name,
    price = s.price
WHEN NOT MATCHED THEN INSERT VALUES
    (s.product_id, s.product_name, s.price);

MERGE INTO dim_products d
USING staging_products s
ON d.product_id = s.product_id
WHEN MATCHED AND d.price != s.price THEN UPDATE SET
    product_name = s.product_name,
    price = s.price,
    updated_at = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN INSERT VALUES
    (s.product_id, s.product_name, s.price, CURRENT_TIMESTAMP);

-- 配置表结构
CREATE TABLE sync_config (
    table_name VARCHAR PRIMARY KEY,
    max_id BIGINT
);

-- 初始化配置
INSERT INTO sync_config VALUES ('dim_products', 0);

-- 增量同步
MERGE INTO dim_products d
USING staging_products s
ON d.product_id = s.product_id
WHEN NOT MATCHED AND s.product_id > (
    SELECT max_id FROM sync_config WHERE table_name = 'dim_products'
) THEN INSERT VALUES
    (s.product_id, s.product_name, s.price, CURRENT_TIMESTAMP);

-- 更新配置
UPDATE sync_config 
SET max_id = (SELECT MAX(product_id) FROM dim_products)
WHERE table_name = 'dim_products';

-- 目标表结构
CREATE TABLE dim_customers (
    customer_id INTEGER,
    name VARCHAR,
    email VARCHAR,
    is_active BOOLEAN,
    valid_from DATE,
    valid_to DATE
);

-- MERGE INTO 实现 SCD Type 2
MERGE INTO dim_customers dc
USING staging_customers sc
ON dc.customer_id = sc.customer_id
WHEN MATCHED AND dc.email != sc.email THEN UPDATE SET
    is_active = false,
    valid_to = CURRENT_DATE
WHEN NOT MATCHED THEN INSERT 
    (customer_id, name, email, is_active, valid_from, valid_to)
VALUES 
    (sc.customer_id, sc.name, sc.email, true, CURRENT_DATE, NULL);

SELECT source, COUNT(*), SUM(value)
FROM events
WHERE user_id IN (
    SELECT DISTINCT user_id 
    FROM events 
    WHERE event_date >= date '2024-06-01'
      AND event_date < date '2024-07-01'
)
GROUP BY source;

-- 开启缓存算子（DuckDB 1.5.3+ 推荐）
SET enable_caching_operators = true;

-- 关闭动态过滤器（调试用）
SET dynamic_filters = false;

-- 查看当前设置
SHOW ALL LIKE '%filter%';

import duckdb

con = duckdb.connect("etl_pipeline.duckdb")

# 创建目标汇总表
con.execute("""
CREATE TABLE IF NOT EXISTS daily_orders (
    order_id VARCHAR PRIMARY KEY,
    customer_id VARCHAR,
    product_name VARCHAR,
    amount DECIMAL(10, 2),
    order_date DATE,
    status VARCHAR,
    last_updated TIMESTAMP
)
""")

# 创建每日增量数据（模拟从外部 API 获取）
con.execute("""
CREATE TABLE IF NOT EXISTS daily_staging AS
SELECT 
    'ORD-' || i AS order_id,
    'CUST-' || (i % 1000) AS customer_id,
    'Product-' || (i % 200) AS product_name,
    ROUND(UNIFORM(10, 500)::DECIMAL(10, 2), 2) AS amount,
    CURRENT_DATE - INTERVAL (i % 30) DAY AS order_date,
    CASE 
        WHEN i % 5 = 0 THEN 'cancelled'
        WHEN i % 3 = 0 THEN 'refunded'
        ELSE 'completed'
    END AS status,
    CURRENT_TIMESTAMP AS last_updated
FROM GENERATE_SERIES(1, 10000) AS t(i)
""")

-- 增量合并：更新已存在的订单，插入新订单
MERGE INTO daily_orders d
USING daily_staging s
ON d.order_id = s.order_id
WHEN MATCHED AND d.amount != s.amount OR d.status != s.status THEN UPDATE SET
    customer_id = s.customer_id,
    product_name = s.product_name,
    amount = s.amount,
    status = s.status,
    last_updated = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN INSERT VALUES
    (s.order_id, s.customer_id, s.product_name, s.amount, 
     s.order_date, s.status, s.last_updated);

-- 查询今日完成的订单总额（Dynamic Filter 自动优化）
SELECT product_name, SUM(amount) as total_amount
FROM daily_orders
WHERE order_date = CURRENT_DATE
  AND status = 'completed'
  AND customer_id IN (
      SELECT DISTINCT customer_id 
      FROM daily_orders 
      WHERE order_date = CURRENT_DATE
  )
GROUP BY product_name
ORDER BY total_amount DESC;

-- 清空 staging 表，准备下一批数据
TRUNCATE TABLE daily_staging;

-- 或者使用 COPY 将数据导出为 Parquet
COPY daily_orders TO 'output/daily_orders.parquet' (COMPRESSION ZSTD);