This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new b86c9be5 feat(benchmark): add AWS EC2 setup for the TPC-H harness
(#718)
b86c9be5 is described below
commit b86c9be5755398972af130939ee6dcf143904c25
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Sep 4 16:16:03 2026 -0700
feat(benchmark): add AWS EC2 setup for the TPC-H harness (#718)
---
benchmark/tpch/README.md | 112 +++++++++++++++++
benchmark/tpch/infra/aws/bootstrap.sh | 120 +++++++++++++++++++
benchmark/tpch/infra/aws/sync.sh | 71 +++++++++++
benchmark/tpch/infra/gcp/bootstrap.sh | 6 +-
benchmark/tpch/infra/spark/spark-defaults.conf | 3 +
benchmark/tpch/run.sh | 160 ++++++++++++++++++++++---
benchmark/tpch/src/config.rs | 4 +
benchmark/tpch/src/main.rs | 69 +++++++++--
8 files changed, 516 insertions(+), 29 deletions(-)
diff --git a/benchmark/tpch/README.md b/benchmark/tpch/README.md
index 7f8aa995..f47eb6a5 100644
--- a/benchmark/tpch/README.md
+++ b/benchmark/tpch/README.md
@@ -60,8 +60,18 @@ make bench-tpch QUERIES=1,6,17 SF=10
# Run against cloud-hosted data
make bench-tpch ENGINE=datafusion SF=100 HUDI_DIR=gs://bucket/sf100-hudi
+make bench-tpch ENGINE=datafusion SF=100 HUDI_DIR=s3://bucket/sf100-hudi
```
+Credentials come from the environment for both engines: DataFusion reads the
+`AWS_*` / `GOOGLE_*` / `AZURE_*` variables, and on a cloud VM with an attached
+instance role or service account neither engine needs any variable set. For S3
+outside `us-east-1`, set `AWS_REGION` (`object_store` defaults to `us-east-1`
+when the region is neither configured nor derivable from the URL).
+
+The two sections below cover running on a cloud VM, one per provider; they are
+alternatives, so follow whichever matches your setup.
+
## GCP VM
### One-time setup
@@ -119,3 +129,105 @@ gcloud compute instances start bench-vm
--zone=us-central1-a
The bootstrap script only runs once (guarded by a sentinel file),
so restarting the VM is fast.
+
+## AWS EC2
+
+### One-time setup
+
+Launch an Amazon Linux 2023 instance with an instance profile that grants
+S3 access to the benchmark bucket and includes the
+`AmazonSSMManagedInstanceCore` policy (for Session Manager access). The
+instance needs outbound internet access for the package, crate, PyPI and
+Maven downloads.
+
+Get the repo onto the instance first. A fresh Amazon Linux 2023 image has no
+`git` (bootstrap is what installs it), so either sync from a local checkout,
+which needs nothing preinstalled on the instance:
+
+```bash
+bash benchmark/tpch/infra/aws/sync.sh i-0123456789abcdef0 us-west-2
~/.ssh/key.pem
+```
+
+or install `git` on the instance and clone there:
+
+```bash
+sudo dnf install -y git && git clone <repo-url> ~/hudi-rs
+```
+
+Then run the bootstrap script as the login user (not as user data, which would
+install the toolchain into root's home). It installs Rust, Java, PySpark, and
+the S3A connector, and mounts a local NVMe instance store at `/mnt/nvme` when
+the instance type has one (e.g. `r8gd.4xlarge` for SF100). The packages are
+architecture-neutral, so Graviton and x86 instances both work.
+
+```bash
+cd ~/hudi-rs
+bash benchmark/tpch/infra/aws/bootstrap.sh
+
+exec bash -l # pick up the variables it appended
+echo "$SPARK_HOME" "$AWS_REGION" # both must be non-empty
+```
+
+The new shell matters: the one that ran bootstrap does not yet have those
+variables, and an unset `AWS_REGION` sends DataFusion to `us-east-1` rather
+than the bucket's region.
+
+Both engines resolve instance-profile credentials automatically: DataFusion
+through `object_store`, Spark through the S3A default credential chain, so no
+keys need to be configured. The region is the one thing neither infers, which
+is why bootstrap persists it from the instance metadata.
+
+### Choose where the generated data lands
+
+`generate` writes to `benchmark/tpch/data`, so size the root volume for the
+scale factor (SF100 parquet is roughly 40 GB). Instance types ending in `d`
+(`r8gd`, `r7gd`, `m7gd`) carry a local NVMe instance store, which bootstrap
+mounts at `/mnt/nvme`; where one is present, keeping the generated parquet on
+it is faster than EBS and leaves the root volume alone:
+
+```bash
+# optional, and only before generating
+mountpoint -q /mnt/nvme && ln -sfn /mnt/nvme/tpch-data benchmark/tpch/data
+```
+
+Instance types without a local disk work unchanged; everything simply stays on
+the root volume, so provision it accordingly.
+
+### Run benchmarks on the instance
+
+Same commands as on GCP, with `s3://` data URLs. `create-tables` writes the
+Hudi tables straight to the bucket, so the instance only holds the generated
+parquet. Run these under `tmux`: at SF100 they take hours, and a dropped SSM
+session would otherwise kill them.
+
+```bash
+S3=s3://bucket/sf100-hudi
+
+benchmark/tpch/run.sh generate --scale-factor 100
+# reuses the tables if they are already there; --recreate rebuilds them
+benchmark/tpch/run.sh create-tables --scale-factor 100 --hudi-dir $S3
+
+benchmark/tpch/run.sh bench-datafusion --scale-factor 100 --hudi-dir $S3 \
+ --output-dir benchmark/tpch/results
+benchmark/tpch/run.sh bench-spark --scale-factor 100 --hudi-dir $S3 \
+ --output-dir benchmark/tpch/results
+
+benchmark/tpch/run.sh compare --scale-factor 100 --engines datafusion,spark
+```
+
+`--output-dir` is what makes the `bench-*` commands persist their results, and
+`compare` reads them from there, so omitting it leaves nothing to compare once
+the runs have finished. To check credentials, region and the S3 path before
+committing hours to a full run, benchmark a couple of queries first with
+`--queries 1,6`.
+
+Re-run `sync.sh` from your local checkout whenever local code changes; it
+rebuilds the binary on the instance.
+
+### Stop/start the instance
+
+Stopping wipes the NVMe instance store, taking the mount, any generated data on
+it, and the shuffle directory with it. Re-run the bootstrap script after
+starting again to restore them, then regenerate. The package installs are
+sentinel-guarded and skipped, so that is fast. Tables already written to the
+bucket are unaffected, and `create-tables` reuses them.
diff --git a/benchmark/tpch/infra/aws/bootstrap.sh
b/benchmark/tpch/infra/aws/bootstrap.sh
new file mode 100755
index 00000000..eea872c9
--- /dev/null
+++ b/benchmark/tpch/infra/aws/bootstrap.sh
@@ -0,0 +1,120 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+# Bootstrap an AWS EC2 Amazon Linux 2023 instance with system-level
+# dependencies for TPC-H benchmarks. This script is repo-agnostic.
+#
+# Run it as the login user, not as EC2 user data: it installs into $HOME
+# (rustup, pip --user, SPARK_HOME in .bashrc) and takes ownership of the
+# instance-store mount, so a root-owned install from user data would leave the
+# login user without a toolchain.
+#
+# Re-run it after a stop/start: the instance store is wiped, and the mount
+# section below is deliberately outside the sentinel guard so it is restored.
+#
+# Prerequisites:
+# - Instance profile granting S3 access to the benchmark bucket
+# (object_store and the S3A connector both resolve instance credentials
+# automatically; no keys need to be configured)
+# - Outbound internet access, for the package, crate, PyPI and Maven
+# downloads here and for the Hudi Spark bundle that run.sh fetches
+#
+set -euo pipefail
+
+if [[ $EUID -eq 0 ]]; then
+ echo "Error: run this as the login user (e.g. ec2-user), not as root." >&2
+ exit 1
+fi
+
+# Spark local dirs for shuffle and event logs. This section runs on every
+# invocation (not sentinel-guarded): the instance store is wiped on every
+# stop/start, so re-running the script restores the mount.
+# If the instance has a local NVMe instance store, format and mount it at
+# /mnt/nvme and put the shuffle dir there for faster I/O.
+NVME_DEV=$(lsblk -dno NAME,MODEL | awk '/Instance Storage/ {print "/dev/" $1;
exit}')
+if [[ -n "${NVME_DEV:-}" ]]; then
+ if ! sudo blkid "$NVME_DEV" >/dev/null 2>&1; then
+ sudo mkfs -t xfs -q "$NVME_DEV"
+ fi
+ sudo mkdir -p /mnt/nvme
+ mountpoint -q /mnt/nvme || sudo mount "$NVME_DEV" /mnt/nvme
+ sudo chown "$(id -un):$(id -gn)" /mnt/nvme
+ mkdir -p /mnt/nvme/spark-local
+ ln -sfn /mnt/nvme/spark-local /tmp/spark-local
+ # Target for an optional benchmark/tpch/data symlink. Created here so that
+ # linking to it cannot leave a dangling symlink, which surfaces much later as
+ # an EEXIST from the generator's create_dir_all.
+ mkdir -p /mnt/nvme/tpch-data
+else
+ # No instance store on this type; shuffle stays on the root volume. Clear any
+ # symlink left by a run on an instance that had one, which would otherwise
+ # point at an empty mount point.
+ [[ -L /tmp/spark-local ]] && rm -f /tmp/spark-local
+ mkdir -p /tmp/spark-local
+fi
+mkdir -p /tmp/spark-events
+
+SENTINEL="/var/lib/bootstrap-done"
+[[ -f "$SENTINEL" ]] && exit 0
+
+# System packages
+# clang-devel supplies the libclang that bindgen loads: hudi enables
+# spill-rocksdb by default, and rocksdb generates its bindings at build time.
+sudo dnf install -y gcc gcc-c++ clang-devel make git pkgconfig openssl-devel \
+ protobuf-compiler protobuf-devel java-17-amazon-corretto-headless \
+ python3-pip rsync tmux sysstat
+
+# Rust
+curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
+. "$HOME/.cargo/env"
+
+# arrow/DataFusion kernels rely on LLVM auto-vectorization, and rustc targets a
+# conservative baseline for each architecture (SSE2 on x86-64, plain NEON on
+# aarch64); target the local CPU so the wider vector units are actually used.
+mkdir -p "$HOME/.cargo"
+cat > "$HOME/.cargo/config.toml" <<'EOF'
+[build]
+rustflags = ["-C", "target-cpu=native"]
+EOF
+
+# PySpark
+pip3 install --user pyspark==3.5.8
+
+# S3A connector jars matching the Hadoop 3.3.4 client PySpark 3.5.x bundles
+SPARK_HOME=$(python3 -c "import pyspark; print(pyspark.__path__[0])")
+mkdir -p "$SPARK_HOME/conf" "$SPARK_HOME/jars"
+curl -fL -o "$SPARK_HOME/jars/hadoop-aws-3.3.4.jar" \
+
"https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3.4/hadoop-aws-3.3.4.jar";
+curl -fL -o "$SPARK_HOME/jars/aws-java-sdk-bundle-1.12.262.jar" \
+
"https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.12.262/aws-java-sdk-bundle-1.12.262.jar";
+
+# Persist SPARK_HOME for future sessions, plus the instance's region:
+# object_store takes no region from the instance metadata and falls back to
+# us-east-1, which misroutes reads of a bucket in any other region.
+AWS_REGION=$(curl -fsS -H "X-aws-ec2-metadata-token: $(curl -fsS -X PUT \
+ -H 'X-aws-ec2-metadata-token-ttl-seconds: 60' \
+ http://169.254.169.254/latest/api/token)" \
+ http://169.254.169.254/latest/meta-data/placement/region)
+{
+ echo "export SPARK_HOME=$SPARK_HOME"
+ echo "export AWS_REGION=$AWS_REGION"
+} >> ~/.bashrc
+
+sudo touch "$SENTINEL"
+echo "Bootstrap complete."
diff --git a/benchmark/tpch/infra/aws/sync.sh b/benchmark/tpch/infra/aws/sync.sh
new file mode 100755
index 00000000..2a5ce77f
--- /dev/null
+++ b/benchmark/tpch/infra/aws/sync.sh
@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+# Sync local code to an AWS EC2 instance over SSM and rebuild the benchmark
+# binary. Connects through Session Manager, so the instance needs no public
+# IP or open SSH port; it does need the AmazonSSMManagedInstanceCore policy
+# and an SSH key authorized for the login user (e.g. the instance's key pair).
+#
+# Requires locally: AWS CLI with the Session Manager plugin, and credentials
+# for the instance's account.
+#
+# Usage:
+# bash benchmark/tpch/infra/aws/sync.sh <instance-id> <region> [ssh-key-path]
+#
+# Example:
+# bash benchmark/tpch/infra/aws/sync.sh i-0123456789abcdef0 us-west-2
~/.ssh/bench.pem
+#
+set -euo pipefail
+
+INSTANCE_ID="${1:?Usage: sync.sh <instance-id> <region> [ssh-key-path]}"
+REGION="${2:?Usage: sync.sh <instance-id> <region> [ssh-key-path]}"
+SSH_KEY="${3:-}"
+SSH_USER="${SSH_USER:-ec2-user}"
+
+REPO_ROOT="$(cd "$(dirname "$0")/../../../.." && pwd)"
+
+PROXY_CMD="aws ssm start-session --target %h --region $REGION --document-name
AWS-StartSSHSession --parameters portNumber=%p"
+
+SSH_OPTS=(-o StrictHostKeyChecking=no -o "ProxyCommand=$PROXY_CMD")
+if [[ -n "$SSH_KEY" ]]; then
+ SSH_OPTS+=(-i "$SSH_KEY")
+fi
+
+RSYNC_SSH="ssh -o StrictHostKeyChecking=no -o 'ProxyCommand=$PROXY_CMD'"
+if [[ -n "$SSH_KEY" ]]; then
+ RSYNC_SSH+=" -i '$SSH_KEY'"
+fi
+
+echo "==> Syncing code to $INSTANCE_ID..."
+rsync -az --progress \
+ --exclude='target/' \
+ --exclude='.git/' \
+ --exclude='.context/' \
+ -e "$RSYNC_SSH" \
+ "$REPO_ROOT/" "$SSH_USER@$INSTANCE_ID":~/hudi-rs/
+
+echo "==> Building on the instance..."
+ssh "${SSH_OPTS[@]}" "$SSH_USER@$INSTANCE_ID" \
+ "cd ~/hudi-rs && . \$HOME/.cargo/env && cargo build -p tpch --release"
+
+echo ""
+echo "Ready. Connect with:"
+printf ' ssh'
+printf ' %q' "${SSH_OPTS[@]}"
+printf ' %s\n' "$SSH_USER@$INSTANCE_ID"
diff --git a/benchmark/tpch/infra/gcp/bootstrap.sh
b/benchmark/tpch/infra/gcp/bootstrap.sh
index 874b5692..2a9f52bb 100755
--- a/benchmark/tpch/infra/gcp/bootstrap.sh
+++ b/benchmark/tpch/infra/gcp/bootstrap.sh
@@ -31,8 +31,10 @@ SENTINEL="/var/lib/bootstrap-done"
# System packages
sudo apt-get update
-sudo apt-get install -y build-essential protobuf-compiler pkg-config git curl \
- openjdk-17-jdk-headless python3-pip sysstat tmux glances
+# libclang-dev is what bindgen loads: hudi enables spill-rocksdb by default,
+# and rocksdb generates its bindings at build time.
+sudo apt-get install -y build-essential clang libclang-dev protobuf-compiler \
+ pkg-config git curl openjdk-17-jdk-headless python3-pip sysstat tmux glances
# Rust
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
diff --git a/benchmark/tpch/infra/spark/spark-defaults.conf
b/benchmark/tpch/infra/spark/spark-defaults.conf
index c1d27902..3564c601 100644
--- a/benchmark/tpch/infra/spark/spark-defaults.conf
+++ b/benchmark/tpch/infra/spark/spark-defaults.conf
@@ -26,3 +26,6 @@ spark.local.dir /tmp/spark-local
spark.eventLog.dir /tmp/spark-events
spark.hadoop.fs.gs.impl
com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem
spark.hadoop.google.cloud.auth.type COMPUTE_ENGINE
+# Hadoop 3 dropped the s3:// filesystem, so route it to S3A as well: the
+# harness accepts both s3:// and s3a:// URLs, and object_store maps them alike.
+spark.hadoop.fs.s3.impl
org.apache.hadoop.fs.s3a.S3AFileSystem
diff --git a/benchmark/tpch/run.sh b/benchmark/tpch/run.sh
index 3bd1e454..6a861a4e 100755
--- a/benchmark/tpch/run.sh
+++ b/benchmark/tpch/run.sh
@@ -25,6 +25,73 @@ REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
DEFAULT_SCALE_FACTOR=1
TPCH_BIN="$REPO_ROOT/target/release/tpch"
+HUDI_SPARK_BUNDLE="org.apache.hudi:hudi-spark3.5-bundle_2.12:1.1.1"
+
+# Record what the numbers were produced on. A timing is only comparable against
+# another run on the same hardware, build and data, and none of that is
+# recoverable from the results afterwards.
+write_env_report() {
+ local out_file="$1"
+ local sf="$2"
+ local data_dir="$3"
+
+ local cpu_model cpu_count mem_total
+ if [ -r /proc/cpuinfo ]; then
+ cpu_model=$(awk -F': ' '/^model name|^Model name/ {print $2; exit}'
/proc/cpuinfo)
+ [ -z "$cpu_model" ] && cpu_model=$(lscpu 2>/dev/null | awk -F': +'
'/^Model name/ {print $2; exit}')
+ cpu_count=$(nproc)
+ mem_total=$(awk '/^MemTotal/ {printf "%.0f GiB", $2/1048576}'
/proc/meminfo)
+ else
+ cpu_model=$(sysctl -n machdep.cpu.brand_string 2>/dev/null)
+ cpu_count=$(sysctl -n hw.ncpu 2>/dev/null)
+ mem_total=$(sysctl -n hw.memsize 2>/dev/null | awk '{printf "%.0f GiB",
$1/1073741824}')
+ fi
+
+ # Instance identity, when this is an EC2 box. IMDSv2, and short timeouts so a
+ # non-EC2 machine costs nothing.
+ local instance_type="" instance_region=""
+ local imds_token
+ imds_token=$(curl -fsS --max-time 1 -X PUT \
+ -H 'X-aws-ec2-metadata-token-ttl-seconds: 60' \
+ http://169.254.169.254/latest/api/token 2>/dev/null) || true
+ if [ -n "$imds_token" ]; then
+ instance_type=$(curl -fsS --max-time 1 -H "X-aws-ec2-metadata-token:
$imds_token" \
+ http://169.254.169.254/latest/meta-data/instance-type 2>/dev/null) ||
true
+ instance_region=$(curl -fsS --max-time 1 -H "X-aws-ec2-metadata-token:
$imds_token" \
+ http://169.254.169.254/latest/meta-data/placement/region 2>/dev/null) ||
true
+ fi
+
+ local git_rev git_dirty=""
+ git_rev=$(git -C "$REPO_ROOT" rev-parse --short HEAD 2>/dev/null || echo
unknown)
+ git -C "$REPO_ROOT" diff --quiet 2>/dev/null || git_dirty=" (modified)"
+
+ # Where the data physically sits: an EBS root and an instance store give very
+ # different read numbers for the same command.
+ local data_backing="n/a (cloud storage)"
+ if ! is_cloud_url "$data_dir"; then
+ data_backing=$(df -h "$data_dir" 2>/dev/null | awk 'NR==2 {print $1" "$2}')
+ fi
+
+ {
+ echo "# Benchmark environment"
+ echo "captured: $(date -u '+%Y-%m-%dT%H:%M:%SZ')"
+ echo "scale factor: $sf"
+ echo "data location: $data_dir"
+ echo "data backing: $data_backing"
+ [ -n "$instance_type" ] && echo "instance: $instance_type
($instance_region)"
+ echo "cpu: ${cpu_model:-unknown} x ${cpu_count:-?}"
+ echo "memory: ${mem_total:-unknown}"
+ echo "os: $(uname -srm)"
+ echo "hudi-rs commit: ${git_rev}${git_dirty}"
+ echo "rustc: $(rustc --version 2>/dev/null || echo unknown)"
+ echo "RUSTFLAGS: ${RUSTFLAGS:-<from cargo config>}"
+ echo "cargo build: release"
+ echo "java: $(java -version 2>&1 | head -1)"
+ echo "spark: $("$SPARK_HOME/bin/spark-submit" --version 2>&1 |
awk '/version/ {print $NF; exit}')"
+ echo "hudi bundle: $HUDI_SPARK_BUNDLE"
+ echo "config: config/sf$sf.yaml"
+ } > "$out_file"
+}
build_tpch() {
echo "Building TPC-H tool..."
@@ -51,6 +118,8 @@ setup_spark() {
fi
echo "Configuring Spark at $SPARK_HOME..."
+ # A pip-installed PySpark ships without a conf directory.
+ mkdir -p "$SPARK_HOME/conf"
cp "$SCRIPT_DIR/infra/spark/spark-defaults.conf"
"$SPARK_HOME/conf/spark-defaults.conf"
cp "$SCRIPT_DIR/infra/spark/log4j2.properties"
"$SPARK_HOME/conf/log4j2.properties"
}
@@ -62,6 +131,28 @@ is_cloud_url() {
esac
}
+# A data dir symlinked to an instance store dangles once a stop/start wipes it.
+# Name that, because the failure it otherwise produces is create_dir_all
+# reporting EEXIST for a directory that does not exist.
+require_usable_data_dir() {
+ local data_root="$SCRIPT_DIR/data"
+ if [ -L "$data_root" ] && [ ! -d "$data_root" ]; then
+ echo "Error: $data_root links to $(readlink "$data_root"), which is
missing." >&2
+ echo "Recreate that directory, or remove the symlink to use local
storage." >&2
+ exit 1
+ fi
+}
+
+# Fail with the missing table names rather than letting the engine report a
+# missing path once the run is already under way.
+require_hudi_tables() {
+ local hudi_dir="$1"
+ if ! "$TPCH_BIN" check-tables --hudi-base "$hudi_dir"; then
+ echo "Error: run 'create-tables' first, or point --hudi-dir at existing
tables." >&2
+ exit 1
+ fi
+}
+
usage() {
cat <<EOF
Usage: $0 <command> [options]
@@ -76,8 +167,9 @@ Commands:
Options (per command):
--scale-factor N TPC-H scale factor [all commands] (default:
$DEFAULT_SCALE_FACTOR)
--format F Table format: hudi or parquet [bench-*, compare] (default:
auto)
- --hudi-dir D Hudi data directory or cloud URL [bench-*] (default:
data/sf{N}-hudi)
- --parquet-dir D Parquet data directory or cloud URL [bench-*] (default:
data/sf{N}-parquet)
+ --recreate Rebuild tables that already exist [create-tables]
(default: reuse them)
+ --hudi-dir D Hudi data directory or cloud URL [create-tables, bench-*]
(default: data/sf{N}-hudi)
+ --parquet-dir D Parquet data directory or cloud URL [create-tables,
bench-*] (default: data/sf{N}-parquet)
--queries Q Comma-separated query numbers [bench-*] (default: all 22)
--iterations N Number of measured iterations per query [bench-*] (from
config)
--warmup N Number of unmeasured warmup iterations per query [bench-*]
(from config)
@@ -87,6 +179,7 @@ Options (per command):
Examples:
$0 generate --scale-factor 1
$0 create-tables --scale-factor 1
+ $0 create-tables --scale-factor 100 --hudi-dir s3://bucket/sf100-hudi
$0 bench-spark --scale-factor 1 --queries 1,3,6
$0 bench-datafusion --scale-factor 1 --queries 1,3,6
$0 bench-datafusion --scale-factor 100 --hudi-dir gs://bucket/sf100-hudi
@@ -105,6 +198,8 @@ cmd_generate() {
esac
done
+ require_usable_data_dir
+
local parquet_dir="$SCRIPT_DIR/data/sf$sf-parquet"
if [ -d "$parquet_dir" ]; then
echo "Removing existing parquet data at $parquet_dir..."
@@ -117,28 +212,48 @@ cmd_generate() {
cmd_create_tables() {
local sf="$DEFAULT_SCALE_FACTOR"
+ local custom_hudi_dir=""
+ local custom_parquet_dir=""
+ local recreate=0
while [[ $# -gt 0 ]]; do
case "$1" in
--scale-factor) sf="$2"; shift 2 ;;
+ --hudi-dir) custom_hudi_dir="$2"; shift 2 ;;
+ --parquet-dir) custom_parquet_dir="$2"; shift 2 ;;
+ --recreate) recreate=1; shift ;;
*) echo "Unknown option: $1" >&2; usage; exit 1 ;;
esac
done
- local parquet_dir="$SCRIPT_DIR/data/sf$sf-parquet"
- if [ ! -d "$parquet_dir" ]; then
- echo "Error: parquet data not found at $parquet_dir. Run 'generate'
first." >&2
- exit 1
+ local hudi_dir="${custom_hudi_dir:-$SCRIPT_DIR/data/sf$sf-hudi}"
+
+ build_tpch
+
+ # Rebuilding is the expensive step at scale, so existing tables are reused
+ # unless --recreate says otherwise.
+ if [ "$recreate" -eq 0 ] && "$TPCH_BIN" check-tables --hudi-base "$hudi_dir"
2>/dev/null; then
+ echo "Reusing existing Hudi tables at: $hudi_dir"
+ echo "Pass --recreate to rebuild them."
+ return 0
fi
- local hudi_dir="$SCRIPT_DIR/data/sf$sf-hudi"
- if [ -d "$hudi_dir" ]; then
- echo "Removing existing Hudi data at $hudi_dir..."
- rm -rf "$hudi_dir"
+ local parquet_dir="${custom_parquet_dir:-$SCRIPT_DIR/data/sf$sf-parquet}"
+ if ! is_cloud_url "$parquet_dir" && [ ! -d "$parquet_dir" ]; then
+ echo "Error: parquet data not found at $parquet_dir. Run 'generate'
first." >&2
+ exit 1
fi
- build_tpch
setup_spark
- mkdir -p "$hudi_dir"
+
+ # Spark creates the cloud prefix itself; only a local target needs clearing
+ # and pre-creating, and only there would stale files survive a rerun.
+ if ! is_cloud_url "$hudi_dir"; then
+ if [ -d "$hudi_dir" ]; then
+ echo "Removing existing Hudi data at $hudi_dir..."
+ rm -rf "$hudi_dir"
+ fi
+ mkdir -p "$hudi_dir"
+ fi
local sql_file
sql_file="$(mktemp)"
@@ -149,7 +264,7 @@ cmd_create_tables() {
echo "Creating Hudi COW tables from parquet (sf$sf)..."
"$SPARK_HOME/bin/spark-sql" \
- --packages org.apache.hudi:hudi-spark3.5-bundle_2.12:1.1.1 \
+ --packages "$HUDI_SPARK_BUNDLE" \
"${SPARK_ARGS[@]}" \
-f "$sql_file"
@@ -214,6 +329,9 @@ cmd_bench_spark() {
echo "Error: $format data not found at $data_dir." >&2
exit 1
fi
+ if [ "$format" = "hudi" ]; then
+ require_hudi_tables "$data_dir"
+ fi
read_spark_args --scale-factor "$sf" --command bench
setup_spark
@@ -236,7 +354,7 @@ cmd_bench_spark() {
echo "Running Spark SQL benchmark ($format)..."
"$SPARK_HOME/bin/spark-submit" \
- --packages org.apache.hudi:hudi-spark3.5-bundle_2.12:1.1.1 \
+ --packages "$HUDI_SPARK_BUNDLE" \
"${SPARK_ARGS[@]}" \
"$SCRIPT_DIR/infra/spark/bench.py" \
"${bench_args[@]}"
@@ -246,6 +364,7 @@ cmd_bench_spark() {
if [ -n "$output_dir" ]; then
mkdir -p "$output_dir"
parse_args+=(--output-dir "$output_dir" --engine-label spark
--format-label "$format" --display-name "spark+hudi" --scale-factor "$sf")
+ write_env_report "$output_dir/environment.txt" "$sf" "$data_dir"
fi
"$TPCH_BIN" "${parse_args[@]}"
rm -rf "$tmp_dir"
@@ -314,6 +433,10 @@ cmd_bench_datafusion() {
build_tpch
+ if [ "$use_hudi" = true ]; then
+ require_hudi_tables "$hudi_dir"
+ fi
+
local bench_args=(bench --scale-factor "$sf")
[ "$use_hudi" = true ] && bench_args+=(--hudi-dir "$hudi_dir")
[ "$use_parquet" = true ] && bench_args+=(--parquet-dir "$parquet_dir")
@@ -325,6 +448,7 @@ cmd_bench_datafusion() {
mkdir -p "$output_dir"
output_dir="$(cd "$output_dir" && pwd)"
bench_args+=(--output-dir "$output_dir" --engine-label datafusion
--format-label "${format:-hudi}" --display-name "datafusion+hudi-rs")
+ write_env_report "$output_dir/environment.txt" "$sf" "$hudi_dir"
fi
echo "Running DataFusion benchmark..."
@@ -364,6 +488,14 @@ cmd_compare() {
"$TPCH_BIN" compare \
--results-dir "$SCRIPT_DIR/results" \
--runs "$runs"
+
+ # Print alongside the chart so a copied result carries the conditions that
+ # produced it.
+ local env_file="$SCRIPT_DIR/results/environment.txt"
+ if [ -f "$env_file" ]; then
+ echo ""
+ cat "$env_file"
+ fi
}
# --- Main ---
diff --git a/benchmark/tpch/src/config.rs b/benchmark/tpch/src/config.rs
index e446fb1f..24186651 100644
--- a/benchmark/tpch/src/config.rs
+++ b/benchmark/tpch/src/config.rs
@@ -170,6 +170,9 @@ impl ScaleFactorConfig {
let Some(table) = self.tables.get(name) else {
continue;
};
+ // The catalog registration outlives the data it points at, so
+ // without this a rerun fails on the previous run's entry.
+ writeln!(sql, "DROP TABLE IF EXISTS {name};").unwrap();
writeln!(sql, "CREATE TABLE {name} USING hudi").unwrap();
writeln!(sql, "LOCATION '{hudi_base}/{name}'").unwrap();
writeln!(sql, "TBLPROPERTIES (").unwrap();
@@ -208,6 +211,7 @@ impl ScaleFactorConfig {
// Register Hudi tables
for &name in TABLE_ORDER {
if self.tables.contains_key(name) {
+ writeln!(sql, "DROP TABLE IF EXISTS {name};").unwrap();
writeln!(
sql,
"CREATE TABLE {name} USING hudi LOCATION
'{hudi_base}/{name}';"
diff --git a/benchmark/tpch/src/main.rs b/benchmark/tpch/src/main.rs
index 73362a41..c02285a6 100644
--- a/benchmark/tpch/src/main.rs
+++ b/benchmark/tpch/src/main.rs
@@ -86,6 +86,12 @@ enum Commands {
#[arg(long)]
hudi_base: String,
},
+ /// Verify every TPC-H Hudi table is readable at a base path
+ CheckTables {
+ /// Hudi tables base path (e.g., /opt/hudi or s3://bucket/path)
+ #[arg(long)]
+ hudi_base: String,
+ },
/// Render benchmark SQL (table registrations + query iterations)
RenderBenchSql {
/// TPC-H scale factor (loads config/sf{N}.yaml)
@@ -354,6 +360,17 @@ async fn main() -> Result<()> {
print!("{}", cfg.render_ctas_sql(&parquet_base, &hudi_base));
Ok(())
}
+ Commands::CheckTables { hudi_base } => {
+ let missing = missing_hudi_tables(&hudi_base).await?;
+ if missing.is_empty() {
+ Ok(())
+ } else {
+ Err(datafusion::error::DataFusionError::Plan(format!(
+ "Hudi tables not found at {hudi_base}: {}",
+ missing.join(", ")
+ )))
+ }
+ }
Commands::RenderBenchSql {
scale_factor,
hudi_base,
@@ -484,24 +501,50 @@ fn load_query(query_num: usize, scale_factor: f64) ->
std::result::Result<String
Ok(sql.replace("${Q11_FRACTION}", &q11_fraction))
}
+/// Names of the TPC-H tables that cannot be opened under `base_dir`.
+///
+/// Opens each through the same `HudiDataSource` the benchmark uses, so a table
+/// that passes here is one the benchmark can actually read, not merely a
+/// directory that exists.
+async fn missing_hudi_tables(base_dir: &str) -> Result<Vec<String>> {
+ let resolved =
resolve_path(base_dir).map_err(datafusion::error::DataFusionError::Plan)?;
+
+ let mut missing = Vec::new();
+ for table_name in TPCH_TABLES {
+ let table_uri = hudi_table_uri(&resolved, table_name)?;
+ if HudiDataSource::new(&table_uri).await.is_err() {
+ missing.push((*table_name).to_string());
+ }
+ }
+ Ok(missing)
+}
+
+/// Build the URI for one Hudi table under a local path or cloud URL.
+fn hudi_table_uri(resolved_base: &str, table_name: &str) -> Result<String> {
+ if is_cloud_url(resolved_base) {
+ Ok(format!(
+ "{}/{table_name}",
+ resolved_base.trim_end_matches('/')
+ ))
+ } else {
+ let table_path = Path::new(resolved_base).join(table_name);
+ Ok(url::Url::from_file_path(&table_path)
+ .map_err(|_| {
+ datafusion::error::DataFusionError::Plan(format!(
+ "Failed to create file URL for {}",
+ table_path.display()
+ ))
+ })?
+ .to_string())
+ }
+}
+
/// Register all 8 TPC-H Hudi tables. Supports local paths and cloud URLs.
async fn register_hudi_tables(ctx: &SessionContext, base_dir: &str) ->
Result<()> {
let resolved =
resolve_path(base_dir).map_err(datafusion::error::DataFusionError::Plan)?;
for table_name in TPCH_TABLES {
- let table_uri = if is_cloud_url(&resolved) {
- format!("{}/{table_name}", resolved.trim_end_matches('/'))
- } else {
- let table_path = Path::new(&resolved).join(table_name);
- url::Url::from_file_path(&table_path)
- .map_err(|_| {
- datafusion::error::DataFusionError::Plan(format!(
- "Failed to create file URL for {}",
- table_path.display()
- ))
- })?
- .to_string()
- };
+ let table_uri = hudi_table_uri(&resolved, table_name)?;
let hudi = HudiDataSource::new(&table_uri).await?;
ctx.register_table(*table_name, Arc::new(hudi))?;
}
