CREATE TABLE IF NOT EXISTS orders_raw (
    order_id VARCHAR,
    user_id VARCHAR,
    product_id VARCHAR,
    category VARCHAR,
    amount DECIMAL(12,2),
    status VARCHAR,        -- paid, refunded, pending, cancelled
    event_time TIMESTAMP,
    country VARCHAR,
    utm_source VARCHAR,
    -- ... ~200 fields total
    _loaded_at TIMESTAMP DEFAULT now()
);

import duckdb
import polars as pl
from pathlib import Path
from datetime import datetime, timedelta

DB_PATH = "/data/analytics.duckdb"
PARQUET_DIR = "/data/parquet/orders"

con = duckdb.connect(DB_PATH)

def ingest_batch(df: pl.DataFrame):
    """Write Polars DataFrame into DuckDB"""
    con.register("_batch", df.to_arrow())
    
    # Append only, no upserts needed
    con.execute("""
        INSERT INTO orders_raw 
        SELECT *, now() AS _loaded_at FROM _batch
    """)
    
    # Archive every 5M rows or every 24 hours
    row_count = con.execute(
        "SELECT count(*) FROM orders_raw "
        "WHERE event_time < now() - interval '6 hours'"
    ).fetchone()[0]
    
    if row_count > 5_000_000:
        partition_key = datetime.now().strftime("%Y%m%d_%H")
        con.execute(f"""
            COPY (
                SELECT * FROM orders_raw 
                WHERE event_time < now() - interval '6 hours'
            ) TO '{PARQUET_DIR}/{partition_key}.parquet'
            (FORMAT PARQUET, COMPRESSION ZSTD)
        """)
        # Clean up archived data
        con.execute("""
            DELETE FROM orders_raw 
            WHERE event_time < now() - interval '6 hours'
        """)

-- Hourly pre-aggregation table
CREATE TABLE IF NOT EXISTS gmv_hourly AS
SELECT 
    date_trunc('hour', event_time) AS hour,
    category,
    country,
    status,
    count(*) AS order_count,
    sum(amount) AS gmv,
    count(DISTINCT user_id) AS unique_buyers,
    sum(CASE WHEN status = 'paid' THEN amount ELSE 0 END) AS paid_gmv,
    sum(CASE WHEN status = 'refunded' THEN amount ELSE 0 END) AS refund_amount
FROM orders_raw
WHERE event_time >= date_trunc('hour', now()) - interval '48 hours'
GROUP BY ALL;

-- Unique constraint for upsert
CREATE UNIQUE INDEX idx_gmv_hourly ON gmv_hourly (hour, category, country, status);

INSERT OR REPLACE INTO gmv_hourly
SELECT 
    date_trunc('hour', event_time) AS hour,
    category,
    country,
    status,
    count(*) AS order_count,
    sum(amount) AS gmv,
    count(DISTINCT user_id) AS unique_buyers,
    sum(CASE WHEN status = 'paid' THEN amount ELSE 0 END) AS paid_gmv,
    sum(CASE WHEN status = 'refunded' THEN amount ELSE 0 END) AS refund_amount
FROM orders_raw
WHERE event_time >= date_trunc('hour', now()) - interval '2 hours'
GROUP BY ALL
ON CONFLICT (hour, category, country, status) DO UPDATE SET
    order_count = excluded.order_count,
    gmv = excluded.gmv,
    unique_buyers = excluded.unique_buyers,
    paid_gmv = excluded.paid_gmv,
    refund_amount = excluded.refund_amount;

import streamlit as st
import duckdb
import plotly.express as px
import pandas as pd
from datetime import datetime, timedelta

st.set_page_config(layout="wide", page_title="GMV Real-Time Monitor")

con = duckdb.connect("/data/analytics.duckdb", read_only=True)

@st.cache_data(ttl=60)
def load_realtime_metrics():
    """Load today vs yesterday comparison"""
    return con.execute("""
        WITH today AS (
            SELECT 
                count(*) AS orders,
                sum(amount) AS gmv,
                count(DISTINCT user_id) AS buyers
            FROM orders_raw
            WHERE date_trunc('day', event_time) = date_trunc('day', now())
        ),
        yesterday AS (
            SELECT 
                count(*) AS orders,
                sum(amount) AS gmv,
                count(DISTINCT user_id) AS buyers
            FROM orders_raw
            WHERE date_trunc('day', event_time) = date_trunc('day', now() - interval '1 day')
        )
        SELECT 
            t.orders, t.gmv, t.buyers,
            y.orders AS y_orders, y.gmv AS y_gmv, y.buyers AS y_buyers,
            CASE WHEN y.gmv > 0 
                THEN round((t.gmv - y.gmv) / y.gmv * 100, 1) 
                ELSE 0 END AS gmv_growth_pct
        FROM today t, yesterday y
    """).fetchdf()

@st.cache_data(ttl=300)
def load_hourly_trend():
    """Last 48 hours GMV trend"""
    return con.execute("""
        SELECT 
            hour,
            sum(gmv) AS total_gmv,
            sum(order_count) AS total_orders
        FROM gmv_hourly
        WHERE hour >= now() - interval '48 hours'
        GROUP BY hour
        ORDER BY hour
    """).fetchdf()

@st.cache_data(ttl=300)
def load_top_categories():
    """Today's category ranking"""
    return con.execute("""
        SELECT 
            category,
            count(*) AS orders,
            sum(amount) AS gmv,
            count(DISTINCT user_id) AS buyers
        FROM orders_raw
        WHERE date_trunc('day', event_time) = date_trunc('day', now())
        GROUP BY category
        ORDER BY gmv DESC
        LIMIT 10
    """).fetchdf()

# ── Top Metric Cards ──
metrics = load_realtime_metrics()
col1, col2, col3, col4 = st.columns(4)
col1.metric("Today GMV", f"¥{metrics['gmv'][0]:,.0f}", 
            f"{metrics['gmv_growth_pct'][0]:+.1f}%")
col2.metric("Today Orders", f"{metrics['orders'][0]:,}",
            f"{metrics['orders'][0] - metrics['y_orders'][0]:+,}")
col3.metric("Buyers", f"{metrics['buyers'][0]:,}",
            f"{metrics['buyers'][0] - metrics['y_buyers'][0]:+,}")
col4.metric("Avg Order Value", f"¥{metrics['gmv'][0]/max(metrics['orders'][0],1):,.0f}")

# ── Trend Chart ──
st.subheader("48-Hour GMV Trend")
df_trend = load_hourly_trend()
fig = px.line(df_trend, x='hour', y='total_gmv',
              title='Hourly GMV Trend')
st.plotly_chart(fig, use_container_width=True)

# ── Category Ranking ──
st.subheader("Today's Category Top 10")
df_cat = load_top_categories()
fig_bar = px.bar(df_cat, x='category', y='gmv',
                  title='GMV by Category')
st.plotly_chart(fig_bar, use_container_width=True)

streamlit run dashboard.py --server.port 8501 --server.maxUploadSize 10

-- Increase checkpoint threshold before batch writes
SET checkpoint_threshold = '500MB';

-- Or disable auto-checkpoint entirely (batch-only scenarios)
SET automatic_checkpoint = false;

-- Manual checkpoint after batch completes
CHECKPOINT;

-- Restore defaults
SET checkpoint_threshold = '16MB';

-- Increase memory limit (default is 80% of RAM)
SET memory_limit = '6GB';

-- Set temp directory to avoid filling /tmp
SET temp_directory = '/data/tmp';

-- Set parallelism to match CPU cores
SET threads = 4;

-- External merge sort to save memory
SET max_temp_directory_size = '10GB';