# サーバー側：Pythonから実行（pip install fastapi uvicorn duckdb）
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import duckdb
import uvicorn

app = FastAPI()
db_conn = duckdb.connect('my_database.db')

class QueryRequest(BaseModel):
    sql: str

@app.post("/query")
async def execute_query(request: QueryRequest):
    try:
        result = db_conn.execute(request.sql)
        columns = [desc[0] for desc in result.description] if result.description else []
        rows = [list(row) for row in result.fetchall()]
        return {"columns": columns, "rows": rows}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

if __name__ == "__main__":
    # ポート9494でサービス開始
    uvicorn.run(app, host="0.0.0.0", port=9494)

# クライアント側：別マシンのPythonから実行（pip install requests pandas）
import requests
import pandas as pd

class DuckDBClient:
    """HTTP経由でリモートDuckDBサーバーに接続するクライアント"""

    def __init__(self, host: str, port: int = 9494):
        self.base_url = f"http://{host}:{port}"
        self._last_sql: str = ""

    def execute(self, sql: str) -> "DuckDBClient":
        self._last_sql = sql
        return self

    def df(self) -> pd.DataFrame:
        response = requests.post(
            f"{self.base_url}/query",
            json={"sql": self._last_sql},
            timeout=30
        )
        response.raise_for_status()
        data = response.json()
        return pd.DataFrame(data["rows"], columns=data["columns"])

# 接続後は通常のSQLで操作可能
conn = DuckDBClient("192.168.1.100", 9494)
result = conn.execute("SELECT COUNT(*) FROM events WHERE created_at > '2026-01-01'").df()
print(result)

import requests
import pandas as pd

class DuckDBClient:
    """HTTP経由でリモートDuckDBサーバーに接続するクライアント"""

    def __init__(self, host: str, port: int = 9494):
        self.base_url = f"http://{host}:{port}"
        self._last_sql: str = ""

    def execute(self, sql: str) -> "DuckDBClient":
        self._last_sql = sql
        return self

    def df(self) -> pd.DataFrame:
        response = requests.post(
            f"{self.base_url}/query",
            json={"sql": self._last_sql},
            timeout=30
        )
        response.raise_for_status()
        data = response.json()
        return pd.DataFrame(data["rows"], columns=data["columns"])

# クライアント接続
conn = DuckDBClient("192.168.1.100", 9494)

# 通常のduckdb接続と同じAPIで操作できる
df = conn.execute("""
    SELECT
        date_trunc('hour', created_at) AS hour,
        COUNT(*) AS event_count
    FROM events
    WHERE created_at > '2026-01-01'
    GROUP BY 1
    ORDER BY 1
""").df()

print(df.head())

-- PostgreSQLへの接続
CONNECT 'postgresql://user:password@pghost:5432/mydb' AS pg;

-- MySQL/MariaDBへの接続
CONNECT 'mysql://user:password@mysqlhost:3306/mydb' AS mysql;

-- 接続後はエイリアスを使ってクロスDBクエリが可能
SELECT
    pg.orders.order_id,
    pg.orders.amount,
    mysql.customers.customer_name
FROM pg.orders
JOIN mysql.customers ON pg.orders.customer_id = mysql.customers.id
WHERE pg.orders.created_at > '2026-01-01';

-- 環境変数から接続文字列を構築する例
SET pg_connection = getenv('POSTGRES_DSN');
CONNECT $pg_connection AS pg;

import duckdb
import schedule

def sync_and_analyze():
    conn = duckdb.connect('my_analytics.db')
    conn.execute("LOAD quack;")

    # PostgreSQLから差分データを取得してローカルに追記
    conn.execute("""
        CONNECT 'postgresql://localhost/production' AS pg;

        INSERT INTO local_events
        SELECT * FROM pg.events
        WHERE created_at > (SELECT MAX(created_at) FROM local_events);
    """)

    # 高速な列指向クエリで集計
    result = conn.execute("""
        SELECT
            date_trunc('day', created_at) AS day,
            event_type,
            COUNT(*) AS cnt,
            AVG(response_time_ms) AS avg_response_ms
        FROM local_events
        WHERE created_at > NOW() - INTERVAL 30 DAYS
        GROUP BY 1, 2
        ORDER BY 1 DESC, 3 DESC
    """).df()

    return result

# 1時間ごとに同期
schedule.every().hour.do(sync_and_analyze)

| id | payload (TEXT型) |
|----|-----------------|
| 1  | '{"user_id":42,"action":"click","ts":1234567890}' |
| 2  | '{"user_id":7,"action":"view","ts":1234567899,"extra":{"page":"home"}}' |

CREATE TABLE events (
    id BIGINT,
    payload VARIANT
);

INSERT INTO events VALUES
    (1, '{"user_id":42,"action":"click","ts":1234567890}'::VARIANT),
    (2, '{"user_id":7,"action":"view","ts":1234567899,"extra":{"page":"home"}}'::VARIANT);

-- フィールド抽出はパースなしで直接実行できる
SELECT
    id,
    payload.user_id::BIGINT AS user_id,
    payload.action::VARCHAR AS action
FROM events;

-- このクエリがv2.0で特に速くなる例
-- payload.action = 'purchase' の行だけを効率的に取得
SELECT
    payload.user_id::BIGINT,
    payload.amount::DECIMAL(10,2),
    payload.ts::TIMESTAMP
FROM events
WHERE payload.action = 'purchase'
  AND payload.ts > '2026-01-01'::TIMESTAMP;

-- VARIANT型を含むテーブルをParquetにエクスポート
COPY events TO 's3://my-bucket/events/2026-08/' (FORMAT PARQUET, PARTITION_BY (date_trunc('day', created_at)));

-- Parquetから読み込み（VARIANT型として自動認識）
CREATE VIEW events_parquet AS
SELECT * FROM read_parquet('s3://my-bucket/events/2026-08/**/*.parquet');

-- 読み込んだParquetに対してVARIANTクエリが動く
SELECT payload.user_id, COUNT(*) FROM events_parquet GROUP BY 1;

-- 型チェック
SELECT variant_typeof(payload) FROM events LIMIT 5;
-- → 'OBJECT', 'OBJECT', ...

-- 存在確認
SELECT * FROM events WHERE variant_has_key(payload, 'extra');

-- キー一覧取得
SELECT variant_keys(payload) FROM events LIMIT 1;
-- → ['user_id', 'action', 'ts', 'extra']

-- ネスト構造へのアクセス
SELECT payload.extra.page::VARCHAR FROM events WHERE variant_has_key(payload, 'extra');

-- 1. ログ受け取りテーブルの定義
CREATE TABLE raw_logs (
    id BIGINT DEFAULT nextval('log_id_seq'),
    received_at TIMESTAMPTZ DEFAULT now(),
    source VARCHAR,
    payload VARIANT
);

-- 2. インデックスの作成（受信時刻での範囲絞り込みを高速化）
CREATE INDEX idx_raw_logs_received ON raw_logs (received_at);

# 3. アプリからのログ投入（Python側）
import duckdb
import json
from datetime import datetime

conn = duckdb.connect('analytics.db')

def ingest_log(source: str, event: dict):
    """アプリケーションログをVARIANT型で投入"""
    conn.execute("""
        INSERT INTO raw_logs (source, payload)
        VALUES (?, ?::VARIANT)
    """, [source, json.dumps(event)])

# 様々なスキーマのイベントを同一テーブルに投入できる
ingest_log('web-api', {
    'user_id': 42,
    'action': 'purchase',
    'amount': 9800,
    'item_ids': [101, 202],
    'ts': '2026-08-17T10:30:00Z'
})

ingest_log('mobile-app', {
    'user_id': 7,
    'action': 'view',
    'screen': 'product_detail',
    'product_id': 101,
    'duration_sec': 45,
    'ts': '2026-08-17T10:30:05Z'
})

-- 4. 分析クエリ（VARIANT型のまま集計）
SELECT
    source,
    payload.action::VARCHAR AS action,
    COUNT(*) AS event_count,
    COUNT(DISTINCT payload.user_id::BIGINT) AS unique_users
FROM raw_logs
WHERE received_at > NOW() - INTERVAL '1 hour'
  AND payload.action IS NOT NULL
GROUP BY 1, 2
ORDER BY 3 DESC;

-- 5. Parquetへのエクスポート（日次バッチ）
COPY (
    SELECT
        id,
        received_at,
        source,
        payload
    FROM raw_logs
    WHERE received_at >= '2026-08-17'::DATE
      AND received_at < '2026-08-18'::DATE
)
TO 's3://my-data-lake/logs/dt=2026-08-17/'
(FORMAT PARQUET, COMPRESSION ZSTD);

タイミング：BEFORE / AFTER
操作：INSERT / UPDATE / DELETE
粒度：FOR EACH ROW / FOR EACH STATEMENT
遷移テーブル：REFERENCING OLD TABLE / NEW TABLE（STATEMENTレベルのみ）

-- 監査ログテーブル
CREATE TABLE audit_log (
    log_id BIGINT DEFAULT nextval('audit_seq'),
    table_name VARCHAR,
    operation VARCHAR,
    old_data VARIANT,
    new_data VARIANT,
    changed_at TIMESTAMPTZ DEFAULT now()
);

-- 注文テーブルへのトリガー（行単位）
CREATE TRIGGER orders_audit
AFTER INSERT OR UPDATE OR DELETE ON orders
FOR EACH ROW
BEGIN
    INSERT INTO audit_log (table_name, operation, old_data, new_data)
    VALUES (
        'orders',
        TG_OP,
        CASE WHEN TG_OP != 'INSERT' THEN row_to_json(OLD)::VARIANT END,
        CASE WHEN TG_OP != 'DELETE' THEN row_to_json(NEW)::VARIANT END
    );
END;

-- リアルタイム集計キャッシュの自動更新
CREATE TABLE daily_summary (
    summary_date DATE PRIMARY KEY,
    total_sales DECIMAL(15,2),
    order_count BIGINT,
    updated_at TIMESTAMPTZ
);

CREATE TRIGGER refresh_daily_summary
AFTER INSERT ON orders
FOR EACH STATEMENT
BEGIN
    INSERT INTO daily_summary (summary_date, total_sales, order_count, updated_at)
    SELECT
        CURRENT_DATE,
        SUM(amount),
        COUNT(*),
        now()
    FROM orders
    WHERE order_date = CURRENT_DATE
    ON CONFLICT (summary_date)
    DO UPDATE SET
        total_sales = EXCLUDED.total_sales,
        order_count = EXCLUDED.order_count,
        updated_at = EXCLUDED.updated_at;
END;

Error: Parser Error: syntax error at or near "FROM"
LINE 1: SELECT user_id amount FROM orders

Error: Parser Error: Expected ',' or 'FROM' after column expression, but got identifier 'amount'
  Did you forget a comma between 'user_id' and 'amount'?
  SELECT user_id amount FROM orders
                ^^^^^^^
  Hint: SELECT user_id, amount FROM orders

-- Spark互換モードの有効化
SET dialect = 'spark';

-- SparkSQL固有の構文が動く
SELECT explode(array(1, 2, 3)) AS val;
SELECT date_format(now(), 'yyyy-MM-dd') AS formatted_date;

-- 仮想的な例：時系列分析用のDSLをSQL内に埋め込む
-- （拡張機能が提供する構文）
SELECT TIMESERIES(
    metric = cpu_usage,
    interval = '5m',
    window = '1h',
    aggregation = avg
) FROM metrics WHERE host = 'server-01';

従来：[I/O要求] → [待機] → [データ受信] → [計算] → [I/O要求] → [待機] → ...
v2.0：[I/O要求1] → [I/O要求2] → [I/O要求3] → ...（ASYNCプール）
              ↓データ到着            ↓               ↓
      [計算1]   →    [計算2]   →   [計算3]   →    ...（REGULARプール）

-- 非同期I/Oの有効化（v2.0ではデフォルト有効）
SET enable_async_io = true;

-- ASYNCスレッド数の調整（ネットワーク帯域が広い環境では増やす）
SET async_io_threads = 64;  -- デフォルト: CPU数に依存

-- 先読みバッファサイズの調整
SET prefetch_buffer_size = '256MB';  -- デフォルト: 64MB

-- S3のリージョンを明示（DNS解決の遅延を回避）
SET s3_region = 'ap-northeast-1';

-- 組織階層をたどる再帰CTE（v2.0で40倍高速化する典型例）
WITH RECURSIVE org_hierarchy AS (
    -- ベースケース：ルートノード
    SELECT employee_id, name, manager_id, 0 AS depth
    FROM employees
    WHERE manager_id IS NULL

    UNION ALL

    -- 再帰ステップ：一階層ずつ下へ
    SELECT e.employee_id, e.name, e.manager_id, h.depth + 1
    FROM employees e
    JOIN org_hierarchy h ON e.manager_id = h.employee_id
    WHERE h.depth < 10  -- 無限ループ防止
)
SELECT * FROM org_hierarchy ORDER BY depth, name;

# 1. バックアップ（必須）
cp my_database.db my_database.db.backup_v1.5

# 2. チェックサムを記録（ロールバック確認用）
md5sum my_database.db > my_database.md5

# 3. v2.0でDBを開いて変換の確認
duckdb my_database.db -c "SELECT * FROM pragma_database_list();"

# 4. ストレージフォーマットv2.0への明示的な変換
duckdb my_database.db -c "CHECKPOINT;"

# v1.5のバイナリに戻してバックアップを使用
cp my_database.db.backup_v1.5 my_database.db

-- 要注意①：曖昧なキャスト
-- v1.5では暗黙変換されていたケースが、v2.0では明示的なキャストを要求する場合がある
-- Before（動くかもしれない）
SELECT '2026-01-01' + INTERVAL '1 day';
-- After（明示的に）
SELECT '2026-01-01'::DATE + INTERVAL '1 day';

-- 要注意②：予約語の追加
-- VARIANTが予約語になったため、カラム名・テーブル名に使っている場合はクォートが必要
-- Before
CREATE TABLE t (variant TEXT);  -- エラーになる可能性
-- After
CREATE TABLE t ("variant" TEXT);  -- ダブルクォートでエスケープ

-- 要注意③：関数名の変更（一部）
-- 拡張機能の関数は各拡張機能の移行ガイドを確認すること

# v1.5の最新パッチへアップデート
pip install --upgrade duckdb

# 既存クエリのテストスイートを実行して差異がないことを確認
python -m pytest tests/database/ -v

-- 非同期I/Oのみ有効化（v1.5.x backportまたはv2.0 RC）
SET enable_async_io = true;

-- パフォーマンス計測用クエリ（S3が主なデータソースの場合）
EXPLAIN ANALYZE
SELECT COUNT(*) FROM read_parquet('s3://my-bucket/data/**/*.parquet');

複数メンバーがDuckDBを同時利用する必要がある？
  → YES → Quackを検討（v2.0 GA後）
  → NO  → 組み込みのまま継続

JSON文字列でスキーマが不定なデータを扱っている？
  → YES → VARIANTへの移行を検討（v1.5でも利用可能）
  → NO  → 既存スキーマを維持

長時間稼働のDuckDBサーバーでイベント駆動の処理が必要？
  → YES → トリガーを検討（v2.0 GA後）
  → NO  → バッチ処理で代替

import duckdb
import time
import psutil
import os

def benchmark_query(db_path: str, query: str, label: str):
    """クエリの実行時間とメモリピークを計測"""
    process = psutil.Process(os.getpid())
    mem_before = process.memory_info().rss / 1024 / 1024  # MB

    conn = duckdb.connect(db_path)
    start = time.perf_counter()
    result = conn.execute(query).fetchall()
    elapsed = time.perf_counter() - start

    mem_after = process.memory_info().rss / 1024 / 1024
    mem_peak = mem_after - mem_before

    print(f"[{label}]")
    print(f"  実行時間: {elapsed:.3f}秒")
    print(f"  メモリ増加: {mem_peak:.1f}MB")
    print(f"  結果行数: {len(result)}")
    conn.close()

# 移行前後で同じクエリを計測
TEST_QUERY = """
    SELECT date_trunc('day', ts), COUNT(*), AVG(amount)
    FROM events
    WHERE ts > '2026-01-01'
    GROUP BY 1
    ORDER BY 1
"""

benchmark_query("my_database.db", TEST_QUERY, "v2.0移行後")

分析者A (Python/Jupyter) ─┐
分析者B (SQL Client)     ──┤→ Quack Server (EC2 t3.xlarge) → S3 Parquet
分析者C (BI Tool)        ─┘   DuckDB + httpfs + quack拡張

# サーバー起動スクリプト（EC2上で実行）
import duckdb

conn = duckdb.connect('/data/shared_analytics.db')
conn.execute("INSTALL httpfs; LOAD httpfs;")
conn.execute("INSTALL quack; LOAD quack;")

# S3アクセス設定
conn.execute("""
    SET s3_region = 'ap-northeast-1';
    SET s3_access_key_id = getenv('AWS_ACCESS_KEY_ID');
    SET s3_secret_access_key = getenv('AWS_SECRET_ACCESS_KEY');
    SET enable_async_io = true;
    SET async_io_threads = 64;
""")

# Quackサーバー起動（ブロッキング）
conn.execute("SELECT quack_serve('/data/shared_analytics.db', host='0.0.0.0', port=9494);")

Webサービス → アプリログ → DuckDB（VARIANT型受け取り）
                              ↓ AFTER INSERT トリガー
                         リアルタイム集計テーブル更新
                              ↓ 日次バッチ
                         S3 Parquet（分析用）

アプリ → PostgreSQL（OLTP、書き込み/参照）
              ↓ CONNECTで差分同期
         DuckDB（分析専用、CONNECTで接続）
              ↓
         BI Tool / 分析クエリ

import duckdb
from apscheduler.schedulers.blocking import BlockingScheduler

def sync_from_postgres():
    conn = duckdb.connect('analytics.db')
    conn.execute("LOAD quack;")

    conn.execute(f"""
        CONNECT 'postgresql://{PG_DSN}' AS pg;

        -- 差分のみを取り込む
        INSERT OR REPLACE INTO local_orders
        SELECT * FROM pg.orders
        WHERE updated_at > (
            SELECT COALESCE(MAX(updated_at), '1970-01-01') FROM local_orders
        );
    """)
    conn.close()

scheduler = BlockingScheduler()
scheduler.add_job(sync_from_postgres, 'interval', minutes=5)
scheduler.start()

# 1. v1.5の最新をインストール（v2.0 RC待ちの間に基盤を整える）
pip install --upgrade duckdb

# 2. 非同期I/Oの効果を現在のワークロードで確認
python -c "
import duckdb, time
conn = duckdb.connect()
conn.execute(\"SET enable_async_io = true; INSTALL httpfs; LOAD httpfs;\")
t = time.perf_counter()
conn.execute(\"SELECT COUNT(*) FROM read_parquet('s3://your-bucket/your-data/**/*.parquet')\").fetchall()
print(f'実行時間: {time.perf_counter() - t:.2f}秒')
"

# 3. VARIANT型の基本操作を手元で試す
python -c "
import duckdb
conn = duckdb.connect()
conn.execute(\"\"\"
    CREATE TABLE test AS
    SELECT '{\"user_id\": 1, \"action\": \"click\"}'::VARIANT AS payload;
    SELECT payload.user_id::BIGINT, payload.action::VARCHAR FROM test;
\"\"\").df().to_csv('/dev/stdout', index=False)
"

現在のDuckDBで「複数人が同時接続したい」という要求があるか？
  → NO  → 組み込みモードで継続、v2.0 GAを待つ
  → YES → Quack RC版を検証環境でテスト → GA後に本番投入

S3上のCSV/Parquetへのクエリが遅くて困っているか？
  → NO  → 現状維持
  → YES → 今すぐ非同期I/OをONにする（v1.5でも有効）

JSONカラムにスキーマが不定なデータを格納しているか？
  → NO  → 既存スキーマを維持
  → YES → VARIANTへの段階的移行を開始（v1.5から利用可能）