import duckdb

# No setup needed — create a DB or work in-memory
con = duckdb.connect()  # In-memory

# Query CSV directly (no import step!)
con.execute("""
    SELECT 
        category,
        COUNT(*) as orders,
        SUM(amount) as revenue,
        AVG(amount) as avg_order
    FROM read_csv_auto('orders_2026.csv')
    WHERE date >= '2026-01-01'
    GROUP BY category
    ORDER BY revenue DESC
""").df()  # Returns a pandas DataFrame

import duckdb

con = duckdb.connect()

# Install the HTTP extension for S3 access
con.execute("INSTALL httpfs; LOAD httpfs;")
con.execute("""
    SET s3_region = 'us-east-1';
    SET s3_access_key_id = '...';
    SET s3_secret_access_key = '...';
""")

# Query Parquet files directly from S3 (with predicate pushdown!)
result = con.execute("""
    SELECT 
        user_id,
        SUM(purchase_amount) as ltv,
        COUNT(*) as transactions,
        MAX(event_date) as last_purchase
    FROM read_parquet('s3://my-data-lake/events/year=2026/month=*/day=*/*.parquet')
    WHERE event_type = 'purchase'
      AND event_date >= '2026-01-01'
    GROUP BY user_id
    HAVING SUM(purchase_amount) > 1000
    ORDER BY ltv DESC
    LIMIT 1000
""").df()

import duckdb
import pandas as pd

# Load a pandas DataFrame
df = pd.read_parquet("events.parquet")  # 10GB file

# DuckDB can directly query pandas DataFrames!
# (It reads the Arrow zero-copy — no data duplication)
result = duckdb.sql("""
    SELECT 
        date_trunc('week', event_time) as week,
        platform,
        COUNT(DISTINCT user_id) as dau,
        COUNT(*) as events
    FROM df
    GROUP BY 1, 2
    ORDER BY 1, 2
""").df()

# Benchmark: pandas groupby on 10M rows → 8.2 seconds
# DuckDB on same data → 0.4 seconds (20x faster)

import duckdb

rel = (
    duckdb.read_parquet("events.parquet")
    .filter("event_type = 'purchase'")
    .project("user_id, amount, event_date")
    .aggregate(
        "user_id",
        "SUM(amount) as total_spend, COUNT(*) as orders, MAX(event_date) as last_order"
    )
    .filter("total_spend > 100")
    .order("total_spend DESC")
    .limit(100)
)

df = rel.df()

import duckdb

# Create a persistent DuckDB database
con = duckdb.connect("analytics.duckdb")

# Create tables with column-store format
con.execute("""
    CREATE TABLE IF NOT EXISTS events (
        event_id BIGINT PRIMARY KEY,
        user_id VARCHAR,
        event_type VARCHAR,
        event_time TIMESTAMP,
        properties JSON,
    )
""")

# Bulk insert from Parquet (extremely fast)
con.execute("""
    INSERT INTO events 
    SELECT * FROM read_parquet('raw_events/*.parquet')
""")

# Now query with full SQL including window functions
con.execute("""
    SELECT 
        user_id,
        event_type,
        event_time,
        LAG(event_time) OVER (PARTITION BY user_id ORDER BY event_time) as prev_event,
        DATEDIFF('minute', 
            LAG(event_time) OVER (PARTITION BY user_id ORDER BY event_time),
            event_time
        ) as minutes_since_last
    FROM events
    WHERE event_time >= NOW() - INTERVAL 7 DAYS
""").df()

# Transform 50GB of raw JSON logs into clean Parquet 
# in 12 minutes on a single m6i.2xlarge (8 CPU, 32GB)
import duckdb

con = duckdb.connect()
con.execute("""
    COPY (
        SELECT 
            json_extract_string(log, '$.user_id') as user_id,
            json_extract_string(log, '$.event') as event_type,
            CAST(json_extract_string(log, '$.timestamp') AS TIMESTAMP) as event_time,
            json_extract(log, '$.properties') as properties
        FROM read_ndjson_auto('s3://raw-logs/2026-02-22/*.ndjson.gz')
        WHERE json_extract_string(log, '$.event') IS NOT NULL
    )
    TO 's3://processed-data/events/date=2026-02-22/'
    (FORMAT PARQUET, PARTITION_BY (event_type), COMPRESSION ZSTD)
""")

# Ship analytics inside your application — no external warehouse needed
from fastapi import FastAPI
import duckdb

app = FastAPI()

# Shared connection pool for the app
@app.on_event("startup")
async def startup():
    app.state.db = duckdb.connect("app_analytics.duckdb", read_only=True)

@app.get("/api/analytics/revenue")
async def get_revenue(period: str = "30d"):
    interval_map = {"7d": "7 DAYS", "30d": "30 DAYS", "90d": "90 DAYS"}
    interval = interval_map.get(period, "30 DAYS")
    
    result = app.state.db.execute(f"""
        SELECT 
            date_trunc('day', order_date) as date,
            SUM(amount) as revenue,
            COUNT(*) as orders
        FROM orders
        WHERE order_date >= NOW() - INTERVAL {interval}
        GROUP BY 1
        ORDER BY 1
    """).fetchall()
    
    return {"data": result}

# In a Jupyter notebook — query 10GB of data locally
%load_ext duckdb_magic  

%%duckdb
SELECT 
    cohort_month,
    months_since_signup,
    COUNT(DISTINCT user_id) as users,
    SUM(revenue) as revenue
FROM read_parquet('~/data/user_cohorts/*.parquet')
GROUP BY 1, 2
ORDER BY 1, 2