# Project export: [Table 9B] - Rollup + DuckDB Query Planner

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: Rollups + DuckDB for blazingly fast queries (~40ms for the 5 example queries)
- Devpost: https://devpost.com/software/applovin-query-planner-rollup-duckdb-query-planner
- GitHub: https://github.com/georgeIshaq/Calhacks_AppLovin_Challenge
- Video: https://www.youtube.com/embed/8-OKzEXq6B4?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — George (12 commits)

## Devpost submission (written by the team)

### Overview

🏆 1,600× Faster than Baseline — 39ms vs 62 seconds on 5 example queries

### Inspiration

Ad networks process massive volumes of event data. AppLovin needed instant query execution on 245M rows of ad events (~20GB). Traditional full-scan approaches were too slow; we needed a smarter architecture that could handle arbitrary queries without sacrificing speed.

### What it does

A two-tier query engine that executes ad-hoc analytics queries on massive event datasets in <100ms: 11 pre-aggregated rollup tables (Arrow IPC, ~340MB) handle (hopefully!) 80–90% of queries in <10ms via direct lookup Sorted DuckDB fallback (~2.5GB) handles complex multi-dimensional queries in <400ms Smart query router analyzes query patterns and routes to the optimal data structure Supports 20+ concurrent queries with <200ms total execution time

### How we built it

Phase 1: Rollup Design Analyzed baseline query patterns to identify high-impact dimensions (day, country, advertiser, publisher, type) Pre-aggregated data into 11 rollup tables covering 80–90% of query patterns Each rollup stores pre-computed aggregates: sum, count, avg, min, max Phase 2: Build Optimization Single-pass incremental folding over 49 CSV files (reduces peak memory to ~10GB) Parallel CSV reading (8 threads) + Polars multi-threading Arrow IPC format with LZ4 compression (3:1 ratio, instant memory-mapping) Phase 3: Query Execution Query router parses dimensions and filters, matches against rollup definitions Rollup queries: memory-mapped load → filter → aggregate (done) Fallback queries: DuckDB sorted scans (sorted by week, country, type for efficient range queries) Tech Stack: Python 3.9+, Polars, DuckDB, Apache Arrow, LZ4 compression

### Challenges we ran into

Memory explosion during build → Solved with incremental folding (merge accumulators every 50 batches) Build time >15 min on M3 Air → Attempted multiprocessing but file-level parallelism added merge overhead; reverted to sequential single-pass (more efficient) Choosing which dimensions to pre-aggregate → Analyzed baseline patterns; 11 rollups cover 80–90% with diminishing returns beyond Query routing accuracy → Built intelligent matcher that handles partial dimension sets and falls back to DuckDB safely Storage format trade-offs → Arrow IPC (memory-mappable, instant zero-copy reads) vs Parquet; chose IPC for 5–10× faster load

### Accomplishments we're proud of

✅ 1,600× speedup (39ms vs 62s) on example queries ✅ Sub-100ms query execution on 245M rows ✅ Compact storage (~3GB rollups + fallback vs ~20GB raw) ✅ Incremental folding keeps peak memory at ~10GB (safe on 12–16GB systems) ✅ Intelligent routing — no manual query tuning needed ✅ Validated against baseline — results are pixel-perfect identical

### What we learned

Pre-aggregation is powerful but requires careful dimension selection (diminishing returns beyond 11 rollups) Sequential single-pass is faster than parallelism for merge-heavy workloads (join overhead outweighs parallelism gains) Physical layout beats indexes — sorting DuckDB by (week, country, type) eliminates need for indexes, helps massively in reducing build time Memory-mapped I/O matters — Arrow IPC's zero-copy semantics cut load time from 2–3s to 450ms Constraints drive design — 16GB RAM limit forced us toward incremental folding and bounded memory strategies

### What's next

for AppLovin Query Planner Dynamic rollup generation — auto-detect optimal rollup dimensions from query workloads Adaptive routing — learn query patterns over time and reorganize rollups on-the-fly Distributed execution — scale to multi-machine clusters for even larger datasets Real-time updates — stream new events into rollups without rebuilding Query optimizer — cost-based planner to choose between rollup vs fallback more intelligently Quick Start

## README (from the GitHub repository)

# CalHacks AppLovin Database Challenge - Optimized Solution

**🏆 1,600× Faster than Baseline on the 5 example queries** - 39ms vs 62 seconds

Pre-aggregated rollup tables + intelligent query routing = sub-100ms query execution.

---

## Quick Start

```bash
pip install -r requirements.txt
python3 prepare.py --data-dir ./data    # One-time build: ~11-15 min on m3 air, should be faster on m2 pro
python3 run.py                           # Query execution on 5 example queries: ~40ms
```

---

## For Judges: Testing Your Queries

```bash
# Step 1: Prepare
python3 prepare.py --data-dir <your-data-path>

# Step 2: Run with your 20 queries (queries should be a json array)
python3 run.py --query-file ./your_queries.json

# Results appear in ./results/q1.csv through q20.csv
```

**Your JSON file should be an array of query objects:**
```json
[
  {
    "select": ["day", {"SUM": "bid_price"}],
    "from": "events",
    "where": [{"col": "type", "op": "eq", "val": "impression"}],
    "group_by": ["day"]
  },
  {
    "select": ["country", {"AVG": "total_price"}],
    "from": "events",
    "where": [{"col": "type", "op": "eq", "val": "purchase"}],
    "group_by": ["country"]
  }
  // ... your remaining 18 queries
]
```

**Expected output:**
- Execution time: <200ms for 20 queries
- Summary showing which rollup handled each query
- Individual CSV files for each query result

📖 See [queries/README.md](./queries/README.md) for complete query format reference.

---

## Performance

| Metric | Baseline (M3 Air) | Optimized | Speedup |
|--------|-------------------|-----------|---------|
| **5 queries** | 61.9s | 39ms | **1,600×** |
| Per query avg | 12.4s | 8ms | **1,550×** |
| Memory | ~4GB | ~10GB build, ~2-4GB query | Within limits |
| Disk | ~20GB raw | ~3GB | 85% smaller |

**Query breakdown m3 mac air:**
- Q1: 13.5s → 7ms (day aggregation)
- Q2: 11.4s → 8ms (publisher-country)
- Q3: 11.0s → 9ms (country average)
- Q4: 11.7s → 10ms (multi-group)
- Q5: 14.2s → 5ms (minute-level)

---

## Architecture

**Two-tier system:**
1. **Rollup tables** (11 pre-aggregated) → handles 80-90% of queries in <10ms
2. **DuckDB fallback** (sorted layout) → handles complex queries in <20ms

**Smart router** → analyzes query → routes to optimal data structure

---

## System Requirements

- Python 3.9+, 12GB+ RAM, 5GB disk
- Dependencies: `polars`, `duckdb`, `pyarrow`
- Apple Silicon optimized (works on Linux/Windows too)

---

## Project Structure

```
prepare.py              # Build rollups (11-15 min)
run.py                  # Execute queries (<100ms)
src/core/               # Query engine
  ├── rollup_builder.py
  ├── query_router.py
  ├── query_executor.py
  └── fallback_executor.py
rollups/                # 11 pre-aggregated tables (~340MB)
fallback.duckdb         # Sorted DuckDB table (~2.5GB)
queries/                # Example query templates
baseline/               # Original baseline for comparison
```

---

## Validation

```bash
python3 validate_setup.py           # Check setup

# Compare with baseline
cd baseline && python3 main.py --data-dir ../data --out-dir ./baseline_results
cd .. && python3 run.py --output-dir ./optimized_results
diff -r baseline/baseline_results optimized_results    # Should be identical
```

---

## Key Optimizations

1. **Pre-aggregation** - 11 rollup tables eliminate 99.9% of data scanning
2. **Incremental folding** - Process data in batches (~10GB peak during build)
3. **Sorted DuckDB** - Physical layout optimization (no indexes needed)
4. **Smart routing** - Automatic selection of optimal rollup per query
5. **Apple Silicon** - Multi-threaded Polars (8 cores) + DuckDB (7 cores)

---

## Troubleshooting

| Issue | Fix |
|-------|-----|
| "Rollup directory not found" | Run `python3 prepare.py` first |
| Build >10 min | Normal on M3 Air (11-15 min typical), should be faster on M2 pro|
| Out of memory | Close other apps during build (needs ~10GB free RAM) |
| Wrong results | Run `validate_setup.py` |

---

## Documentation

- [queries/README.md](./queries/README.md) - Query format reference

---

## Architecture Deep Dive

### Data Flow

```
Raw CSV (20GB, 245M rows)
    ↓
prepare.py (11-15 min build)
    ↓
    ├─→ 11 Rollup Tables (Arrow IPC, ~340MB)
    │   ├─ day_type, hour_type, minute_type, week_type
    │   ├─ country_type, advertiser_type, publisher_type
    │   └─ day_country_type, day_advertiser_type, hour_country_type, day_publisher_country_type
    │
    └─→ DuckDB Fallback (sorted table, ~2.5GB)
        Sorted by: (week, country, type)
    ↓
run.py (<100ms execution)
    ↓
    ├─→ Query Router → analyze dimensions & filters
    │
    ├─→ 80-90% queries → Rollup Executor (<10ms)
    │   - Direct lookup in pre-aggregated table
    │   - Apply additional filters
    │   - Return results
    │
    └─→ 10-20% queries → DuckDB Fallback (<400ms)
        - Scan sorted table (efficient range queries)
        - Complex multi-dimensional aggregations
        - Return results
```

### Rollup Table Design

Each rollup pre-aggregates data by specific dimensions and event type:

| Rollup Name | Dimensions | Aggregates | Rows | Use Case |
|-------------|------------|------------|------|----------|
| day_type | day, type | bid_price, total_price, count | ~1,500 | Daily trends (Q1) |
| hour_type | hour, type | bid_price, total_price, count | ~100 | Hourly patterns |
| minute_type | minute, type | bid_price, total_price, count | ~240 | Minute-level (Q5) |
| week_type | week, type | bid_price, total_price, count | ~220 | Weekly trends |
| country_type | country, type | bid_price, total_price, count | ~100 | Geographic (Q3) |
| advertiser_type | advertiser_id, type | bid_price, total_price, count | ~4K | Advertiser performance |
| publisher_type | publisher_id, type | bid_price, total_price, count | ~4K | Publisher performance |
| day_country_type | day, country, type | bid_price, total_price, count | ~40K | Daily geo trends |
| day_advertiser_type | day, advertiser_id, type | bid_price, total_price, count | ~1M | Daily advertiser stats |
| hour_country_type | hour, country, type | bid_price, total_price, count | ~2K | Hourly geo patterns |
| day_publisher_country_type | day, publisher_id, country, type | bid_price, total_price, count | ~900K | Publisher geo (Q2) |

**Aggregates stored per dimension combination:**
- `sum_bid_price`, `sum_total_price` (for SUM queries)
- `count` (for COUNT queries)
- `avg_bid_price`, `avg_total_price` (for AVG queries)
- `min/max` values (for MIN/MAX queries)

### Query Routing Logic

**Decision tree for query routing:**

```python
1. Parse query dimensions and filters
2. Extract: group_by dimensions + WHERE filter columns
3. Match against rollup definitions:
   
   If group_by == ["day"] and "type" in filters:
       → Use day_type rollup
   
   If group_by == ["day", "publisher_id", "country"] and "type" in filters:
       → Use day_publisher_country_type rollup
   
   If no exact match:
       → Use DuckDB fallback
       
4. Execute query on selected data source
5. Apply additional filters (date ranges, etc.)
6. Return results
```

### Build Process Details

**Phase 1: Rollup Building (10-11 min)**

```python
For each CSV file (49 files):
    1. Read batch with PyArrow (256MB blocks, 8 threads)
    2. Add time dimensions (day, hour, minute, week)
    3. Aggregate by all 11 rollup dimensions simultaneously
    4. Every 50 batches:
        - Merge with accumulator (incremental folding)
        - Keep memory bounded (~10GB peak)
    5. Write final rollups as Arrow IPC (LZ4 compression)
```

**Phase 2: DuckDB Fallback (4-5 min)**

```python
1. Read all CSVs into DuckDB table
2. Add time dimensions (day, hour, minute, week)
3. Sort by (week, country, type) - physical layout optimization
4. Write to fallback.duckdb (~2.5GB)
```

**Why this is fast:**
- Single pass through data (read once)
- Parallel processing (8 threads for CSV reading)
- Incremental folding prevents memory explosion
- Arrow IPC = zero-copy reads during que

[README truncated for size]

## Detected evidence (automated analysis)

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

## Codebase structure (from repository index)

### Files (87 of 87)

```
__MACOSX/._baseline
__MACOSX/._data-lite
__MACOSX/baseline/.___init__.py
__MACOSX/baseline/._.DS_Store
__MACOSX/baseline/._assembler.py
__MACOSX/baseline/._inputs.py
__MACOSX/baseline/._main.py
__MACOSX/baseline/._README.md
__MACOSX/baseline/._requirements.txt
.gitignore
baseline/__init__.py
baseline/assembler.py
baseline/inputs.py
baseline/main.py
baseline/README.md
baseline/requirements.txt
BUILD_OPTIMIZATION_ANALYSIS.md
DATA_SETUP.md
docs/ARCHITECTURE.md
docs/ASSUMPTION_VALIDATION.md
docs/CRITICAL_TEST_RESULTS.md
docs/FALLBACK_DECISION.md
docs/IMPLEMENTATION_CHECKLIST.md
docs/IMPLEMENTATION_PLAN.md
docs/OPTIMIZATION_STRATEGY.md
docs/PRELIMINARY_TEST_RESULTS.md
docs/REALITY_CHECK_SUB_1S.md
FINAL_TEST.md
JUDGE_SETUP.md
prepare.py
PROGRESS.md
queries_test/test_q01_us_impressions.json
queries_test/test_q02_hourly_avg_bid.json
queries_test/test_q03_advertiser_purchases.json
queries_test/test_q04_asian_countries.json
queries_test/test_q05_publisher_performance.json
queries_test/test_q06_weekly_advertiser.json
queries_test/test_q07_european_hours.json
queries_test/test_q08_bid_stats.json
queries_test/test_q09_christmas_minutes.json
queries_test/test_q10_canada_january.json
queries/example_queries.json
queries/inputs.py
queries/q1_daily_revenue.json
queries/q2_publisher_japan.json
queries/q3_avg_purchase_country.json
queries/q4_advertiser_event_counts.json
queries/q5_minutely_spend.json
queries/README.md
QUICK_START.md
README.md
requirements.txt
rollups_test/country_type.arrow
run.py
setup_venvs.sh
src/__init__.py
src/analysis/compression_bench.py
src/analysis/correctness_test.py
src/analysis/data_distribution.py
src/analysis/mem_calc.py
src/analysis/memory_reality_check.py
src/analysis/predicate_stats.py
src/baseline_runner.py
src/core/__init__.py
src/core/data_loader.py
src/core/fallback_executor.py
src/core/query_executor.py
src/core/query_router.py
src/core/rollup_builder.py
src/core/rollup_loader.py
src/core/storage.py
src/query_parser.py
src/tests/phase0/test_01_arrow_ipc_perf.py
src/tests/phase0/test_02_null_handling.py
src/tests/phase0/test_03_large_rollup_perf.py
src/tests/phase0/test_04_partitioned_rollup.py
src/tests/phase0/test_05_multi_rollup_routing.py
STATUS.md
test_coalesce.py
test_e2e_queries.py
test_join_behavior.py
test_loader.py
test_merge.py
test_q16_baseline.py
test_router.py
validate_setup.py
VENV_GUIDE.md
```

### Dependencies

- baseline/requirements.txt: duckdb@>=1.1.1, pandas@>=2.2.0
- requirements.txt: duckdb@>=0.9.0, memory_profiler@>=0.61.0, numpy@>=1.24.0, pandas@>=2.0.0, polars@>=0.19.0, psutil@>=5.9.0, pyarrow@>=14.0.0, pytest@>=7.4.0, pytest-benchmark@>=4.0.0, python-dateutil@>=2.8.2, tqdm@>=4.66.0, tzlocal@>=5.0.0

### Recent commits (newest first)

- Clearer usage guide
- Small tweaks again
- Read me update again
- Documentation updates
- DuckDB fallback
- Fix order problem
- Working solution
- Phase 2
- Phase 1
- Tests to validate approach
- Tons of stuff (some of this should probably be in gitignore but idc)
- init commit

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

### FINAL_TEST.md

```markdown
# Final Test Summary

## What We Just Fixed:

1. ✅ **Q5 Routing**: Now correctly routes to `minute_type` rollup
2. ✅ **Date Format Conversion**: Converts calendar dates (2024-06-01) to day-of-year format (2024-153)
3. ✅ **Q5 Execution**: Successfully executes and returns 1,440 results
4. 🔄 **Q2 Rollup Added**: `day_publisher_country_type` for Q2 support

## Current Results (4/5 working):

- **Q1**: 1.6ms ✅ (366 rows)
- **Q2**: Failed ❌ (no rollup yet)
- **Q3**: 0.6ms ✅ (12 rows)
- **Q4**: 1.8ms ✅ (6,616 rows)
- **Q5**: 8.0ms ✅ (1,440 rows)

**Total: 12ms for 4 queries (83× under 1s budget!)**

## Next Step:

Rebuild rollups to include the new `day_publisher_country_type` rollup for Q2:

```bash
python prepare.py --data-dir data --rollup-dir rollups
```

This will take ~11-12 minutes but will add the missing rollup.

Then test all 5 queries:

```bash
python run.py --rollup-dir rollups --out-dir results
```

**Expected final results: 5/5 queries working, <20ms total!**

```

### VENV_GUIDE.md

```markdown
# Virtual Environment Guide

This project uses **two separate virtual environments**:

## 1. Baseline Environment (`baseline/venv`)
- **Location**: `baseline/venv/`
- **Purpose**: Run the DuckDB baseline solution
- **Dependencies**: Minimal (just DuckDB and pandas)

## 2. Main Project Environment (`venv`)
- **Location**: `venv/` (project root)
- **Purpose**: Your optimized solution development
- **Dependencies**: Full toolkit (DuckDB, pandas, polars, pyarrow, performance monitoring, etc.)

---

## Quick Setup (Automated)

Run the setup script from the project root:

```bash
./setup_venvs.sh
```

This will create and configure both virtual environments automatically.

---

## Manual Setup

If you prefer to set up manually:

### Baseline Environment
```bash
cd baseline
python3 -m venv venv
source venv/bin/activate
pip install --upgrade pip
pip install -r requirements.txt
deactivate
cd ..
```

### Main Project Environment
```bash
python3 -m venv venv
source venv/bin/activate
pip install --upgrade pip
pip install -r requirements.txt
deactivate
```

---

## Usage

### Running the Baseline
```bash
cd baseline
source venv/bin/activate
python main.py
deactivate
```

### Working on Your Optimized Solution
```bash
# From project root
source venv/bin/activate
python src/baseline_runner.py --data-dir data/data
# ... do your development work
deactivate
```

### Quick Tips
- Always activate the appropriate venv before running code
- Use `deactivate` to exit a virtual environment
- Your shell prompt will show `(venv)` when a venv is active
- The two venvs are independent - changes in one don't affect the other

---

## Why Separate Virtual Environments?

1. **Isolation**: Baseline stays clean and lightweight
2. **Dependency Management**: Avoid conflicts between baseline and your solution
3. **Fair Comparison**: Baseline runs in its original environment
4. **Flexibility**: You can experiment with different libraries without affecting the baseline

```

### requirements.txt

```
# Core dependencies
duckdb>=0.9.0
pandas>=2.0.0
numpy>=1.24.0

# Data processing
pyarrow>=14.0.0
polars>=0.19.0

# Performance monitoring
psutil>=5.9.0
memory_profiler>=0.61.0

# Testing
pytest>=7.4.0
pytest-benchmark>=4.0.0

# Utilities
python-dateutil>=2.8.2
tqdm>=4.66.0
tzlocal>=5.0.0  # For timezone auto-detection

```

### baseline/requirements.txt

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

### baseline/main.py

```python
#!/usr/bin/env python3
"""
DuckDB Baseline Benchmark Demo
------------------------------

Reads data from a given folder (CSV or Parquet),
adds derived day/minute columns,
executes JSON queries, and reports timings.

Usage:
  python main.py --data-dir ./data --out-dir ./out
"""

import duckdb
import time
from pathlib import Path
import csv
import argparse
from assembler import assemble_sql
from inputs import queries
# from judges import queries


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


# -------------------
# Load Data
# -------------------
def load_data(con, data_dir: Path):
    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}")


# -------------------
# Run Queries
# -------------------
def run(queries, data_dir: Path, out_dir: Path):
    # Ensure directories exist
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    out_dir.mkdir(parents=True, exist_ok=True)

    con = duckdb.connect(DB_PATH)
    load_data(con, data_dir)

    out_dir.mkdir(parents=True, exist_ok=True)
    results = []
    for i, q in enumerate(queries, 1):
        sql = assemble_sql(q)
        print(f"\n🟦 Query {i}:\n{q}\n")
        t0 = time.time()
        res = con.execute(sql)
        cols = [d[0] for d in res.description]
        rows = res.fetchall()
        dt = time.time() - t0

        print(f"✅ Rows: {len(rows)} | Time: {dt:.3f}s")

        out_path = out_dir / f"q{i}.csv"
        with out_path.open("w", newline="") as f:
            w = csv.writer(f)
            w.writerow(cols)
            w.writerows(rows)

        results.append({"query": i, "rows": len(rows), "time": dt})
    con.close()

    print("\nSummary:")
    for r in results:
        print(f"Q{r['query']}: {r['time']:.3f}s ({r['rows']} rows)")
    print(f"Total time: {sum(r['time'] for r in results):.3f}s")


# -------------------
# Main Entry Point
# -------------------
if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="DuckDB Baseline Benchmark Demo — runs benchmark queries on input CSV data."
    )
    parser.add_argument(
        "--data-dir",
        type=Path,
        required=True,
        help="The folder where the input CSV is provided"
    )
    parser.add_argument(
        "--out-dir",
        type=Path,
        required=True,
        help="Where to output query results-full"
    )

    args = parser.parse_args()
    run(queries, args.data_dir, args.out_dir)

```

### test_router.py

```python
#!/usr/bin/env python3
"""Test router to see why Q2 isn't matching"""

from src.core.query_router import QueryRouter
import json

router = QueryRouter()

# Load Q2
with open('queries/q2_publisher_japan.json') as f:
    q2 = json.load(f)

print('='*60)
print('Q2 Query:')
print('='*60)
print(json.dumps(q2, indent=2))
print()

print('='*60)
print('Router Catalog:')
print('='*60)
for name, (dims, rows) in router.catalog.items():
    print(f"  {name}: {dims} ({rows:,} rows)")
print()

print('='*60)
print('Routing Q2:')
print('='*60)
rollup_name, pattern = router.route_query(q2)
print()
print(f'✅ Result: {rollup_name}')
print(f'   Pattern: {pattern}')

```

### test_join_behavior.py

```python
import polars as pl

# Test outer join behavior with composite keys
batch1 = pl.DataFrame({
    'advertiser_id': [1, 2, 3],
    'type': ['serve', 'serve', 'click'],
    'row_count': [100, 200, 300]
})

batch2 = pl.DataFrame({
    'advertiser_id': [2, 3, 4],
    'type': ['serve', 'click', 'impression'],
    'row_count': [50, 75, 125]
})

print("Batch 1:")
print(batch1)
print("\nBatch 2:")
print(batch2)

# Join on BOTH keys
keys = ['advertiser_id', 'type']
merged = batch1.join(batch2, on=keys, how="full", suffix="_batch")

print("\nAfter FULL join on ['advertiser_id', 'type']:")
print(merged)
print(f"\nRows: {len(merged)}")

# The question: Does row (4, 'impression') from batch2 appear with NULL batch1 columns?
# Or does it appear with NULL keys?

# Let's check if any keys are NULL
print("\nRows with NULL advertiser_id:")
print(merged.filter(pl.col('advertiser_id').is_null()))

print("\nRows with NULL type:")
print(merged.filter(pl.col('type').is_null()))

```

### test_merge.py

```python
#!/usr/bin/env python3
"""Test the merge accumulator logic"""

import polars as pl

# Simulate two batches with overlapping keys
batch1 = pl.DataFrame({
    'advertiser_id': [1, 2, 3],
    'type': ['serve', 'serve', 'impression'],
    'row_count': [100, 200, 300],
})

batch2 = pl.DataFrame({
    'advertiser_id': [2, 3, 4],  # 2 and 3 overlap with batch1
    'type': ['serve', 'impression', 'serve'],
    'row_count': [50, 75, 125],
})

print("Batch 1:")
print(batch1)
print("\nBatch 2:")
print(batch2)

# Merge using outer join
keys = ['advertiser_id', 'type']
merged = batch1.join(batch2, on=keys, how="outer", suffix="_batch")
print("\nAfter outer join:")
print(merged)

# Sum the counts
result = merged.with_columns([
    (pl.col('row_count').fill_null(0) + pl.col('row_count_batch').fill_null(0)).alias('row_count')
]).select(keys + ['row_count'])

print("\nFinal result:")
print(result)
print(f"\nTotal rows: {result.height}")

# Expected:
# (1, serve): 100
# (2, serve): 200 + 50 = 250
# (3, impression): 300 + 75 = 375
# (4, serve): 125
# Total: 4 rows

```

### test_coalesce.py

```python
import polars as pl

# Simulate the exact merge_accumulator logic
batch1 = pl.DataFrame({
    'advertiser_id': [1, 2, 3],
    'type': ['serve', 'serve', 'click'],
    'row_count': [100, 200, 300]
})

batch2 = pl.DataFrame({
    'advertiser_id': [2, 3, 4],
    'type': ['serve', 'click', 'impression'],
    'row_count': [50, 75, 125]
})

keys = ['advertiser_id', 'type']
agg_cols = ['row_count']

# Outer join
merged = batch1.join(batch2, on=keys, how="full", suffix="_batch")
print("After outer join:")
print(merged)
print()

# Build merge expressions (with coalesce)
merge_exprs = []
for key in keys:
    key_batch = f"{key}_batch"
    if key_batch in merged.columns:
        print(f"✅ Coalescing {key} with {key_batch}")
        merge_exprs.append(
            pl.coalesce([pl.col(key), pl.col(key_batch)]).alias(key)
        )

for col in agg_cols:
    col_batch = f"{col}_batch"
    merge_exprs.append(
        (pl.col(col).fill_null(0) + pl.col(col_batch).fill_null(0)).alias(col)
    )

# Apply merges
result = merged.with_columns(merge_exprs).select(keys + agg_cols)

print("\n✅ Final result after coalesce:")
print(result)
print(f"\nRows: {len(result)}")
print(f"Rows with NULL advertiser_id: {result.filter(pl.col('advertiser_id').is_null()).height}")
print(f"Rows with NULL type: {result.filter(pl.col('type').is_null()).height}")

```

### setup_venvs.sh

```shell
#!/bin/bash
# Setup script for CalHacks AppLovin Challenge
# Creates separate virtual environments for baseline and optimized solution

set -e  # Exit on error

echo "=========================================="
echo "Setting up virtual environments"
echo "=========================================="
echo ""

# Get the project root directory
PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$PROJECT_ROOT"

# Setup baseline venv
echo "📦 Setting up baseline virtual environment..."
cd baseline
if [ -d "venv" ]; then
    echo "  ⚠️  baseline/venv already exists, skipping creation"
else
    python3 -m venv venv
    echo "  ✅ Created baseline/venv"
fi

echo "  📥 Installing baseline dependencies..."
source venv/bin/activate
pip install --upgrade pip
pip install -r requirements.txt
deactivate
echo "  ✅ Baseline dependencies installed"
echo ""

# Setup main project venv
cd "$PROJECT_ROOT"
echo "📦 Setting up main project virtual environment..."
if [ -d "venv" ]; then
    echo "  ⚠️  venv already exists, skipping creation"
else
    python3 -m venv venv
    echo "  ✅ Created venv"
fi

echo "  📥 Installing main project dependencies..."
source venv/bin/activate
pip install --upgrade pip
pip install -r requirements.txt
deactivate
echo "  ✅ Main project dependencies installed"
echo ""

echo "=========================================="
echo "✅ Setup complete!"
echo "=========================================="
echo ""
echo "To activate virtual environments:"
echo ""
echo "  For baseline:"
echo "    cd baseline && source venv/bin/activate"
echo ""
echo "  For main project:"
echo "    source venv/bin/activate"
echo ""
echo "To deactivate:"
echo "    deactivate"
echo ""

```

[51 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]