# 最新版 DuckDB（v1.5.2+）
pip install duckdb --upgrade

# 验证版本
python -c "import duckdb; print(duckdb.__version__)"
# 应输出 1.5.2 或更高

import duckdb
import os

# 创建数据库连接
con = duckdb.connect()

# 安装并加载 Delta 扩展
con.execute("INSTALL delta;")
con.execute("LOAD delta;")

# 清理之前的演示数据
if os.path.exists("./sales_delta"):
    import shutil
    shutil.rmtree("./sales_delta")

# 附加一个 Delta 目录作为 DuckDB schema
con.execute("""
    ATTACH './sales_delta' AS sales (TYPE DELTA);
""")

# 创建表并写入数据
con.execute("""
    CREATE TABLE sales.orders (
        order_id INTEGER,
        product VARCHAR,
        amount DECIMAL(10,2),
        order_date DATE
    );
""")

# 插入第一批数据
con.execute("""
    INSERT INTO sales.orders VALUES
        (1, '笔记本电脑', 5999.00, '2026-05-01'),
        (2, '机械键盘', 899.00, '2026-05-01'),
        (3, '显示器', 2499.00, '2026-05-02'),
        (4, '鼠标', 199.00, '2026-05-02'),
        (5, '耳机', 699.00, '2026-05-03');
""")

print("✅ 第一批数据写入完成（Version 1）")

# 插入第二批数据（会创建 Version 2）
con.execute("""
    INSERT INTO sales.orders VALUES
        (6, '平板电脑', 3999.00, '2026-05-04'),
        (7, '充电器', 149.00, '2026-05-04'),
        (8, '移动硬盘', 499.00, '2026-05-05');
""")

print("✅ 第二批数据写入完成（Version 2）")

# 查询当前数据
result = con.execute("SELECT * FROM sales.orders ORDER BY order_id").fetchdf()
print("\n📊 当前数据（Version 2）：")
print(result)

# 查询 Version 1 的数据（只有前5条）
result_v1 = con.execute("""
    SELECT * FROM sales.orders (VERSION AS OF 1)
    ORDER BY order_id;
""").fetchdf()

print("📜 Version 1（第一批数据）：")
print(result_v1)

# 获取当前时间戳
import datetime
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

# 查询某个时间点的数据状态
result_ts = con.execute(f"""
    SELECT * FROM sales.orders (TIMESTAMP AS OF '2026-05-03 23:59:59'::TIMESTAMP)
    ORDER BY order_id;
""").fetchdf()

print(f"\n📜 2026-05-03 时的数据状态：")
print(result_ts)

# 查看 Delta 表的所有版本
history = con.execute("""
    SELECT 
        version,
        timestamp,
        operation,
        operation_parameters
    FROM sales.orders ('HISTORY')
    ORDER BY version;
""").fetchdf()

print("\n📋 Delta 版本历史：")
print(history)

# 更新：给所有订单金额加个 10% 的税（模拟）
con.execute("""
    UPDATE sales.orders 
    SET amount = amount * 1.1
    WHERE order_date >= '2026-05-04';
""")

# 删除：取消某笔订单
con.execute("""
    DELETE FROM sales.orders 
    WHERE order_id = 7;
""")

print("✅ UPDATE + DELETE 完成（Version 3）")

# 验证结果
result = con.execute("""
    SELECT * FROM sales.orders ORDER BY order_id
""").fetchdf()
print("\n📊 更新后的数据：")
print(result)

# 假设有新的日数据到达
import pandas as pd
import numpy as np

# 模拟 1000 条新订单
np.random.seed(42)
new_orders = pd.DataFrame({
    'order_id': range(100, 1100),
    'product': np.random.choice(
        ['笔记本电脑', '机械键盘', '显示器', '鼠标', '耳机', 
         '平板电脑', '充电器', '移动硬盘', '摄像头', '音箱'],
        1000
    ),
    'amount': np.round(np.random.uniform(50, 8000, 1000), 2),
    'order_date': pd.date_range('2026-05-06', periods=1000, freq='H')
})

# 保存为 Parquet
new_orders.to_parquet('./new_orders.parquet')

# 用 DuckDB 批量写入 Delta
con.execute("""
    INSERT INTO sales.orders
    SELECT * FROM read_parquet('./new_orders.parquet');
""")

print("✅ 1000 条新订单从 Parquet 写入 Delta 完成")

# 查询汇总
summary = con.execute("""
    SELECT 
        order_date::DATE AS day,
        COUNT(*) AS orders,
        ROUND(SUM(amount)::NUMERIC, 0) AS revenue
    FROM sales.orders
    GROUP BY day
    ORDER BY day;
""").fetchdf()

print("\n📊 每日订单汇总：")
print(summary)

-- 创建 UC 密钥
CREATE SECRET uc_secret (
    TYPE UC,
    TOKEN 'your-token-here'
);

-- 附加 Unity Catalog
ATTACH 'http://localhost:8080' AS uc_catalog (TYPE UC);

-- 查询 UC 中的表
SELECT * FROM uc_catalog.my_schema.orders;

-- 跨 Catalog JOIN
SELECT 
    o.*,
    p.product_category
FROM uc_catalog.my_schema.orders o
JOIN local_schema.products p ON o.product_id = p.product_id;

# 每日自动同步脚本示例
cat << 'EOF' > daily_sync.sh
#!/bin/bash
# 每天凌晨 2 点执行：业务CSV → Delta
duckdb -c "
INSTALL delta;
LOAD delta;
ATTACH './data_warehouse' AS dw (TYPE DELTA);
INSERT INTO dw.daily_sales
SELECT * FROM read_csv_auto('/data/sales/$(date -d 'yesterday' +%Y-%m-%d).csv');
"
EOF

# 添加到 crontab
# 0 2 * * * /path/to/daily_sync.sh