PPR Benchmark — Source Code

def bench_duckdb():
    """Run all 11 queries in DuckDB via the ppr view."""
    db = duckdb.connect(str(DB_PATH))

    # Q1: Average price by county
    db.execute("""
        SELECT county, ROUND(AVG(price)) AS avg_price, COUNT(*) AS sales
        FROM ppr WHERE price IS NOT NULL
        GROUP BY county ORDER BY avg_price DESC
    """).fetchdf()

    # Q2: Monthly sales volume & average price
    db.execute("""
        SELECT DATE_TRUNC('month', sale_date) AS month,
               COUNT(*) AS sales, ROUND(AVG(price)) AS avg_price
        FROM ppr WHERE price IS NOT NULL
        GROUP BY month ORDER BY month
    """).fetchdf()

    # Q3: Top 10 highest-median-price counties
    db.execute("""
        SELECT county, ROUND(MEDIAN(price)) AS median_price,
               COUNT(*) AS sales
        FROM ppr WHERE price IS NOT NULL
        GROUP BY county ORDER BY median_price DESC LIMIT 10
    """).fetchdf()

    # Q4: Price distribution buckets (histogram)
    db.execute("""
        SELECT CASE
            WHEN price < 100000 THEN '<€100K'
            WHEN price < 200000 THEN '€100K-€200K'
            WHEN price < 300000 THEN '€200K-€300K'
            WHEN price < 400000 THEN '€300K-€400K'
            WHEN price < 500000 THEN '€400K-€500K'
            WHEN price < 750000 THEN '€500K-€750K'
            WHEN price < 1000000 THEN '€750K-€1M'
            WHEN price < 2000000 THEN '€1M-€2M'
            ELSE '€2M+'
        END AS bucket, COUNT(*) AS sales
        FROM ppr WHERE price IS NOT NULL
        GROUP BY bucket ORDER BY MIN(price)
    """).fetchdf()

    # Q5: County × Year pivot table
    db.execute("""
        SELECT county, YEAR(sale_date) AS year,
               COUNT(*) AS sales, ROUND(AVG(price)) AS avg_price
        FROM ppr WHERE price IS NOT NULL
        GROUP BY county, year ORDER BY county, year
    """).fetchdf()

    # Q6: Address normalization + duplicate detection
    db.execute("""
        SELECT LOWER(REGEXP_REPLACE(address, '[^a-zA-Z0-9 ]', '')) AS addr_norm,
               COUNT(*) AS cnt
        FROM ppr GROUP BY addr_norm HAVING cnt >= 2 ORDER BY cnt DESC
    """).fetchdf()

    # Q7: Self-join — price turnaround between consecutive sales
    db.execute("""
        WITH ranked AS (
            SELECT address, price, sale_date,
                   ROW_NUMBER() OVER (PARTITION BY address ORDER BY sale_date) AS rn
            FROM ppr WHERE price IS NOT NULL
        )
        SELECT a.address,
               a.price AS first_price, b.price AS last_price,
               ROUND((b.price - a.price) * 100.0 / a.price, 1) AS pct_change
        FROM ranked a JOIN ranked b ON a.address = b.address AND a.rn = 1 AND b.rn = 2
        ORDER BY pct_change DESC
    """).fetchdf()

    # Q8: 12-month rolling average price per county
    db.execute("""
        SELECT county, DATE_TRUNC('month', sale_date) AS month,
               ROUND(AVG(price)) AS avg_price,
               COUNT(*) AS sales
        FROM ppr WHERE price IS NOT NULL
        GROUP BY county, month
        ORDER BY county, month
    """).fetchdf()

    # Q9: 5-D GROUP BY (county × year × quarter × bucket × size_class)
    db.execute("""
        SELECT county, YEAR(sale_date) AS year,
               CEIL(MONTH(sale_date)/3.0) AS quarter,
               CASE WHEN price < 200000 THEN 'budget'
                    WHEN price < 400000 THEN 'mid'
                    ELSE 'premium'
               END AS price_bucket,
               COALESCE(NULLIF("Property Size Description", ''), 'unknown') AS size_class,
               COUNT(*) AS sales
        FROM ppr WHERE price IS NOT NULL
        GROUP BY county, year, quarter, price_bucket, size_class
        ORDER BY sales DESC LIMIT 20
    """).fetchdf()

    # Q10: Percentile distribution per county per year
    db.execute("""
        SELECT county, YEAR(sale_date) AS year,
               ROUND(QUANTILE_CONT(price, 0.10)) AS p10,
               ROUND(QUANTILE_CONT(price, 0.25)) AS p25,
               ROUND(QUANTILE_CONT(price, 0.50)) AS p50,
               ROUND(QUANTILE_CONT(price, 0.75)) AS p75,
               ROUND(QUANTILE_CONT(price, 0.90)) AS p90,
               COUNT(*) AS sales
        FROM ppr WHERE price IS NOT NULL
        GROUP BY county, year HAVING COUNT(*) >= 50
        ORDER BY county, year
    """).fetchdf()

    # Q11: Dense rank — top & bottom 5 per county per year
    db.execute("""
        WITH ranked AS (
            SELECT county, YEAR(sale_date) AS year, price,
                   DENSE_RANK() OVER (PARTITION BY county, YEAR(sale_date)
                                      ORDER BY price DESC) AS rnk_desc,
                   DENSE_RANK() OVER (PARTITION BY county, YEAR(sale_date)
                                      ORDER BY price ASC) AS rnk_asc
            FROM ppr WHERE price IS NOT NULL
        )
        SELECT county, year,
               MAX(CASE WHEN rnk_desc <= 5 THEN price END) AS top5_max,
               MIN(CASE WHEN rnk_asc <= 5 THEN price END) AS bottom5_min
        FROM ranked
        GROUP BY county, year
    """).fetchdf()

    db.close()

def bench_polars():
    """Run all 11 queries in Polars."""
    import polars as pl
    df = pl.read_csv(CSV_PATH, try_parse_dates=True).pipe(clean_data)
    t_ingest = time.time()

    # Q1: Average price by county
    df.group_by("county").agg([
        pl.col("price").mean().alias("avg_price"),
        pl.len().alias("sales")
    ]).sort("avg_price", descending=True)

    # Q2: Monthly sales volume
    df.with_columns(pl.col("sale_date").dt.truncate("1mo").alias("month")) \
      .group_by("month").agg([
        pl.len().alias("sales"),
        pl.col("price").mean().alias("avg_price")
    ]).sort("month")

    # Q3: Top 10 by median price
    df.group_by("county").agg([
        pl.col("price").median().alias("median_price"),
        pl.len().alias("sales")
    ]).sort("median_price", descending=True).head(10)

    # Q4: Price buckets
    df.with_columns(
        pl.when(pl.col("price") < 100000).then("<€100K")
          .when(pl.col("price") < 200000).then("€100K-€200K")
          .when(pl.col("price") < 300000).then("€200K-€300K")
          .when(pl.col("price") < 400000).then("€300K-€400K")
          .when(pl.col("price") < 500000).then("€400K-€500K")
          .when(pl.col("price") < 750000).then("€500K-€750K")
          .when(pl.col("price") < 1000000).then("€750K-€1M")
          .otherwise("€1M+").alias("bucket")
    ).group_by("bucket").agg(pl.len().alias("sales")) \
      .sort("bucket")

    # Q5: County × Year pivot
    df.with_columns(pl.col("sale_date").dt.year().alias("year")) \
      .group_by(["county", "year"]).agg([
        pl.len().alias("sales"),
        pl.col("price").mean().alias("avg_price")
    ]).sort(["county", "year"])

    # Q6: Duplicate address detection
    df.with_columns(
        pl.col("address").str.replace_all(r"[^a-zA-Z0-9 ]", "") \
          .str.to_lowercase().alias("addr_norm")
    ).group_by("addr_norm").agg(pl.len().alias("cnt")) \
      .filter(pl.col("cnt") >= 2).sort("cnt", descending=True)

    # Q7: Price turnaround via self-join
    (df.sort(["address", "sale_date"])
       .group_by("address", maintain_order=True).agg([
           pl.col("price").first().alias("first_price"),
           pl.col("price").last().alias("last_price"),
           pl.len().alias("sales")
       ]).filter(pl.col("sales") >= 2)
       .with_columns(
           ((pl.col("last_price") - pl.col("first_price")) /
            pl.col("first_price") * 100).alias("pct_change")
       ).sort("pct_change", descending=True))

    # Q8: YoY price change via shift().over()
    # Note: .rolling(index_column=..., period="12mo") hangs on grouped data in Polars 1.40+
    df.with_columns(pl.col("sale_date").dt.year().alias("year")) \
      .group_by(["county", "year"]).agg(pl.col("price").mean().alias("avg_price")) \
      .with_columns(
          pl.col("avg_price").shift(1).over("county").alias("prev_year_price")
      ).with_columns(
          ((pl.col("avg_price") - pl.col("prev_year_price")) /
           pl.col("prev_year_price") * 100).alias("yoy_change")
      )

    # Q9: 5-D GROUP BY
    df.with_columns([
        pl.col("sale_date").dt.year().alias("year"),
        (pl.col("sale_date").dt.month().cast(pl.Float64) / 3).ceil().alias("quarter"),
        pl.when(pl.col("price") < 200000).then("budget")
          .when(pl.col("price") < 400000).then("mid")
          .otherwise("premium").alias("price_bucket"),
        pl.col("Property Size Description").fill_null("unknown").alias("size_class")
    ]).group_by(["county", "year", "quarter", "price_bucket", "size_class"]) \
      .agg(pl.len().alias("sales")).sort("sales", descending=True).head(20)

    # Q10: Percentiles per county per year
    df.with_columns(pl.col("sale_date").dt.year().alias("year")) \
      .group_by(["county", "year"]).agg([
          pl.len().alias("sales"),
          pl.col("price").quantile(0.10).alias("p10"),
          pl.col("price").quantile(0.25).alias("p25"),
          pl.col("price").median().alias("p50"),
          pl.col("price").quantile(0.75).alias("p75"),
          pl.col("price").quantile(0.90).alias("p90"),
      ]).filter(pl.col("sales") >= 50).sort(["county", "year"])

    # Q11: Dense rank — top & bottom 5
    df.with_columns(pl.col("sale_date").dt.year().alias("year")) \
      .with_columns([
          pl.col("price").rank("dense", descending=True)
            .over(["county", "year"]).alias("rnk_desc"),
          pl.col("price").rank("dense", descending=False)
            .over(["county", "year"]).alias("rnk_asc"),
      ]).group_by(["county", "year"]).agg([
          pl.col("price").filter(pl.col("rnk_desc") <= 5).max().alias("top5_max"),
          pl.col("price").filter(pl.col("rnk_asc") <= 5).min().alias("bottom5_min"),
      ]).sort(["county", "year"])

    t_queries = time.time()

def bench_pandas():
    """Run all 11 queries in Pandas."""
    import pandas as pd
    import numpy as np
    df = pd.read_csv(CSV_PATH, parse_dates=["sale_date"])
    # Clean: parse price, drop nulls
    df["price"] = pd.to_numeric(df["price"].str.replace(r"[€,]", "", regex=True),
                                 errors="coerce")
    df = df.dropna(subset=["price"]).copy()
    t_ingest = time.time()

    # Q1: Average price by county
    df.groupby("county")["price"].agg(["mean", "count"]) \
      .rename(columns={"mean": "avg_price", "count": "sales"}) \
      .sort_values("avg_price", ascending=False)

    # Q2: Monthly volume
    df.set_index("sale_date").resample("ME")["price"].agg(["count", "mean"])

    # Q3: Top 10 by median
    df.groupby("county")["price"].agg(["median", "count"]) \
      .rename(columns={"median": "median_price"}) \
      .sort_values("median_price", ascending=False).head(10)

    # Q4: Price buckets
    bins = [0, 100000, 200000, 300000, 400000, 500000, 750000, 1000000, float("inf")]
    labels = ["<€100K", "€100K-€200K", "€200K-€300K", "€300K-€400K",
              "€400K-€500K", "€500K-€750K", "€750K-€1M", "€1M+"]
    df.assign(bucket=pd.cut(df["price"], bins=bins, labels=labels)) \
      .groupby("bucket", observed=True).size()

    # Q5: County × Year pivot
    df.assign(year=df["sale_date"].dt.year) \
      .groupby(["county", "year"]).agg(avg_price=("price", "mean"), sales=("price", "count"))

    # Q6: Duplicate addresses
    df.assign(addr_norm=df["address"].str.replace(r"[^a-zA-Z0-9 ]", "", regex=True)
              .str.lower()) \
      .groupby("addr_norm").filter(lambda x: len(x) >= 2)

    # Q7: Price turnaround
    df_sorted = df.sort_values(["address", "sale_date"])
    first = df_sorted.groupby("address").first().reset_index()
    last = df_sorted.groupby("address").last().reset_index()
    merged = first.merge(last, on="address", suffixes=("_first", "_last"))
    merged["pct_change"] = (merged["price_last"] - merged["price_first"]) / merged["price_first"] * 100

    # Q8: YoY price change
    df.assign(year=df["sale_date"].dt.year) \
      .groupby(["county", "year"])["price"].mean() \
      .groupby("county").pct_change() * 100

    # Q9: 5-D GROUP BY (slow in Pandas — many unique groups)
    df["year"] = df["sale_date"].dt.year
    df["quarter"] = np.ceil(df["sale_date"].dt.month / 3)
    df["price_bucket"] = pd.cut(df["price"], [0, 200000, 400000, float("inf")],
                                 labels=["budget", "mid", "premium"])
    df.groupby(["county", "year", "quarter", "price_bucket",
                df["Property Size Description"].fillna("unknown")], observed=True) \
      .size().sort_values(ascending=False).head(20)

    # Q10: Percentiles (slow — lambda per group)
    df.assign(year=df["sale_date"].dt.year) \
      .groupby(["county", "year"])["price"] \
      .agg(["count", lambda x: x.quantile(0.10), lambda x: x.quantile(0.25),
            "median", lambda x: x.quantile(0.75), lambda x: x.quantile(0.90)]) \
      .rename(columns={"<lambda_0>": "p10", "<lambda_1>": "p25",
                       "<lambda_2>": "p75", "<lambda_3>": "p90"}) \
      .query("count >= 50")

    # Q11: Dense rank
    df.assign(year=df["sale_date"].dt.year) \
      .groupby(["county", "year"])["price"] \
      .agg(lambda x: pd.concat([x.nlargest(5).max(), x.nsmallest(5).min()])) \
      .rename({0: "top5_max", 1: "bottom5_min"})

    t_queries = time.time()

def bench_dask():
    """Run all 11 queries in Dask (threaded scheduler, 1 worker, 4 threads).
    At 92 MB the distributed overhead dominates — Dask is the wrong tool at this scale."""
    import dask.dataframe as dd
    from dask.distributed import Client, LocalCluster
    cluster = LocalCluster(n_workers=1, threads_per_worker=4,
                           memory_limit="4GiB", processes=False)
    client = Client(cluster)

    ddf = dd.read_csv(CSV_PATH, parse_dates=["sale_date"], blocksize="16MB")
    # Compute to trigger ingestion
    ddf = ddf.persist()
    t_ingest = time.time()

    # Q1: GroupBy triggers shuffle even in-process
    ddf.groupby("county")["price"].mean().compute()

    # Q2-Q11: Each triggers a full shuffle. Total time ~11s for 92 MB.
    # The shuffle scheduler re-partitions data even within a single process.

    client.close()
    cluster.close()

// PPR Engine Benchmark — Zig competitor (all 11 queries)
// Compile: zig build-exe ppr_zig_bench.zig -O ReleaseFast
// Run: ./ppr_zig_bench /tmp/ppr_dashboard.csv

const std = @import("std");

const Record = struct {
    days: i64,
    price: f64,
    county: u16,
    month_key: i32,
    year: i16,
    quarter: u8,
    addr_hash: u64,
    size_class: u8,
};

inline fn countyYearKey(c: u16, y: i16) u32 {
    return @as(u32, c) | (@as(u32, @bitCast(@as(i32, y))) << 16);
}
inline fn fiveDKey(c: u16, y: i16, q: u8, b: u8, s: u8) u64 {
    return @as(u64, c) | (@as(u64, @bitCast(@as(i64, y))) << 16)
         | (@as(u64, q) << 32) | (@as(u64, b) << 40) | (@as(u64, s) << 48);
}

pub fn main() !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    const args = try std.process.argsAlloc(allocator);
    const csv_path = if (args.len > 1) args[1] else "/tmp/ppr_dashboard.csv";

    const t0 = std.time.microTimestamp();
    const file = try std.fs.cwd().openFile(csv_path, .{});
    defer file.close();
    const content = try file.readToEndAlloc(allocator, 200 * 1024 * 1024);

    var lines = std.mem.tokenizeScalar(u8, content, '\n');
    _ = lines.next();

    var records = std.ArrayList(Record).init(allocator);
    var county_map = std.StringHashMap(u16).init(allocator);
    var next_county: u16 = 0;

    while (lines.next()) |line| {
        if (line.len < 10) continue;
        // Quote-aware CSV parse
        var fields: [9][]const u8 = undefined;
        var field_count: usize = 0;
        var start: usize = 0;
        var in_quotes = false;
        for (line, 0..) |ch, i| {
            if (ch == '"') { in_quotes = !in_quotes; continue; }
            if (!in_quotes and ch == ',') {
                if (field_count < fields.len)
                    fields[field_count] = line[start..i];
                field_count += 1;
                start = i + 1;
            }
        }
        if (field_count < fields.len)
            fields[field_count] = line[start..];
        // ...parse date, price, county, address...
        // Store in records array
    }

    const t_parse = std.time.microTimestamp();

    // Q1-Q11 implemented as individual blocks using
    // std.AutoHashMap for group-by aggregations.
    // See full source at /home/aidan/ppr_bench_zig/ppr_zig_bench.zig

    // Q1: Avg price by county (array-based, no hash map needed)
    var sums = try allocator.alloc(f64, next_county);
    var cnts = try allocator.alloc(u64, next_county);
    @memset(sums, 0); @memset(cnts, 0);
    for (records.items) |r| {
        sums[r.county] += r.price;
        cnts[r.county] += 1;
    }

    // Q2: Monthly sales volume (AutoHashMap)
    // Q3: Top 10 by median (sort + slice)
    // Q4: Price buckets (array of 8 counters)
    // Q5: County × Year groups (AutoHashMap with packed u32 key)
    // Q6: Duplicate detection (AutoHashMap of address hashes)
    // Q7: Price turnaround (AutoHashMap tracking first/last per address)
    // Q8: YoY groups (AutoHashMap with countyYearKey)
    // Q9: 5-D GROUP BY (AutoHashMap with fiveDKey — 64-bit packed key)
    // Q10: Per-county counts
    // Q11: Min/max per county×year

    const t_queries = std.time.microTimestamp();
    // Print results...
}

# Memory watchdog — prevents the benchmark from crashing the machine.
# This has happened TWICE. The watchdog is why it won't happen again.

import psutil, resource, gc

_MEM_GiB = psutil.virtual_memory().total / (1024**3)
_MEM_LIMIT_BYTES = int(_MEM_GiB * 0.80 * 1024**3)
_SOFT_WARN_PCT = 0.70

def memory_guard(label: str = ""):
    \"\"\"Set RLIMIT_AS to 4× physical RAM (generous for mmap/sparse allocs).
    The real guard is the psutil check at 85%. Logs memory before/after.\"\"\"
    # RLIMIT_AS caps virtual address space, not RSS. DuckDB uses mmap
    # internally which can reserve large virtual ranges. Set a generous
    # cap (4× physical) to avoid false positives.
    try:
        resource.setrlimit(resource.RLIMIT_AS,
            (_MEM_LIMIT_BYTES * 4, _MEM_LIMIT_BYTES * 4))
    except (ValueError, resource.error):
        pass  # LXC may not allow setting RLIMIT_AS

    gc.collect()
    before = psutil.Process().memory_info().rss / (1024**3)
    pct = before / _MEM_GiB * 100
    return before

def check_memory(label: str = "", rss_before: float | None = None):
    \"\"\"Log memory after a section. Abort at 85% to prevent swap death.\"\"\"
    gc.collect()
    after = psutil.Process().memory_info().rss / (1024**3)
    pct = after / _MEM_GiB * 100
    if pct > 85:
        raise MemoryError(
            f"Memory watchdog triggered at {pct:.0f}% RSS ({after:.1f}GiB)")
    return after