diff --git a/scripts/benchmark-rowfn.sh b/scripts/benchmark-rowfn.sh new file mode 100755 index 00000000000..7536e37ec87 --- /dev/null +++ b/scripts/benchmark-rowfn.sh @@ -0,0 +1,488 @@ +#!/usr/bin/env bash + +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright the Vortex contributors + +set -Eeu -o pipefail + +script_directory=$(dirname "$(realpath "${BASH_SOURCE[0]}")") + +usage() { + cat >&2 <<'EOF' +Usage: benchmark-rowfn.sh [OPTIONS] + +Options: + --suite NAME Select a preset or benchmark label. Repeatable; defaults to full. + --filter PATTERN Pass a Divan benchmark filter. Repeatable. + --build-only Build and record benchmark executables without measuring. + --measure-only Measure previously recorded benchmark executables without building. + --config NAME repository (16 CGUs/no LTO, default) or primary (1 CGU/fat LTO). + --target-root PATH Parent for reusable baseline and candidate Cargo targets. + --baseline-target PATH Reusable Cargo target for the baseline revision. + --candidate-target PATH + Reusable Cargo target for the candidate revision. + --codegen-units N Override the selected configuration. + --lto VALUE Override LTO with false, thin, or fat. + --rustflags FLAGS Override RUSTFLAGS; defaults to -C target-cpu=native. + --build-jobs N Jobs per concurrent revision build; defaults to 8 and cannot exceed 8. + --bench-cpu N Logical CPU used for every timed process; defaults to 4. + --warm-runs N Warm runs per revision; defaults to 2. + --measured-pairs N Alternating measured pairs; defaults to 7. + --sample-count N Divan sample count; defaults to 100. + --min-time SECONDS Divan minimum time; defaults to 0.25. + --max-time SECONDS Divan maximum time; defaults to 0.5. + --lock-file PATH Global timed-run lock; defaults to /tmp/vortex-rowfn-benchmark.lock. + --list-suites Print presets and benchmark labels, then exit. +EOF +} + +suite_catalog=( + "array-binary_ops|vortex-array|binary_ops|array,numeric,design-a-matrix,full" + "array-compare|vortex-array|compare|array,compare,full" + "array-row_fn_executor|vortex-array|row_fn_executor|array,framework,full" + "array-strict_validity|vortex-array|strict_validity|array,framework,full" + "array-like|vortex-array|like|array,full" + "array-take_filter|vortex-array|take_filter|array,full" + "array-varbinview_compact|vortex-array|varbinview_compact|array,full" + "tensor-l2_norm|vortex-tensor|l2_norm|tensor,full" + "tensor-inner_product|vortex-tensor|inner_product|tensor,full" + "tensor-cosine_similarity|vortex-tensor|cosine_similarity|tensor,full" + "tensor-normalized|vortex-tensor|normalized|tensor,full" + "spatial-binary_predicates|vortex-spatial|binary_predicates|spatial,full" + "spatial-distance|vortex-spatial|distance|spatial,full" + "spatial-envelope|vortex-spatial|envelope|spatial,full" + "spatial-predicate_bbox|vortex-spatial|predicate_bbox|spatial,full" +) + +requested_suites=() +filters=() +run_build=true +run_measure=true +configuration=repository +target_root= +baseline_target_override= +candidate_target_override= +codegen_units_override= +lto_override= +rustflags_override= +build_jobs=8 +bench_cpu=4 +warm_runs=2 +measured_pairs=7 +sample_count=100 +min_time=0.25 +max_time=0.5 +lock_file=/tmp/vortex-rowfn-benchmark.lock + +while [[ $# -gt 0 ]]; do + case $1 in + --suite) requested_suites+=("$2"); shift 2 ;; + --filter) filters+=("$2"); shift 2 ;; + --build-only) + if [[ $run_build == false ]]; then + echo "--build-only and --measure-only are mutually exclusive." >&2 + exit 1 + fi + run_measure=false + shift + ;; + --measure-only) + if [[ $run_measure == false ]]; then + echo "--build-only and --measure-only are mutually exclusive." >&2 + exit 1 + fi + run_build=false + shift + ;; + --config) configuration=$2; shift 2 ;; + --target-root) target_root=$2; shift 2 ;; + --baseline-target) baseline_target_override=$2; shift 2 ;; + --candidate-target) candidate_target_override=$2; shift 2 ;; + --codegen-units) codegen_units_override=$2; shift 2 ;; + --lto) lto_override=$2; shift 2 ;; + --rustflags) rustflags_override=$2; shift 2 ;; + --build-jobs) build_jobs=$2; shift 2 ;; + --bench-cpu) bench_cpu=$2; shift 2 ;; + --warm-runs) warm_runs=$2; shift 2 ;; + --measured-pairs) measured_pairs=$2; shift 2 ;; + --sample-count) sample_count=$2; shift 2 ;; + --min-time) min_time=$2; shift 2 ;; + --max-time) max_time=$2; shift 2 ;; + --lock-file) lock_file=$2; shift 2 ;; + --list-suites) + echo "Presets: full array framework numeric design-a-matrix compare tensor spatial" + printf '%s\n' "${suite_catalog[@]}" | cut -d '|' -f 1 + exit 0 + ;; + -h|--help) usage; exit 0 ;; + --*) echo "Unknown option: $1" >&2; usage; exit 1 ;; + *) break ;; + esac +done + +if [[ $# -ne 3 ]]; then + usage + exit 1 +fi +if [[ $(uname -m) != x86_64 ]]; then + echo "RowFn native performance decisions require an x86_64 host." >&2 + exit 1 +fi +if ((build_jobs < 1 || build_jobs > 8)); then + echo "--build-jobs must be between 1 and 8 so two builds cannot exceed 16 jobs." >&2 + exit 1 +fi +if [[ $run_measure == true ]]; then + command -v flock >/dev/null || { echo "benchmark-rowfn.sh requires flock." >&2; exit 1; } +fi + +baseline=$(realpath "$1") +candidate=$(realpath "$2") +output=$(realpath -m "$3") +if [[ -e $output ]]; then + echo "Output path already exists: $output" >&2 + exit 1 +fi + +case $configuration in + primary) codegen_units=1; lto=fat ;; + repository) codegen_units=16; lto=false ;; + *) echo "Unknown configuration: $configuration" >&2; exit 1 ;; +esac +codegen_units=${codegen_units_override:-$codegen_units} +lto=${lto_override:-$lto} +rustflags=${rustflags_override:--C target-cpu=native} + +if ((${#requested_suites[@]} == 0)); then + requested_suites=(full) +fi +selected_suites=() +declare -A selected_labels=() +for request in "${requested_suites[@]}"; do + matched=false + for entry in "${suite_catalog[@]}"; do + IFS='|' read -r label _ _ groups <<<"$entry" + if [[ $request == "$label" || ,$groups, == *,$request,* ]]; then + matched=true + if [[ -z ${selected_labels[$label]:-} ]]; then + selected_suites+=("$entry") + selected_labels[$label]=1 + fi + fi + done + if [[ $matched == false ]]; then + echo "Unknown suite or benchmark label: $request" >&2 + exit 1 + fi +done + +common_suites=() +skipped_suites=() +for entry in "${selected_suites[@]}"; do + IFS='|' read -r label package bench _ <<<"$entry" + baseline_source="$baseline/$package/benches/$bench.rs" + candidate_source="$candidate/$package/benches/$bench.rs" + + if [[ -f $baseline_source && -f $candidate_source ]]; then + common_suites+=("$entry") + elif [[ -f $baseline_source ]]; then + skipped_suites+=("$label (baseline only)") + elif [[ -f $candidate_source ]]; then + skipped_suites+=("$label (candidate only)") + else + skipped_suites+=("$label (missing from both revisions)") + fi +done +if ((${#common_suites[@]} == 0)); then + echo "No requested benchmark targets exist in both revisions; no comparison is possible." >&2 + printf 'Skipped: %s\n' "${skipped_suites[@]}" >&2 + exit 1 +fi +selected_suites=("${common_suites[@]}") +if ((${#skipped_suites[@]} != 0)); then + printf 'Skipping one-sided benchmark target: %s\n' "${skipped_suites[@]}" >&2 +fi + +common_git_dir=$(git -C "$candidate" rev-parse --path-format=absolute --git-common-dir) +repository_root=$(dirname "$common_git_dir") +if [[ -n $target_root && (-n $baseline_target_override || -n $candidate_target_override) ]]; then + echo "--target-root cannot be combined with revision-specific target paths." >&2 + exit 1 +fi +if [[ -z $target_root ]]; then + target_root="$repository_root/target/rowfn-benchmark/$(basename "$output")" +fi +target_root=$(realpath -m "$target_root") +baseline_target=$(realpath -m "${baseline_target_override:-$target_root/baseline}") +candidate_target=$(realpath -m "${candidate_target_override:-$target_root/candidate}") +if [[ $baseline_target == "$candidate_target" ]]; then + echo "Baseline and candidate must use different Cargo target directories." >&2 + exit 1 +fi + +mkdir -p "$output" +if [[ $run_build == true ]]; then + mkdir -p "$output/build" "$baseline_target" "$candidate_target" +fi +if [[ $run_measure == true ]]; then + mkdir -p "$output/warm" "$output/measured" +fi +parser="$script_directory/rowfn_benchmark.py" + +{ + echo "RowFn benchmark machine record" + echo "Date: $(date --iso-8601=seconds)" + echo "Host: $(hostname)" + echo "Kernel: $(uname -srvmo)" + echo "Benchmark CPU: $bench_cpu" + echo "Configuration: $configuration" + echo "Cargo profile: bench, $codegen_units codegen units, LTO $lto" + echo "RUSTFLAGS: $rustflags" + echo "Warm runs: $warm_runs" + echo "Measured pairs: $measured_pairs" + echo "Divan: TSC timer, $sample_count samples, min $min_time s, max $max_time s" + if ((${#skipped_suites[@]} == 0)); then + echo "Skipped one-sided benchmark targets: none" + else + printf 'Skipped one-sided benchmark target: %s\n' "${skipped_suites[@]}" + fi + echo + echo "Baseline toolchain:" + (cd "$baseline" && rustc -vV && cargo -V) + echo + echo "Candidate toolchain:" + (cd "$candidate" && rustc -vV && cargo -V) + echo + lscpu + echo + rg -m1 '^microcode' /proc/cpuinfo || true + for path in \ + /sys/devices/system/cpu/cpu"$bench_cpu"/cpufreq/scaling_governor \ + /sys/devices/system/cpu/cpu"$bench_cpu"/cpufreq/energy_performance_preference \ + /sys/devices/system/cpu/cpufreq/boost; do + [[ -r $path ]] && echo "$path: $(<"$path")" + done +} >"$output/machine.txt" + +build_revision() { + local worktree=$1 + local target=$2 + local log=$3 + + ( + cd "$worktree" + export CARGO_TARGET_DIR=$target + export CARGO_PROFILE_BENCH_CODEGEN_UNITS=$codegen_units + export CARGO_PROFILE_BENCH_LTO=$lto + export RUSTFLAGS=$rustflags + for package in vortex-array vortex-tensor vortex-spatial; do + local command=(cargo bench --no-run -j "$build_jobs" -p "$package") + local has_bench=false + for entry in "${selected_suites[@]}"; do + IFS='|' read -r _ suite_package bench _ <<<"$entry" + if [[ $suite_package == "$package" ]]; then + command+=(--bench "$bench") + has_bench=true + fi + done + if [[ $has_bench == true ]]; then + "${command[@]}" + fi + done + ) >"$log" 2>&1 +} + +if [[ $run_build == true ]]; then + echo "Building baseline and candidate with $build_jobs jobs each." + build_revision "$baseline" "$baseline_target" "$output/build/baseline.txt" & + baseline_pid=$! + build_revision "$candidate" "$candidate_target" "$output/build/candidate.txt" & + candidate_pid=$! + baseline_status=0 + candidate_status=0 + wait "$baseline_pid" || baseline_status=$? + wait "$candidate_pid" || candidate_status=$? + if [[ $baseline_status -ne 0 || $candidate_status -ne 0 ]]; then + echo "Benchmark build failed; see $output/build/." >&2 + exit 1 + fi +fi + +find_benchmark() { + local target=$1 + local name=$2 + local binary + + binary=$(find "$target/release/deps" -maxdepth 1 -type f -executable -name "$name-*" \ + -printf '%T@ %p\n' | sort -nr | head -n 1 | cut -d ' ' -f 2-) + [[ -n $binary ]] || { echo "Cannot find benchmark $name under $target." >&2; exit 1; } + echo "$binary" +} + +declare -A baseline_binaries=() +declare -A candidate_binaries=() +build_settings=( + --setting "configuration=$configuration" + --setting "codegen_units=$codegen_units" + --setting "lto=$lto" + --setting "rustflags=$rustflags" +) + +record_build() { + local revision=$1 + local worktree=$2 + local target=$3 + local metadata="$target/rowfn-benchmark-build.json" + local arguments=( + record-build + --output "$metadata" + --worktree "$worktree" + --target "$target" + "${build_settings[@]}" + ) + + for entry in "${selected_suites[@]}"; do + IFS='|' read -r label _ bench _ <<<"$entry" + local binary + binary=$(find_benchmark "$target" "$bench") + arguments+=(--binary "$label=$binary") + done + python3 "$parser" "${arguments[@]}" + echo "Recorded $revision build metadata: $metadata" +} + +if [[ $run_build == true ]]; then + record_build baseline "$baseline" "$baseline_target" + record_build candidate "$candidate" "$candidate_target" +fi + +load_binaries() { + local worktree=$1 + local target=$2 + local metadata="$target/rowfn-benchmark-build.json" + local arguments=( + validate-build + --metadata "$metadata" + --worktree "$worktree" + --target "$target" + "${build_settings[@]}" + ) + + for entry in "${selected_suites[@]}"; do + IFS='|' read -r label _ _ _ <<<"$entry" + arguments+=(--suite "$label") + done + python3 "$parser" "${arguments[@]}" +} + +baseline_binary_output=$(load_binaries "$baseline" "$baseline_target") +candidate_binary_output=$(load_binaries "$candidate" "$candidate_target") +mapfile -t baseline_binary_records <<<"$baseline_binary_output" +mapfile -t candidate_binary_records <<<"$candidate_binary_output" +for record in "${baseline_binary_records[@]}"; do + label=${record%%=*} + baseline_binaries[$label]=${record#*=} +done +for record in "${candidate_binary_records[@]}"; do + label=${record%%=*} + candidate_binaries[$label]=${record#*=} +done + +manifest_args=( + manifest + --output "$output/manifest.json" + --machine-record "$output/machine.txt" + --baseline-worktree "$baseline" + --candidate-worktree "$candidate" + --baseline-target "$baseline_target" + --candidate-target "$candidate_target" + --setting "configuration=$configuration" + --setting "codegen_units=$codegen_units" + --setting "lto=$lto" + --setting "rustflags=$rustflags" + --setting "bench_cpu=$bench_cpu" + --setting "warm_runs=$warm_runs" + --setting "measured_pairs=$measured_pairs" + --setting "sample_count=$sample_count" + --setting "min_time=$min_time" + --setting "max_time=$max_time" +) +for filter in "${filters[@]}"; do + manifest_args+=(--filter "$filter") +done +for entry in "${selected_suites[@]}"; do + IFS='|' read -r label _ _ _ <<<"$entry" + manifest_args+=( + --suite "$label" + --baseline-binary "$label=${baseline_binaries[$label]}" + --candidate-binary "$label=${candidate_binaries[$label]}" + ) +done +python3 "$parser" "${manifest_args[@]}" + +if [[ $run_measure == false ]]; then + echo "Build evidence: $output" + echo "Baseline target: $baseline_target" + echo "Candidate target: $candidate_target" + exit 0 +fi + +run_suite() { + local revision=$1 + local label=$2 + local destination=$3 + local binary + local command + + if [[ $revision == baseline ]]; then + binary=${baseline_binaries[$label]} + else + binary=${candidate_binaries[$label]} + fi + command=( + taskset -c "$bench_cpu" "$binary" + --bench --timer tsc --sample-count "$sample_count" + --min-time "$min_time" --max-time "$max_time" --color never + "${filters[@]}" + ) + echo "Running $label ($revision) -> $destination" + "${command[@]}" >"$destination" 2>&1 +} + +echo "Waiting for the global timed benchmark lock: $lock_file" +exec {benchmark_lock}>"$lock_file" +flock "$benchmark_lock" +if pgrep -x cargo >/dev/null || pgrep -x rustc >/dev/null; then + echo "Cargo or rustc is active after acquiring the benchmark lock; refusing to measure." >&2 + exit 1 +fi + +for ((round = 1; round <= warm_runs; round++)); do + for entry in "${selected_suites[@]}"; do + IFS='|' read -r label _ _ _ <<<"$entry" + if ((round % 2 == 1)); then + run_suite baseline "$label" "$output/warm/$label-baseline-$round.txt" + run_suite candidate "$label" "$output/warm/$label-candidate-$round.txt" + else + run_suite candidate "$label" "$output/warm/$label-candidate-$round.txt" + run_suite baseline "$label" "$output/warm/$label-baseline-$round.txt" + fi + done +done + +for ((pair = 1; pair <= measured_pairs; pair++)); do + for entry in "${selected_suites[@]}"; do + IFS='|' read -r label _ _ _ <<<"$entry" + if ((pair % 2 == 1)); then + run_suite baseline "$label" "$output/measured/$label-baseline-$pair.txt" + run_suite candidate "$label" "$output/measured/$label-candidate-$pair.txt" + else + run_suite candidate "$label" "$output/measured/$label-candidate-$pair.txt" + run_suite baseline "$label" "$output/measured/$label-baseline-$pair.txt" + fi + done +done + +python3 "$parser" summarize "$output" +echo "Raw results: $output" +echo "Summary: $output/summary.md" diff --git a/scripts/rowfn_benchmark.py b/scripts/rowfn_benchmark.py new file mode 100755 index 00000000000..35c08918d75 --- /dev/null +++ b/scripts/rowfn_benchmark.py @@ -0,0 +1,482 @@ +#!/usr/bin/env python3 + +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright the Vortex contributors + +"""Capture and summarize evidence from ``benchmark-rowfn.sh`` runs.""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import re +import statistics +import subprocess +from collections.abc import Iterable +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from pathlib import Path + +RESULT_FILE = re.compile(r"^(?P.+)-(?Pbaseline|candidate)-(?P\d+)\.txt$") +TREE_ROW = re.compile(r"^(?P(?:│ | )*)(?:├─ |╰─ )(?P.*)$") +TIMING = re.compile(r"(?P\d+(?:\.\d+)?)\s*(?Pps|ns|µs|us|ms|s)\s*$") +UNIT_TO_NS = { + "ps": 0.001, + "ns": 1.0, + "µs": 1_000.0, + "us": 1_000.0, + "ms": 1_000_000.0, + "s": 1_000_000_000.0, +} + + +@dataclass(frozen=True) +class BenchmarkSummary: + suite: str + benchmark: str + pairs: int + baseline_median_ns: float + candidate_median_ns: float + median_ratio: float + minimum_ratio: float + maximum_ratio: float + ratio_mad: float + + +def run_git(worktree: Path, *args: str, binary: bool = False) -> str | bytes: + """Run one read-only Git command in ``worktree``.""" + + result = subprocess.run( + ["git", "-C", str(worktree), *args], + check=True, + capture_output=True, + text=not binary, + ) + return result.stdout if binary else result.stdout.strip() + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as file: + for chunk in iter(lambda: file.read(1024 * 1024), b""): + digest.update(chunk) + + return digest.hexdigest() + + +def toolchain_record(worktree: Path) -> dict[str, str]: + """Capture the tools selected from a revision's working directory.""" + + def version(*command: str) -> str: + result = subprocess.run( + command, + cwd=worktree, + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + return {"rustc": version("rustc", "-vV"), "cargo": version("cargo", "-V")} + + +def revision_record(worktree: Path, target: Path, binaries: Iterable[str]) -> dict[str, object]: + """Describe the exact revision, dirty patch, targets, and benchmark executables.""" + + status = str(run_git(worktree, "status", "--short")).splitlines() + diff = run_git(worktree, "diff", "--binary", "HEAD", binary=True) + assert isinstance(diff, bytes) + + untracked = run_git(worktree, "ls-files", "--others", "--exclude-standard", "-z", binary=True) + assert isinstance(untracked, bytes) + dirty_digest = hashlib.sha256(diff) + dirty_digest.update(untracked) + for relative_path in filter(None, untracked.decode().split("\0")): + path = worktree / relative_path + if path.is_file(): + dirty_digest.update(relative_path.encode()) + dirty_digest.update(bytes.fromhex(sha256_file(path))) + + executable_records: dict[str, object] = {} + for entry in binaries: + label, separator, raw_path = entry.partition("=") + if not separator: + raise ValueError(f"expected LABEL=PATH for benchmark binary, got {entry!r}") + path = Path(raw_path).resolve() + executable_records[label] = { + "path": str(path), + "sha256": sha256_file(path), + "size": path.stat().st_size, + } + + return { + "worktree": str(worktree.resolve()), + "head": run_git(worktree, "rev-parse", "HEAD"), + "changed_paths": status, + "tracked_diff_sha256": hashlib.sha256(diff).hexdigest(), + "dirty_state_sha256": dirty_digest.hexdigest(), + "target": str(target.resolve()), + "binaries": executable_records, + } + + +def build_identity(worktree: Path, target: Path, settings: dict[str, str]) -> dict[str, object]: + revision = revision_record(worktree, target, []) + revision.pop("binaries") + return { + "settings": settings, + "toolchain": toolchain_record(worktree), + "revision": revision, + } + + +def binary_records(entries: Iterable[str]) -> dict[str, object]: + records: dict[str, object] = {} + for entry in entries: + label, separator, raw_path = entry.partition("=") + if not separator: + raise ValueError(f"expected LABEL=PATH for benchmark binary, got {entry!r}") + path = Path(raw_path).resolve() + records[label] = { + "path": str(path), + "sha256": sha256_file(path), + "size": path.stat().st_size, + } + return records + + +def write_build_record(args: argparse.Namespace) -> None: + output = Path(args.output) + worktree = Path(args.worktree) + target = Path(args.target) + settings = dict(setting.split("=", 1) for setting in args.setting) + identity = build_identity(worktree, target, settings) + binaries = binary_records(args.binary) + + if output.exists(): + previous = json.loads(output.read_text(encoding="utf-8")) + previous_identity = {key: previous.get(key) for key in identity} + if previous_identity == identity: + binaries = {**previous.get("binaries", {}), **binaries} + + record = { + "schema_version": 1, + "created_at": datetime.now(UTC).isoformat(), + **identity, + "binaries": binaries, + } + output.write_text(f"{json.dumps(record, indent=2, sort_keys=True)}\n", encoding="utf-8") + + +def validated_build_binaries(args: argparse.Namespace) -> dict[str, str]: + metadata = Path(args.metadata) + if not metadata.is_file(): + raise ValueError(f"build metadata does not exist: {metadata}") + + record = json.loads(metadata.read_text(encoding="utf-8")) + if record.get("schema_version") != 1: + raise ValueError(f"unsupported build metadata schema in {metadata}") + + settings = dict(setting.split("=", 1) for setting in args.setting) + identity = build_identity(Path(args.worktree), Path(args.target), settings) + mismatches = [key for key in identity if record.get(key) != identity[key]] + if mismatches: + fields = ", ".join(mismatches) + raise ValueError(f"stale benchmark build metadata ({fields} changed): {metadata}") + + binaries = record.get("binaries", {}) + resolved: dict[str, str] = {} + for suite in args.suite: + stored = binaries.get(suite) + if stored is None: + raise ValueError(f"benchmark suite {suite!r} was not recorded in {metadata}") + path = Path(stored["path"]) + if not path.is_file(): + raise ValueError(f"recorded benchmark binary does not exist: {path}") + current = { + "path": str(path.resolve()), + "sha256": sha256_file(path), + "size": path.stat().st_size, + } + if current != stored: + raise ValueError(f"recorded benchmark binary changed: {path}") + resolved[suite] = str(path.resolve()) + + return resolved + + +def validate_build_record(args: argparse.Namespace) -> None: + for suite, path in validated_build_binaries(args).items(): + print(f"{suite}={path}") + + +def write_manifest(args: argparse.Namespace) -> None: + settings = dict(setting.split("=", 1) for setting in args.setting) + manifest = { + "schema_version": 1, + "created_at": datetime.now(UTC).isoformat(), + "settings": settings, + "suites": args.suite, + "filters": args.filter, + "machine_record": str(Path(args.machine_record).resolve()), + "baseline": revision_record( + Path(args.baseline_worktree), + Path(args.baseline_target), + args.baseline_binary, + ), + "candidate": revision_record( + Path(args.candidate_worktree), + Path(args.candidate_target), + args.candidate_binary, + ), + } + output = Path(args.output) + output.write_text(f"{json.dumps(manifest, indent=2, sort_keys=True)}\n", encoding="utf-8") + + +def timing_ns(field: str) -> float: + match = TIMING.search(field.strip()) + if match is None: + raise ValueError(f"cannot parse Divan timing from {field!r}") + + return float(match.group("value")) * UNIT_TO_NS[match.group("unit")] + + +def parse_divan(path: Path) -> dict[str, float]: + """Return benchmark paths and median nanoseconds from one Divan table.""" + + parents: dict[int, str] = {} + timings: dict[str, float] = {} + for line in path.read_text(encoding="utf-8").splitlines(): + fields = re.split(r"\s+│\s+", line) + tree_match = TREE_ROW.match(fields[0]) + if tree_match is None: + continue + + depth = len(tree_match.group("prefix")) // 3 + body = tree_match.group("body").rstrip() + timing_match = TIMING.search(body) + name = body[: timing_match.start()].rstrip() if timing_match else body.strip() + parents = {level: parent for level, parent in parents.items() if level < depth} + + if timing_match is None: + parents[depth] = name + continue + if len(fields) < 3: + raise ValueError(f"timed Divan row has no median column in {path}: {line}") + + components = [parents[level] for level in sorted(parents) if level < depth] + benchmark = "/".join([*components, name]) + if benchmark in timings: + raise ValueError(f"duplicate benchmark {benchmark!r} in {path}") + timings[benchmark] = timing_ns(fields[2]) + + if not timings: + raise ValueError(f"no Divan benchmark timings found in {path}") + + return timings + + +def read_measurements(directory: Path) -> dict[tuple[str, str, int, str], float]: + measurements: dict[tuple[str, str, int, str], float] = {} + for path in sorted(directory.glob("*.txt")): + match = RESULT_FILE.match(path.name) + if match is None: + continue + suite = match.group("suite") + revision = match.group("revision") + pair = int(match.group("pair")) + for benchmark, median_ns in parse_divan(path).items(): + measurements[suite, revision, pair, benchmark] = median_ns + + if not measurements: + raise ValueError(f"no measured result files found in {directory}") + + return measurements + + +def summarize(measurements: dict[tuple[str, str, int, str], float]) -> list[BenchmarkSummary]: + inventories: dict[tuple[str, str], set[str]] = {} + for suite, revision, _, benchmark in measurements: + inventories.setdefault((suite, revision), set()).add(benchmark) + + suites = {suite for suite, _ in inventories} + comparable = { + (suite, benchmark) + for suite in suites + for benchmark in inventories.get((suite, "baseline"), set()) & inventories.get((suite, "candidate"), set()) + } + groups = { + (suite, pair, benchmark) for suite, _, pair, benchmark in measurements if (suite, benchmark) in comparable + } + incomplete = [ + group + for group in groups + if (group[0], "baseline", group[1], group[2]) not in measurements + or (group[0], "candidate", group[1], group[2]) not in measurements + ] + if incomplete: + raise ValueError(f"unpaired benchmark measurements: {sorted(incomplete)!r}") + if not groups: + raise ValueError("unpaired benchmark measurements: no comparable benchmarks") + + by_benchmark: dict[tuple[str, str], list[tuple[float, float]]] = {} + for suite, pair, benchmark in sorted(groups): + baseline = measurements[suite, "baseline", pair, benchmark] + candidate = measurements[suite, "candidate", pair, benchmark] + by_benchmark.setdefault((suite, benchmark), []).append((baseline, candidate)) + + summaries = [] + for (suite, benchmark), pairs in sorted(by_benchmark.items()): + baseline_values = [baseline for baseline, _ in pairs] + candidate_values = [candidate for _, candidate in pairs] + ratios = [candidate / baseline for baseline, candidate in pairs] + median_ratio = statistics.median(ratios) + summaries.append( + BenchmarkSummary( + suite=suite, + benchmark=benchmark, + pairs=len(pairs), + baseline_median_ns=statistics.median(baseline_values), + candidate_median_ns=statistics.median(candidate_values), + median_ratio=median_ratio, + minimum_ratio=min(ratios), + maximum_ratio=max(ratios), + ratio_mad=statistics.median(abs(ratio - median_ratio) for ratio in ratios), + ) + ) + + return summaries + + +def inventory_differences( + measurements: dict[tuple[str, str, int, str], float], +) -> list[tuple[str, str, str]]: + """Return benchmarks that exist in only one revision.""" + + inventories: dict[tuple[str, str], set[str]] = {} + for suite, revision, _, benchmark in measurements: + inventories.setdefault((suite, revision), set()).add(benchmark) + + differences = [] + for suite in sorted({suite for suite, _ in inventories}): + baseline = inventories.get((suite, "baseline"), set()) + candidate = inventories.get((suite, "candidate"), set()) + differences.extend((suite, "baseline only", benchmark) for benchmark in baseline - candidate) + differences.extend((suite, "candidate only", benchmark) for benchmark in candidate - baseline) + + return sorted(differences) + + +def format_ns(value: float) -> str: + for divisor, unit in ((1_000_000_000, "s"), (1_000_000, "ms"), (1_000, "µs")): + if value >= divisor: + return f"{value / divisor:.3f} {unit}" + + return f"{value:.3f} ns" + + +def write_summary( + output_directory: Path, + summaries: list[BenchmarkSummary], + differences: Iterable[tuple[str, str, str]] = (), +) -> None: + csv_path = output_directory / "ratios.csv" + with csv_path.open("w", encoding="utf-8", newline="") as file: + writer = csv.DictWriter(file, fieldnames=list(asdict(summaries[0]))) + writer.writeheader() + writer.writerows(asdict(summary) for summary in summaries) + + markdown = [ + "# RowFn benchmark comparison", + "", + "Ratios are paired candidate/baseline medians. Lower is faster.", + "", + "| Suite | Benchmark | Pairs | Baseline | Candidate | Ratio | Change | MAD |", + "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: |", + ] + for summary in sorted(summaries, key=lambda result: result.median_ratio, reverse=True): + change = (summary.median_ratio - 1.0) * 100.0 + markdown.append( + f"| {summary.suite} | `{summary.benchmark}` | {summary.pairs} " + f"| {format_ns(summary.baseline_median_ns)} " + f"| {format_ns(summary.candidate_median_ns)} " + f"| {summary.median_ratio:.6f} | {change:+.2f}% | {summary.ratio_mad:.6f} |" + ) + differences = list(differences) + if differences: + markdown.extend( + [ + "", + "## Unpaired benchmark inventory", + "", + "These benchmarks were recorded for only one revision and are excluded from ratios.", + "", + ] + ) + markdown.extend(f"- `{suite}/{benchmark}`: {revision}." for suite, revision, benchmark in differences) + markdown.append("") + (output_directory / "summary.md").write_text("\n".join(markdown), encoding="utf-8") + + +def summarize_directory(args: argparse.Namespace) -> None: + output_directory = Path(args.output_directory) + measurements = read_measurements(output_directory / "measured") + summaries = summarize(measurements) + write_summary(output_directory, summaries, inventory_differences(measurements)) + + +def argument_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(required=True) + + manifest = subparsers.add_parser("manifest", help="capture revisions and executable hashes") + manifest.add_argument("--output", required=True) + manifest.add_argument("--machine-record", required=True) + manifest.add_argument("--baseline-worktree", required=True) + manifest.add_argument("--candidate-worktree", required=True) + manifest.add_argument("--baseline-target", required=True) + manifest.add_argument("--candidate-target", required=True) + manifest.add_argument("--setting", action="append", default=[]) + manifest.add_argument("--suite", action="append", default=[]) + manifest.add_argument("--filter", action="append", default=[]) + manifest.add_argument("--baseline-binary", action="append", default=[]) + manifest.add_argument("--candidate-binary", action="append", default=[]) + manifest.set_defaults(function=write_manifest) + + record_build = subparsers.add_parser("record-build", help="record reusable benchmark binaries") + record_build.add_argument("--output", required=True) + record_build.add_argument("--worktree", required=True) + record_build.add_argument("--target", required=True) + record_build.add_argument("--setting", action="append", default=[]) + record_build.add_argument("--binary", action="append", default=[]) + record_build.set_defaults(function=write_build_record) + + validate_build = subparsers.add_parser("validate-build", help="validate a reusable build") + validate_build.add_argument("--metadata", required=True) + validate_build.add_argument("--worktree", required=True) + validate_build.add_argument("--target", required=True) + validate_build.add_argument("--setting", action="append", default=[]) + validate_build.add_argument("--suite", action="append", default=[]) + validate_build.set_defaults(function=validate_build_record) + + summary = subparsers.add_parser("summarize", help="write ratios.csv and summary.md") + summary.add_argument("output_directory") + summary.set_defaults(function=summarize_directory) + + return parser + + +def main() -> None: + parser = argument_parser() + args = parser.parse_args() + try: + args.function(args) + except (OSError, subprocess.CalledProcessError, ValueError) as error: + parser.error(str(error)) + + +if __name__ == "__main__": + main() diff --git a/scripts/tests/test_rowfn_benchmark.py b/scripts/tests/test_rowfn_benchmark.py new file mode 100644 index 00000000000..ace0797c6a3 --- /dev/null +++ b/scripts/tests/test_rowfn_benchmark.py @@ -0,0 +1,197 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright the Vortex contributors + +import importlib.util +import sys +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT = REPO_ROOT / "scripts" / "rowfn_benchmark.py" + + +def load_module(): + spec = importlib.util.spec_from_file_location("rowfn_benchmark", SCRIPT) + assert spec is not None + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def write_divan(path: Path, rows: list[str]) -> None: + path.write_text( + "\n".join( + [ + "Timer precision: 20 ns", + "bench fastest │ slowest │ median │ mean │ samples │ iters", + *rows, + "", + ] + ), + encoding="utf-8", + ) + + +class RowFnBenchmarkTest(unittest.TestCase): + def setUp(self) -> None: + self.module = load_module() + self.temporary_directory = tempfile.TemporaryDirectory() + self.directory = Path(self.temporary_directory.name) + + def tearDown(self) -> None: + self.temporary_directory.cleanup() + + def test_parse_divan_preserves_nested_benchmark_names_and_converts_units(self) -> None: + output = self.directory / "result.txt" + write_divan( + output, + [ + "├─ non_nullable │ │ │ │ │", + "│ ├─ 2 17.18 µs │ 18 µs │ 17.33 µs │ 17.4 µs │ 100 │ 100", + "│ ╰─ 32 6.709 µs │ 8 µs │ 6.829 µs │ 7 µs │ 100 │ 100", + "╰─ nullable │ │ │ │ │", + " ╰─ 2 799.7 ns │ 1 µs │ 979.7 ns │ 986 ns │ 100 │ 100", + ], + ) + + self.assertEqual( + self.module.parse_divan(output), + { + "non_nullable/2": 17_330.0, + "non_nullable/32": 6_829.0, + "nullable/2": 979.7, + }, + ) + + def test_summarize_writes_paired_ratios_and_slowest_first_markdown(self) -> None: + measured = self.directory / "measured" + measured.mkdir() + results = { + "numeric-baseline-1.txt": ["├─ add 10 ns │ 10 ns │ 10 ns │ 10 ns │ 100 │ 100"], + "numeric-candidate-1.txt": ["├─ add 12 ns │ 12 ns │ 12 ns │ 12 ns │ 100 │ 100"], + "numeric-baseline-2.txt": ["├─ add 20 ns │ 20 ns │ 20 ns │ 20 ns │ 100 │ 100"], + "numeric-candidate-2.txt": ["├─ add 18 ns │ 18 ns │ 18 ns │ 18 ns │ 100 │ 100"], + "numeric-baseline-3.txt": ["├─ add 10 ns │ 10 ns │ 10 ns │ 10 ns │ 100 │ 100"], + "numeric-candidate-3.txt": ["├─ add 11 ns │ 11 ns │ 11 ns │ 11 ns │ 100 │ 100"], + "numeric-baseline-4.txt": ["├─ mul 10 ns │ 10 ns │ 10 ns │ 10 ns │ 100 │ 100"], + "numeric-candidate-4.txt": ["├─ mul 9 ns │ 9 ns │ 9 ns │ 9 ns │ 100 │ 100"], + } + for name, rows in results.items(): + write_divan(measured / name, rows) + + summaries = self.module.summarize(self.module.read_measurements(measured)) + self.module.write_summary(self.directory, summaries) + + add = next(summary for summary in summaries if summary.benchmark == "add") + self.assertEqual(add.pairs, 3) + self.assertAlmostEqual(add.median_ratio, 1.1) + self.assertAlmostEqual(add.ratio_mad, 0.1) + + csv_output = (self.directory / "ratios.csv").read_text(encoding="utf-8") + markdown = (self.directory / "summary.md").read_text(encoding="utf-8") + self.assertIn("suite,benchmark,pairs", csv_output) + self.assertLess(markdown.index("`add`"), markdown.index("`mul`")) + + def test_summarize_rejects_unpaired_measurements(self) -> None: + measured = self.directory / "measured" + measured.mkdir() + write_divan( + measured / "numeric-baseline-1.txt", + ["╰─ add 10 ns │ 10 ns │ 10 ns │ 10 ns │ 100 │ 100"], + ) + + with self.assertRaisesRegex(ValueError, "unpaired benchmark measurements"): + self.module.summarize(self.module.read_measurements(measured)) + + def test_summarize_excludes_and_reports_revision_only_benchmarks(self) -> None: + measured = self.directory / "measured" + measured.mkdir() + results = { + "numeric-baseline-1.txt": ["├─ add 10 ns │ 10 ns │ 10 ns │ 10 ns │ 100 │ 100"], + "numeric-candidate-1.txt": [ + "├─ add 11 ns │ 11 ns │ 11 ns │ 11 ns │ 100 │ 100", + "╰─ candidate 5 ns │ 5 ns │ 5 ns │ 5 ns │ 100 │ 100", + ], + } + for name, rows in results.items(): + write_divan(measured / name, rows) + + measurements = self.module.read_measurements(measured) + summaries = self.module.summarize(measurements) + differences = self.module.inventory_differences(measurements) + self.module.write_summary(self.directory, summaries, differences) + + self.assertEqual([summary.benchmark for summary in summaries], ["add"]) + self.assertEqual(differences, [("numeric", "candidate only", "candidate")]) + markdown = (self.directory / "summary.md").read_text(encoding="utf-8") + self.assertIn("`numeric/candidate`: candidate only.", markdown) + + def test_build_record_validates_identity_and_executable(self) -> None: + target = self.directory / "target" + target.mkdir() + binary = target / "binary_ops-123" + binary.write_bytes(b"first binary") + metadata = target / "rowfn-benchmark-build.json" + identity = { + "settings": {"codegen_units": "1", "lto": "fat"}, + "toolchain": {"rustc": "rustc 1.97.1", "cargo": "cargo 1.97.1"}, + "revision": {"head": "abc123", "dirty_state_sha256": "clean"}, + } + arguments = SimpleNamespace( + output=str(metadata), + worktree=str(self.directory), + target=str(target), + setting=["codegen_units=1", "lto=fat"], + binary=[f"numeric={binary}"], + ) + + with mock.patch.object(self.module, "build_identity", return_value=identity): + self.module.write_build_record(arguments) + + validation = SimpleNamespace( + metadata=str(metadata), + worktree=str(self.directory), + target=str(target), + setting=["codegen_units=1", "lto=fat"], + suite=["numeric"], + ) + with mock.patch.object(self.module, "build_identity", return_value=identity): + self.assertEqual( + self.module.validated_build_binaries(validation), + {"numeric": str(binary.resolve())}, + ) + + changed_identities = { + "settings": {**identity, "settings": {"codegen_units": "16", "lto": "false"}}, + "toolchain": { + **identity, + "toolchain": {"rustc": "rustc 1.98.0", "cargo": "cargo 1.98.0"}, + }, + "revision": { + **identity, + "revision": {"head": "def456", "dirty_state_sha256": "changed"}, + }, + } + for field, changed_identity in changed_identities.items(): + with ( + self.subTest(field=field), + mock.patch.object(self.module, "build_identity", return_value=changed_identity), + self.assertRaisesRegex(ValueError, f"{field} changed"), + ): + self.module.validated_build_binaries(validation) + + binary.write_bytes(b"second binary") + with ( + mock.patch.object(self.module, "build_identity", return_value=identity), + self.assertRaisesRegex(ValueError, "binary changed"), + ): + self.module.validated_build_binaries(validation) + + +if __name__ == "__main__": + unittest.main() diff --git a/vortex-array/Cargo.toml b/vortex-array/Cargo.toml index 2af2eacf238..68bd189ef11 100644 --- a/vortex-array/Cargo.toml +++ b/vortex-array/Cargo.toml @@ -134,6 +134,10 @@ harness = false name = "binary_ops" harness = false +[[bench]] +name = "row_fn_executor" +harness = false + [[bench]] name = "kleene_bool" harness = false @@ -213,6 +217,10 @@ harness = false name = "validity_is_valid" harness = false +[[bench]] +name = "strict_validity" +harness = false + [[bench]] name = "dict_unreferenced_mask" harness = false diff --git a/vortex-array/benches/row_fn_executor.rs b/vortex-array/benches/row_fn_executor.rs new file mode 100644 index 00000000000..84a2e029412 --- /dev/null +++ b/vortex-array/benches/row_fn_executor.rs @@ -0,0 +1,332 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Compares owned-output, sink-writing, and hand-written primitive row loops. + +#![expect(clippy::unwrap_used)] + +use std::sync::LazyLock; + +use divan::Bencher; +use vortex_array::ArrayRef; +use vortex_array::Canonical; +use vortex_array::IntoArray; +use vortex_array::VortexSessionExecute; +use vortex_array::array_session; +use vortex_array::arrays::ConstantArray; +use vortex_array::arrays::PrimitiveArray; +use vortex_array::arrays::scalar_fn::ScalarFnFactoryExt; +use vortex_array::dtype::DType; +use vortex_array::dtype::NativePType; +use vortex_array::scalar::Scalar; +use vortex_array::scalar_fn::EmptyOptions; +use vortex_array::scalar_fn::ScalarFnId; +use vortex_array::scalar_fn::unstable::row::InitializedElement; +use vortex_array::scalar_fn::unstable::row::OutputSink; +use vortex_array::scalar_fn::unstable::row::RowFn; +use vortex_array::scalar_fn::unstable::row::RowVisitor; +use vortex_array::scalar_fn::unstable::row::UninitElementSink; +use vortex_array::validity::Validity; +use vortex_buffer::Buffer; +use vortex_buffer::BufferMut; +use vortex_error::VortexError; +use vortex_error::VortexResult; +use vortex_error::vortex_err; +use vortex_session::VortexSession; +use vortex_session::registry::CachedId; + +const ROWS: usize = 65_536; + +static SESSION: LazyLock = LazyLock::new(array_session); + +fn main() { + LazyLock::force(&SESSION); + divan::main(); +} + +#[derive(Clone)] +struct RowWrappingAdd; + +impl RowFn for RowWrappingAdd { + type Options = EmptyOptions; + + const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"]; + const FALLIBLE: bool = false; + + fn id(&self) -> ScalarFnId { + static ID: CachedId = CachedId::new("bench.row_wrapping_add"); + *ID + } + + fn dispatch>( + &self, + _options: &Self::Options, + _args: &[DType], + visitor: V, + ) -> VortexResult { + visitor.visit::<(i64, i64), i64>(|(lhs, rhs)| lhs.wrapping_add(rhs)) + } +} + +#[derive(Clone)] +struct RowCheckedAdd; + +impl RowFn for RowCheckedAdd { + type Options = EmptyOptions; + + const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"]; + const FALLIBLE: bool = true; + + fn id(&self) -> ScalarFnId { + static ID: CachedId = CachedId::new("bench.row_checked_add"); + *ID + } + + fn dispatch>( + &self, + _options: &Self::Options, + _args: &[DType], + visitor: V, + ) -> VortexResult { + visitor.visit_deferred::<(i64, i64), i64, bool>( + |(lhs, rhs)| lhs.overflowing_add(rhs), + |failed| { + if failed { + return Err(checked_add_error()); + } + Ok(()) + }, + ) + } +} + +/// Keep error construction out of the benchmarked success path. +#[cold] +#[inline(never)] +fn checked_add_error() -> VortexError { + vortex_err!("integer overflow in row checked add") +} + +/// A benchmark sink that writes one `i64` per row. +struct I64Sink( + /// The output values written by the row loop. + BufferMut, +); + +// SAFETY: every row is initialized by `BufferMut::zeroed`, and the sink exposes exactly that +// initialized slice. The `()` write token therefore proves no additional invariant. +unsafe impl OutputSink for I64Sink { + type Rows<'a> = &'a mut [i64]; + type Row<'a> = &'a mut i64; + type WriteToken = (); + + fn output_dtype(_options: &Options, _args: &[DType]) -> VortexResult { + Ok(DType::from(i64::PTYPE)) + } + + fn with_capacity(rows: usize) -> VortexResult { + Ok(Self(BufferMut::zeroed(rows))) + } + + fn rows(&mut self) -> Self::Rows<'_> { + self.0.as_mut_slice() + } + + fn row_count(rows: &Self::Rows<'_>) -> usize { + rows.len() + } + + unsafe fn row_unchecked<'a>(rows: &'a mut Self::Rows<'_>, index: usize) -> Self::Row<'a> { + // SAFETY: required by this method's contract. + unsafe { rows.get_unchecked_mut(index) } + } + + unsafe fn finish(self) -> VortexResult { + Ok(PrimitiveArray::new(self.0.freeze(), Validity::NonNullable).into_array()) + } +} + +#[derive(Clone)] +struct RowSinkWrappingAdd; + +impl RowFn for RowSinkWrappingAdd { + type Options = EmptyOptions; + + const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"]; + const FALLIBLE: bool = false; + + fn id(&self) -> ScalarFnId { + static ID: CachedId = CachedId::new("bench.row_sink_wrapping_add"); + *ID + } + + fn dispatch>( + &self, + _options: &Self::Options, + _args: &[DType], + visitor: V, + ) -> VortexResult { + visitor.visit_into::<(i64, i64), I64Sink, _>(|(lhs, rhs), out| { + *out = lhs.wrapping_add(rhs); + }) + } +} + +#[derive(Clone)] +struct RowSinkCheckedAdd; + +impl RowFn for RowSinkCheckedAdd { + type Options = EmptyOptions; + + const ARG_NAMES: &'static [&'static str] = &["lhs", "rhs"]; + const FALLIBLE: bool = true; + + fn id(&self) -> ScalarFnId { + static ID: CachedId = CachedId::new("bench.row_sink_checked_add"); + *ID + } + + fn dispatch>( + &self, + _options: &Self::Options, + _args: &[DType], + visitor: V, + ) -> VortexResult { + visitor.visit_into::<(i64, i64), UninitElementSink, _>( + |(lhs, rhs), output| -> VortexResult { + let value = lhs.checked_add(rhs).ok_or_else(checked_add_error)?; + // SAFETY: `output` is the `UninitElementSink` row supplied for this callback. + Ok(unsafe { InitializedElement::write(output, value) }) + }, + ) + } +} + +fn inputs() -> (ArrayRef, ArrayRef) { + let lhs = (0..ROWS) + .map(|index| index as i64) + .collect::>() + .into_array(); + let rhs = (0..ROWS) + .map(|index| (index % 17) as i64) + .collect::>() + .into_array(); + (lhs, rhs) +} + +fn constant_inputs() -> (ArrayRef, ArrayRef) { + let (lhs, _) = inputs(); + let rhs = ConstantArray::new(Scalar::from(7i64), ROWS).into_array(); + (lhs, rhs) +} + +fn nullable_inputs() -> (ArrayRef, ArrayRef) { + let lhs = PrimitiveArray::new( + (0..ROWS).map(|index| index as i64).collect::>(), + Validity::from_iter((0..ROWS).map(|index| !index.is_multiple_of(5))), + ) + .into_array(); + let rhs = PrimitiveArray::new( + (0..ROWS) + .map(|index| (index % 17) as i64) + .collect::>(), + Validity::from_iter((0..ROWS).map(|index| !index.is_multiple_of(7))), + ) + .into_array(); + (lhs, rhs) +} + +fn bench_row_fn(bencher: Bencher, function: F, make_inputs: fn() -> (ArrayRef, ArrayRef)) +where + F: RowFn, +{ + bencher + .with_inputs(make_inputs) + .bench_local_values(|(lhs, rhs)| { + let mut ctx = SESSION.create_execution_ctx(); + function + .clone() + .try_new_array(ROWS, EmptyOptions, [lhs, rhs]) + .unwrap() + .into_array() + .execute::(&mut ctx) + .unwrap() + }); +} + +#[divan::bench] +fn row_wrapping_add(bencher: Bencher) { + bench_row_fn(bencher, RowWrappingAdd, inputs); +} + +#[divan::bench] +fn row_sink_wrapping_add(bencher: Bencher) { + bench_row_fn(bencher, RowSinkWrappingAdd, inputs); +} + +#[divan::bench] +fn row_sink_wrapping_add_constant(bencher: Bencher) { + bench_row_fn(bencher, RowSinkWrappingAdd, constant_inputs); +} + +#[divan::bench] +fn row_sink_wrapping_add_nullable(bencher: Bencher) { + bench_row_fn(bencher, RowSinkWrappingAdd, nullable_inputs); +} + +#[divan::bench] +fn handrolled_sink_wrapping_add(bencher: Bencher) { + bencher + .with_inputs(inputs) + .bench_local_values(|(lhs, rhs)| { + let mut ctx = SESSION.create_execution_ctx(); + let lhs = lhs + .execute::(&mut ctx) + .unwrap() + .into_buffer::(); + let rhs = rhs + .execute::(&mut ctx) + .unwrap() + .into_buffer::(); + let mut output = BufferMut::zeroed(ROWS); + for ((out, lhs), rhs) in output + .as_mut_slice() + .iter_mut() + .zip(lhs.as_slice()) + .zip(rhs.as_slice()) + { + *out = lhs.wrapping_add(*rhs); + } + PrimitiveArray::new(output.freeze(), Validity::NonNullable).into_array() + }); +} + +#[divan::bench] +fn row_checked_add(bencher: Bencher) { + bench_row_fn(bencher, RowCheckedAdd, inputs); +} + +#[divan::bench] +fn row_wrapping_add_constant(bencher: Bencher) { + bench_row_fn(bencher, RowWrappingAdd, constant_inputs); +} + +#[divan::bench] +fn row_checked_add_constant(bencher: Bencher) { + bench_row_fn(bencher, RowCheckedAdd, constant_inputs); +} + +#[divan::bench] +fn row_checked_add_nullable(bencher: Bencher) { + bench_row_fn(bencher, RowCheckedAdd, nullable_inputs); +} + +#[divan::bench] +fn row_sink_checked_add_nullable(bencher: Bencher) { + bench_row_fn(bencher, RowSinkCheckedAdd, nullable_inputs); +} + +#[divan::bench] +fn row_wrapping_add_nullable(bencher: Bencher) { + bench_row_fn(bencher, RowWrappingAdd, nullable_inputs); +} diff --git a/vortex-array/benches/strict_validity.rs b/vortex-array/benches/strict_validity.rs new file mode 100644 index 00000000000..9a713220fab --- /dev/null +++ b/vortex-array/benches/strict_validity.rs @@ -0,0 +1,216 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Benchmarks how the row lifting applies validity to a dense kernel. +//! +//! Both arms run the same kernel over the same nullable input and differ only in how the output's +//! nulls are restored: +//! +//! - `lazy` is what the lifting behind [`RowFn`] does: conjoin the input validities (itself lazy) +//! and hand the resulting boolean array straight to `mask`. +//! - `eager` is the strategy it replaced: materialize the conjunction into a `Mask`, copy it into a +//! `BitBuffer`, wrap that in a `BoolArray`, and mask with it. +//! +//! `chain` composes three of the same function, which is where a per-call materialization +//! compounds. + +#![expect(clippy::unwrap_used)] +#![expect(clippy::cast_possible_truncation)] + +use std::sync::LazyLock; + +use divan::Bencher; +use vortex_array::ArrayRef; +use vortex_array::Canonical; +use vortex_array::ExecutionCtx; +use vortex_array::IntoArray; +use vortex_array::VortexSessionExecute; +use vortex_array::array_session; +use vortex_array::arrays::BoolArray; +use vortex_array::arrays::PrimitiveArray; +use vortex_array::arrays::scalar_fn::ScalarFnFactoryExt; +use vortex_array::builtins::ArrayBuiltins; +use vortex_array::dtype::DType; +use vortex_array::dtype::PType; +use vortex_array::expr::Expression; +use vortex_array::expr::union_child_validities; +use vortex_array::scalar_fn::Arity; +use vortex_array::scalar_fn::ChildName; +use vortex_array::scalar_fn::EmptyOptions; +use vortex_array::scalar_fn::ExecutionArgs; +use vortex_array::scalar_fn::ScalarFnId; +use vortex_array::scalar_fn::ScalarFnVTable; +use vortex_array::scalar_fn::unstable::row::RowExecution; +use vortex_array::scalar_fn::unstable::row::RowFn; +use vortex_array::scalar_fn::unstable::row::RowVisitor; +use vortex_array::validity::Validity; +use vortex_buffer::Buffer; +use vortex_error::VortexResult; +use vortex_session::VortexSession; +use vortex_session::registry::CachedId; + +const SIZES: &[usize] = &[65_536, 1 << 20]; + +static SESSION: LazyLock = LazyLock::new(array_session); + +fn main() { + LazyLock::force(&SESSION); + divan::main(); +} + +/// The shared kernel: double every lane, ignoring validity. +fn doubled(input: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult { + let input = input.clone().execute::(ctx)?; + let values: Buffer = input + .as_slice::() + .iter() + .map(|value| value.wrapping_mul(2)) + .collect(); + Ok(PrimitiveArray::new(values, Validity::NonNullable).into_array()) +} + +/// Dense row function: the lifting decides how validity is applied. +/// +/// Its [`reduce_encoded`](RowFn::reduce_encoded) answers with [`doubled`] before the row loop is +/// reached, so this arm runs exactly the kernel [`EagerDouble`] runs and the two differ only in +/// validity. The row closure below is what makes it a [`RowFn`] at all, and never executes. +#[derive(Clone)] +struct LazyDouble; + +impl RowFn for LazyDouble { + type Options = EmptyOptions; + + const ARG_NAMES: &'static [&'static str] = &["input"]; + const FALLIBLE: bool = false; + + fn id(&self) -> ScalarFnId { + static ID: CachedId = CachedId::new("bench.lazy_double"); + *ID + } + + fn dispatch>( + &self, + _options: &Self::Options, + _args: &[DType], + visitor: V, + ) -> VortexResult { + visitor.visit::<(i32,), i32>(|(value,)| value.wrapping_mul(2)) + } + + fn reduce_encoded( + &self, + _options: &Self::Options, + args: &[ArrayRef], + ctx: &mut ExecutionCtx, + ) -> VortexResult> { + doubled(&args[0], ctx).map(|output| Some(RowExecution::Output(output))) + } +} + +/// The same function, applying validity the way the adapter used to: materialize a mask first. +#[derive(Clone)] +struct EagerDouble; + +impl ScalarFnVTable for EagerDouble { + type Options = EmptyOptions; + + fn id(&self) -> ScalarFnId { + static ID: CachedId = CachedId::new("bench.eager_double"); + *ID + } + + fn arity(&self, _options: &Self::Options) -> Arity { + Arity::Exact(1) + } + + fn child_name(&self, _options: &Self::Options, _child_idx: usize) -> ChildName { + ChildName::from("input") + } + + fn return_dtype(&self, _options: &Self::Options, args: &[DType]) -> VortexResult { + Ok(DType::Primitive(PType::I32, args[0].nullability())) + } + + fn execute( + &self, + _options: &Self::Options, + args: &dyn ExecutionArgs, + ctx: &mut ExecutionCtx, + ) -> VortexResult { + let input = args.get(0)?; + let valid = input.validity()?.execute_mask(args.row_count(), ctx)?; + let values = doubled(&input, ctx)?; + + if valid.all_true() { + return values.cast(DType::Primitive(PType::I32, input.dtype().nullability())); + } + + let mask = BoolArray::new(valid.to_bit_buffer(), Validity::NonNullable).into_array(); + values.mask(mask) + } + + fn validity( + &self, + _options: &Self::Options, + expression: &Expression, + ) -> VortexResult> { + union_child_validities(expression) + } + + fn is_strict(&self, _options: &Self::Options) -> bool { + true + } + + fn is_fallible(&self, _options: &Self::Options) -> bool { + false + } +} + +/// A nullable i32 column with array-backed validity (~10% nulls). +fn nullable_input(len: usize) -> ArrayRef { + PrimitiveArray::new( + (0..len as i32).collect::>(), + Validity::from_iter((0..len).map(|index| !index.is_multiple_of(10))), + ) + .into_array() +} + +fn bench_depth(bencher: Bencher, function: F, len: usize, depth: usize) +where + F: ScalarFnVTable + Clone, +{ + bencher + .with_inputs(|| nullable_input(len)) + .bench_local_values(|input| { + let mut ctx = SESSION.create_execution_ctx(); + let mut array = input; + for _ in 0..depth { + array = function + .clone() + .try_new_array(len, EmptyOptions, [array]) + .unwrap() + .into_array(); + } + array.execute::(&mut ctx).unwrap() + }); +} + +#[divan::bench(args = SIZES)] +fn lazy(bencher: Bencher, len: usize) { + bench_depth(bencher, LazyDouble, len, 1); +} + +#[divan::bench(args = SIZES)] +fn eager(bencher: Bencher, len: usize) { + bench_depth(bencher, EagerDouble, len, 1); +} + +#[divan::bench(args = SIZES)] +fn lazy_chain(bencher: Bencher, len: usize) { + bench_depth(bencher, LazyDouble, len, 3); +} + +#[divan::bench(args = SIZES)] +fn eager_chain(bencher: Bencher, len: usize) { + bench_depth(bencher, EagerDouble, len, 3); +}