# Project export: Applovin Query Challenge

This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.

## Project metadata

- Hackathon: Cal Hacks 12.0
- Tagline: We are participating in Applovin's Query Challenge, optimizing queries on a large ad dataset.
- Devpost: https://devpost.com/software/applovin-query-challenge
- GitHub: https://github.com/ArrowDigby/CalHacks12.0
- Video: https://www.youtube.com/embed/dJibkgGI0PY?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 2 GitHub contributor(s) — bluething001 (4 commits), ArrowDigby (1 commits)

## Devpost submission (written by the team)

### Inspiration

As a team, we were interested in learning more about databases, how queries are made, and different methods of storing data. Applovin's challenge seemed like a great opportunity to dip our toes into the world of data through optimizing queries.

### What it does

Our scripts utilize data aggregation, caching, and DuckDB to speed up queries on a given dataset.

### How we built it

Building off of the starting scripts, we tested and implemented a number of optimizations, learning and testing the viability of different techniques, such as indexing, partitioning, and sorting.

### Challenges we ran into

Many of the techniques we wanted to implement weren't feasible, since preprocessing gigabytes of information took a significant amount of time. We were time constrained to around 5-10 minutes of data preprocessing time, and combining techniques like sorting and indexing took too long.

### Accomplishments we're proud of

Any queries that fall into our aggregated data tables are extremely fast, and we spent a good amount of time testing which columns to aggregate, coming up with a pretty good set of aggregations (we think).

### What we learned

As a team, we gained a lot of experience working together and loved the opportunity to learn about how databases organize data and how architectural decisions influence performance.

### What's next

We are excited and eager to dive deeper into learning more about data!

## README (from the GitHub repository)

# Query Engine Benchmark

## Setup

Install dependencies:
```bash
pip install -r requirements.txt
```

Requirements: `duckdb>=1.1.1`, `pandas>=2.2.0`

## Adding Your Queries

Add your benchmark queries to `inputs.py` using the provided format:
```python
queries = [
    {
        "select": ["day", {"SUM": "total_price"}],
        "from": "events",
        "where": [{"col": "type", "op": "eq", "val": "click"}],
        "group_by": ["day"],
        "order_by": [{"col": "day", "dir": "asc"}]
    },
    # Add more queries here...
]
```

## Running the Benchmark

### Step 1: Prepare Data and Build Rollups
```bash
python prepare_and_build.py --data-dir /path/to/data
```

This will:
- Load CSV files into DuckDB
- Create optimized rollup tables
- Takes ~8-9 minutes on the full dataset (on an Apple M1)

### Step 2: Run Queries
```bash
python step3_run_queries.py --out-dir /path/to/output --truth-dir /path/to/truth
```

This will:
- Execute all benchmark queries
- Output results to CSV files
- Compare with ground truth (if provided)
- Display performance metrics

## Example

```bash
# Prepare data (one-time setup)
python prepare_and_build.py --data-dir ./data

# Run queries
python step3_run_queries.py --out-dir ./output --truth-dir ./truth
```

## File Structure

```
CalHacks12.0/
├── prepare_and_build.py   # Data preparation + rollup building (REQUIRED)
├── step3_run_queries.py   # Query execution (REQUIRED)
├── inputs.py              # Query definitions (ADD YOUR QUERIES HERE)
├── assembler.py           # SQL query builder
├── requirements.txt       # Dependencies
├── step1_prepare_data.py  # Optional: Run data prep separately
├── step2_build_rollups.py # Optional: Run rollup building separately
└── tmp/
    └── baseline.duckdb    # DuckDB database (created on first run)
```

## Optional: Modular Workflow

For development/debugging, you can run steps separately:
```bash
# Step 1 only: Prepare data
python step1_prepare_data.py --data-dir ./data

# Step 2 only: Build rollups (after step 1)
python step2_build_rollups.py

# Step 3: Run queries
python step3_run_queries.py --out-dir ./output --truth-dir ./truth
```

## Cleanup

To start fresh:
```bash
rm -rf tmp/baseline.duckdb*
```


## Detected evidence (automated analysis)

Indexed codebase: 8 recognized source files, 59 KB.
- Python (language) — detected in the code

## Codebase structure (from repository index)

### Files (11 of 11)

```
__init__.py
.DS_Store
.gitignore
assembler.py
inputs.py
prepare_and_build.py
README.md
requirements.txt
step1_prepare_data.py
step2_build_rollups.py
step3_run_queries.py
```

### Dependencies

- requirements.txt: duckdb@>=1.1.1, pandas@>=2.2.0

### Recent commits (newest first)

- removed unique user count (took too long to preprocess)
- finalized scripts and wrote readme
- separated process into steps
- potentially working aggregation
- Add files via upload

## Key source files (fetched from GitHub, selected and truncated for size)

### requirements.txt

```
duckdb>=1.1.1
pandas>=2.2.0
```

### inputs.py

```python
#!/usr/bin/env python3

"""
Source of queries to test
"""

queries = [
    {
        "select": ["day", {"SUM": "bid_price"}],
        "from": "events",
        "where": [ {"col": "type", "op": "eq", "val": "impression"} ],
        "group_by": ["day"],
    },
    {
        "select": ["publisher_id", {"SUM": "bid_price"}],
        "from": "events",
        "where": [
            {"col": "type", "op": "eq", "val": "impression"},
            {"col": "country", "op": "eq", "val": "JP"},
            {"col": "day", "op": "between", "val": ["2024-10-20", "2024-10-23"]}
        ],
        "group_by": ["publisher_id"],
    },
    {
        "select": ["country", {"AVG": "total_price"}],
        "from": "events",
        "where": [{"col": "type", "op": "eq", "val": "purchase"}],
        "group_by": ["country"],
        "order_by": [{"col": "AVG(total_price)", "dir": "desc"}]
    },
    {
        "select": ["advertiser_id", "type", {"COUNT": "*"}],
        "from": "events",
        "group_by": ["advertiser_id", "type"],
        "order_by": [{"col": "COUNT(*)", "dir": "desc"}]
    },
    {
        "select": ["minute", {"SUM": "bid_price"}],
        "from": "events",
        "where": [
            {"col": "type", "op": "eq", "val": "impression"},
            {"col": "day", "op": "eq", "val": "2024-06-01"}
        ],
        "group_by": ["minute"],
        "order_by": [{"col": "minute", "dir": "asc"}]
    }
]
```

### assembler.py

```python
# Asbemble JSON query to SQL
# Note -- your solution may or may
# not need to use something similar depending on how you
# do query scheduling

def assemble_sql(q):
    select = _select_to_sql(q.get("select", []))
    from_tbl = q["from"]
    where = _where_to_sql(q.get("where"))
    group_by = _group_by_to_sql(q.get("group_by"))
    order_by = _order_by_to_sql(q.get("order_by"))
    sql = f"SELECT {select} FROM {from_tbl} {where} {group_by} {order_by}"
    if q.get("limit"):
        sql += f" LIMIT {q['limit']}"
    return sql.strip()


def _where_to_sql(where):
    if not where:
        return ""
    parts = []
    for cond in where:
        col, op, val = cond["col"], cond["op"], cond["val"]
        if op == "eq":
            parts.append(f"{col} = '{val}'")
        if op == "neq":
            parts.append(f"{col} != '{val}'")
        elif op in ("lt", "lte", "gt", "gte"):
            sym = {"lt": "<", "lte": "<=", "gt": ">", "gte": ">="}[op]
            parts.append(f"{col} {sym} {val}")
        elif op == "between":
            low, high = val
            parts.append(f"{col} BETWEEN '{low}' AND '{high}'")
        elif op == "in":
            vals = ", ".join(f"'{v}'" for v in val)
            parts.append(f"{col} IN ({vals})")
    return "WHERE " + " AND ".join(parts)


def _select_to_sql(select):
    parts = []
    for item in select:
        if isinstance(item, str):
            parts.append(item)
        elif isinstance(item, dict):
            for func, col in item.items():
                parts.append(f"{func.upper()}({col})")
    return ", ".join(parts)


def _group_by_to_sql(group_by):
    if not group_by: return ""
    return "GROUP BY " + ", ".join(group_by)


def _order_by_to_sql(order_by):
    if not order_by: return ""
    parts = [f"{o['col']} {o.get('dir', 'asc').upper()}" for o in order_by]
    return "ORDER BY " + ", ".join(parts)

```

### step1_prepare_data.py

```python
#!/usr/bin/env python3
"""
Step 1: Prepare Data
--------------------
Reads CSV files and creates:
1. Persisted table (events_persisted) in DuckDB
2. View (events_raw) pointing to the persisted table for fallback queries

Usage:
  python step1_prepare_data.py --data-dir ./data
"""

import duckdb
import time
import argparse
from pathlib import Path
import os

# Configuration
DB_PATH = Path("tmp/baseline.duckdb")
TABLE_NAME = "events"
PERSISTED_TABLE = "events_persisted"


def load_data(con, data_dir: Path):
    """Load CSV data into a view with proper typing and derived columns."""
    csv_files = list(data_dir.glob("events_part_*.csv"))

    if csv_files:
        print(f"🟩 Loading {len(csv_files)} CSV parts from {data_dir} ...")
        con.execute(f"""
            CREATE OR REPLACE VIEW {TABLE_NAME} AS
            WITH raw AS (
              SELECT *
              FROM read_csv(
                '{data_dir}/events_part_*.csv',
                AUTO_DETECT = FALSE,
                HEADER = TRUE,
                union_by_name = TRUE,
                COLUMNS = {{
                  'ts': 'VARCHAR',
                  'type': 'VARCHAR',
                  'auction_id': 'VARCHAR',
                  'advertiser_id': 'VARCHAR',
                  'publisher_id': 'VARCHAR',
                  'bid_price': 'VARCHAR',
                  'user_id': 'VARCHAR',
                  'total_price': 'VARCHAR',
                  'country': 'VARCHAR'
                }}
              )
            ),
            casted AS (
              SELECT
                to_timestamp(TRY_CAST(ts AS DOUBLE) / 1000.0)    AS ts,
                type,
                auction_id,
                TRY_CAST(advertiser_id AS INTEGER)        AS advertiser_id,
                TRY_CAST(publisher_id  AS INTEGER)        AS publisher_id,
                NULLIF(bid_price, '')::DOUBLE             AS bid_price,
                TRY_CAST(user_id AS BIGINT)               AS user_id,
                NULLIF(total_price, '')::DOUBLE           AS total_price,
                country
              FROM raw
            )
            SELECT
              ts,
              DATE_TRUNC('week', ts)              AS week,
              DATE(ts)                            AS day,
              DATE_TRUNC('hour', ts)              AS hour,
              STRFTIME(ts, '%Y-%m-%d %H:%M')      AS minute,
              type,
              auction_id,
              advertiser_id,
              publisher_id,
              bid_price,
              user_id,
              total_price,
              country
            FROM casted;
        """)
        print(f"🟩 Loading complete")
    else:
        raise FileNotFoundError(f"No events_part_*.csv found in {data_dir}")


def create_persisted_table(con):
    """Create a physical table from the events view."""
    print("🟩 Creating persisted table from events view...")
    t0 = time.time()
    con.execute(f"CREATE OR REPLACE TABLE {PERSISTED_TABLE} AS SELECT * FROM {TABLE_NAME};")
    print(f"   ✓ Persisted table created in {time.time() - t0:.1f}s")


def create_fallback_view(con):
    """Create a view that points to the persisted table for queries that don't use rollups."""
    print("🟩 Creating fallback view (events_raw → events_persisted)...")
    con.execute(f"""
        CREATE OR REPLACE VIEW events_raw AS
        SELECT * FROM {PERSISTED_TABLE};
    """)
    print("   ✓ View created")


def main():
    parser = argparse.ArgumentParser(
        description="Step 1: Prepare data - create persisted table in DuckDB"
    )
    parser.add_argument(
        "--data-dir",
        type=Path,
        required=True,
        help="The folder where the input CSV files are located"
    )

    args = parser.parse_args()

    # Ensure database directory exists
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)

    # Connect to DuckDB
    con = duckdb.connect(DB_PATH)
    
    # Tune DuckDB for optimal performance
    threads = os.cpu_count() or 4
    con.execute(f"PRAGMA threads={threads};")
    mem_limit = os.environ.get("DUCKDB_MEMORY_LIMIT", "14GB")
    con.execute(f"PRAGMA memory_limit='{mem_limit}';")
    con.execute("SET preserve_insertion_order=false;")

    print("=" * 60)
    print("STEP 1: PREPARE DATA")
    print("=" * 60)
    print(f"📊 Using {threads} threads and {mem_limit} memory limit")

    # Load and prepare data
    load_data(con, args.data_dir)
    create_persisted_table(con)
    create_fallback_view(con)

    con.close()

    print("\n✅ Step 1 complete! Data is prepared and ready.")
    print(f"   Next: Run step2_build_rollups.py")


if __name__ == "__main__":
    main()


```

### step2_build_rollups.py

```python
#!/usr/bin/env python3
"""
Step 2: Build Rollups
---------------------
Creates pre-aggregated rollup tables for fast query execution.

Creates rollups:
- by_day (day, type)
- by_country_day (day, country, type)
- by_publisher_day (day, publisher_id, type)
- by_advertiser_day (day, advertiser_id, type)
- by_publisher_country_day (day, publisher_id, country, type)
- by_advertiser_country_day (day, advertiser_id, country, type)
- by_publisher_advertiser_day (day, publisher_id, advertiser_id, type)
- by_minute (minute, day, type)
- by_country (country, type)
- by_publisher (publisher_id, type)
- by_advertiser (advertiser_id, type)

Usage:
  python step2_build_rollups.py
"""

import duckdb
import time
from pathlib import Path

# Configuration
DB_PATH = Path("tmp/baseline.duckdb")
PERSISTED_TABLE = "events_persisted"

# Rollup table names
ROLLUP_BY_DAY = "by_day"
ROLLUP_BY_COUNTRY_DAY = "by_country_day"
ROLLUP_BY_PUBLISHER_DAY = "by_publisher_day"
ROLLUP_BY_ADVERTISER_DAY = "by_advertiser_day"
ROLLUP_BY_PUBLISHER_COUNTRY_DAY = "by_publisher_country_day"
ROLLUP_BY_ADVERTISER_COUNTRY_DAY = "by_advertiser_country_day"
ROLLUP_BY_PUBLISHER_ADVERTISER_DAY = "by_publisher_advertiser_day"
ROLLUP_BY_MINUTE = "by_minute"
ROLLUP_BY_COUNTRY = "by_country"
ROLLUP_BY_PUBLISHER = "by_publisher"
ROLLUP_BY_ADVERTISER = "by_advertiser"


def build_day_rollups(con):
    """Build rollups with day dimension."""
    print("🟩 Building day-level rollups...")
    
    print("   Building by_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_DAY} AS
        SELECT day, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY day, type;
    """)
    print(f"   ✓ by_day created in {time.time() - t0:.1f}s")
    
    print("   Building by_country_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_COUNTRY_DAY} AS
        SELECT day, country, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY day, country, type;
    """)
    print(f"   ✓ by_country_day created in {time.time() - t0:.1f}s")
    
    print("   Building by_publisher_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_PUBLISHER_DAY} AS
        SELECT day, publisher_id, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY day, publisher_id, type;
    """)
    print(f"   ✓ by_publisher_day created in {time.time() - t0:.1f}s")
    
    print("   Building by_advertiser_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_ADVERTISER_DAY} AS
        SELECT day, advertiser_id, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY day, advertiser_id, type;
    """)
    print(f"   ✓ by_advertiser_day created in {time.time() - t0:.1f}s")
    
    print("   Building by_publisher_country_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_PUBLISHER_COUNTRY_DAY} AS
        SELECT day, publisher_id, country, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY day, publisher_id, country, type;
    """)
    print(f"   ✓ by_publisher_country_day created in {time.time() - t0:.1f}s")
    
    print("   Building by_advertiser_country_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_ADVERTISER_COUNTRY_DAY} AS
        SELECT day, advertiser_id, country, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY day, advertiser_id, country, type;
    """)
    print(f"   ✓ by_advertiser_country_day created in {time.time() - t0:.1f}s")
    
    print("   Building by_publisher_advertiser_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_PUBLISHER_ADVERTISER_DAY} AS
        SELECT day, publisher_id, advertiser_id, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY day, publisher_id, advertiser_id, type;
    """)
    print(f"   ✓ by_publisher_advertiser_day created in {time.time() - t0:.1f}s")


def build_minute_rollups(con):
    """Build minute-level rollups for fine-grained time analysis."""
    print("🟩 Building minute-level rollups...")
    
    print("   Building by_minute...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_MINUTE} AS
        SELECT minute, day, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY minute, day, type;
    """)
    print(f"   ✓ by_minute created in {time.time() - t0:.1f}s")


def build_dimension_rollups(con):
    """Build dimension-only rollups (no day)."""
    print("🟩 Building dimension-only rollups (no day) ...")
    
    print("   Building by_country...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_COUNTRY} AS
        SELECT country, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY country, type;
    """)
    print(f"   ✓ by_country created in {time.time() - t0:.1f}s")
    
    print("   Building by_publisher...")
    t0
[truncated — 1490 more characters]
```

### prepare_and_build.py

```python
#!/usr/bin/env python3
"""
Combined Data Preparation and Rollup Building
----------------------------------------------
This script combines Step 1 (data preparation) and Step 2 (rollup building)
into a single workflow for convenience.

Creates:
1. Persisted table (events_persisted) in DuckDB
2. View (events_raw) pointing to the persisted table
3. All 11 rollup tables for optimized query performance

Rollups created:
- by_day (day, type)
- by_country_day (day, country, type)
- by_publisher_day (day, publisher_id, type)
- by_advertiser_day (day, advertiser_id, type)
- by_publisher_country_day (day, publisher_id, country, type)
- by_advertiser_country_day (day, advertiser_id, country, type)
- by_publisher_advertiser_day (day, publisher_id, advertiser_id, type)
- by_minute (minute, day, type)
- by_country (country, type)
- by_publisher (publisher_id, type)
- by_advertiser (advertiser_id, type)

Usage:
  python prepare_and_build.py --data-dir ./data
"""

import duckdb
import time
import argparse
from pathlib import Path
import os

# Configuration
DB_PATH = Path("tmp/baseline.duckdb")
TABLE_NAME = "events"
PERSISTED_TABLE = "events_persisted"

# Rollup table names
ROLLUP_BY_DAY = "by_day"
ROLLUP_BY_COUNTRY_DAY = "by_country_day"
ROLLUP_BY_PUBLISHER_DAY = "by_publisher_day"
ROLLUP_BY_ADVERTISER_DAY = "by_advertiser_day"
ROLLUP_BY_PUBLISHER_COUNTRY_DAY = "by_publisher_country_day"
ROLLUP_BY_ADVERTISER_COUNTRY_DAY = "by_advertiser_country_day"
ROLLUP_BY_PUBLISHER_ADVERTISER_DAY = "by_publisher_advertiser_day"
ROLLUP_BY_MINUTE = "by_minute"
ROLLUP_BY_COUNTRY = "by_country"
ROLLUP_BY_PUBLISHER = "by_publisher"
ROLLUP_BY_ADVERTISER = "by_advertiser"


# ============================================================
# STEP 1: DATA PREPARATION
# ============================================================

def load_data(con, data_dir: Path):
    """Load CSV data into a view with proper typing and derived columns."""
    csv_files = list(data_dir.glob("events_part_*.csv"))

    if csv_files:
        print(f"🟩 Loading {len(csv_files)} CSV parts from {data_dir} ...")
        con.execute(f"""
            CREATE OR REPLACE VIEW {TABLE_NAME} AS
            WITH raw AS (
              SELECT *
              FROM read_csv(
                '{data_dir}/events_part_*.csv',
                AUTO_DETECT = FALSE,
                HEADER = TRUE,
                union_by_name = TRUE,
                COLUMNS = {{
                  'ts': 'VARCHAR',
                  'type': 'VARCHAR',
                  'auction_id': 'VARCHAR',
                  'advertiser_id': 'VARCHAR',
                  'publisher_id': 'VARCHAR',
                  'bid_price': 'VARCHAR',
                  'user_id': 'VARCHAR',
                  'total_price': 'VARCHAR',
                  'country': 'VARCHAR'
                }}
              )
            ),
            casted AS (
              SELECT
                to_timestamp(TRY_CAST(ts AS DOUBLE) / 1000.0)    AS ts,
                type,
                auction_id,
                TRY_CAST(advertiser_id AS INTEGER)        AS advertiser_id,
                TRY_CAST(publisher_id  AS INTEGER)        AS publisher_id,
                NULLIF(bid_price, '')::DOUBLE             AS bid_price,
                TRY_CAST(user_id AS BIGINT)               AS user_id,
                NULLIF(total_price, '')::DOUBLE           AS total_price,
                country
              FROM raw
            )
            SELECT
              ts,
              DATE_TRUNC('week', ts)              AS week,
              DATE(ts)                            AS day,
              DATE_TRUNC('hour', ts)              AS hour,
              STRFTIME(ts, '%Y-%m-%d %H:%M')      AS minute,
              type,
              auction_id,
              advertiser_id,
              publisher_id,
              bid_price,
              user_id,
              total_price,
              country
            FROM casted;
        """)
        print(f"🟩 Loading complete")
    else:
        raise FileNotFoundError(f"No events_part_*.csv found in {data_dir}")


def create_persisted_table(con):
    """Create a physical table from the events view."""
    print("🟩 Creating persisted table from events view...")
    t0 = time.time()
    con.execute(f"CREATE OR REPLACE TABLE {PERSISTED_TABLE} AS SELECT * FROM {TABLE_NAME};")
    print(f"   ✓ Persisted table created in {time.time() - t0:.1f}s")


def create_fallback_view(con):
    """Create a view that points to the persisted table for queries that don't use rollups."""
    print("🟩 Creating fallback view (events_raw → events_persisted)...")
    con.execute(f"""
        CREATE OR REPLACE VIEW events_raw AS
        SELECT * FROM {PERSISTED_TABLE};
    """)
    print("   ✓ View created")


# ============================================================
# STEP 2: BUILD ROLLUPS
# ============================================================

def build_day_rollups(con):
    """Build rollups with day dimension."""
    print("🟩 Building day-level rollups...")
    
    print("   Building by_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_DAY} AS
        SELECT day, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY day, type;
    """)
    print(f"   ✓ by_day created in {time.time() - t0:.1f}s")
    
    print("   Building by_country_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR REPLACE TABLE {ROLLUP_BY_COUNTRY_DAY} AS
        SELECT day, country, type,
               COUNT(*) AS cnt,
               SUM(bid_price) AS sum_bid,
               SUM(total_price) AS sum_total
        FROM {PERSISTED_TABLE}
        GROUP BY day, country, type;
    """)
    print(f"   ✓ by_country_day created in {time.time() - t0:.1f}s")
    
    print("   Building by_publisher_day...")
    t0 = time.time()
    con.execute(f"""
        CREATE OR R
[truncated — 7112 more characters]
```

### step3_run_queries.py

```python
#!/usr/bin/env python3
"""
Step 3: Run Queries
-------------------
Executes benchmark queries using optimized routing and caching.

Features:
- Automatic rollup routing for fast queries
- Result caching for repeated queries
- Query performance tracking
- Terminal output only (no CSV files)

Usage:
  python step3_run_queries.py
"""

import duckdb
import time
from pathlib import Path
import argparse
import json
import copy
import csv
import pandas as pd
from assembler import assemble_sql
from inputs import queries

# Configuration
DB_PATH = Path("tmp/baseline.duckdb")

# Rollup table names
ROLLUP_BY_DAY = "by_day"
ROLLUP_BY_COUNTRY_DAY = "by_country_day"
ROLLUP_BY_PUBLISHER_DAY = "by_publisher_day"
ROLLUP_BY_ADVERTISER_DAY = "by_advertiser_day"
ROLLUP_BY_PUBLISHER_COUNTRY_DAY = "by_publisher_country_day"
ROLLUP_BY_ADVERTISER_COUNTRY_DAY = "by_advertiser_country_day"
ROLLUP_BY_PUBLISHER_ADVERTISER_DAY = "by_publisher_advertiser_day"
ROLLUP_BY_MINUTE = "by_minute"
ROLLUP_BY_COUNTRY = "by_country"
ROLLUP_BY_PUBLISHER = "by_publisher"
ROLLUP_BY_ADVERTISER = "by_advertiser"


# -------------------
# Query Routing + Cache helpers
# -------------------
def _is_simple_agg(select_items):
    allowed = {"COUNT", "SUM", "AVG"}
    for item in select_items:
        if isinstance(item, str):
            continue
        if isinstance(item, dict):
            func, _ = next(iter(item.items()))
            if func.upper() not in allowed:
                return False
    return True


def _normalize_query_for_cache(q: dict) -> str:
    def normalize(obj):
        if isinstance(obj, dict):
            return {k: normalize(obj[k]) for k in sorted(obj)}
        if isinstance(obj, list):
            return [normalize(v) for v in obj]
        return obj
    return json.dumps(normalize(q), separators=(",", ":"))


def pick_source(q: dict) -> str:
    group_by = set(q.get("group_by") or [])
    selects = q.get("select", [])
    where = q.get("where") or []
    
    # If no GROUP BY, this is a raw data query (filtering/selecting, not aggregating)
    if not group_by:
        return "events_raw"
    
    if not _is_simple_agg(selects):
        return "events_raw"
    
    # Extract columns referenced in WHERE clause
    where_cols = set()
    for cond in where:
        where_cols.add(cond["col"])
    
    needs_day = "day" in group_by
    
    # Rollup schema reference (columns that exist in each rollup):
    # by_day: day, type, cnt, sum_bid, sum_total
    # by_country_day: day, country, type, cnt, sum_bid, sum_total
    # by_publisher_day: day, publisher_id, type, cnt, sum_bid, sum_total
    # by_advertiser_day: day, advertiser_id, type, cnt, sum_bid, sum_total
    # by_publisher_country_day: day, publisher_id, country, type, cnt, sum_bid, sum_total
    # by_advertiser_country_day: day, advertiser_id, country, type, cnt, sum_bid, sum_total
    # by_publisher_advertiser_day: day, publisher_id, advertiser_id, type, cnt, sum_bid, sum_total
    # by_minute: minute, day, type, cnt, sum_bid, sum_total
    # by_country: country, type, cnt, sum_bid, sum_total
    # by_publisher: publisher_id, type, cnt, sum_bid, sum_total
    # by_advertiser: advertiser_id, type, cnt, sum_bid, sum_total
    
    # Helper function to check if rollup has all required columns
    def _rollup_has_columns(rollup_name: str, required_cols: set) -> bool:
        """Check if a rollup has all required columns (group_by + where filters)."""
        rollup_schemas = {
            ROLLUP_BY_DAY: {"day", "type"},
            ROLLUP_BY_COUNTRY_DAY: {"day", "country", "type"},
            ROLLUP_BY_PUBLISHER_DAY: {"day", "publisher_id", "type"},
            ROLLUP_BY_ADVERTISER_DAY: {"day", "advertiser_id", "type"},
            ROLLUP_BY_PUBLISHER_COUNTRY_DAY: {"day", "publisher_id", "country", "type"},
            ROLLUP_BY_ADVERTISER_COUNTRY_DAY: {"day", "advertiser_id", "country", "type"},
            ROLLUP_BY_PUBLISHER_ADVERTISER_DAY: {"day", "publisher_id", "advertiser_id", "type"},
            ROLLUP_BY_MINUTE: {"minute", "day", "type"},
            ROLLUP_BY_COUNTRY: {"country", "type"},
            ROLLUP_BY_PUBLISHER: {"publisher_id", "type"},
            ROLLUP_BY_ADVERTISER: {"advertiser_id", "type"},
        }
        rollup_cols = rollup_schemas.get(rollup_name, set())
        return required_cols.issubset(rollup_cols)
    
    # Check for minute-level queries first
    if "minute" in group_by:
        # Minute-level queries can use by_minute rollup if only grouping by minute + type
        if group_by in ({"minute"}, {"minute", "type"}, {"minute", "day"}, {"minute", "day", "type"}):
            # Check if rollup has all required columns (group_by + where filters)
            required_cols = group_by | where_cols
            if _rollup_has_columns(ROLLUP_BY_MINUTE, required_cols):
                return ROLLUP_BY_MINUTE
        return "events_raw"  # Need full data for complex minute queries
    
    if needs_day:
        # Query explicitly groups by day - use matching rollup
        dims = group_by - {"day"}
        required_cols = group_by | where_cols
        
        # 3-dimensional rollups (day + 2 dimensions)
        if dims == {"publisher_id", "country"} or dims == {"publisher_id", "country", "type"}:
            if _rollup_has_columns(ROLLUP_BY_PUBLISHER_COUNTRY_DAY, required_cols):
                return ROLLUP_BY_PUBLISHER_COUNTRY_DAY
        if dims == {"advertiser_id", "country"} or dims == {"advertiser_id", "country", "type"}:
            if _rollup_has_columns(ROLLUP_BY_ADVERTISER_COUNTRY_DAY, required_cols):
                return ROLLUP_BY_ADVERTISER_COUNTRY_DAY
        if dims == {"publisher_id", "advertiser_id"} or dims == {"publisher_id", "advertiser_id", "type"}:
            if _rollup_has_columns(ROLLUP_BY_PUBLISHER_ADVERTISER_DAY, required_cols):
                return ROLLUP_BY_PUBLISHER_ADVERTISER_DAY
        
        # 2-dimensional rollups (day + 1 dimension)
        if dims == {"country"} or dims == {"country", 
[truncated — 23550 more characters]
```