From ff168aaa4723fdfd31309bcdb49fbe243a6cc016 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 20:58:19 +0000 Subject: [PATCH 1/5] ci: weekly TPC-H SF1000 with memory monitoring Adds a weekly (and manually dispatchable) TPC-H SF=1000 run on NVMe for DataFusion and DuckDB over Parquet and Vortex, one machine per engine. Parquet is generated with tpchgen-cli 3.0.0 (parallel, ZSTD(3)) because the in-repo generator writes each table single-threaded, then data-gen converts to Vortex. A stdlib-only scripts/bench-memory.py samples system memory and the largest process throughout, runs each query in its own process (OOM score raised so the runner survives) and records per-query status, wall time and peak RSS, rendered into the job summary. Signed-off-by: Joe Isaacs Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01FDifRZhxcVd5gvt3dsDhu5 --- .github/workflows/tpch-sf1000.yml | 243 +++++++++++++++++++ scripts/bench-memory.py | 386 ++++++++++++++++++++++++++++++ 2 files changed, 629 insertions(+) create mode 100644 .github/workflows/tpch-sf1000.yml create mode 100644 scripts/bench-memory.py diff --git a/.github/workflows/tpch-sf1000.yml b/.github/workflows/tpch-sf1000.yml new file mode 100644 index 00000000000..52a97e1e301 --- /dev/null +++ b/.github/workflows/tpch-sf1000.yml @@ -0,0 +1,243 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright the Vortex contributors + +# Weekly TPC-H at SF=1000 on NVMe for DataFusion and DuckDB over Parquet and Vortex. +# +# At this scale the data (~6B lineitem rows) is larger than the runner's memory, so the run is +# as much a memory check as a benchmark: every query runs in its own process, a query that is +# OOM-killed or times out is recorded and the rest still run, and system memory is sampled +# throughout data generation, conversion and every query. + +name: TPC-H SF1000 + +concurrency: + group: tpch-sf1000 + cancel-in-progress: false + +on: + schedule: + - cron: "17 2 * * 6" # weekly, Saturday + workflow_dispatch: + inputs: + engine: + description: "Engines to run, each on its own machine" + type: choice + options: + - both + - datafusion + - duckdb + default: both + scale_factor: + description: "TPC-H scale factor" + type: string + default: "1000.0" + machine_type: + description: "EC2 instance type for the benchmark machines" + type: string + default: c8gd.metal-24xl + queries: + description: "Space-separated query numbers (empty runs all 22)" + type: string + default: "" + iterations: + description: "Timed iterations per query" + type: string + default: "1" + query_timeout_minutes: + description: "Per-query timeout in minutes" + type: string + default: "45" + +permissions: + contents: read + +env: + SCALE_FACTOR: ${{ inputs.scale_factor || '1000.0' }} + +jobs: + build: + if: github.repository == 'vortex-data/vortex' + timeout-minutes: 60 + runs-on: runs-on=${{ github.run_id }}/runner=bench-dedicated/family=c8gd.8xlarge/image=ubuntu24-full-arm64-pre-v2/tag=build/extras=s3-cache + steps: + - uses: runs-on/action@v2 + with: + sccache: s3 + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + - uses: ./.github/actions/setup-prebuild + with: + repo-token: ${{ secrets.GITHUB_TOKEN }} + enable-sccache: "true" + - name: Build binaries + shell: bash + env: + RUSTFLAGS: "-C target-cpu=native -C force-frame-pointers=yes" + run: | + cargo build --bin data-gen --bin datafusion-bench --bin duckdb-bench --profile release_debug + - name: Install tpchgen-cli + shell: bash + # The in-repo generator writes each table from a single thread, which takes hours for + # SF=1000 lineitem. tpchgen-cli generates in parallel from the same tpchgen-arrow 3.0 + # crate, so the schema matches. + run: cargo install tpchgen-cli --version 3.0.0 --locked --root target/tpchgen-cli + - name: Upload binaries + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7 + with: + name: tpch-sf1000-binaries + path: | + target/release_debug/data-gen + target/release_debug/datafusion-bench + target/release_debug/duckdb-bench + target/release_debug/build/vortex-duckdb-cache/duckdb-lib-*-prebuilt/libduckdb.so + target/tpchgen-cli/bin/tpchgen-cli + + bench: + needs: build + timeout-minutes: 720 + strategy: + fail-fast: false + matrix: + engine: ${{ fromJSON(inputs.engine == 'datafusion' && '["datafusion"]' || inputs.engine == 'duckdb' && '["duckdb"]' || '["datafusion", "duckdb"]') }} + runs-on: runs-on=${{ github.run_id }}/runner=bench-dedicated/family=${{ inputs.machine_type || 'c8gd.metal-24xl' }}/image=ubuntu24-full-arm64-pre-v2/tag=tpch-sf1000-${{ matrix.engine }} + env: + DATA_DIR: vortex-bench/data/tpch/${{ inputs.scale_factor || '1000.0' }} + OUT_DIR: ${{ github.workspace }}/tpch-sf1000-out + PHASE_FILE: ${{ github.workspace }}/tpch-sf1000-out/phase + MONITOR_CSV: ${{ github.workspace }}/tpch-sf1000-out/memory.csv + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + persist-credentials: false + - name: Download binaries + uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8 + with: + name: tpch-sf1000-binaries + path: target/ + - name: Make binaries executable + shell: bash + run: chmod +x target/release_debug/data-gen target/release_debug/*-bench target/tpchgen-cli/bin/tpchgen-cli + + - uses: ./.github/actions/system-info + + - name: Check free disk + shell: bash + # Parquet and Vortex are each about 0.25 GB per scale factor unit, and DuckDB spills + # large joins to disk next to its database. + run: | + set -Eeuo pipefail + mkdir -p "$DATA_DIR" + need_gb=$(python3 -c "print(int(float('$SCALE_FACTOR') * 0.8))") + avail_gb=$(( $(df -B1 --output=avail "$DATA_DIR" | tail -1) / 1024 / 1024 / 1024 )) + echo "Need ~${need_gb} GB free under $DATA_DIR, have ${avail_gb} GB" + if (( avail_gb < need_gb )); then + echo "::error::Not enough disk for SF=$SCALE_FACTOR: ${avail_gb} GB free, need ~${need_gb} GB" + exit 1 + fi + + - name: Start memory monitor + shell: bash + run: | + if [[ -x /usr/bin/time ]]; then + echo "TIME_CMD=/usr/bin/time -v" >> "$GITHUB_ENV" + fi + mkdir -p "$OUT_DIR" + echo idle > "$PHASE_FILE" + nohup python3 scripts/bench-memory.py monitor \ + --output "$MONITOR_CSV" --phase-file "$PHASE_FILE" --interval 2 \ + > "$OUT_DIR/monitor.log" 2>&1 & + echo $! > "$OUT_DIR/monitor.pid" + + - name: Generate Parquet + shell: bash + run: | + set -Eeuo pipefail + echo gen-parquet > "$PHASE_FILE" + ${TIME_CMD:-} target/tpchgen-cli/bin/tpchgen-cli parquet \ + --scale-factor "$SCALE_FACTOR" \ + --output-dir "$DATA_DIR/parquet" \ + --compression 'ZSTD(3)' \ + --row-group-bytes 67108864 \ + --num-threads "$(nproc)" \ + --no-progress + echo idle > "$PHASE_FILE" + du -sh "$DATA_DIR/parquet"/* + + - name: Convert to Vortex + shell: bash + env: + RUST_BACKTRACE: full + run: | + set -Eeuo pipefail + echo convert-vortex > "$PHASE_FILE" + # data-gen skips the Parquet tables that already exist and only converts. + ${TIME_CMD:-} target/release_debug/data-gen tpch \ + --formats parquet,vortex \ + --opt scale-factor="$SCALE_FACTOR" + echo idle > "$PHASE_FILE" + du -sh "$DATA_DIR"/vortex-file-compressed/* + + - name: Run ${{ matrix.engine }} queries + id: queries + shell: bash + env: + RUST_BACKTRACE: "1" + QUERIES: ${{ inputs.queries }} + ITERATIONS: ${{ inputs.iterations || '1' }} + QUERY_TIMEOUT_MINUTES: ${{ inputs.query_timeout_minutes || '45' }} + run: | + set -Eeuo pipefail + # Leave the job time for the summary and uploads once the queries have used 9 hours. + # Word splitting of $QUERIES is intended. + # shellcheck disable=SC2086 + python3 scripts/bench-memory.py run-queries \ + --binary "target/release_debug/${{ matrix.engine }}-bench" \ + --engine "${{ matrix.engine }}" \ + --formats parquet vortex \ + ${QUERIES:+--queries $QUERIES} \ + --iterations "$ITERATIONS" \ + --timeout "$(( QUERY_TIMEOUT_MINUTES * 60 ))" \ + --budget "$(( 9 * 3600 ))" \ + --opt scale-factor="$SCALE_FACTOR" \ + --out-dir "$OUT_DIR" \ + --phase-file "$PHASE_FILE" \ + --fail-on-error + + - name: Stop memory monitor + if: always() + shell: bash + run: | + if [[ -f "$OUT_DIR/monitor.pid" ]]; then + kill "$(cat "$OUT_DIR/monitor.pid")" || true + fi + sudo -n dmesg 2>/dev/null | grep -iE "out of memory|oom-kill|killed process" > "$OUT_DIR/dmesg-oom.txt" || true + df -h "$DATA_DIR" || true + + - name: Summarize memory usage + if: always() + shell: bash + run: | + python3 scripts/bench-memory.py summarize \ + --monitor-csv "$MONITOR_CSV" \ + --queries-jsonl "$OUT_DIR/queries.jsonl" \ + --title "TPC-H SF=$SCALE_FACTOR ${{ matrix.engine }} on ${{ inputs.machine_type || 'c8gd.metal-24xl' }}" \ + | tee -a "$GITHUB_STEP_SUMMARY" + if [[ -s "$OUT_DIR/dmesg-oom.txt" ]]; then + { + echo '
Kernel OOM log' + echo + echo '```' + cat "$OUT_DIR/dmesg-oom.txt" + echo '```' + echo '
' + } >> "$GITHUB_STEP_SUMMARY" + fi + + - name: Upload results + if: always() + uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7 + with: + name: tpch-sf1000-${{ matrix.engine }} + path: tpch-sf1000-out/ + if-no-files-found: ignore diff --git a/scripts/bench-memory.py b/scripts/bench-memory.py new file mode 100644 index 00000000000..51cc901fa12 --- /dev/null +++ b/scripts/bench-memory.py @@ -0,0 +1,386 @@ +#!/usr/bin/env python3 + +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright the Vortex contributors + +"""Memory monitoring for large-scale benchmark runs. + +Three subcommands, all stdlib-only so they run before any Python environment is set up: + +- ``monitor`` samples system memory and the largest process into a CSV until it is terminated. + Each sample is tagged with the contents of a phase file, which the other steps rewrite to + attribute memory to data generation, conversion, or an individual query. +- ``run-queries`` runs every query of a benchmark binary in its own process, so one query + running out of memory or time is recorded and the rest still run. Each query gets its own peak + RSS (from ``wait4``) and a status of ``ok``, ``oom-killed``, ``alloc-failed``, ``timeout``, + ``error`` or ``skipped`` (the time budget ran out first). +- ``summarize`` renders the monitor CSV and query records as GitHub Markdown. +""" + +import argparse +import csv +import json +import os +import signal +import subprocess +import sys +import threading +import time +from collections import defaultdict +from pathlib import Path +from typing import Any + +KIB = 1024 +GIB = 1024**3 + + +def read_meminfo() -> dict[str, int]: + """Return /proc/meminfo in bytes.""" + info = {} + with open("/proc/meminfo") as f: + for line in f: + key, rest = line.split(":", 1) + info[key] = int(rest.split()[0]) * KIB + return info + + +def memory_pressure() -> str: + """Return the 10s average of the "some" memory pressure stall, if the kernel exposes PSI.""" + try: + with open("/proc/pressure/memory") as f: + for line in f: + if line.startswith("some"): + return line.split()[1].split("=", 1)[1] + except OSError: + pass + return "" + + +def largest_process() -> tuple[int, str, int]: + """Return (pid, command name, RSS bytes) of the process with the largest resident set.""" + best = (0, "", 0) + for entry in os.scandir("/proc"): + if not entry.name.isdigit(): + continue + try: + with open(f"/proc/{entry.name}/status") as f: + name, rss = "", 0 + for line in f: + if line.startswith("Name:"): + name = line.split(None, 1)[1].strip() + elif line.startswith("VmRSS:"): + rss = int(line.split()[1]) * KIB + break + except OSError: + continue + if rss > best[2]: + best = (int(entry.name), name, rss) + return best + + +def read_phase(path: Path) -> str: + try: + return path.read_text().strip() or "idle" + except OSError: + return "idle" + + +def cmd_monitor(args: argparse.Namespace) -> None: + stop = threading.Event() + signal.signal(signal.SIGTERM, lambda *_: stop.set()) + signal.signal(signal.SIGINT, lambda *_: stop.set()) + + fields = [ + "timestamp", + "phase", + "mem_total_gib", + "mem_used_gib", + "mem_available_gib", + "page_cache_gib", + "swap_used_gib", + "psi_some_avg10", + "top_pid", + "top_comm", + "top_rss_gib", + ] + with open(args.output, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=fields) + writer.writeheader() + while not stop.is_set(): + mem = read_meminfo() + pid, comm, rss = largest_process() + writer.writerow( + { + "timestamp": f"{time.time():.1f}", + "phase": read_phase(args.phase_file), + "mem_total_gib": f"{mem['MemTotal'] / GIB:.2f}", + "mem_used_gib": f"{(mem['MemTotal'] - mem['MemAvailable']) / GIB:.2f}", + "mem_available_gib": f"{mem['MemAvailable'] / GIB:.2f}", + "page_cache_gib": f"{mem.get('Cached', 0) / GIB:.2f}", + "swap_used_gib": f"{(mem.get('SwapTotal', 0) - mem.get('SwapFree', 0)) / GIB:.2f}", + "psi_some_avg10": memory_pressure(), + "top_pid": pid, + "top_comm": comm, + "top_rss_gib": f"{rss / GIB:.2f}", + } + ) + f.flush() + stop.wait(args.interval) + + +def list_queries(binary: str, benchmark: str, options: list[str]) -> list[int]: + cmd = [binary, benchmark, "--print-queries", *options] + out = subprocess.run(cmd, check=True, capture_output=True, text=True).stdout + return [int(line) for line in out.split() if line.strip()] + + +def kernel_oom_killed(pid: int) -> bool: + """Whether the kernel log records the OOM killer taking this pid.""" + try: + log = subprocess.run(["sudo", "-n", "dmesg"], capture_output=True, text=True, check=False).stdout + except OSError: + return False + return any(f"Killed process {pid} " in line or f"Killed process {pid}," in line for line in log.splitlines()) + + +def classify(returncode: int, timed_out: bool, pid: int, stderr_tail: str) -> str: + if timed_out: + return "timeout" + if returncode == 0: + return "ok" + if "memory allocation of" in stderr_tail or "Out of Memory Error" in stderr_tail: + return "alloc-failed" + if returncode == -signal.SIGKILL and kernel_oom_killed(pid): + return "oom-killed" + if returncode == -signal.SIGKILL: + # Without dmesg access a SIGKILL we did not send is almost always the OOM killer. + return "oom-killed" + return "error" + + +def raise_oom_score() -> None: + # Make the benchmark the OOM killer's first choice so a runaway query cannot take the + # runner agent down with it. Raising the score needs no privileges. + try: + with open("/proc/self/oom_score_adj", "w") as f: + f.write("1000") + except OSError: + pass + + +def run_query(cmd: list[str], stdout_path: Path, stderr_path: Path, timeout_s: int) -> tuple[int, bool, int, int]: + """Run one query process; return (returncode, timed_out, pid, peak RSS bytes).""" + with open(stdout_path, "w") as out, open(stderr_path, "w") as err: + proc = subprocess.Popen(cmd, stdout=out, stderr=err, start_new_session=True, preexec_fn=raise_oom_score) + timed_out = threading.Event() + + def kill() -> None: + timed_out.set() + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + timer = threading.Timer(timeout_s, kill) + timer.start() + try: + # wait4 reports this child's own peak RSS, unlike the cumulative RUSAGE_CHILDREN. + _, status, usage = os.wait4(proc.pid, 0) + finally: + timer.cancel() + proc.returncode = os.waitstatus_to_exitcode(status) + return proc.returncode, timed_out.is_set(), proc.pid, usage.ru_maxrss * KIB + + +def tail(path: Path, lines: int = 20) -> str: + try: + return "\n".join(path.read_text(errors="replace").splitlines()[-lines:]) + except OSError: + return "" + + +def cmd_run_queries(args: argparse.Namespace) -> int: + args.out_dir.mkdir(parents=True, exist_ok=True) + options = [arg for opt in args.opt for arg in ("--opt", opt)] + queries = args.queries or list_queries(args.binary, args.benchmark, options) + records_path = args.out_dir / "queries.jsonl" + deadline = time.monotonic() + args.budget if args.budget else None + failures = 0 + + for fmt in args.formats: + for query in queries: + label = f"{args.engine}:{fmt}:q{query:02d}" + args.phase_file.write_text(label) + stem = args.out_dir / f"{args.engine}-{fmt}-q{query:02d}" + cmd = [ + args.binary, + args.benchmark, + "--formats", + fmt, + "--queries", + str(query), + "--iterations", + str(args.iterations), + "--display-format", + "gh-json", + "--hide-progress-bar", + "--track-memory", + *options, + ] + if args.engine == "duckdb": + cmd.append("--delete-duckdb-database") + + if deadline is not None and time.monotonic() > deadline: + failures += 1 + print(f"::warning title={label} skipped::time budget of {args.budget}s exhausted", flush=True) + with open(records_path, "a") as f: + f.write( + json.dumps({"engine": args.engine, "format": fmt, "query": query, "status": "skipped"}) + "\n" + ) + continue + + subprocess.run(["sync"], check=False) + subprocess.run( + ["sudo", "-n", "sh", "-c", "echo 3 > /proc/sys/vm/drop_caches"], check=False, capture_output=True + ) + + print(f"::group::{label}", flush=True) + start = time.monotonic() + returncode, timed_out, pid, peak_rss = run_query( + cmd, stem.with_suffix(".jsonl"), stem.with_suffix(".log"), args.timeout + ) + elapsed = time.monotonic() - start + stderr_tail = tail(stem.with_suffix(".log")) + status = classify(returncode, timed_out, pid, stderr_tail) + print(stderr_tail) + print("::endgroup::", flush=True) + + record = { + "engine": args.engine, + "format": fmt, + "query": query, + "status": status, + "returncode": returncode, + "wall_s": round(elapsed, 1), + "peak_rss_gib": round(peak_rss / GIB, 2), + } + with open(records_path, "a") as f: + f.write(json.dumps(record) + "\n") + + if status == "ok": + print(f"{label}: ok in {elapsed:.1f}s, peak RSS {peak_rss / GIB:.1f} GiB", flush=True) + else: + failures += 1 + print( + f"::error title={label} {status}::exit {returncode} after {elapsed:.0f}s, " + f"peak RSS {peak_rss / GIB:.1f} GiB", + flush=True, + ) + args.phase_file.write_text("idle") + + return 1 if failures and args.fail_on_error else 0 + + +def peak_by_phase(samples: list[dict[str, str]]) -> dict[str, dict[str, str]]: + peaks: dict[str, dict[str, str]] = {} + for row in samples: + phase = row["phase"] + if phase not in peaks or float(row["mem_used_gib"]) > float(peaks[phase]["mem_used_gib"]): + peaks[phase] = row + return peaks + + +def cmd_summarize(args: argparse.Namespace) -> None: + samples: list[dict[str, str]] = [] + if args.monitor_csv.exists(): + with open(args.monitor_csv) as f: + samples = list(csv.DictReader(f)) + + records: list[dict[str, Any]] = [] + for path in args.queries_jsonl: + if path.exists(): + records.extend(json.loads(line) for line in path.read_text().splitlines() if line.strip()) + + out = [f"## {args.title}", ""] + if samples: + total = samples[0]["mem_total_gib"] + overall = max(samples, key=lambda row: float(row["mem_used_gib"])) + swap = max(float(row["swap_used_gib"]) for row in samples) + out += [ + f"System memory **{total} GiB**; peak used **{overall['mem_used_gib']} GiB** during " + f"`{overall['phase']}` (largest process `{overall['top_comm']}` at {overall['top_rss_gib']} GiB); " + f"peak swap {swap:.2f} GiB.", + "", + ] + + phase_peaks = peak_by_phase(samples) + non_query = {phase: row for phase, row in phase_peaks.items() if phase.count(":") != 2} + if non_query: + out += ["| Phase | Peak used (GiB) | Largest process | Its RSS (GiB) |", "|---|---:|---|---:|"] + for phase, row in non_query.items(): + out.append(f"| `{phase}` | {row['mem_used_gib']} | `{row['top_comm']}` | {row['top_rss_gib']} |") + out.append("") + + if records: + by_status: dict[str, int] = defaultdict(int) + for record in records: + by_status[record["status"]] += 1 + out += [ + "Queries: " + ", ".join(f"{count} {status}" for status, count in sorted(by_status.items())), + "", + "| Engine | Format | Query | Status | Wall (s) | Peak RSS (GiB) | Peak system used (GiB) |", + "|---|---|---:|---|---:|---:|---:|", + ] + for record in records: + phase = f"{record['engine']}:{record['format']}:q{record['query']:02d}" + system_peak = phase_peaks.get(phase, {}).get("mem_used_gib", "") + status = record["status"] if record["status"] == "ok" else f"**{record['status']}**" + out.append( + f"| {record['engine']} | {record['format']} | {record['query']} | {status} | " + f"{record.get('wall_s', '')} | {record.get('peak_rss_gib', '')} | {system_peak} |" + ) + out.append("") + + print("\n".join(out)) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + sub = parser.add_subparsers(dest="command", required=True) + + monitor = sub.add_parser("monitor", help="Sample memory into a CSV until terminated") + monitor.add_argument("--output", type=Path, required=True) + monitor.add_argument("--phase-file", type=Path, required=True) + monitor.add_argument("--interval", type=float, default=2.0) + + run = sub.add_parser("run-queries", help="Run each query in its own process and record its memory") + run.add_argument("--binary", required=True) + run.add_argument("--engine", required=True, choices=["datafusion", "duckdb"]) + run.add_argument("--benchmark", default="tpch") + run.add_argument("--formats", nargs="+", required=True) + run.add_argument("--queries", type=int, nargs="*") + run.add_argument("--iterations", type=int, default=1) + run.add_argument("--timeout", type=int, default=3600, help="Per-query timeout in seconds") + run.add_argument("--budget", type=int, help="Skip the remaining queries after this many seconds") + run.add_argument("--opt", action="append", default=[], help="Benchmark option, e.g. scale-factor=1000.0") + run.add_argument("--out-dir", type=Path, required=True) + run.add_argument("--phase-file", type=Path, required=True) + run.add_argument("--fail-on-error", action="store_true") + + summarize = sub.add_parser("summarize", help="Render the monitor CSV and query records as Markdown") + summarize.add_argument("--monitor-csv", type=Path, required=True) + summarize.add_argument("--queries-jsonl", type=Path, nargs="*", default=[]) + summarize.add_argument("--title", default="Memory usage") + + args = parser.parse_args() + if args.command == "monitor": + cmd_monitor(args) + return 0 + if args.command == "run-queries": + return cmd_run_queries(args) + cmd_summarize(args) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) From 1abe99906acc8375904a990b001e7a3787f0f2bc Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 01:14:48 +0000 Subject: [PATCH 2/5] ci: pass --runner to SF1000 bench queries The bench binaries refuse the default runner ID in CI. Label records with the EC2 instance runs-on provisioned, as vx-bench does. Signed-off-by: Joe Isaacs Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01FDifRZhxcVd5gvt3dsDhu5 --- scripts/bench-memory.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/scripts/bench-memory.py b/scripts/bench-memory.py index 51cc901fa12..90b6d3d73b0 100644 --- a/scripts/bench-memory.py +++ b/scripts/bench-memory.py @@ -225,6 +225,8 @@ def cmd_run_queries(args: argparse.Namespace) -> int: "gh-json", "--hide-progress-bar", "--track-memory", + "--runner", + args.runner, *options, ] if args.engine == "duckdb": @@ -365,6 +367,12 @@ def main() -> int: run.add_argument("--opt", action="append", default=[], help="Benchmark option, e.g. scale-factor=1000.0") run.add_argument("--out-dir", type=Path, required=True) run.add_argument("--phase-file", type=Path, required=True) + run.add_argument( + "--runner", + # Matches vx-bench: label records with the EC2 instance runs-on actually provisioned. + default=f"ec2_{os.environ['RUNS_ON_INSTANCE_TYPE']}" if "RUNS_ON_INSTANCE_TYPE" in os.environ else "local", + help="Benchmark runner ID; the bench binaries refuse the default in CI", + ) run.add_argument("--fail-on-error", action="store_true") summarize = sub.add_parser("summarize", help="Render the monitor CSV and query records as Markdown") From 1c56e7b6b315b58a2cc47cd36d88591934a0c113 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 04:17:41 +0000 Subject: [PATCH 3/5] ci: kill SF1000 queries before memory runs out DataFusion Q18 at SF1000 took the whole runner down (the agent got a shutdown signal) instead of only the query, losing the job's summary and artifacts. Poll available memory while each query runs and kill it below a threshold, and stop systemd-oomd, which kills whole cgroups regardless of the query's raised OOM score. Signed-off-by: Joe Isaacs Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01FDifRZhxcVd5gvt3dsDhu5 --- .github/workflows/tpch-sf1000.yml | 5 +++ scripts/bench-memory.py | 67 ++++++++++++++++++++----------- 2 files changed, 49 insertions(+), 23 deletions(-) diff --git a/.github/workflows/tpch-sf1000.yml b/.github/workflows/tpch-sf1000.yml index 52a97e1e301..f0f7a64e94e 100644 --- a/.github/workflows/tpch-sf1000.yml +++ b/.github/workflows/tpch-sf1000.yml @@ -142,6 +142,10 @@ jobs: if [[ -x /usr/bin/time ]]; then echo "TIME_CMD=/usr/bin/time -v" >> "$GITHUB_ENV" fi + # systemd-oomd kills whole cgroups under memory pressure, which takes the runner agent + # down with a runaway query and loses the job. Leave OOM handling to the query + # watchdog and the kernel, which picks the query first by its raised OOM score. + sudo -n systemctl stop systemd-oomd.service systemd-oomd.socket || true mkdir -p "$OUT_DIR" echo idle > "$PHASE_FILE" nohup python3 scripts/bench-memory.py monitor \ @@ -202,6 +206,7 @@ jobs: --opt scale-factor="$SCALE_FACTOR" \ --out-dir "$OUT_DIR" \ --phase-file "$PHASE_FILE" \ + --min-available-gib 8 \ --fail-on-error - name: Stop memory monitor diff --git a/scripts/bench-memory.py b/scripts/bench-memory.py index 90b6d3d73b0..2d17dbb8078 100644 --- a/scripts/bench-memory.py +++ b/scripts/bench-memory.py @@ -143,9 +143,11 @@ def kernel_oom_killed(pid: int) -> bool: return any(f"Killed process {pid} " in line or f"Killed process {pid}," in line for line in log.splitlines()) -def classify(returncode: int, timed_out: bool, pid: int, stderr_tail: str) -> str: - if timed_out: +def classify(returncode: int, reason: str | None, pid: int, stderr_tail: str) -> str: + if reason == "timeout": return "timeout" + if reason == "memory": + return "oom-killed" if returncode == 0: return "ok" if "memory allocation of" in stderr_tail or "Out of Memory Error" in stderr_tail: @@ -168,28 +170,37 @@ def raise_oom_score() -> None: pass -def run_query(cmd: list[str], stdout_path: Path, stderr_path: Path, timeout_s: int) -> tuple[int, bool, int, int]: - """Run one query process; return (returncode, timed_out, pid, peak RSS bytes).""" +def run_query( + cmd: list[str], stdout_path: Path, stderr_path: Path, timeout_s: int, min_available: int +) -> tuple[int, str | None, int, int]: + """Run one query process; return (returncode, kill reason, pid, peak RSS bytes). + + The query is killed when it times out, or when system available memory drops below + ``min_available``: under heavy pressure systemd-oomd or the runner agent can be killed before + the kernel OOM killer reaches the query, which loses the whole job, so the query goes first. + """ with open(stdout_path, "w") as out, open(stderr_path, "w") as err: proc = subprocess.Popen(cmd, stdout=out, stderr=err, start_new_session=True, preexec_fn=raise_oom_score) - timed_out = threading.Event() - - def kill() -> None: - timed_out.set() - try: - os.killpg(proc.pid, signal.SIGKILL) - except ProcessLookupError: - pass - - timer = threading.Timer(timeout_s, kill) - timer.start() - try: + deadline = time.monotonic() + timeout_s + reason = None + while True: # wait4 reports this child's own peak RSS, unlike the cumulative RUSAGE_CHILDREN. - _, status, usage = os.wait4(proc.pid, 0) - finally: - timer.cancel() + pid, status, usage = os.wait4(proc.pid, os.WNOHANG) + if pid: + break + if reason is None: + if time.monotonic() > deadline: + reason = "timeout" + elif read_meminfo()["MemAvailable"] < min_available: + reason = "memory" + if reason is not None: + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + time.sleep(0.1) proc.returncode = os.waitstatus_to_exitcode(status) - return proc.returncode, timed_out.is_set(), proc.pid, usage.ru_maxrss * KIB + return proc.returncode, reason, proc.pid, usage.ru_maxrss * KIB def tail(path: Path, lines: int = 20) -> str: @@ -248,12 +259,16 @@ def cmd_run_queries(args: argparse.Namespace) -> int: print(f"::group::{label}", flush=True) start = time.monotonic() - returncode, timed_out, pid, peak_rss = run_query( - cmd, stem.with_suffix(".jsonl"), stem.with_suffix(".log"), args.timeout + returncode, reason, pid, peak_rss = run_query( + cmd, + stem.with_suffix(".jsonl"), + stem.with_suffix(".log"), + args.timeout, + int(args.min_available_gib * GIB), ) elapsed = time.monotonic() - start stderr_tail = tail(stem.with_suffix(".log")) - status = classify(returncode, timed_out, pid, stderr_tail) + status = classify(returncode, reason, pid, stderr_tail) print(stderr_tail) print("::endgroup::", flush=True) @@ -363,6 +378,12 @@ def main() -> int: run.add_argument("--queries", type=int, nargs="*") run.add_argument("--iterations", type=int, default=1) run.add_argument("--timeout", type=int, default=3600, help="Per-query timeout in seconds") + run.add_argument( + "--min-available-gib", + type=float, + default=4.0, + help="Kill a query when system available memory falls below this", + ) run.add_argument("--budget", type=int, help="Skip the remaining queries after this many seconds") run.add_argument("--opt", action="append", default=[], help="Benchmark option, e.g. scale-factor=1000.0") run.add_argument("--out-dir", type=Path, required=True) From cc05252218d15ba2f1e1ef26935394eed1659835 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 16:36:45 +0000 Subject: [PATCH 4/5] ci: report SF1000 query failures without failing the job DataFusion does not yet fit Q18 (Parquet and Vortex) or Q21 (Parquet) in memory at SF1000. Record OOM kills, timeouts and errors in the summary and keep the weekly job green so it tracks both engines. Signed-off-by: Joe Isaacs Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01FDifRZhxcVd5gvt3dsDhu5 --- .github/workflows/tpch-sf1000.yml | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/.github/workflows/tpch-sf1000.yml b/.github/workflows/tpch-sf1000.yml index f0f7a64e94e..ae5fd6ade60 100644 --- a/.github/workflows/tpch-sf1000.yml +++ b/.github/workflows/tpch-sf1000.yml @@ -6,7 +6,9 @@ # At this scale the data (~6B lineitem rows) is larger than the runner's memory, so the run is # as much a memory check as a benchmark: every query runs in its own process, a query that is # OOM-killed or times out is recorded and the rest still run, and system memory is sampled -# throughout data generation, conversion and every query. +# throughout data generation, conversion and every query. Query failures are reported in the job +# summary rather than failing the job: DataFusion does not yet fit Q18 (Parquet and Vortex) or Q21 +# (Parquet) in memory at this scale. name: TPC-H SF1000 @@ -206,8 +208,7 @@ jobs: --opt scale-factor="$SCALE_FACTOR" \ --out-dir "$OUT_DIR" \ --phase-file "$PHASE_FILE" \ - --min-available-gib 8 \ - --fail-on-error + --min-available-gib 8 - name: Stop memory monitor if: always() From 5aae9156802529db75c9384842482347c8cba07e Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 16:40:08 +0000 Subject: [PATCH 5/5] Fix DataFusion Vortex scan over-decoding; add a spilling SF1000 pass vortex-datafusion: a Vortex scan runs `concurrency * available_parallelism` splits per stream, and DataFusion opens one stream per partition, one per core by default, so Vortex's default concurrency of 4 multiplied with it. TPC-H SF1000 on 96 cores kept ~37k decoded splits in flight: DataFusion Q1 on Vortex peaked at 115 GiB against Parquet's 37 GiB. Default the scan concurrency to 1 under DataFusion when the table options leave it unset; `vortex.scan_concurrency` still overrides it. At SF1000 this cut Q1 to 32 GiB and Q5 from 146 to 52 GiB, sped Q5 and Q8 from ~18s to ~12s, and let Vortex Q9 complete. datafusion-bench: opt-in DATAFUSION_BENCH_MEMORY_LIMIT_GIB (FairSpillPool behind TrackConsumersPool), DATAFUSION_BENCH_SPILL_DIR and DATAFUSION_BENCH_MAX_SPILL_GIB. Without a pool DataFusion never spills. The weekly workflow runs DataFusion a second time with sort-merge joins (which spill, unlike hash join builds), an 80 GiB memory limit and a 500 GiB spill cap, labelled with run-queries' new --variant. bench-memory.py also reports DataFusion's resources-exhausted error as its own status. Signed-off-by: Joe Isaacs Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01FDifRZhxcVd5gvt3dsDhu5 --- .github/workflows/tpch-sf1000.yml | 50 +++++++++++++++++++--- benchmarks/datafusion-bench/src/lib.rs | 27 ++++++++++++ scripts/bench-memory.py | 21 +++++---- vortex-datafusion/src/persistent/format.rs | 3 +- vortex-datafusion/src/persistent/opener.rs | 14 ++++-- 5 files changed, 96 insertions(+), 19 deletions(-) diff --git a/.github/workflows/tpch-sf1000.yml b/.github/workflows/tpch-sf1000.yml index ae5fd6ade60..6a819e21e2c 100644 --- a/.github/workflows/tpch-sf1000.yml +++ b/.github/workflows/tpch-sf1000.yml @@ -8,7 +8,8 @@ # OOM-killed or times out is recorded and the rest still run, and system memory is sampled # throughout data generation, conversion and every query. Query failures are reported in the job # summary rather than failing the job: DataFusion does not yet fit Q18 (Parquet and Vortex) or Q21 -# (Parquet) in memory at this scale. +# (Parquet) in memory at this scale with hash joins, so it also runs a second pass with sort-merge +# joins and spilling. name: TPC-H SF1000 @@ -125,12 +126,12 @@ jobs: - name: Check free disk shell: bash - # Parquet and Vortex are each about 0.25 GB per scale factor unit, and DuckDB spills - # large joins to disk next to its database. + # Parquet and Vortex are each about 0.25 GB per scale factor unit, DuckDB spills large + # joins next to its database, and DataFusion's spill pass may write up to 500 GiB. run: | set -Eeuo pipefail mkdir -p "$DATA_DIR" - need_gb=$(python3 -c "print(int(float('$SCALE_FACTOR') * 0.8))") + need_gb=$(python3 -c "print(int(float('$SCALE_FACTOR') * 0.6) + 550)") avail_gb=$(( $(df -B1 --output=avail "$DATA_DIR" | tail -1) / 1024 / 1024 / 1024 )) echo "Need ~${need_gb} GB free under $DATA_DIR, have ${avail_gb} GB" if (( avail_gb < need_gb )); then @@ -194,7 +195,6 @@ jobs: QUERY_TIMEOUT_MINUTES: ${{ inputs.query_timeout_minutes || '45' }} run: | set -Eeuo pipefail - # Leave the job time for the summary and uploads once the queries have used 9 hours. # Word splitting of $QUERIES is intended. # shellcheck disable=SC2086 python3 scripts/bench-memory.py run-queries \ @@ -204,12 +204,50 @@ jobs: ${QUERIES:+--queries $QUERIES} \ --iterations "$ITERATIONS" \ --timeout "$(( QUERY_TIMEOUT_MINUTES * 60 ))" \ - --budget "$(( 9 * 3600 ))" \ + --budget "$(( 3 * 3600 ))" \ --opt scale-factor="$SCALE_FACTOR" \ --out-dir "$OUT_DIR" \ --phase-file "$PHASE_FILE" \ --min-available-gib 8 + # DataFusion's hash join keeps its whole build side in memory and cannot spill, so at + # SF1000 the largest joins run out of memory. This second pass uses sort-merge joins, which + # spill, under a memory limit, with spill files on the local NVMe. The limit leaves headroom + # for scan buffers the pool does not track, and DataFusion's default 100 GB spill cap is + # too small for these joins. + - name: Run datafusion queries with spilling + if: ${{ !cancelled() && matrix.engine == 'datafusion' }} + shell: bash + env: + RUST_BACKTRACE: "1" + QUERIES: ${{ inputs.queries }} + ITERATIONS: ${{ inputs.iterations || '1' }} + QUERY_TIMEOUT_MINUTES: ${{ inputs.query_timeout_minutes || '45' }} + DATAFUSION_OPTIMIZER_PREFER_HASH_JOIN: "false" + DATAFUSION_BENCH_MEMORY_LIMIT_GIB: "80" + DATAFUSION_BENCH_MAX_SPILL_GIB: "500" + DATAFUSION_BENCH_SPILL_DIR: ${{ github.workspace }}/df-spill + run: | + set -Eeuo pipefail + mkdir -p "$DATAFUSION_BENCH_SPILL_DIR" + # Word splitting of $QUERIES is intended. + # shellcheck disable=SC2086 + python3 scripts/bench-memory.py run-queries \ + --binary target/release_debug/datafusion-bench \ + --engine datafusion \ + --variant spill \ + --formats parquet vortex \ + ${QUERIES:+--queries $QUERIES} \ + --iterations "$ITERATIONS" \ + --timeout "$(( QUERY_TIMEOUT_MINUTES * 60 ))" \ + --budget "$(( 5 * 3600 ))" \ + --opt scale-factor="$SCALE_FACTOR" \ + --out-dir "$OUT_DIR" \ + --phase-file "$PHASE_FILE" \ + --min-available-gib 8 + du -sh "$DATAFUSION_BENCH_SPILL_DIR" || true + rm -rf "$DATAFUSION_BENCH_SPILL_DIR" + - name: Stop memory monitor if: always() shell: bash diff --git a/benchmarks/datafusion-bench/src/lib.rs b/benchmarks/datafusion-bench/src/lib.rs index 4eec073e7fc..32e8c243c47 100644 --- a/benchmarks/datafusion-bench/src/lib.rs +++ b/benchmarks/datafusion-bench/src/lib.rs @@ -4,6 +4,7 @@ pub mod metrics; pub mod tracer; +use std::num::NonZeroUsize; use std::sync::Arc; use datafusion::datasource::file_format::FileFormat; @@ -12,6 +13,8 @@ use datafusion::datasource::file_format::parquet::ParquetFormat; use datafusion::datasource::provider::DefaultTableFactory; use datafusion::execution::SessionStateBuilder; use datafusion::execution::cache::cache_manager::CacheManagerConfig; +use datafusion::execution::memory_pool::FairSpillPool; +use datafusion::execution::memory_pool::TrackConsumersPool; use datafusion::execution::runtime_env::RuntimeEnvBuilder; use datafusion::prelude::SessionConfig; use datafusion::prelude::SessionContext; @@ -33,6 +36,30 @@ pub fn get_session_context() -> SessionContext { rt_builder = rt_builder.with_cache_manager(CacheManagerConfig::default()); + // Without a memory pool DataFusion never spills, so large joins and aggregations grow until + // the process is killed. A limit makes aggregations and sorts spill to disk, and operators + // that cannot spill (such as hash join builds) fail with a resources-exhausted error. + if let Some(limit_gib) = std::env::var("DATAFUSION_BENCH_MEMORY_LIMIT_GIB") + .ok() + .and_then(|v| v.parse::().ok()) + { + let pool = TrackConsumersPool::new( + FairSpillPool::new(limit_gib << 30), + NonZeroUsize::new(10).expect("non-zero"), + ); + rt_builder = rt_builder.with_memory_pool(Arc::new(pool)); + } + if let Ok(spill_dir) = std::env::var("DATAFUSION_BENCH_SPILL_DIR") { + rt_builder = rt_builder.with_temp_file_path(spill_dir); + } + // DataFusion caps spill files at 100 GB by default, which sort-merge joins exceed at SF1000. + if let Some(max_spill_gib) = std::env::var("DATAFUSION_BENCH_MAX_SPILL_GIB") + .ok() + .and_then(|v| v.parse::().ok()) + { + rt_builder = rt_builder.with_max_temp_directory_size(max_spill_gib << 30); + } + let rt = rt_builder .build_arc() .expect("could not build runtime environment"); diff --git a/scripts/bench-memory.py b/scripts/bench-memory.py index 2d17dbb8078..34776e5e623 100644 --- a/scripts/bench-memory.py +++ b/scripts/bench-memory.py @@ -150,6 +150,8 @@ def classify(returncode: int, reason: str | None, pid: int, stderr_tail: str) -> return "oom-killed" if returncode == 0: return "ok" + if "Resources exhausted" in stderr_tail: + return "resources-exhausted" if "memory allocation of" in stderr_tail or "Out of Memory Error" in stderr_tail: return "alloc-failed" if returncode == -signal.SIGKILL and kernel_oom_killed(pid): @@ -216,13 +218,15 @@ def cmd_run_queries(args: argparse.Namespace) -> int: queries = args.queries or list_queries(args.binary, args.benchmark, options) records_path = args.out_dir / "queries.jsonl" deadline = time.monotonic() + args.budget if args.budget else None + # The variant keeps records and memory phases of differently configured runs of one engine apart. + engine = f"{args.engine}-{args.variant}" if args.variant else args.engine failures = 0 for fmt in args.formats: for query in queries: - label = f"{args.engine}:{fmt}:q{query:02d}" + label = f"{engine}:{fmt}:q{query:02d}" args.phase_file.write_text(label) - stem = args.out_dir / f"{args.engine}-{fmt}-q{query:02d}" + stem = args.out_dir / f"{engine}-{fmt}-q{query:02d}" cmd = [ args.binary, args.benchmark, @@ -247,9 +251,7 @@ def cmd_run_queries(args: argparse.Namespace) -> int: failures += 1 print(f"::warning title={label} skipped::time budget of {args.budget}s exhausted", flush=True) with open(records_path, "a") as f: - f.write( - json.dumps({"engine": args.engine, "format": fmt, "query": query, "status": "skipped"}) + "\n" - ) + f.write(json.dumps({"engine": engine, "format": fmt, "query": query, "status": "skipped"}) + "\n") continue subprocess.run(["sync"], check=False) @@ -267,13 +269,13 @@ def cmd_run_queries(args: argparse.Namespace) -> int: int(args.min_available_gib * GIB), ) elapsed = time.monotonic() - start - stderr_tail = tail(stem.with_suffix(".log")) - status = classify(returncode, reason, pid, stderr_tail) - print(stderr_tail) + # Classify on a long tail: a backtrace can push the error message far from the end. + status = classify(returncode, reason, pid, tail(stem.with_suffix(".log"), 400)) + print(tail(stem.with_suffix(".log"))) print("::endgroup::", flush=True) record = { - "engine": args.engine, + "engine": engine, "format": fmt, "query": query, "status": status, @@ -373,6 +375,7 @@ def main() -> int: run = sub.add_parser("run-queries", help="Run each query in its own process and record its memory") run.add_argument("--binary", required=True) run.add_argument("--engine", required=True, choices=["datafusion", "duckdb"]) + run.add_argument("--variant", help="Label for this configuration, e.g. spill") run.add_argument("--benchmark", default="tpch") run.add_argument("--formats", nargs="+", required=True) run.add_argument("--queries", type=int, nargs="*") diff --git a/vortex-datafusion/src/persistent/format.rs b/vortex-datafusion/src/persistent/format.rs index 3c171789f41..d88bfcc09cc 100644 --- a/vortex-datafusion/src/persistent/format.rs +++ b/vortex-datafusion/src/persistent/format.rs @@ -223,7 +223,8 @@ extensions_options! { /// This does not affect the overall parallelism /// across partitions, which is controlled by DataFusion's execution configuration. /// - /// Leave as `None` to use Vortex's scan default. Override per session + /// Leave as `None` for one split per core per stream: DataFusion already runs one stream + /// per partition, so Vortex's own default would multiply with it. Override per session /// with `SET vortex.scan_concurrency = `, or per table with /// `OPTIONS(scan_concurrency '')`. pub scan_concurrency: Option, default = None diff --git a/vortex-datafusion/src/persistent/opener.rs b/vortex-datafusion/src/persistent/opener.rs index 62455fd7b6b..2ea052f94ac 100644 --- a/vortex-datafusion/src/persistent/opener.rs +++ b/vortex-datafusion/src/persistent/opener.rs @@ -70,6 +70,10 @@ use crate::persistent::cache::CachedVortexMetadata; use crate::persistent::reader::VortexReaderFactory; use crate::persistent::stream::PrunableStream; +/// Per-core scan concurrency when the table options leave it unset. It is lower than the Vortex +/// scan default because DataFusion already runs one scan stream per partition. +const DATAFUSION_DEFAULT_SCAN_CONCURRENCY: usize = 1; + #[derive(Clone)] pub(crate) struct VortexOpener { /// The partition this opener is assigned to. Only used for labeling metrics. @@ -409,9 +413,13 @@ impl FileOpener for VortexOpener { scan_builder = scan_builder.with_limit(limit); } - if let Some(concurrency) = scan_concurrency { - scan_builder = scan_builder.with_concurrency(concurrency); - } + // The scan runs `concurrency * available_parallelism` splits per stream, and DataFusion + // already opens one stream per partition (one per core by default). Vortex's default of + // 4 multiplies with that: at SF1000 on 96 cores it kept ~37k decoded splits in flight + // and Q1 peaked at 115 GiB against Parquet's 37 GiB. One split per core per stream + // keeps every core busy without the second multiplier. + scan_builder = scan_builder + .with_concurrency(scan_concurrency.unwrap_or(DATAFUSION_DEFAULT_SCAN_CONCURRENCY)); // Set before the byte-range translation below, which computes natural splits for // the fields the scan's projection and filter reference.