DuckDB 物化视图 + 增量刷新：把 BI 看板查询提速 50 倍

CREATE MATERIALIZED VIEW mv_daily_sales AS
SELECT 
    DATE(order_date) AS sale_date,
    category,
    shop_name,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue,
    AVG(amount) AS avg_order_value,
    COUNT(DISTINCT customer_id) AS unique_customers
FROM read_csv_auto('orders.csv')
GROUP BY DATE(order_date), category, shop_name;

import duckdb
from datetime import datetime

con = duckdb.connect(":memory:")

# ========== 第一步：创建物化视图 ==========
con.execute("""
    CREATE MATERIALIZED IF NOT EXISTS mv_daily_sales (
        sale_date DATE,
        category VARCHAR,
        shop_name VARCHAR,
        order_count BIGINT,
        total_revenue DOUBLE,
        avg_order_value DOUBLE,
        unique_customers BIGINT
    ) AS
    SELECT 
        DATE(order_date)::DATE AS sale_date,
        category,
        shop_name,
        COUNT(*)::BIGINT AS order_count,
        ROUND(SUM(amount), 2)::DOUBLE AS total_revenue,
        ROUND(AVG(amount), 2)::DOUBLE AS avg_order_value,
        COUNT(DISTINCT customer_id)::BIGINT AS unique_customers
    FROM read_csv_auto('orders.csv')
    GROUP BY DATE(order_date), category, shop_name
""")

# ========== 第二步：元数据管理 ==========
# 用一张小表记录上次刷新时间
con.execute("""
    CREATE TABLE IF NOT EXISTS mv_refresh_meta (
        key VARCHAR PRIMARY KEY,
        value TIMESTAMP
    )
""")

# 初始化或获取上次刷新时间
result = con.execute("""
    SELECT COALESCE(
        (SELECT value FROM mv_refresh_meta WHERE key = 'last_refresh'),
        TIMESTAMP '1970-01-01'
    )
""").fetchone()
last_refresh = result[0]

# ========== 第三步：增量刷新 ==========
# 找出上次刷新后新增的数据
new_data = con.execute(f"""
    SELECT DISTINCT DATE(order_date)::DATE as sale_date
    FROM read_csv_auto('orders.csv')
    WHERE order_date > '{last_refresh}'
    ORDER BY sale_date DESC
    LIMIT 30
""").fetchall()

if new_data:
    # 对每个新日期执行增量插入
    for (sale_date,) in new_data:
        con.execute("""
            INSERT INTO mv_daily_sales
            SELECT 
                DATE(order_date)::DATE AS sale_date,
                category,
                shop_name,
                COUNT(*)::BIGINT AS order_count,
                ROUND(SUM(amount), 2)::DOUBLE AS total_revenue,
                ROUND(AVG(amount), 2)::DOUBLE AS avg_order_value,
                COUNT(DISTINCT customer_id)::BIGINT AS unique_customers
            FROM read_csv_auto('orders.csv')
            WHERE DATE(order_date) = ?
            GROUP BY DATE(order_date), category, shop_name
        """, (sale_date,))
    
    # 更新刷新时间戳
    now = datetime.now()
    con.execute("DELETE FROM mv_refresh_meta WHERE key = 'last_refresh'")
    con.execute(f"""
        INSERT INTO mv_refresh_meta VALUES ('last_refresh', '{now}')
    """)
    
    print(f"✅ 增量刷新完成，处理了 {len(new_data)} 个日期的数据")
else:
    print("没有新数据需要刷新")

-- 一键增量刷新：只处理最近 7 天且未存在于物化视图中的数据
INSERT INTO mv_daily_sales
SELECT 
    DATE(order_date)::DATE AS sale_date,
    category,
    shop_name,
    COUNT(*)::BIGINT AS order_count,
    ROUND(SUM(amount), 2)::DOUBLE AS total_revenue,
    ROUND(AVG(amount), 2)::DOUBLE AS avg_order_value,
    COUNT(DISTINCT customer_id)::BIGINT AS unique_customers
FROM read_csv_auto('orders.csv')
WHERE DATE(order_date) >= CURRENT_DATE - INTERVAL '7' DAY
  AND DATE(order_date) NOT IN (SELECT DISTINCT sale_date FROM mv_daily_sales)
GROUP BY DATE(order_date), category, shop_name;

import schedule
import time
import duckdb

REFRESH_SQL = """
INSERT INTO mv_daily_sales
SELECT ... -- 增量刷新 SQL
"""

def refresh_mv():
    con = duckdb.connect(":memory:")
    con.execute(REFRESH_SQL)
    con.close()
    print(f"[{datetime.now()}] 刷新完成")

# 每天凌晨 2 点刷新
schedule.every().day.at("02:00").do(refresh_mv)

while True:
    schedule.run_pending()
    time.sleep(60)