[DO NOT MERGE] graphframes-rs like flow

examples/extensions/benchmarks/sem_benchmark/.gitignore

@@ -0,0 +1,3 @@
+data/
+work/
+target/

examples/extensions/benchmarks/sem_benchmark/Makefile

@@ -0,0 +1,84 @@
+# Build the Sail + Nutmeg stack for sem_benchmark on a fresh VPS and run it.
+#
+# Prerequisites (Ubuntu/Debian, see README.md for details):
+#   curl, git, protobuf-compiler, libprotobuf-dev, clang, libclang-dev, pkg-config
+#   rustup (1.97.1+), uv, python3.12 (a uv-managed interpreter is fine)
+#
+#   make venv          # .venv with pyspark[connect] + maturin + gfrs-poc
+#   make build         # nutmeg wheel + sail host, both with RUSTFLAGS=-C target-cpu=native
+#   make smoke         # tiny example-directed graph, both engines, 1 run
+#   make bench DS=cit-Patents ENGINE=both       # hash joins (sail default)
+#   make bench-smj DS=cit-Patents ENGINE=both   # sort-merge joins (--use-smj)
+#   make clean
+
+REPO_ROOT := $(shell git rev-parse --show-toplevel)
+HERE := $(REPO_ROOT)/examples/extensions/benchmarks/sem_benchmark
+
+VENV ?= $(REPO_ROOT)/.venv
+PYTHON ?= $(VENV)/bin/python
+# Respect CARGO_TARGET_DIR when the environment sets it, like cargo does.
+HOST_TARGET_DIR ?= $(or $(CARGO_TARGET_DIR),$(REPO_ROOT)/target)
+SAIL_BIN ?= $(HOST_TARGET_DIR)/release/sail
+RUSTFLAGS ?= -C target-cpu=native
+CARGO_BUILD_JOBS ?= 4
+
+WORK_DIR ?= $(HERE)/work
+RESULTS_DIR ?= $(HERE)/results
+DATA_DIR ?= $(HERE)/data
+MAX_MEMORY ?= 24G
+PARTITIONS ?= 8
+THREADS ?= 8
+DS ?= example-directed
+ENGINE ?= both
+RUNS ?= 3
+TOL ?= 1e-5
+
+RUN_ARGS = \
+	  --work-dir $(WORK_DIR) \
+	  --max-memory $(MAX_MEMORY) \
+	  --partitions $(PARTITIONS) \
+	  --threads $(THREADS) \
+	  --sail-binary $(SAIL_BIN) \
+	  --results-dir $(RESULTS_DIR) \
+	  --data-dir $(DATA_DIR) \
+	  --tol $(TOL)
+
+.PHONY: venv build nutmeg-wheel sail-host smoke bench bench-smj clean
+
+venv:
+	uv venv --python 3.12 $(VENV)
+	uv pip install --python $(PYTHON) "pyspark[connect]==4.0.1" maturin==1.9.6
+	uv pip install --python $(PYTHON) --no-deps $(REPO_ROOT)/examples/extensions/graph-algorithms/gfrs-poc
+
+nutmeg-wheel: venv
+	mkdir -p $(HERE)/target/wheels
+	cd $(REPO_ROOT) && \
+	RUSTFLAGS="$(RUSTFLAGS)" CARGO_BUILD_JOBS=$(CARGO_BUILD_JOBS) \
+	CARGO_TARGET_DIR=$(HERE)/target/nutmeg \
+	$(PYTHON) -m maturin build --release --locked \
+	  --manifest-path $(REPO_ROOT)/examples/extensions/nutmeg/Cargo.toml \
+	  --interpreter $(PYTHON) --out $(HERE)/target/wheels
+	uv pip install --python $(PYTHON) --reinstall $(HERE)/target/wheels/sail_nutmeg-*.whl
+
+sail-host:
+	cd $(REPO_ROOT) && \
+	PYO3_PYTHON=$(PYTHON) \
+	RUSTFLAGS="$(RUSTFLAGS)" CARGO_BUILD_JOBS=$(CARGO_BUILD_JOBS) \
+	cargo build --release --locked -p sail-cli
+
+build: nutmeg-wheel sail-host
+
+smoke: build
+	$(PYTHON) $(HERE)/main.py \
+	  --engine both --dataset example-directed --runs 1 --warmup 0 $(RUN_ARGS)
+
+bench: build
+	$(PYTHON) $(HERE)/main.py \
+	  --engine $(ENGINE) --dataset $(DS) --runs $(RUNS) $(RUN_ARGS)
+
+bench-smj: build
+	$(PYTHON) $(HERE)/main.py \
+	  --engine $(ENGINE) --dataset $(DS) --runs $(RUNS) --use-smj $(RUN_ARGS)
+
+clean:
+	rm -rf $(WORK_DIR) $(RESULTS_DIR)

examples/extensions/benchmarks/sem_benchmark/README.md

@@ -0,0 +1,105 @@
+# sem_benchmark: Nutmeg pagerank vs PySpark Pregel pagerank (gfrs-poc) on Sail
+
+A POC harness that compares two PageRank implementations against the same Sail
+server, the same two parquet input files per dataset and the same resource
+envelope, and reports in the graphframes-rs benchmark style.
+
+| `--engine` | Implementation | Notes |
+| --- | --- | --- |
+| `pregel` | [`gfrs_poc.pagerank`](../../../graph-algorithms/gfrs-poc) — pure PySpark port of the graphframes-rs Pregel engine | GraphX-style deltas, decreasing active frontier, parquet checkpoints (no purge) |
+| `nutmeg` | `sail_nutmeg` native kernel `pagerankDelta` (Banda) | driver-resident CSR, tolerance-scaled residual pushes |
+
+Both run until convergence with `tol=1e-5` (damping 0.85) and write exactly one
+parquet result file (`id`, `pagerank`) under `--work-dir`.
+
+## Harness shape
+
+- **Inputs**: two parquet files per dataset on local disk
+  (`<data-dir>/<dataset>/<dataset>-{v,e}.parquet`, downloaded from
+  `datasets.ldbcouncil.org` on first use).
+- **Cold runs**: every run (warmup included) starts a **fresh Sail server
+  process** and stops it afterwards; every measured run is a "first run".
+- **Server settings per run** (from `main.py` flags):
+  - `SAIL_MODE=local`
+  - `SAIL_RUNTIME__MEMORY_POOL__TYPE=fair`,
+    `SAIL_RUNTIME__MEMORY_POOL__FAIR__MAX_SIZE=<--max-memory>`
+  - `SAIL_RUNTIME__TEMPORARY_FILES__PATHS=["<--work-dir>/spill"]` (point this at
+    fast local NVM)
+  - `--use-smj` -> `SAIL_OPTIMIZER__PREFER_HASH_JOIN=false` (this is also the
+    graphframes-rs default)
+- **pregel checkpoints** land under `<--work-dir>/<engine>_<dataset>/run_<i>/gfrs/<uuid>/`
+  (a fresh uuid per run, like graphframes-rs). **Nothing is purged**, so disk
+  usage grows with the iteration count — keep `--work-dir` on a large volume.
+- **Output**: `results/<engine>/<scale>/<dataset>/max_mem_<mem>_<hash|smj>_parts_<n>/`
+  with `benchmark.json` (params, per-run wall time / peak RSS / peak disk /
+  iterations / residual, stats, DTW-aligned series) plus `wall_time`, `rss` and
+  `disk` gnuplot scripts and PNGs when gnuplot is installed.
+
+The `datasets.py`, `monitor.py`, `stats.py` and `plotting.py` files are copied
+as-is from `graphframes-rs/benches/python/`.
+
+## What to do on a VPS
+
+Prerequisites (Ubuntu/Debian; the same inventory as
+`examples/extensions/scripts/Dockerfile.linux` minus GEOS — Sedona is not needed):
+
+```bash
+sudo apt-get update && sudo apt-get install -y \
+  curl git build-essential clang libclang-dev libssl-dev pkg-config \
+  libprotobuf-dev protobuf-compiler
+curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --default-toolchain 1.97.1
+source "$HOME/.cargo/env"
+curl -LsSf https://astral.sh/uv/install.sh | sh
+```
+
+Then, from the branch checkout (put the repo and the build targets on a fast
+disk; the Rust builds need tens of GiB):
+
+```bash
+cd sail
+make -C examples/extensions/benchmarks/sem_benchmark venv   # .venv: pyspark[connect], maturin, gfrs-poc
+make -C examples/extensions/benchmarks/sem_benchmark build  # nutmeg wheel + sail host (release, target-cpu=native)
+```
+
+Smoke test (tiny 10-vertex graph, both engines, one run, no warmup):
+
+```bash
+make -C examples/extensions/benchmarks/sem_benchmark smoke
+```
+
+Measured runs (adjust `WORK_DIR` to your NVM mount; sizes from the LDBC catalog:
+cit-Patents ~16M edges, graph500-24 ~260M edges, twitter_mpi ~1.4B edges):
+
+```bash
+S=examples/extensions/benchmarks/sem_benchmark
+make -C $S bench      DS=cit-Patents ENGINE=both WORK_DIR=/mnt/nvm/sem RUNS=3 MAX_MEMORY=24G
+make -C $S bench-smj  DS=cit-Patents ENGINE=both WORK_DIR=/mnt/nvm/sem RUNS=3 MAX_MEMORY=24G
+make -C $S bench      DS=graph500-24  ENGINE=both WORK_DIR=/mnt/nvm/sem RUNS=3 MAX_MEMORY=24G
+make -C $S bench      DS=twitter_mpi  ENGINE=both WORK_DIR=/mnt/nvm/sem RUNS=3 MAX_MEMORY=24G
+```
+
+Every Makefile variable (`WORK_DIR`, `RESULTS_DIR`, `DATA_DIR`, `MAX_MEMORY`,
+`PARTITIONS`, `THREADS`, `DS`, `ENGINE`, `RUNS`, `TOL`) can be overridden on the
+command line. `make -C $S clean` removes the workdir and results.
+
+## Reading the numbers
+
+- `wall_time_s` covers **input DataFrame handles -> completed result parquet
+  write** (server startup, verification and cleanup are excluded), matching the
+  boundary of the existing extension benchmarks.
+- `peak_rss_kb` samples the **Sail server process** (`/proc/<pid>/status VmRSS`);
+  `peak_disk_bytes` is the exact `du`-style size of the run's workdir tree.
+- For `nutmeg`, per-run `details.native` carries the kernel-reported
+  `iterations`, `converged` and `residual`; for `pregel`, `details.iterations`
+  and `details.active_counts` come from the engine itself.
+
+## Notes / POC limitations
+
+- No checkpoint purge: the PySpark Connect API has no FS list/delete, so
+  `gfrs-poc` keeps every `state-<i>` / `aggregated-messages-<i>` directory.
+  Expect roughly `2 * iterations * (state size)` on disk per measured run.
+- `--nutmeg-kernel pagerank` switches the nutmeg side to the Grust reference
+  kernel (successive-iterate stopping); the default `pagerankDelta` is the
+  apples-to-apples delta variant.
+- `--nutmeg-memory` is prepaid from the same fair pool as `--max-memory`, so it
+  must stay smaller (the harness rejects invalid pairs up front).

examples/extensions/benchmarks/sem_benchmark/datasets.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""LDBC Graphalytics dataset catalog + parquet download (pure stdlib).
+
+The catalog mirrors the official dataset table
+(https://ldbcouncil.org/benchmarks/graphalytics/datasets/): name, scale
+class, node/edge counts and package size. Counts are stored both as the
+human-readable strings from the page ("3M") and as parsed integers used for
+e.g. the SSSP landmark (25th percentile vertex id) and throughput numbers.
+"""
+
+from __future__ import annotations
+
+import sys
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+BASE_URL = "https://datasets.ldbcouncil.org/graphalytics-parquet"
+
+# name -> (scale, nodes_str, edges_str, size_str)
+_CATALOG = [
+    ("cit-Patents", "XS", "3M", "16M", "119.1 MB"),
+    ("com-friendster", "XL", "65M", "1B", "6.7 GB"),
+    ("datagen-7_5-fb", "S", "633k", "34M", "162.3 MB"),
+    ("datagen-7_6-fb", "S", "754k", "42M", "200.0 MB"),
+    ("datagen-7_7-zf", "S", "13M", "32M", "434.5 MB"),
+    ("datagen-7_8-zf", "S", "16M", "41M", "544.3 MB"),
+    ("datagen-7_9-fb", "S", "1M", "85M", "401.2 MB"),
+    ("datagen-8_0-fb", "M", "1M", "107M", "502.5 MB"),
+    ("datagen-8_1-fb", "M", "2M", "134M", "625.4 MB"),
+    ("datagen-8_2-zf", "M", "43M", "106M", "1.4 GB"),
+    ("datagen-8_3-zf", "M", "53M", "130M", "1.7 GB"),
+    ("datagen-8_4-fb", "M", "3M", "269M", "1.2 GB"),
+    ("datagen-8_5-fb", "L", "4M", "332M", "1.5 GB"),
+    ("datagen-8_6-fb", "L", "5M", "421M", "1.9 GB"),
+    ("datagen-8_7-zf", "L", "145M", "340M", "4.6 GB"),
+    ("datagen-8_8-zf", "L", "168M", "413M", "5.3 GB"),
+    ("datagen-8_9-fb", "L", "10M", "848M", "3.7 GB"),
+    ("datagen-9_0-fb", "XL", "12M", "1B", "4.6 GB"),
+    ("datagen-9_1-fb", "XL", "16M", "1B", "5.8 GB"),
+    ("datagen-9_2-zf", "XL", "434M", "1B", "13.7 GB"),
+    ("datagen-9_3-zf", "XL", "555M", "1B", "17.4 GB"),
+    ("datagen-9_4-fb", "XL", "29M", "2B", "14.0 GB"),
+    ("datagen-sf3k-fb", "XL", "33M", "2B", "12.7 GB"),
+    ("datagen-sf10k-fb", "2XL", "100M", "9B", "40.5 GB"),
+    ("dota-league", "S", "61k", "50M", "114.3 MB"),
+    ("graph500-22", "S", "2M", "64M", "202.4 MB"),
+    ("graph500-23", "M", "4M", "129M", "410.6 MB"),
+    ("graph500-24", "M", "8M", "260M", "847.7 MB"),
+    ("graph500-25", "L", "17M", "523M", "1.7 GB"),
+    ("graph500-26", "XL", "32M", "1B", "3.4 GB"),
+    ("graph500-27", "XL", "63M", "2B", "7.1 GB"),
+    ("graph500-28", "2XL", "121M", "4B", "14.4 GB"),
+    ("graph500-29", "2XL", "232M", "8B", "29.6 GB"),
+    ("graph500-30", "3XL", "447M", "17B", "60.8 GB"),
+    ("kgs", "XS", "832k", "17M", "65.7 MB"),
+    ("twitter_mpi", "XL", "52M", "1B", "5.7 GB"),
+    ("wiki-Talk", "2XS", "2M", "5M", "34.9 MB"),
+    ("example-directed", "-", "10", "17", "1.0 KB"),
+    ("example-undirected", "-", "9", "12", "1.0 KB"),
+    ("test-bfs-directed", "-", "<100", "<100", "<2.0 KB"),
+    ("test-bfs-undirected", "-", "<100", "<100", "<2.0 KB"),
+    ("test-cdlp-directed", "-", "<100", "<100", "<2.0 KB"),
+    ("test-cdlp-undirected", "-", "<100", "<100", "<2.0 KB"),
+    ("test-pr-directed", "-", "<100", "<100", "<2.0 KB"),
+    ("test-pr-undirected", "-", "<100", "<100", "<2.0 KB"),
+    ("test-lcc-directed", "-", "<100", "<100", "<2.0 KB"),
+    ("test-lcc-undirected", "-", "<100", "<100", "<2.0 KB"),
+    ("test-wcc-directed", "-", "<100", "<100", "<2.0 KB"),
+    ("test-wcc-undirected", "-", "<100", "<100", "<2.0 KB"),
+    ("test-sssp-directed", "-", "<100", "<100", "<2.0 KB"),
+    ("test-sssp-undirected", "-", "<100", "<100", "<2.0 KB"),
+]
+
+CATALOG = {name: (scale, nodes_s, edges_s, size_s) for name, scale, nodes_s, edges_s, size_s in _CATALOG}
+
+
+def parse_count(s: str) -> int | None:
+    """Parse '3M' -> 3_000_000, '633k' -> 633_000, '1B' -> 1_000_000_000, '10' -> 10."""
+    s = s.strip()
+    if not s or s.startswith("<"):
+        return None
+    mult = 1
+    if s[-1] in "kK":
+        mult = 1_000
+        s = s[:-1]
+    elif s[-1] in "mM":
+        mult = 1_000_000
+        s = s[:-1]
+    elif s[-1] in "bB":
+        mult = 1_000_000_000
+        s = s[:-1]
+    try:
+        return int(float(s) * mult)
+    except ValueError:
+        return None
+
+
+def info(name: str) -> dict:
+    scale, nodes_s, edges_s, size_s = CATALOG[name]
+    return {
+        "name": name,
+        "scale": scale,
+        "nodes_str": nodes_s,
+        "edges_str": edges_s,
+        "size": size_s,
+        "vertices": parse_count(nodes_s),
+        "edges": parse_count(edges_s),
+    }
+
+
+def list_datasets() -> str:
+    rows = [["name", "scale", "nodes", "edges", "size"]]
+    for name, scale, nodes_s, edges_s, size_s in _CATALOG:
+        rows.append([name, scale, nodes_s, edges_s, size_s])
+    widths = [max(len(r[i]) for r in rows) for i in range(5)]
+    lines = []
+    for r in rows:
+        lines.append("  ".join(cell.ljust(w) for cell, w in zip(r, widths)))
+    return "\n".join(lines)
+
+
+def ensure_dataset(name: str, data_dir: Path, retries: int = 3) -> None:
+    """Download <name>-v.parquet / <name>-e.parquet into data_dir/<name>/ if missing."""
+    ds_dir = data_dir / name
+    ds_dir.mkdir(parents=True, exist_ok=True)
+    for suffix in ("v", "e"):
+        target = ds_dir / f"{name}-{suffix}.parquet"
+        if target.exists() and target.stat().st_size > 0:
+            continue
+        url = f"{BASE_URL}/{name}-{suffix}.parquet"
+        print(f"Downloading {url}")
+        tmp = target.with_suffix(".parquet.part")
+        for attempt in range(1, retries + 1):
+            try:
+                req = urllib.request.Request(url, headers={"User-Agent": "graphframes-rs-bench"})
+                with urllib.request.urlopen(req, timeout=120) as resp, open(tmp, "wb") as f:
+                    while True:
+                        chunk = resp.read(1 << 16)
+                        if not chunk:
+                            break
+                        f.write(chunk)
+                tmp.rename(target)
+                print(f"  -> {target} ({target.stat().st_size / 1e6:.1f} MB)")
+                break
+            except (urllib.error.URLError, TimeoutError, OSError) as e:
+                print(f"  attempt {attempt}/{retries} failed: {e}", file=sys.stderr)
+                if attempt < retries:
+                    time.sleep(5)
+                else:
+                    raise SystemExit(f"failed to download {url}")