# Project export: Duck Duck Goose

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: A fast, intelligent query engine built on DuckDB.
- Devpost: https://devpost.com/software/duck-duck-goose-jelxr9
- GitHub: https://github.com/KevinL10/applovin-query
- Video: https://www.youtube.com/embed/_QnZ6bNqfr4?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Result: winner (AppLovin: Query Planner Challenge)
- Team: 1 GitHub contributor(s) — KevinL10 (26 commits)

## Devpost submission (written by the team)

No Devpost description available.

## README (from the GitHub repository)

# Calhacks 2025 – Query Challenge

A submission to AppLovin's query planner challenge at Cal Hacks 2025. See an example run [here](https://asciinema.org/a/50woSW9207CSufW3cdFxAjj89).

**Summary**: we achieve a **~1300x speedup** compared to the baseline implementation (~45s to ~30ms) on an M3 Pro with 18 GB RAM. Preprocessing takes 300.5 seconds (~5 minutes).

![baseline](/assets/baseline-comparison.png)

**Figure 1**: A comparison of the baseline implementation to our submission (`v1`). **Note the log scale**: v1 is over three orders of magnitude faster.

On our custom set of 25 additional validation queries, we achieve a ~120x speedup (~205s to ~1.72s).

![baseline](/assets/validation.png)

# Setup

This project uses `uv` to manage Python packages, which can be installed [here](https://docs.astral.sh/uv/).

To evaluate a set of queries (provided as a JSON file):

```sh
PYTHONPATH=. uv run benchmark/run.py \
            --data-dir data/data \
            --out-dir results/v1-baseline-full \
            --solution=v1 \
            --queries queries/baseline.json \
            --verbose \
            --num-runs=1
```

You can pass the directory to the expected results for correctness checks. For example:

```sh
PYTHONPATH=. uv run benchmark/run.py \
            --data-dir data/data \
            --out-dir results/v1-baseline-full \
            --solution=v1 \
            --queries queries/baseline.json   \
            --expected-dir results/results-full \
            --verbose  \
            --num-runs=1
```

# Design

We use four main techniques to optimize query performance.

## Efficient on-disk storage layout

Instead of storing the data in CSV files and re-reading via `read_csv` on each query, we can pre-process the table into parquet files for significantly more efficient IO. In particular, we allow each thread to store its own parquet shard.

![parquet layout](/assets/parquet-layout.png)

With the default `ROW_GROUP_SIZE = 122880`, each of the 10 shards (for each thread) has around ~160 row groups.

```
SELECT num_row_groups FROM parquet_file_metadata('tmp/events.parquet/data_0.parquet');
> 163
```

This means that there's a fair chunk of overhead spent on reading per-group metadata. We can afford to store much more data in memory - given we have ~10 threads, we should aim for `num_row_groups` around ~20.

![row group comparisons](/assets/row_group_comparisons.png)

Using `ROW_GROUP_SIZE = 1M`, we see significant improvements on the majority of queries (22x improvement on baseline from optimized layout alone).

## Materialized views

Materialized views let us pre-aggregate common query patterns and can easily improve query times 10-100x (note: DuckDB doesn't have materialized views, so we just create new tables).

Although we could create arbitrarily many pre-aggregated tables to account for various query patterns, I chose a small set of general MVs (see [`src/mv.py`](src/mv.py) for the definitions). In particular, I have a set of "slow/large" MVs and a corresponding set of faster/smaller ones for common use cases.

![tables in duckdb](/assets/tables.png)

For each of the ID types (`advertiser_id`, `publisher_id`, `user_id`), I've constructed a slow/large table grouped by `(type, day, country, <id>)`. This answers common questions like:

- How much in total bids did a particular advertiser make on a certain day?
- How many impressions were served to users in JP?

Even though these tables are fairly large (`advertiser_full` is ~10M rows), they are still significantly smaller than `events` with 245M rows.

On the faster side, I have a set of small MVs (e.g. `mv_day_fast`, `mv_time_fast` and `mv_auction_fast`) that have ~10K rows and act as a quick cache if a query can be written to hit them.

_Note_: this is a tradeoff between pre-processing time and query performance. I've only set up a handful of common MVs, but have tried to make it as easy as possible to add new MVs. All it takes is updating the registry in [`src/mv.py`](src/mv.py) – the corresponding SQL will be automatically generated.

## Intelligent query planning & rewriting
![query planning](/assets/query-planning.png)


The query planner (`src/planner.py`) is responsible for rewriting the query to hit a particular materialized view. This happens in two steps:

1. Find the set of feasible MVs that support the query
2. Score each feasible MV based on estimated cost heuristics


### Feasibility Check

We can only substitute a table for an MV if:

- the group-by of the MV fully covers the group-by of the query and any filters
- all select columns/aggregates can be derived from the MV

For the last point, we make a micro-optimization to derive `AVG` from `SUM` and `COUNT`, which lets us cut down on the number of bytes we need to save.

### Intelligent scoring

The key idea here is that the cost of the query is approximately linear to the # of filtered rows and the # of rows that need to be rolled-up. Luckily, we can pre-compute these statistics during the pre-loading phase.

In the preload, we batch together a number of `COUNT` and `COUNT(DISTINCT)` calls to cache:

1. the number of rows in the MV
2. the number of distinct values per column
3. the top-k most common values and their count per column (to improve estimates)

Then, estimating the number of filtered rows looks like:

```python
num_filtered_rows = num_rows
for f in filters:
  if f.value in top_k:
    num_filtered_rows *= top_k[f.col][f.value] / distinct[f.col][f.value]
  else:
    num_filtered_rows *= 1 / distinct[f.col][f.value]
```

Next, there's additional cost if the MV has higher granularity than the query. For example, if the query groups by `(type, day, hour)` but the closest MV groups by `(type, day, hour, minute)`, DuckDB will need to rollup the values in the minute. We can approximate this rollup cost as:

```python
num_rolled_up = 1
for col in mv.group_by:
  if col not in query.group_by:
    num_rolled_up *= distinct[col]
```

Finally, I computed the total cost as a linear combination of the two. These are all rough hyperparameters and require more tuning in the future.

### Example

This is taken directly from the `v1` solution when trying to optimize the following query:

Query:

```
{'select': ['day', {'SUM': 'bid_price'}], 'from': 'events', 'where': [{'col': 'type', 'op': 'eq', 'val': 'impression'}], 'group_by': ['day']}
```

Logs from `verbose=True`:

```
Considering mv_auction_id_full: cost 80316988.0
Considering mv_advertiser_id_full: cost 7132445.0
Considering mv_publisher_id_full: cost 4868461.0
Considering mv_user_id_full: cost 72091914.0
Considering mv_day_fast: cost 366.0    <--- planner rewrites query to use `mv_day_fast`
Considering mv_time: cost 6252631.0
```

## Efficient caching and warmup

The last significant optimization was warming up the [DuckDB buffer manager](https://duckdb.org/2024/07/09/memory-management) and OS cache. This lets subsequent queries hit warm pages and avoid direct disk reads.

To do this, I ran `ANALYZE mv_...` for each of the materialized views. Any similar operation that scans all of the data (e.g. `COUNT(*)`) should have the same effect.

I timed this on my validation suite; the variation makes it hard to tell, but this step offers some speedup and doesn't hurt subsequent performance in any way.

**Miscellaneous**: using enums for types; re-computing AVG from SUM/CNT; ZSTD compression.

# Implementation

## Benchmarking

We implemented a generic `Solution` class (`benchmark/solution.py`), which both the `baseline` and `v1` solutions inherit from.

Additionally, we provide options to:

- benchmark over multiple runs (`--num-runs`)
- plot stats (`benchmark/plot_stats.py`)
- compare against a ground-truth directory of results (`--expected-dir`)

## Materialized Views

To avoid duplicated SQL and code, we created a custom `MaterializedView` class that can automatically generate the corresponding `CREATE TABLE` SQL statement. This lets us manage the set of MVs solely from Python

[README truncated for size]

## Detected evidence (automated analysis)

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

## Codebase structure (from repository index)

### Files (23 of 23)

```
.gitignore
.python-version
benchmark/__init__.py
benchmark/baseline.py
benchmark/plot_stats.py
benchmark/run.py
benchmark/solution.py
duck.txt
duck2.txt
pyproject.toml
queries/baseline.json
queries/cache_warmup.json
queries/test.json
queries/test2.json
queries/validation.json
README.md
src/assembler.py
src/baseline_assembler.py
src/mv.py
src/planner.py
src/queries.py
src/v1.py
uv.lock
```

### Dependencies

- pyproject.toml: duckdb@>=1.4.1, matplotlib@>=3.10.7, pytz@>=2025.2, tqdm@>=4.67.1

### Recent commits (newest first)

- feat: query planning
- feat: add parquet layout
- feat: add example run
- fix: misc protection
- fix: typo
- feat: add tables image
- feat: graphs
- feat: readme
- feat: materialized views
- feat: storage layout
- feat: readme
- feat: better cache warmup
- fix: timings
- feat: add cache warmup
- feat: misc
- feat: qol
- feat: improve cache computation; validation - 2.59
- feat: fast and slow buckets
- feat: implement basic selectivity
- feat: move to planner class

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

### pyproject.toml

```
[project]
name = "applovin-query"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
    "duckdb>=1.4.1",
    "matplotlib>=3.10.7",
    "pytz>=2025.2",
    "tqdm>=4.67.1",
]

```

### benchmark/solution.py

```python
from pathlib import Path


class Solution:
    """An abstract solution to the query challenge."""

    def __init__(self, db_path: Path):
        pass

    def load_data(self, data_dir: Path) -> None:
        pass

    def run(
        self, queries: list[dict], out_dir: Path, verbose: bool = False
    ) -> list[float]:
        """Run the given queries and write the results to the given directory.

        Returns a list of timings for each query."""
        pass

    def close(self) -> None:
        """Close any resources (e.g. duckdb connection)."""
        pass

```

### src/queries.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"}],
    },
]

```

### src/baseline_assembler.py

```python
# NOTE: this is copied directly from the provided files.


# 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)

```

### benchmark/plot_stats.py

```python
import argparse
import json
from pathlib import Path
from statistics import mean
import numpy as np
import matplotlib.pyplot as plt


def natkey(qname: str):
    try:
        return (
            int(qname[1:]) if qname.startswith("q") and qname[1:].isdigit() else 10**9
        )
    except Exception:
        return 10**9


def main():
    parser = argparse.ArgumentParser(
        description="Plot per-query timings (seconds) from one or more JSON files."
    )
    parser.add_argument(
        "--paths", nargs="+", type=Path, help="Paths to timings JSON files"
    )
    args = parser.parse_args()

    stats = []
    for p in args.paths:
        with p.open("r") as f:
            stats.append(json.load(f))

    labels = [stat["solution"] for stat in stats]
    num_runs = stats[0]["num_runs"]
    datasets = [
        {
            k: v["timings"]
            for k, v in data.items()
            if isinstance(v, dict) and "timings" in v and k.startswith("q")
        }
        for data in stats
    ]

    queries = []
    for i in range(1, 1000):
        if f"q{i}" in datasets[0]:
            queries.append(f"q{i}")
        else:
            break

    avgs, err_low, err_up = [], [], []
    for d in datasets:
        a, el, eu = [], [], []
        for q in queries:
            if q in d:
                t = d[q]
                m, mn, mx = mean(t), min(t), max(t)
                a.append(m)
                el.append(m - mn)
                eu.append(mx - m)
            else:
                a.append(np.nan)
                el.append(0.0)
                eu.append(0.0)
        avgs.append(np.array(a, dtype=float))
        err_low.append(np.array(el, dtype=float))
        err_up.append(np.array(eu, dtype=float))

        print(avgs)

    n_files, n_q = len(datasets), len(queries)
    idx = np.arange(n_q)
    group_w = 0.85
    bar_w = group_w / n_files

    fig = plt.figure()
    for i in range(n_files):
        pos = idx - group_w / 2 + (i + 0.5) * bar_w
        y = avgs[i]
        el = err_low[i].copy()
        eu = err_up[i].copy()
        mask = np.isnan(y)
        el[mask] = 0.0
        eu[mask] = 0.0
        plt.bar(
            pos, y, width=bar_w, yerr=np.vstack([el, eu]), capsize=3, label=labels[i]
        )

    plt.xticks(idx, queries)
    plt.tick_params(axis="x", labelbottom=False)
    plt.xlabel("Query")
    plt.ylabel("Time (s)")
    plt.yscale("log")
    plt.title(f"Average time per query (num_runs={num_runs})")
    plt.legend()
    plt.tight_layout()
    plt.show()


if __name__ == "__main__":
    main()

```

### src/mv.py

```python
from dataclasses import dataclass, field
from typing import Tuple, Set


def metric_col_name(op: str, col: str | None) -> str:
    """Calculate the column name for a metric."""
    op = op.lower()
    if op == "count" and (col is None or col == "*"):
        return "count_rows"
    base = (col or "rows").replace(".", "_")
    return f"{op}_{base}"


@dataclass(frozen=True)
class Agg:
    op: str
    column: str | None = None  # we need to represent count(*) as a special case


@dataclass
class MaterializedView:
    name: str
    group_by: Tuple[str, ...]
    aggs: Set[Agg]

    num_rows: int | None = None

    # Mapping from column to number of distinct values
    num_distinct: dict[str, int] = field(default_factory=dict)

    col_to_topk: dict[str, dict[str, int]] = field(default_factory=dict)

    @property
    def has_stats(self) -> bool:
        return self.num_distinct and self.col_to_topk and self.num_rows

    def generate_create_sql(self) -> str:
        """Generate CREATE TABLE SQL for this materialized view."""
        select_parts = list(self.group_by)
        for agg in self.aggs:
            if agg.column is None:
                select_parts.append("COUNT(*) AS count_rows")
            else:
                select_parts.append(
                    f"{agg.op}({agg.column}) AS {metric_col_name(agg.op, agg.column)}"
                )

        # sql = f"""CREATE OR REPLACE TABLE {self.name} AS
        sql = f"""CREATE TABLE IF NOT EXISTS {self.name} AS
SELECT
{',\n'.join(select_parts)}
FROM events
GROUP BY {', '.join(str(i + 1) for i in range(len(self.group_by)))};"""

        return sql


MV_REGISTRY: list[MaterializedView] = (
    [
        MaterializedView(
            name=f"mv_{col}_full",
            group_by=("type", "day", "country", col),
            aggs={
                Agg("SUM", "bid_price"),
                Agg("SUM", "total_price"),
                Agg("COUNT", None),
                Agg("COUNT", "bid_price"),
                Agg("COUNT", "total_price"),
            },
        )
        # Note: auction_id is fairly distinct, so creating a separate
        # auction_id table effectively replicates all ~200M rows of events.
        for col in ["auction_id", "advertiser_id", "publisher_id", "user_id"]
        # for col in ["advertiser_id", "publisher_id", "user_id"]
    ]
    + [
        MaterializedView(
            name="mv_day_fast",
            group_by=("type", "day"),
            aggs={
                Agg("SUM", "bid_price"),
                Agg("SUM", "total_price"),
                Agg("COUNT", None),
                Agg("COUNT", "bid_price"),
                Agg("COUNT", "total_price"),
            },
        ),
        MaterializedView(
            name="mv_time",
            group_by=("type", "country", "week", "day", "hour", "minute"),
            aggs={
                Agg("SUM", "bid_price"),
                Agg("SUM", "total_price"),
                Agg("COUNT", None),
                Agg("COUNT", "bid_price"),
                Agg("COUNT", "total_price"),
            },
        ),
        MaterializedView(
            name="mv_time_fast",
            group_by=("type", "day", "hour", "minute"),
            aggs={
                Agg("SUM", "bid_price"),
                Agg("SUM", "total_price"),
                Agg("COUNT", None),
                Agg("COUNT", "bid_price"),
                Agg("COUNT", "total_price"),
            },
        ),
    ]
    + [
        MaterializedView(
            name=f"mv_{col}_fast",
            group_by=("type", col),
            aggs={
                Agg("SUM", "bid_price"),
                Agg("SUM", "total_price"),
                Agg("COUNT", None),
                Agg("COUNT", "bid_price"),
                Agg("COUNT", "total_price"),
            },
        )
        for col in ("advertiser_id", "user_id", "publisher_id")
    ]
)

# MV_REGISTRY = []

if __name__ == "__main__":
    for mv in MV_REGISTRY:
        print(mv.generate_create_sql())

```

### benchmark/baseline.py

```python
from benchmark.solution import Solution
from pathlib import Path
import duckdb
import time
import csv
import json
from src.baseline_assembler import assemble_sql


class BaselineSolution(Solution):
    """A baseline solution to the query challenge."""

    def __init__(self, db_path: Path):
        self.db_path = db_path
        self.table_name = "events"
        self.con = duckdb.connect(self.db_path)

    def load_data(self, data_dir: Path) -> None:
        """Load the data from the given directory.

        Note: we're allowed to take arbitrary time in this step."""
        csv_files = list(data_dir.glob("events_part_*.csv"))

        if csv_files:
            print(f"🟩 Loading {len(csv_files)} CSV parts from {data_dir} ...")
            self.con.execute(f"""
                CREATE OR REPLACE VIEW {self.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 run(
        self, queries: list[dict], out_dir: Path, verbose: bool = False
    ) -> list[float]:
        """Run the given queries and write the results to the given directory."""

        results = []
        for i, q in enumerate(queries, 1):
            sql = assemble_sql(q)
            if verbose:
                print(f"\n🟦 Query {i}:\n{sql}\n")

            t0 = time.time()
            res = self.con.execute(sql)
            cols = [d[0] for d in res.description]
            rows = res.fetchall()
            dt = time.time() - t0

            if verbose:
                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})

        if verbose:
            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")

        return [r["time"] for r in results]

    def close(self) -> None:
        self.con.close()


if __name__ == "__main__":
    solution = BaselineSolution()

```

### src/assembler.py

```python
"""Materialized-view aware assembler."""

from src.mv import MV_REGISTRY, MaterializedView, metric_col_name
from src.queries import queries


def assemble_sql_for_mv(q: dict, mv: MaterializedView) -> str:
    """Assembles the SQL, but using the given materialized view.

    The nice thing is that we only need to update the select clause."""

    select_sql = _select_over_mv(q.get("select", []), mv)
    # We already checked that the materialized view is usable, so we can just use it here.
    from_tbl = mv.name

    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_sql} FROM {from_tbl} {where} {group_by} {order_by}"
    if q.get("limit"):
        sql += f" LIMIT {q['limit']}"

    return sql.strip()


def _compute_agg_alias_expr(op: str, col: str | None) -> tuple[str, str]:
    """Compute the alias and new expression for an aggregate function over the MV.

    Rules:
      SUM(x)      -> SUM(sum_x)           AS sum_x
      COUNT(*)    -> SUM(count_rows)      AS count_rows
      COUNT(x)    -> SUM(count_x)         AS count_x
      AVG(x)      -> SUM(sum_x)::DOUBLE / NULLIF(SUM(count_x), 0) AS avg_x
      MIN(x)      -> MIN(min_x)           AS min_x
      MAX(x)      -> MAX(max_x)           AS max_x
    """
    op_u = (op or "").upper()
    if op_u == "AVG":
        sum_col = metric_col_name("sum", col)
        cnt_col = metric_col_name("count", col)
        alias = metric_col_name("avg", col)
        expr = f"SUM({sum_col})::DOUBLE / NULLIF(SUM({cnt_col}), 0)"
        return expr, alias

    if op_u in {"SUM", "COUNT"}:
        alias = metric_col_name(op_u, None if (op_u == "COUNT" and col == "*") else col)
        expr = f"SUM({alias})"
        # To make things cleaner we keep SUM(sum_bid_price) as sum_bid_price.
        # TODO: check with slack to see if the column names have to also match.
        return expr, alias

    if op_u in {"MIN", "MAX"}:
        alias = metric_col_name(op_u, col)
        expr = f"{op_u}({alias})"
        return expr, alias

    raise ValueError(f"Unsupported aggregate: {op_u}({col})")


def _select_over_mv(select: list, mv: MaterializedView) -> str:
    """Convert the select list based on the materialized view."""
    parts: list[str] = []

    for item in select:
        if isinstance(item, str):
            parts.append(item)
        elif isinstance(item, dict):
            op, col = next(iter(item.items()))
            if op.upper() == "COUNT" and col == "*":
                col = None

            expr, alias = _compute_agg_alias_expr(op, col)
            parts.append(f"{expr} AS {alias}")
        else:
            raise ValueError(f"Bad select item: {item}")
    return ", ".join(parts) if parts else "*"


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}'")
        elif 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 _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 = []
    for o in order_by:
        dir_ = o.get("dir", "asc").upper()
        col = o["col"]

        if "(" in col and ")" in col:
            op, col = col.split("(")[0], col.split("(")[1].split(")")[0]
            expr, _ = _compute_agg_alias_expr(op, col)
            # In order by we don't care about the alias; we just need to order by
            # teh expression
            key = f"{expr}"
        else:
            key = col

        parts.append(f"{key} {dir_}")

    return "ORDER BY " + ", ".join(parts)


if __name__ == "__main__":
    view = MV_REGISTRY[0]
    query = queries[0]
    print(assemble_sql_for_mv(query, view))
    print(assemble_sql_for_mv(queries[2], view))

```

### benchmark/run.py

```python
"""Benchmark a particular solution against the baseline."""

import argparse
import json
from pathlib import Path
import csv
import time

from benchmark.baseline import BaselineSolution
from benchmark.solution import Solution
from src.v1 import V1Solution
# from src.queries import queries

SOLUTION_CLASSES = {
    "baseline": BaselineSolution,
    "v1": V1Solution,
}


def normalize_row(row):
    """Normalizes floats to nearest 6 decimal places for comparisons."""
    result = []
    for cell in row:
        try:
            cell = float(cell)
            result.append(round(cell, 6))
        except ValueError:
            result.append(cell)
    return tuple(result)


def run_solution(
    solution_type: str,
    queries_path: Path,
    data_dir: Path,
    out_dir: Path,
    expected_dir: Path | None = None,
    num_runs: int = 1,
    verbose: bool = False,
) -> None:
    """Run the given solution for a given number of times and compare against expected results.

    Expected results in the format `q<i>.csv` for each query `i`."""

    with queries_path.open() as f:
        queries = json.load(f)

    timings = []
    db_path = Path(f"tmp/{data_dir.name}.duckdb")
    db_path.parent.mkdir(parents=True, exist_ok=True)
    solution = SOLUTION_CLASSES[solution_type](db_path=db_path)
    print(f"Using database at {db_path}")

    out_dir.mkdir(parents=True, exist_ok=True)
    t0 = time.perf_counter()
    solution.load_data(data_dir)
    print("load time: ", time.perf_counter() - t0)

    for round in range(num_runs):
        print(f"Running round {round + 1} of {num_runs}...")
        timings.append(solution.run(queries, out_dir, verbose))

    if expected_dir:
        for round in range(1, len(queries) + 1):
            expected_path = expected_dir / f"q{round}.csv"
            actual_path = out_dir / f"q{round}.csv"

            expected_rows = list(csv.reader(expected_path.open()))[1:]
            actual_rows = list(csv.reader(actual_path.open()))[1:]

            expected_rows = set([normalize_row(row) for row in expected_rows])
            actual_rows = set([normalize_row(row) for row in actual_rows])

            assert len(expected_rows) == len(actual_rows)
            assert expected_rows == actual_rows
            print(f"✅ Correct for query {round}!")

    stats_file = out_dir / "stats.json"
    stats = {"num_runs": num_runs, "solution": solution_type}
    for q in range(len(queries)):
        timings_q = [run[q] for run in timings]
        stats[f"q{q + 1}"] = {
            "avg_time": sum(timings_q) / len(timings_q),
            "min_time": min(timings_q),
            "max_time": max(timings_q),
            "timings": timings_q,
        }

    timings_total = [sum(run) for run in timings]
    stats["total"] = {
        "avg_time": sum(timings_total) / len(timings_total),
        "min_time": min(timings_total),
        "max_time": max(timings_total),
        "timings": timings_total,
    }

    with stats_file.open("w") as f:
        json.dump(stats, f)

    print(f"Stats written to {stats_file}")
    print("Total timings:", timings_total)
    if verbose:
        print(f"Stats: {json.dumps(stats, indent=2)}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Query Challenge Benchmark")
    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"
    )
    parser.add_argument(
        "--expected-dir",
        type=Path,
        required=False,
        help="Where to output expected results",
    )
    parser.add_argument(
        "--num-runs", type=int, default=1, help="How many times to run the benchmark"
    )
    parser.add_argument(
        "--solution", type=str, default="baseline", help="The solution to benchmark"
    )
    parser.add_argument("--verbose", action="store_true", help="Print verbose output")
    parser.add_argument(
        "--queries",
        type=Path,
        required=False,
        default=Path("queries/baseline.json"),
        help="Path to the the queries JSON file",
    )

    args = parser.parse_args()
    print(f"Benchmarking solution {args.solution}...")

    run_solution(
        args.solution,
        args.queries,
        args.data_dir,
        args.out_dir,
        args.expected_dir,
        args.num_runs,
        args.verbose,
    )

```

### src/v1.py

```python
from benchmark.solution import Solution
from pathlib import Path
import duckdb
import time
import csv
from src.planner import Planner
from src.mv import MV_REGISTRY
import tempfile
import json
import tqdm


class V1Solution(Solution):
    """A generic solution to the query challenge."""

    def __init__(self, db_path: Path):
        self.db_path = db_path
        self.table_name = "events"
        self.con = duckdb.connect(self.db_path)
        self.planner = Planner(self.con)
        self.parquet_file = db_path.parent / Path("events.parquet")

        self.warmup_queries = Path("queries/cache_warmup.json")
        self.warmup_queries = json.load(self.warmup_queries.open())

    def load_data(
        self, data_dir: Path, use_parquet=True, warmup_cache=True, verbose=False
    ) -> None:
        """Load the data from the given directory.

        Note: we're allowed to take arbitrary time in this step."""
        csv_files = list(data_dir.glob("events_part_*.csv"))

        if csv_files:
            print(f"🟩 Loading {len(csv_files)} CSV parts from {data_dir} ...")
            self.con.execute(f"""
                CREATE OR REPLACE VIEW {self.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,
                    TRY_CAST(type AS ENUM('impression','serve','click','purchase')) AS 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;
            """)

            if use_parquet:
                if not self.parquet_file.exists():
                    # if True:
                    print("generating parquet files (takes ~60 seconds)")
                    t0 = time.perf_counter()
                    self.con.execute(f"""
                    COPY (SELECT * FROM {self.table_name}) TO '{self.parquet_file.absolute()}' (FORMAT PARQUET, COMPRESSION ZSTD, PER_THREAD_OUTPUT, ROW_GROUP_SIZE 1000000);
                    """)
                    print(time.perf_counter() - t0)
                else:
                    print("parquet files already generated - skipping")

                self.con.execute(f"""
                    CREATE OR REPLACE VIEW events AS
                    SELECT *
                    FROM read_parquet('{self.parquet_file.absolute()}');
                """)

            for mv in MV_REGISTRY:
                t0 = time.time()
                print(f"creating materialized view {mv.name} (takes ~10-60 seconds)")
                self.con.execute(mv.generate_create_sql())
                dt = time.time() - t0
                print(f"🟩 {mv.name} created in {dt:.3f}s")

                print(f"computing stats for {mv.name} (takes ~10-60 seconds)")
                self.planner.compute_mv_stats(mv)

            if warmup_cache:
                print("Warming up cache")

                for mv in MV_REGISTRY:
                    t0 = time.time()
                    print(f"analyzing {mv.name} (takes ~10-60 seconds)")
                    self.con.execute(f"ANALYZE {mv.name};")
                    self.con.execute(f"SELECT COUNT(*) FROM {mv.name}")
                    dt = time.time() - t0
                    print(f"🟩 {mv.name} analyzed in {dt:.3f}s")
        else:
            raise FileNotFoundError(f"No events_part_*.csv found in {data_dir}")

    def run(
        self, queries: list[dict], out_dir: Path, verbose: bool = False
    ) -> list[float]:
        """Run the given queries and write the results to the given directory."""

        results = []
        for i, q in enumerate(queries, 1):
            sql = self.planner.translate_query(q, verbose)

            if verbose:
                print(f"\n🟦 Query {i}:\n{sql}\n")
            t0 = time.time()
            res = self.con.execute(sql)
            cols = [d[0] for d in res.description]
            rows = res.fetchall()
            dt = time.time() - t0

            if verbose:
                print(f"✅ Rows: {len(rows)} | Time: {dt}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})

        if verbose:
            print("\nSummary:")
    
[truncated — 268 more characters]
```

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