import duckdb

# 传统方式：DuckDB → CSV/JSON → Pandas
conn = duckdb.connect()
result = conn.execute("SELECT * FROM large_table")
df = result.fetchdf()  # 内部经历了 DuckDB → Python → Pandas 的序列化

import duckdb
import pyarrow as pa

# 创建内存数据库
conn = duckdb.connect()

# 加载数据
conn.execute("""
    CREATE TABLE sales AS SELECT * FROM read_csv_auto('sales_large.csv')
""")

# 以 Arrow 格式获取查询结果
result = conn.execute("""
    SELECT 
        region,
        date_trunc('month', sale_date) AS month,
        SUM(amount) AS total_sales,
        COUNT(*) AS transaction_count
    FROM sales
    WHERE sale_date >= '2025-01-01'
    GROUP BY region, month
    ORDER BY region, month
""")

# 零拷贝：直接以 Arrow Table 形式返回
arrow_table = result.fetch_arrow_table()
print(f"Rows: {arrow_table.num_rows}, Columns: {arrow_table.num_columns}")
print(f"Schema: {arrow_table.schema}")

import pyarrow as pa
import pyarrow.dataset as ds
import duckdb

# 创建 PyArrow 表
arr = pa.array([1, 2, 3, 4, 5])
table = pa.table({
    'id': pa.array([1, 2, 3, 4, 5]),
    'name': pa.array(['Alice', 'Bob', 'Charlie', 'Diana', 'Eve']),
    'score': pa.array([95.5, 87.3, 92.1, 78.9, 88.4])
})

# DuckDB 直接查询 PyArrow 表（零拷贝！）
conn = duckdb.connect()
result = conn.execute("""
    SELECT 
        name,
        score,
        RANK() OVER (ORDER BY score DESC) AS rank
    FROM table
    WHERE score > 85
    ORDER BY score DESC
""").fetch_arrow_table()

print(result)

import duckdb
import pyarrow as pa
import pyarrow.ipc as ipc
import tempfile

# 创建示例数据并写入 Arrow IPC 文件
table = pa.table({
    'timestamp': pa.array([1000, 2000, 3000, 4000]),
    'temperature': pa.array([22.5, 23.1, 21.8, 24.2])
})

with tempfile.NamedTemporaryFile(suffix='.arrow', delete=False) as f:
    writer = ipc.new_file(f, table.schema)
    writer.write_table(table)
    writer.close()
    ipc_path = f.name

# DuckDB 直接查询 Arrow IPC 文件
conn = duckdb.connect()
conn.execute(f"CREATE TABLE temps AS SELECT * FROM read_arrow_ipc('{ipc_path}')")
result = conn.execute("SELECT AVG(temperature) as avg_temp FROM temps").fetchone()
print(f"平均温度: {result[0]}°C")

import duckdb
import pyarrow as pa
import pyarrow.csv as csv

# 流式读取 CSV → 转换为 Arrow 流 → DuckDB 即时查询
conn = duckdb.connect()

# DuckDB 自身就可以高效读取 CSV，但这里展示 Arrow 流程
read_options = csv.ReadOptions(block_size=1024 * 1024 * 10)  # 10MB 块
csv_stream = csv.open_csv('ultra_large.csv', read_options=read_options)

# 逐批读取并查询
for batch in csv_stream:
    # 零拷贝：DuckDB 直接查询 Arrow RecordBatch
    result = conn.execute("""
        SELECT count(*) as cnt, sum(amount) as total
        FROM batch
        WHERE status = 'completed'
    """).fetchone()
    print(f"Batch result: {result}")

import adbc_driver_duckdb.dbapi as duckdb_adbc

# 通过 ADBC 连接 DuckDB
conn = duckdb_adbc.connect()

# 创建表
conn.execute("""
    CREATE TABLE orders AS
    SELECT range AS order_id, 
           random() * 1000 AS amount,
           CASE WHEN random() > 0.5 THEN 'completed' ELSE 'pending' END AS status
    FROM range(1000000)
""")

# 查询并以 Arrow 格式获取结果
cur = conn.cursor()
cur.execute("""
    SELECT status, count(*) as cnt, sum(amount) as total
    FROM orders
    GROUP BY status
""")

# 零拷贝获取 Arrow 数据
for batch in cur.fetch_record_batches():
    print(batch)

# Python 端：DuckDB 处理数据 → Arrow 格式输出
import duckdb
import pyarrow as pa
import pyarrow.ipc as ipc

conn = duckdb.connect()

# 商业数据清洗
conn.execute("""
    CREATE VIEW cleaned_sales AS
    SELECT 
        sale_id,
        customer_id,
        amount,
        sale_date
    FROM read_parquet('raw_sales/*.parquet')
    WHERE amount > 0 AND customer_id IS NOT NULL
""")

# 输出到 Arrow IPC 文件（中间交换格式）
result = conn.execute("SELECT * FROM cleaned_sales")
arrow_table = result.fetch_arrow_table()
with open('exchange_data.arrow', 'wb') as f:
    writer = ipc.new_file(f, arrow_table.schema)
    writer.write_table(arrow_table)
    writer.close()
print(f"Exported {arrow_table.num_rows} rows to Arrow IPC file")

# Java/Rust 端可以直接读取这个 Arrow 文件（零拷贝）

import duckdb
import pyarrow as pa
import pyarrow.parquet as pq
from sklearn.ensemble import RandomForestRegressor
import numpy as np

# 数据库引擎：DuckDB 处理 10 亿行日志数据
conn = duckdb.connect()

# 特征工程 - 全部在 DuckDB 中 SQL 完成
features = conn.execute("""
    SELECT
        customer_id,
        -- 时间特征
        date_diff('day', last_purchase_date, current_date) AS days_since_last_purchase,
        -- 聚合特征
        COUNT(*) AS total_orders,
        SUM(amount) AS total_spent,
        AVG(amount) AS avg_order_value,
        STDDEV(amount) AS order_amount_volatility,
        -- 分类特征编码
        CASE payment_method
            WHEN 'credit_card' THEN 1
            WHEN 'debit_card' THEN 2
            WHEN 'paypal' THEN 3
            ELSE 0
        END AS payment_method_code,
        -- 目标变量
        CASE WHEN churned = true THEN 1 ELSE 0 END AS label
    FROM customer_events
    WHERE event_date >= '2025-01-01'
    GROUP BY customer_id, last_purchase_date, payment_method, churned
""").fetch_arrow_table()  # 零拷贝转换为 Arrow

# Arrow → NumPy（零拷贝转换）
X = np.column_stack([
    features.column('days_since_last_purchase').to_numpy(),
    features.column('total_orders').to_numpy(),
    features.column('total_spent').to_numpy(),
    features.column('avg_order_value').to_numpy(),
    features.column('order_amount_volatility').to_numpy(),
    features.column('payment_method_code').to_numpy(),
])
y = features.column('label').to_numpy()

# 训练模型
model = RandomForestRegressor(n_estimators=100)
model.fit(X, y)

# 进程 A：数据生产者（DuckDB 处理 → Arrow 写入共享内存）
import duckdb
import pyarrow as pa

conn = duckdb.connect()
result = conn.execute("SELECT * FROM daily_aggregation")
arrow_table = result.fetch_arrow_table()

# 将 Arrow 表写入共享内存或文件
import pyarrow.ipc as ipc
with open('/dev/shm/data.arrow', 'wb') as f:
    writer = ipc.new_file(f, arrow_table.schema)
    writer.write_table(arrow_table)
    writer.close()

# 进程 B：数据消费者（毫秒级读取）
import pyarrow.ipc as ipc
with open('/dev/shm/data.arrow', 'rb') as f:
    reader = ipc.open_file(f)
    table = reader.read_all()
print(f"Read {table.num_rows} rows with zero copy from shared memory")

# 最佳实践示例
conn.execute("""
    SELECT 
        -- 只选需要的列
        customer_id,
        total_amount
    FROM orders
    WHERE date >= '2025-01-01'  -- 谓词下推，减少数据量
    ORDER BY total_amount DESC
    LIMIT 1000
""")