DuckDB Realtime Log Aggregation: Partitioned materialized tables + Arrow Cache for 620x Speedup

CREATE TABLE IF NOT EXISTS mv_dau AS
SELECT DATE(ts) AS event_date, COUNT(DISTINCT user_id) AS dau
FROM user_events
GROUP BY DATE(ts);

import duckdb
import time

con = duckdb.connect("log_warehouse.db", read_only=False)

# Create partitioned table—note the PARTITION BY clause
con.execute("""
CREATE TABLE IF NOT EXISTS user_events (
    event_id BIGINT,
    user_id BIGINT,
    event_type VARCHAR,
    page VARCHAR,
    ts TIMESTAMP,
    duration_ms BIGINT,
    session_id VARCHAR
) PARTITION BY (DATE(ts))
""")
print("Partitioned table created")

# Main DAU aggregation view (partitioned)
con.execute("""
CREATE materialized table IF NOT EXISTS mv_daily_dau
PARTITION BY (event_date) AS
SELECT
    DATE(ts) AS event_date,
    COUNT(DISTINCT user_id) AS dau,
    COUNT(*) AS total_events
FROM user_events
GROUP BY DATE(ts)
""")

# Page statistics view
con.execute("""
CREATE materialized table IF NOT EXISTS mv_page_stats
PARTITION BY (event_date) AS
SELECT
    DATE(ts) AS event_date,
    page,
    COUNT(*) AS pv,
    ROUND(AVG(duration_ms), 2) AS avg_duration_ms,
    COUNT(DISTINCT user_id) AS uv
FROM user_events
GROUP BY DATE(ts), page
""")
print("Partitioned materialized tables created")

# Set Arrow memory limit (adjust based on machine config)
con.execute("SET arrow_memory_limit = '4GB'")
con.execute("SET enable_arrow_cache = true")

# After first query, subsequent identical queries hit cache directly
result = con.execute("""
    SELECT event_date, dau, total_events
    FROM mv_daily_dau
    WHERE event_date >= '2026-08-01'
    ORDER BY event_date DESC
    LIMIT 30
""").fetchdf()
print(result)

# Enable auto refresh (default behavior)
# DuckDB automatically detects if materialized tables are stale before queries
# If new data is written, it triggers incremental refresh automatically
con.execute("SET automatic_materialized_view_refresh = true")

# Queries return in milliseconds; refresh happens transparently
df = con.execute("SELECT * FROM mv_daily_dau ORDER BY event_date DESC LIMIT 7").fetchdf()

import schedule
import time

def refresh_all_views():
    """Batch refresh at 2 AM daily"""
    con.execute("REFRESH materialized table mv_daily_dau")
    con.execute("REFRESH materialized table mv_page_stats")
    print(f"[{time.strftime('%Y-%m-%d %H:%M')}] Batch refresh complete")

# Execute daily at 2 AM
schedule.every().day.at("02:00").do(refresh_all_views)

while True:
    schedule.run_pending()
    time.sleep(60)

# Batch refresh historical data at midnight
con.execute("REFRESH materialized table mv_daily_dau")

# During daytime, only refresh today's incremental partition
con.execute("""
    REFRESH materialized table mv_daily_dau
    WHERE event_date = CURRENT_DATE
""")

# View metadata for all materialized tables
con.execute("SELECT * FROM duckdb_materialized_views()").fetchdf()

# Sample output:
# view_name    | schema | query                                      | created
# -------------|--------|--------------------------------------------|--------
# mv_daily_dau | main   | SELECT DATE(ts)...                         | ...
# mv_page_stats| main   | SELECT DATE(ts)...                         | ...

# View Arrow cache statistics
con.execute("SELECT * FROM arrow_cache_stats()").fetchdf()

# If hit rate is below 80%, consider:
# 1. Increase arrow_memory_limit
# 2. Reduce query complexity
# 3. Increase refresh frequency

# Delete partition data older than 90 days (keep 3 months)
con.execute("""
    DELETE FROM user_events
    WHERE DATE(ts) < CURRENT_DATE - INTERVAL '90' DAY
""")

# Re-analyze partition metadata after cleanup
con.execute("ANALYZE user_events")

import duckdb
import pandas as pd
import time

# Initialize connection
con = duckdb.connect("log_warehouse.db")

# 1. Create partitioned table
con.execute("""
CREATE TABLE IF NOT EXISTS user_events (
    event_id BIGINT,
    user_id BIGINT,
    event_type VARCHAR,
    page VARCHAR,
    ts TIMESTAMP,
    duration_ms BIGINT,
    session_id VARCHAR
) PARTITION BY (DATE(ts))
""")

# 2. Create partitioned materialized tables
con.execute("""
CREATE materialized table IF NOT EXISTS mv_daily_dau AS
SELECT DATE(ts) AS event_date,
       COUNT(DISTINCT user_id) AS dau,
       COUNT(*) AS total_events
FROM user_events
GROUP BY DATE(ts)
""")

# 3. Enable Arrow cache
con.execute("SET arrow_memory_limit = '4GB'")
con.execute("SET enable_arrow_cache = true")

# 4. Simulate data insertion (replace with real data source in production)
print("Simulating data write...")
con.execute("""
INSERT INTO user_events
SELECT
    generate_series AS event_id,
    (random() * 100000)::BIGINT AS user_id,
    CASE random()
        WHEN 0 THEN 'page_view' WHEN 1 THEN 'click' WHEN 2 THEN 'purchase'
    END AS event_type,
    CASE (random() * 5)::INT
        WHEN 0 THEN '/home' WHEN 1 THEN '/products'
        WHEN 2 THEN '/cart' WHEN 3 THEN '/checkout'
        ELSE '/profile'
    END AS page,
    TIMESTAMP '2026-01-01' + INTERVAL (random() * 200) DAY AS ts,
    (random() * 5000)::BIGINT AS duration_ms,
    'session_' || (random() * 10000)::BIGINT AS session_id
FROM generate_series(1, 1000000)
""")

# 5. Performance benchmark
print("\n=== Performance Test ===")

# Full table scan
start = time.time()
result = con.execute("""
    SELECT DATE(ts) AS event_date,
           COUNT(DISTINCT user_id) AS dau
    FROM user_events
    GROUP BY DATE(ts)
    ORDER BY event_date DESC
    LIMIT 7
""").fetchdf()
print(f"Full table scan: {time.time() - start:.2f} seconds")

# materialized table (first time)
start = time.time()
result = con.execute("SELECT * FROM mv_daily_dau ORDER BY event_date DESC LIMIT 7").fetchdf()
print(f"materialized table (first): {time.time() - start:.2f} seconds")

# materialized table (cache hit)
start = time.time()
result = con.execute("SELECT * FROM mv_daily_dau ORDER BY event_date DESC LIMIT 7").fetchdf()
print(f"materialized table (cached): {time.time() - start:.4f} seconds")

# 6. Check materialized table status
print("\n=== materialized table Status ===")
print(con.execute("SELECT * FROM duckdb_materialized_views()").fetchdf())