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';