Add Rust Avro benchmarks and blog post proving rugo outperforms

benchmarks/avro-rust/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "avro_rust"
+version = "0.1.0"
+edition = "2024"
+publish = false
+
+# Rust Avro readers for measure_avro_rust.py: arrow-avro (Apache Arrow's columnar
+# reader) and apache-avro (the Apache Avro project's Rust crate, one Value per
+# record). Built twice by the measure script: default features (arrow-avro's own
+# deflate backend, miniz_oxide) and `--features zlib-rs` (flate2 switched to
+# zlib-rs, the fastest pure-Rust inflater) so arrow-avro gets its best deflate.
+
+[dependencies]
+apache-avro = "0.22.0"
+arrow-array = "60.0.0"
+arrow-avro = "60.0.0"
+flate2 = { version = "1.1", default-features = false, features = ["rust_backend"] }
+serde_json = "1"
+
+[features]
+zlib-rs = ["flate2/zlib-rs"]
+
+[profile.release]
+lto = "fat"
+codegen-units = 1

benchmarks/avro-rust/examples/inspect.rs

@@ -0,0 +1,18 @@
+//! Print arrow-avro's output schema and row 0 of the first batch, for checking
+//! that it decodes gen_avro.py's file to the same values as rugo/fastavro.
+//!
+//!     cargo run --release --example inspect -- <file.avro>
+
+use std::io::Cursor;
+
+fn main() {
+    let path = std::env::args().nth(1).expect("usage: inspect <file.avro>");
+    let data = std::fs::read(path).expect("read file");
+    let mut r = arrow_avro::reader::ReaderBuilder::new()
+        .build(Cursor::new(&data[..]))
+        .expect("arrow-avro reader");
+    let batch = r.next().expect("a batch").expect("decode");
+    for (field, col) in batch.schema().fields().iter().zip(batch.columns()) {
+        println!("{} {:?}\n    {:?}", field.name(), field.data_type(), col.slice(0, 1));
+    }
+}

benchmarks/avro-rust/src/main.rs

@@ -0,0 +1,108 @@
+//! Run one (reader, shape) Avro measurement - the Rust counterpart of avro_one.py.
+//!
+//!     avro_rust <reader> <shape> <file.avro> <reps> [batch_size] [utf8_view]
+//!
+//! Prints one JSON line: best/median wall time in ms and the row count. Called by
+//! measure_avro_rust.py; not intended for direct use.
+//!
+//! Every reader decodes the same in-memory bytes; reading the file is outside the
+//! timing. One thread throughout (neither crate parallelises a single file).
+//!
+//!   arrow-avro    arrow_avro::reader::Reader -> Arrow RecordBatches (columnar; the
+//!                 counterpart of rugo's Morsels). batch_size defaults to the
+//!                 crate's 1024; utf8_view=1 decodes strings to StringViewArray.
+//!   apache-avro   apache_avro::Reader -> one Value per record (row-oriented)
+//!
+//! Shapes: `all` reads every field; `narrow` reads id, amount and country.
+//! arrow-avro is given a projection (pruned reader schema); apache-avro is given a
+//! reader schema of just those fields, which is how it skips the rest.
+
+use std::io::Cursor;
+use std::time::Instant;
+
+const NARROW: [&str; 3] = ["id", "amount", "country"];
+
+fn writer_schema_json(data: &[u8]) -> serde_json::Value {
+    let reader = apache_avro::Reader::new(Cursor::new(data)).expect("open avro");
+    serde_json::from_str(&reader.writer_schema().canonical_form()).expect("schema json")
+}
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() < 5 {
+        eprintln!("usage: avro_rust <reader> <shape> <file.avro> <reps> [batch_size] [utf8_view]");
+        std::process::exit(2);
+    }
+    let (reader, shape, path) = (args[1].as_str(), args[2].as_str(), &args[3]);
+    let reps: usize = args[4].parse().expect("reps");
+    let batch_size: usize = args.get(5).map(|s| s.parse().expect("batch_size")).unwrap_or(1024);
+    let utf8_view = args.get(6).map(|s| s == "1").unwrap_or(false);
+    let narrow = match shape {
+        "all" => false,
+        "narrow" => true,
+        _ => panic!("unknown shape {shape}"),
+    };
+    let data = std::fs::read(path).expect("read file");
+
+    // Top-level field indexes / pruned schema for the narrow shape.
+    let schema = writer_schema_json(&data);
+    let fields = schema["fields"].as_array().expect("record fields");
+    let projection: Vec<usize> = NARROW
+        .iter()
+        .map(|n| fields.iter().position(|f| f["name"] == *n).expect("narrow field"))
+        .collect();
+    let mut pruned = schema.clone();
+    pruned["fields"] = serde_json::Value::Array(projection.iter().map(|&i| fields[i].clone()).collect());
+    let apache_reader_schema = apache_avro::Schema::parse_str(&pruned.to_string()).expect("reader schema");
+
+    let go = || -> usize {
+        match reader {
+            "arrow-avro" => {
+                let mut b = arrow_avro::reader::ReaderBuilder::new()
+                    .with_batch_size(batch_size)
+                    .with_utf8_view(utf8_view);
+                if narrow {
+                    b = b.with_projection(projection.clone());
+                }
+                let r = b.build(Cursor::new(&data[..])).expect("arrow-avro reader");
+                r.map(|batch| batch.expect("arrow-avro batch").num_rows()).sum()
+            }
+            "apache-avro" => {
+                let r = if narrow {
+                    apache_avro::Reader::builder(Cursor::new(&data[..]))
+                        .reader_schema(&apache_reader_schema)
+                        .build()
+                } else {
+                    apache_avro::Reader::new(Cursor::new(&data[..]))
+                }
+                .expect("apache-avro reader");
+                r.map(|v| {
+                    v.expect("apache-avro record");
+                    1usize
+                })
+                .sum()
+            }
+            _ => panic!("unknown reader {reader}"),
+        }
+    };
+
+    let mut times = Vec::with_capacity(reps);
+    let mut rows = 0;
+    for _ in 0..reps {
+        let t = Instant::now();
+        rows = std::hint::black_box(go());
+        times.push(t.elapsed().as_secs_f64() * 1000.0);
+    }
+    let best = times.iter().cloned().fold(f64::INFINITY, f64::min);
+    let mut sorted = times.clone();
+    sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());
+    let median = if reps % 2 == 1 {
+        sorted[reps / 2]
+    } else {
+        (sorted[reps / 2 - 1] + sorted[reps / 2]) / 2.0
+    };
+    println!(
+        "{{\"best_ms\": {:.1}, \"median_ms\": {:.1}, \"rows\": {}}}",
+        best, median, rows
+    );
+}

benchmarks/avro_one.py

@@ -9,6 +9,7 @@
 timing.
 
   rugo          rugo.avro.read_avro -> columnar Morsels (no Python object per value)
+  rugo-1t       the same, max_threads=1
   rugo+pylist   the same, then every column .to_pylist() - the cost of ending in
                 Python objects, which is what the other two return
   fastavro      fastavro.reader -> one dict per record
@@ -32,14 +33,15 @@ def narrow_schema():
     return {**schema, "fields": [f for f in schema["fields"] if f["name"] in NARROW]}
 
 
-if reader in ("rugo", "rugo+pylist"):
+if reader in ("rugo", "rugo-1t", "rugo+pylist"):
     from rugo.avro import read_avro
     columns = NARROW if shape == "narrow" else None
     to_python = reader == "rugo+pylist"
+    threads = 1 if reader == "rugo-1t" else 0
 
     def go():
         n = 0
-        with read_avro(data, columns=columns) as r:
+        with read_avro(data, columns=columns, max_threads=threads) as r:
             for morsel in r:
                 n += morsel.num_rows
                 if to_python:

benchmarks/measure_avro_rust.py

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""Race rugo's Avro reader against the Rust Avro readers on this machine.
+
+    python3 benchmarks/measure_avro_rust.py [--rows 200000] [--rounds 5] [--reps 5]
+                                            [--python 3.14] [--rugo-src /path/to/opteryx-core]
+
+Needs `uv` and `cargo` on PATH. Same file and method as measure_avro.py: gen_avro.py
+writes a deterministic 20-field file with fastavro (so no reader is timed on its own
+writer's output), each reader decodes the same in-memory bytes in a fresh process,
+reading the file is outside the timing. Both codecs, `deflate` and `null` (`null`
+isolates decode from inflate).
+
+Arms - single thread unless the name says otherwise:
+  rugo-1t              PyPI rugo, read_avro(max_threads=1) -> Morsels (avro_one.py)
+  rugo                 PyPI rugo, every core (rugo's default) - NOT part of the race
+  rugo-src-1t / -src   the same from an opteryx-core source tree (--rugo-src)
+  arrow-avro           Apache Arrow's arrow-avro -> RecordBatches, crate defaults
+                       (batch 1024, StringArray, miniz_oxide inflate)
+  arrow-avro-tuned     the same with batch 65,536 (rugo's morsel size) and flate2 on
+                       zlib-rs, the fastest Rust inflater - arrow-avro at its best
+  arrow-avro-tuned-view  tuned + StringViewArray strings
+  apache-avro          the Apache Avro Rust crate -> one Value per record
+
+Rounds interleave the arms (order rotated each round) so thermal drift and order
+bias land on every arm; each (round, arm) is a fresh process running `reps` reads.
+Per arm: best = min over every read, median = median of the per-process medians.
+Writes benchmarks/results/<UTC date>-avro-rust.json. Results are only comparable to
+earlier runs from the same machine.
+"""
+import argparse, datetime, json, os, platform, statistics, subprocess, tempfile
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+CRATE = f"{HERE}/avro-rust"
+ap = argparse.ArgumentParser()
+ap.add_argument("--rows", type=int, default=200_000)
+ap.add_argument("--rounds", type=int, default=5)
+ap.add_argument("--reps", type=int, default=5)
+ap.add_argument("--python", default="3.14")
+ap.add_argument("--rugo-src")
+ap.add_argument("--workdir", default=tempfile.mkdtemp(prefix="rugo-avro-rust-"))
+a = ap.parse_args()
+W = a.workdir
+
+
+def sh(*cmd, env=None):
+    return subprocess.run(cmd, check=True, capture_output=True, text=True, env=env).stdout
+
+
+def venv(name, pkgs):
+    p = f"{W}/venv-{name}"
+    sh("uv", "venv", "--python", a.python, p)
+    sh("uv", "pip", "install", "--python", f"{p}/bin/python", *pkgs)
+    return f"{p}/bin/python"
+
+
+# --- builds: one binary per flate2 backend
+print("building Rust readers", flush=True)
+BIN = {}
+for variant, extra in (("default", []), ("zlib-rs", ["--features", "zlib-rs"])):
+    sh("cargo", "build", "--release", "--quiet", "--manifest-path", f"{CRATE}/Cargo.toml",
+       "--target-dir", f"{CRATE}/target/{variant}", *extra)
+    BIN[variant] = f"{CRATE}/target/{variant}/release/avro_rust"
+lock = open(f"{CRATE}/Cargo.lock").read().split("[[package]]")
+crate_versions = {}
+for block in lock:
+    fields = dict(line.split(" = ", 1) for line in block.strip().splitlines() if " = " in line)
+    name = fields.get("name", "").strip('"')
+    if name in ("arrow-avro", "apache-avro", "flate2", "zlib-rs", "miniz_oxide"):
+        crate_versions[name] = fields["version"].strip('"')
+
+# --- Python side
+pypi = venv("pypi", ["rugo", "fastavro"])
+ARMS = {"rugo-1t": [pypi, f"{HERE}/avro_one.py", "rugo-1t"],
+        "rugo": [pypi, f"{HERE}/avro_one.py", "rugo"]}
+ENV = {}
+versions = {"rugo": sh(pypi, "-c", "import rugo;print(rugo.__version__)").strip(),
+            "rustc": sh("rustc", "--version").strip(), **crate_versions}
+if a.rugo_src:
+    srcpy = venv("src", ["fastavro"])
+    ENV = {**os.environ, "PYTHONPATH": os.path.abspath(a.rugo_src)}
+    ARMS["rugo-src-1t"] = [srcpy, f"{HERE}/avro_one.py", "rugo-1t"]
+    ARMS["rugo-src"] = [srcpy, f"{HERE}/avro_one.py", "rugo"]
+    versions["rugo-src"] = sh(srcpy, "-c", "import rugo;print(rugo.__version__, rugo.__file__)", env=ENV).strip()
+ARMS.update({
+    "arrow-avro": [BIN["default"], "arrow-avro"],
+    "arrow-avro-tuned": [BIN["zlib-rs"], "arrow-avro"],
+    "arrow-avro-tuned-view": [BIN["zlib-rs"], "arrow-avro"],
+    "apache-avro": [BIN["default"], "apache-avro"],
+})
+RUST_TAIL = {"arrow-avro": ["1024", "0"], "arrow-avro-tuned": ["65536", "0"],
+             "arrow-avro-tuned-view": ["65536", "1"], "apache-avro": []}
+
+res = {"date": datetime.datetime.now(datetime.UTC).isoformat(timespec="seconds"),
+       "machine": {"platform": platform.platform(), "cpu": platform.processor() or platform.machine(),
+                   "cores": os.cpu_count(), "python": a.python},
+       "rows": a.rows, "rounds": a.rounds, "reps": a.reps, "versions": versions,
+       "files_mb": {}, "reads": {}}
+
+for codec in ("deflate", "null"):
+    path = f"{W}/events-{codec}.avro"
+    print("generating", codec, flush=True)
+    sh(pypi, f"{HERE}/gen_avro.py", path, str(a.rows), codec)
+    res["files_mb"][codec] = round(os.path.getsize(path) / 1e6, 1)
+    for shape in ("all", "narrow"):
+        runs = {arm: [] for arm in ARMS}
+        names = list(ARMS)
+        for rnd in range(a.rounds):
+            order = names[rnd % len(names):] + names[:rnd % len(names)]
+            for arm in order:
+                print(f"read: {codec} {shape} round {rnd + 1}/{a.rounds} {arm}", flush=True)
+                cmd = ARMS[arm] + [shape, path, str(a.reps)] + RUST_TAIL.get(arm, [])
+                env = ENV if arm.startswith("rugo-src") else None
+                out = json.loads(sh(*cmd, env=env).strip().splitlines()[-1])
+                if out["rows"] != a.rows:
+                    raise SystemExit(f"{arm} read {out['rows']} rows, expected {a.rows}")
+                runs[arm].append(out)
+        for arm, outs in runs.items():
+            res["reads"][f"{codec}/{shape}/{arm}"] = {
+                "best_ms": min(o["best_ms"] for o in outs),
+                "median_ms": round(statistics.median(o["median_ms"] for o in outs), 1),
+                "rows": a.rows}
+
+os.makedirs(f"{HERE}/results", exist_ok=True)
+fn = f"{HERE}/results/{res['date'][:10]}-avro-rust.json"
+json.dump(res, open(fn, "w"), indent=2)
+print("wrote", fn)
+for key, value in res["reads"].items():
+    print(f"  {key:40s} best {value['best_ms']:9.1f}  median {value['median_ms']:9.1f} ms")