From 6809036cc149c13ddbf3b70e68e5b9d12d145c83 Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Wed, 15 Jul 2026 16:22:00 -0700 Subject: [PATCH 01/12] Add distributed CTAS benchmark output --- common/testing/validate_results.py | 37 +++-- presto/README.md | 20 ++- .../etc_common/catalog/hive_output.properties | 10 ++ presto/docker/docker-compose.common.yml | 3 +- presto/scripts/run_benchmark.sh | 89 +++++++++++- presto/scripts/run_multiple_benchmarks.sh | 15 +- .../scripts/start_presto_helper_parse_args.sh | 11 ++ presto/slurm/presto-nvl72/README.md | 9 +- presto/slurm/presto-nvl72/functions.sh | 22 ++- presto/slurm/presto-nvl72/launch-run.sh | 19 ++- .../performance_benchmarks/common_fixtures.py | 130 ++++++++++++++++-- .../performance_benchmarks/conftest.py | 2 + tests/test_presto_ctas_results.py | 81 +++++++++++ 13 files changed, 416 insertions(+), 32 deletions(-) create mode 100644 presto/docker/config/template/etc_common/catalog/hive_output.properties create mode 100644 tests/test_presto_ctas_results.py diff --git a/common/testing/validate_results.py b/common/testing/validate_results.py index a19fe143..b3f18823 100644 --- a/common/testing/validate_results.py +++ b/common/testing/validate_results.py @@ -3,7 +3,7 @@ # SPDX-License-Identifier: Apache-2.0 """ -Validate TPC-H/TPC-DS query results against expected parquet files. +Validate TPC-H/TPC-DS query results against expected Parquet files. Comparison logic lives in common/testing/result_comparison.py and is shared with the integration test path so that both paths use identical semantics. @@ -11,6 +11,7 @@ import argparse import json +import re import sys from pathlib import Path @@ -36,7 +37,7 @@ def validate( """Run validation and return a results dict. Args: - results_dir: Directory containing q1.parquet ... q22.parquet result files. + results_dir: Directory containing qN.parquet files or distributed qN datasets. expected_dir: Directory containing expected parquet files. queries: Dict mapping query IDs (e.g. "Q1") to SQL strings. query_numbers: Optional list of query numbers to validate. When provided, @@ -50,16 +51,22 @@ def validate( passed = failed = not_validated = expected_failures = 0 if query_numbers is not None: - result_files = sorted(f for q in query_numbers for f in [results_dir / f"q{q}.parquet"] if f.exists()) + result_files = [] + for q_num in query_numbers: + result_files.extend( + path for path in (results_dir / f"q{q_num}.parquet", results_dir / f"q{q_num}") if path.exists() + ) else: - result_files = sorted(results_dir.glob("q*.parquet")) + result_files = sorted( + path for path in results_dir.iterdir() if re.fullmatch(r"q\d+(?:\.parquet)?", path.name) + ) if not result_files: - print(f"No result parquet files found in {results_dir}", file=sys.stderr) + print(f"No result Parquet files or datasets found in {results_dir}", file=sys.stderr) return {"overall_status": "not-validated", "queries": {}} for result_file in result_files: - query_id = result_file.stem # e.g. "q1" + query_id = result_file.stem # e.g. "q1" for q1 or q1.parquet q_num = int(query_id.lstrip("q")) # Accepted naming conventions for expected files (tried in order): @@ -143,7 +150,7 @@ def validate( def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( - description="Validate TPC-H/TPC-DS query results against expected parquet files.", + description="Validate TPC-H/TPC-DS query results against expected Parquet files or datasets.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) @@ -152,6 +159,12 @@ def parse_args() -> argparse.Namespace: required=True, help="Benchmark output directory (same as pytest --output-dir).", ) + parser.add_argument( + "--actual-results-dir", + default=None, + help="Optional directory containing actual qN Parquet files or dataset directories. " + "Defaults to query_results beneath --output-dir/--tag.", + ) parser.add_argument( "--tag", default=None, @@ -179,11 +192,10 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() -def _write_not_validated(results_dir: Path, reason: str) -> None: +def _write_not_validated(output_path: Path, reason: str) -> None: """Write a not-validated sentinel JSON and print the reason.""" print(f"[Validation] {reason}") results = {"overall_status": "not-validated", "queries": {}} - output_path = results_dir.parent / "validation_results.json" output_path.write_text(json.dumps(results, indent=2)) print(f"[Validation] Results written to {output_path}") @@ -192,14 +204,16 @@ def _write_not_validated(results_dir: Path, reason: str) -> None: args = parse_args() output_dir = Path(args.output_dir) - results_dir = output_dir / args.tag / "query_results" if args.tag else output_dir / "query_results" + report_dir = output_dir / args.tag if args.tag else output_dir + results_dir = Path(args.actual_results_dir) if args.actual_results_dir else report_dir / "query_results" + output_path = report_dir / "validation_results.json" if not results_dir.is_dir(): print(f"Error: results directory not found: {results_dir}", file=sys.stderr) sys.exit(1) if args.reference_results_dir is None: - _write_not_validated(results_dir, "No reference results directory provided; validation skipped.") + _write_not_validated(output_path, "No reference results directory provided; validation skipped.") sys.exit(0) expected_dir = Path(args.reference_results_dir) @@ -216,7 +230,6 @@ def _write_not_validated(results_dir: Path, reason: str) -> None: results = validate(results_dir, expected_dir, queries, query_numbers=query_numbers) # Write validation_results.json next to the query_results/ dir - output_path = results_dir.parent / "validation_results.json" output_path.write_text(json.dumps(results, indent=2)) print(f"[Validation] Results written to {output_path}") diff --git a/presto/README.md b/presto/README.md index c744e2f8..3d9e5594 100644 --- a/presto/README.md +++ b/presto/README.md @@ -50,6 +50,14 @@ All three repositories must be checked out as sibling directories. **Important:* ```bash export PRESTO_DATA_DIR=/path/to/your/benchmark/data ``` + + Benchmark source data is mounted read-only. To have workers write large query + results directly as distributed Parquet datasets, also select a separate + writable directory before starting the cluster: + + ```bash + export PRESTO_OUTPUT_DIR=/path/to/writable/benchmark-results + ``` > **Tip:** Add this export to your `~/.bashrc` to avoid setting it each time. 3. Build dependencies (first time only): @@ -96,10 +104,18 @@ pytest tpch_test.py ``` > This step is necessary because aggregation for statistics collection is not yet supported on GPU. Collecting statistics improves performance and reduces OOM errors for GPU execution. -4. Run benchmarks: +4. Run a benchmark. Add `--write-results-to-file` to write distributed + `benchmark_results_/qN/` Parquet datasets beneath `PRESTO_OUTPUT_DIR` + (`benchmark_results/qN/` when no tag is supplied): + + ```bash + ./run_benchmark.sh -b tpch -s bench_sf100 --write-results-to-file + ``` + +5. For the standard coordinator-returned result mode, run: ```bash ./run_benchmark.sh --help - ./run_benchmark.sh --benchmark tpch --schema-name + ./run_benchmark.sh --benchmark-type tpch --schema-name ``` > **Note:** The `--schema-name` flag is required to specify your target DB schema for the benchmark. diff --git a/presto/docker/config/template/etc_common/catalog/hive_output.properties b/presto/docker/config/template/etc_common/catalog/hive_output.properties new file mode 100644 index 00000000..c779c99c --- /dev/null +++ b/presto/docker/config/template/etc_common/catalog/hive_output.properties @@ -0,0 +1,10 @@ +# Hive catalog dedicated to benchmark CTAS output. Its file metastore root is +# mounted separately from PRESTO_DATA_DIR so source data can remain read-only. +connector.name=hive-hadoop2 +hive.metastore=file +hive.metastore.catalog.dir=file:/var/lib/presto/data/hive/benchmark_output +hive.allow-drop-table=true +# Native workers otherwise stage writes under their container-private /tmp, +# which prevents the coordinator from committing the files. Write directly to +# the shared result mount; interrupted runs are cleaned before the next run. +hive.temporary-staging-directory-enabled=false diff --git a/presto/docker/docker-compose.common.yml b/presto/docker/docker-compose.common.yml index f7805412..cb964c65 100644 --- a/presto/docker/docker-compose.common.yml +++ b/presto/docker/docker-compose.common.yml @@ -3,7 +3,8 @@ services: volumes: - ./.hive_metastore:/var/lib/presto/data/hive/metastore - ../../common/testing/integration_tests/data:/var/lib/presto/data/hive/data/integration_test - - ${PRESTO_DATA_DIR:-/dev/null}:/var/lib/presto/data/hive/data/user_data + - ${PRESTO_DATA_DIR:-/dev/null}:/var/lib/presto/data/hive/data/user_data:ro + - ${PRESTO_OUTPUT_DIR:-/dev/null}:/var/lib/presto/data/hive/benchmark_output - ${LOGS_DIR:-/dev/null}:/opt/presto-server/logs - ./launch_presto_servers.sh:/opt/launch_presto_servers.sh:ro diff --git a/presto/scripts/run_benchmark.sh b/presto/scripts/run_benchmark.sh index 49fd7bae..830ac987 100755 --- a/presto/scripts/run_benchmark.sh +++ b/presto/scripts/run_benchmark.sh @@ -43,6 +43,8 @@ OPTIONS: --profile-script-path Path to a custom profiler functions script. Defaults to ./profiler_functions.sh. --skip-drop-cache Skip dropping system caches before each benchmark query (dropped by default). --skip-analyze-check Skip checking that ANALYZE TABLE has been run on all tables (checked by default). + --write-results-to-file Write query results with distributed Hive CTAS operations instead of returning them + through the coordinator. PRESTO_OUTPUT_DIR must be set and mounted when the cluster starts. -m, --metrics Collect detailed metrics from Presto REST API after each query. Metrics are stored in query-specific directories. --reference-results-dir Path to a directory containing reference (expected) parquet files. @@ -57,6 +59,8 @@ OPTIONS: Set to the full host path of the expected results directory (e.g. PRESTO_EXPECTED_RESULTS_DIR=/data/sf100_expected). Warning (not error) if set but the directory does not exist. + PRESTO_OUTPUT_DIR Writable host directory used for distributed CTAS result datasets. This must be set + before both cluster startup and this script when --write-results-to-file is enabled. -v, --verbose Print debug logs for worker/engine detection (e.g. node URIs, cluster-tag, GPU model). Use when engine is misdetected or the run fails. @@ -68,6 +72,7 @@ EXAMPLES: $0 -b tpch -s bench_sf100 -t gh200_cpu_sf100 $0 -b tpch -s bench_sf100 --profile $0 -b tpch -s bench_sf100 --metrics + PRESTO_OUTPUT_DIR=/results $0 -b tpch -s bench_sf100 --write-results-to-file $0 -b tpch -s bench_sf100 --verbose EOF @@ -191,6 +196,10 @@ parse_args() { SKIP_ANALYZE_CHECK=true shift ;; + --write-results-to-file) + WRITE_RESULTS_TO_FILE=true + shift + ;; -m|--metrics) METRICS=true shift @@ -232,6 +241,26 @@ if [[ -z ${SCHEMA_NAME} ]]; then exit 1 fi +if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then + if [[ -z "${PRESTO_OUTPUT_DIR}" ]]; then + echo "Error: PRESTO_OUTPUT_DIR must be set when --write-results-to-file is enabled." >&2 + exit 1 + fi + mkdir -p "${PRESTO_OUTPUT_DIR}" + PRESTO_OUTPUT_DIR="$(readlink -f "${PRESTO_OUTPUT_DIR}")" + export PRESTO_OUTPUT_DIR + if [[ ! -d "${PRESTO_OUTPUT_DIR}" || ! -w "${PRESTO_OUTPUT_DIR}" ]]; then + echo "Error: PRESTO_OUTPUT_DIR must be a writable directory: ${PRESTO_OUTPUT_DIR}" >&2 + exit 1 + fi + if [[ -n ${TAG} ]]; then + CTAS_RESULTS_SCHEMA="benchmark_results_${TAG,,}" + else + CTAS_RESULTS_SCHEMA="benchmark_results" + fi + CTAS_RESULTS_DIR="${PRESTO_OUTPUT_DIR}/${CTAS_RESULTS_SCHEMA}" +fi + # Fail fast if an explicit --reference-results-dir was given but doesn't exist. if [[ ${EXPLICIT_REFERENCE_DIR} == true && ! -d ${PRESTO_EXPECTED_RESULTS_DIR} ]]; then echo "[Validation] Error: --reference-results-dir not found: ${PRESTO_EXPECTED_RESULTS_DIR}" >&2 @@ -299,6 +328,10 @@ if [[ "${SKIP_ANALYZE_CHECK}" == "true" ]]; then PYTEST_ARGS+=("--skip-analyze-check") fi +if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then + PYTEST_ARGS+=("--write-results-to-file") +fi + source "${SCRIPT_DIR}/../../scripts/py_env_functions.sh" trap delete_python_virtual_env EXIT @@ -321,6 +354,52 @@ echo "Using PRESTO_IMAGE_TAG: $PRESTO_IMAGE_TAG" BENCHMARK_TEST_DIR=${TEST_DIR}/performance_benchmarks pytest -q -s ${BENCHMARK_TEST_DIR}/${BENCHMARK_TYPE}_test.py ${PYTEST_ARGS[*]} +normalize_ctas_result_permissions() { + local host_results_dir=$1 + local container_results_dir="/var/lib/presto/data/hive/benchmark_output/${CTAS_RESULTS_SCHEMA}" + + if [[ ! -d "${host_results_dir}" ]]; then + echo "Error: CTAS result directory not found: ${host_results_dir}" >&2 + return 1 + fi + + # Slurm and non-remapped local files are normally owned by the invoking user. + find "${host_results_dir}" -type f -name '*.parquet' -exec chmod a+r {} + 2>/dev/null || true + local parquet_file + parquet_file="$(find "${host_results_dir}" -type f -name '*.parquet' -print -quit 2>/dev/null || true)" + if [[ -n "${parquet_file}" && -z "$(find "${host_results_dir}" -type f -name '*.parquet' ! -readable -print -quit 2>/dev/null || true)" ]]; then + return 0 + fi + + # Docker-remapped native workers create local files as mode 0600 under their + # remapped identity. Adjust permissions in place inside each worker; no data + # is copied through the coordinator or benchmark client. + if command -v docker >/dev/null 2>&1; then + local container + while IFS= read -r container; do + [[ -n ${container} ]] || continue + docker exec "${container}" find "${container_results_dir}" -type d \ + -exec chmod a+rx {} + 2>/dev/null || true + docker exec "${container}" find "${container_results_dir}" -type f -name '*.parquet' \ + -exec chmod a+r {} + 2>/dev/null || true + done < <(docker ps --format '{{.Names}}' | grep -E '^presto-.*worker') + fi + + parquet_file="$(find "${host_results_dir}" -type f -name '*.parquet' -print -quit 2>/dev/null || true)" + if [[ -z "${parquet_file}" ]]; then + echo "Error: No CTAS Parquet result files found in ${host_results_dir}" >&2 + return 1 + fi + if [[ -n "$(find "${host_results_dir}" -type f -name '*.parquet' ! -readable -print -quit 2>/dev/null || true)" ]]; then + echo "Error: CTAS result files are not readable from the benchmark host: ${host_results_dir}" >&2 + return 1 + fi +} + +if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then + normalize_ctas_result_permissions "${CTAS_RESULTS_DIR}" +fi + # Snapshot logs and engine configs into the benchmark output directory so that # post_results.py has self-contained, run-specific data even when multiple runs # are posted after the fact. @@ -363,10 +442,18 @@ else if [[ -n ${QUERIES} ]]; then VALIDATE_ARGS+=(--queries "${QUERIES}") fi + if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then + VALIDATE_ARGS+=(--actual-results-dir "${CTAS_RESULTS_DIR}") + fi ACTUAL_OUTPUT_DIR="${OUTPUT_DIR:-$(pwd)/benchmark_output}" [[ -n ${TAG} ]] && ACTUAL_OUTPUT_DIR="${ACTUAL_OUTPUT_DIR}/${TAG}" - echo "[Validation] Running validation: ${ACTUAL_OUTPUT_DIR}/query_results vs ${PRESTO_EXPECTED_RESULTS_DIR:-}" + if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then + ACTUAL_RESULTS_DIR="${CTAS_RESULTS_DIR}" + else + ACTUAL_RESULTS_DIR="${ACTUAL_OUTPUT_DIR}/query_results" + fi + echo "[Validation] Running validation: ${ACTUAL_RESULTS_DIR} vs ${PRESTO_EXPECTED_RESULTS_DIR:-}" "${SCRIPT_DIR}/../../scripts/run_py_script.sh" --quiet \ -p "${VALIDATE_SCRIPT}" \ -r "${VALIDATE_REQUIREMENTS}" \ diff --git a/presto/scripts/run_multiple_benchmarks.sh b/presto/scripts/run_multiple_benchmarks.sh index 2243266b..ead5babe 100755 --- a/presto/scripts/run_multiple_benchmarks.sh +++ b/presto/scripts/run_multiple_benchmarks.sh @@ -7,6 +7,7 @@ KVIKIO_ARRAY=(8) DRIVERS_ARRAY=(2) WORKERS_ARRAY=(1) SCHEMA_ARRAY=() +WRITE_RESULTS_TO_FILE=false parse_args() { while [[ $# -gt 0 ]]; do case $1 in @@ -55,6 +56,10 @@ parse_args() { exit 1 fi ;; + --write-results-to-file) + WRITE_RESULTS_TO_FILE=true + shift + ;; *) echo "Error: Unknown argument $1" print_help @@ -76,13 +81,21 @@ if [[ -z ${PRESTO_DATA_DIR} ]]; then exit 1 fi +if [[ "${WRITE_RESULTS_TO_FILE}" == "true" && -z "${PRESTO_OUTPUT_DIR:-}" ]]; then + echo "Error: PRESTO_OUTPUT_DIR is required with --write-results-to-file." + exit 1 +fi + +BENCHMARK_OUTPUT_ARGS=() +[[ "${WRITE_RESULTS_TO_FILE}" == "true" ]] && BENCHMARK_OUTPUT_ARGS+=(--write-results-to-file) + for schema in "${SCHEMA_ARRAY[@]}"; do for kvikio in "${KVIKIO_ARRAY[@]}"; do for drivers in "${DRIVERS_ARRAY[@]}"; do for workers in "${WORKERS_ARRAY[@]}"; do echo "Running combo: num_workers = $workers, kvikio_threads = $kvikio, num_drivers = $drivers, schema = $schema" ./start_native_gpu_presto.sh -w $workers --kvikio-threads $kvikio --num-drivers $drivers - ./run_benchmark.sh -b tpch -s ${schema} --tag "${schema}_${workers}wk_${drivers}dr_${kvikio}kv" + ./run_benchmark.sh -b tpch -s ${schema} --tag "${schema}_${workers}wk_${drivers}dr_${kvikio}kv" "${BENCHMARK_OUTPUT_ARGS[@]}" ./stop_presto.sh done done diff --git a/presto/scripts/start_presto_helper_parse_args.sh b/presto/scripts/start_presto_helper_parse_args.sh index 6b868b94..41b5812c 100644 --- a/presto/scripts/start_presto_helper_parse_args.sh +++ b/presto/scripts/start_presto_helper_parse_args.sh @@ -44,6 +44,7 @@ OPTIONS: ENVIRONMENT VARIABLES: SCCACHE_AUTH_DIR Directory containing sccache auth files (default: ~/.sccache-auth/). + PRESTO_OUTPUT_DIR Optional writable host directory mounted for distributed CTAS benchmark results. EXAMPLES: $SCRIPT_NAME --no-cache @@ -217,6 +218,16 @@ parse_args() { parse_args "$@" +if [[ -n "${PRESTO_OUTPUT_DIR:-}" ]]; then + mkdir -p "${PRESTO_OUTPUT_DIR}" + PRESTO_OUTPUT_DIR="$(readlink -f "${PRESTO_OUTPUT_DIR}")" + if [[ ! -d "${PRESTO_OUTPUT_DIR}" || ! -w "${PRESTO_OUTPUT_DIR}" ]]; then + echo "Error: PRESTO_OUTPUT_DIR must be a writable directory: ${PRESTO_OUTPUT_DIR}" >&2 + exit 1 + fi + export PRESTO_OUTPUT_DIR +fi + if [[ -n ${BUILD_TARGET} && ! ${BUILD_TARGET} =~ ^(coordinator|c|worker|w|all|a)$ ]]; then echo "Error: invalid --build value." print_help diff --git a/presto/slurm/presto-nvl72/README.md b/presto/slurm/presto-nvl72/README.md index 39cdc553..c799e801 100644 --- a/presto/slurm/presto-nvl72/README.md +++ b/presto/slurm/presto-nvl72/README.md @@ -102,11 +102,18 @@ This step only needs to repeat when the worker image's stat tracking changes # Profile queries 5 and 6 on worker 2 ./launch-run.sh -n 8 -s 3000 -p --nsys-worker-id 2 -q 5,6 + +# Have workers write distributed Parquet results to separate shared storage +PRESTO_OUTPUT_DIR=/shared/writable/presto-results \ + ./launch-run.sh -n 8 -s 3000 --write-results-to-file ``` Submits a benchmark sbatch job, polls until completion, and prints a summary. Results land under `result_dir/`. See `./launch-run.sh --help` for the full -flag list (queries filter, output path, GDS toggle, profiling, metrics, …). +flag list (queries filter, output path, CTAS result writing, GDS toggle, +profiling, metrics, …). In CTAS mode, benchmark reports remain in +`result_dir/`, while `benchmark_results_/qN/` Parquet datasets are written +beneath `PRESTO_OUTPUT_DIR` (`benchmark_results/qN/` without a tag). **Prerequisites:** worker + coord images on disk; data from step 1; analyzed metastore from step 2 (either local or shared). diff --git a/presto/slurm/presto-nvl72/functions.sh b/presto/slurm/presto-nvl72/functions.sh index 86808ddd..39d0b673 100644 --- a/presto/slurm/presto-nvl72/functions.sh +++ b/presto/slurm/presto-nvl72/functions.sh @@ -2,6 +2,16 @@ # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +CTAS_CONTAINER_OUTPUT_DIR=/var/lib/presto/data/hive/benchmark_output + +ctas_output_mount_arg() { + if [[ "${WRITE_RESULTS_TO_FILE:-0}" == "1" ]]; then + validate_environment_preconditions PRESTO_OUTPUT_DIR + printf ',%s:%s' "${PRESTO_OUTPUT_DIR}" "${CTAS_CONTAINER_OUTPUT_DIR}" + fi + return 0 +} + # Echo the ",:" bind-mount fragment for the host's # miniforge3 install if it exists, else nothing. Used by every container # that needs to run Python via init_python_virtual_env / run_py_script.sh: @@ -116,6 +126,7 @@ function run_coord_image { local extra_mounts extra_mounts="$(miniforge_mount_arg)" + extra_mounts+="$(ctas_output_mount_arg)" if [[ -n "${CLUSTER_EXTRA_MOUNTS:-}" ]]; then extra_mounts="${extra_mounts},${CLUSTER_EXTRA_MOUNTS}" fi @@ -134,7 +145,7 @@ ${CONFIGS}/etc_common:/opt/presto-server/etc,\ ${CONFIGS}/etc_coordinator/node.properties:/opt/presto-server/etc/node.properties,\ ${CONFIGS}/etc_coordinator/config_native.properties:/opt/presto-server/etc/config.properties,\ ${CONFIGS}/etc_coordinator/catalog/hive.properties:/opt/presto-server/etc/catalog/hive.properties,\ -${DATA}:/var/lib/presto/data/hive/data/user_data,\ +${DATA}:/var/lib/presto/data/hive/data/user_data:ro,\ ${VT_ROOT}/.hive_metastore:/var/lib/presto/data/hive/metastore${extra_mounts} \ -- bash -lc "unset JAVA_HOME; export JAVA_HOME=/usr/lib/jvm/jre-17-openjdk; export PATH=/usr/lib/jvm/jre-17-openjdk/bin:\$PATH; ${script}" >> ${LOGS}/${log_file} 2>&1 & else @@ -148,7 +159,7 @@ ${CONFIGS}/etc_common:/opt/presto-server/etc,\ ${CONFIGS}/etc_coordinator/node.properties:/opt/presto-server/etc/node.properties,\ ${CONFIGS}/etc_coordinator/config_native.properties:/opt/presto-server/etc/config.properties,\ ${CONFIGS}/etc_coordinator/catalog/hive.properties:/opt/presto-server/etc/catalog/hive.properties,\ -${DATA}:/var/lib/presto/data/hive/data/user_data,\ +${DATA}:/var/lib/presto/data/hive/data/user_data:ro,\ ${LOGS}:/var/log/nsys,\ ${VT_ROOT}/.hive_metastore:/var/lib/presto/data/hive/metastore${extra_mounts} \ -- bash -lc "unset JAVA_HOME; export JAVA_HOME=/usr/lib/jvm/jre-17-openjdk; export PATH=/usr/lib/jvm/jre-17-openjdk/bin:\$PATH; ${script}" >> ${LOGS}/${log_file} 2>&1 @@ -292,8 +303,9 @@ function run_worker { driver_mounts="${driver_mounts},${CLUSTER_LIBNVIDIA_ML_HOST_PATH}:${CLUSTER_LIBNVIDIA_ML_CONTAINER_PATH}" fi local worker_extra_mounts="" + worker_extra_mounts="$(ctas_output_mount_arg)" if [[ -n "${CLUSTER_EXTRA_MOUNTS:-}" ]]; then - worker_extra_mounts=",${CLUSTER_EXTRA_MOUNTS}" + worker_extra_mounts+=",${CLUSTER_EXTRA_MOUNTS}" fi # NVIDIA_VISIBLE_DEVICES triggers enroot's 98-nvidia.sh hook, which requires @@ -343,7 +355,7 @@ ${worker_node}:/opt/presto-server/etc/node.properties,\ ${worker_config}:/opt/presto-server/etc/config.properties,\ ${worker_hive}:/opt/presto-server/etc/catalog/hive.properties,\ ${worker_data}:/var/lib/presto/data,\ -${DATA}:/var/lib/presto/data/hive/data/user_data,\ +${DATA}:/var/lib/presto/data/hive/data/user_data:ro,\ ${VT_ROOT}/.hive_metastore:/var/lib/presto/data/hive/metastore,\ ${WORKER_ENV_FILE}:${vt_worker_env_file},\ ${LOGS}:${vt_cufile_log_dir},\ @@ -537,6 +549,7 @@ function run_queries { [[ "${ENABLE_METRICS}" == "1" ]] && extra_args+=("-m") [[ "${ENABLE_NSYS}" == "1" ]] && extra_args+=("-p" "--profile-script-path" "${container_script_dir}/profiler_functions.sh") [[ -n "${QUERIES:-}" ]] && extra_args+=("-q" "${QUERIES}") + [[ "${WRITE_RESULTS_TO_FILE:-0}" == "1" ]] && extra_args+=("--write-results-to-file") source "${SCRIPT_DIR}/defaults.env" @@ -571,6 +584,7 @@ function run_queries { export PORT=$PORT; \ export HOSTNAME=$COORD; \ export PRESTO_DATA_DIR=/var/lib/presto/data/hive/data/user_data; \ + export PRESTO_OUTPUT_DIR=${CTAS_CONTAINER_OUTPUT_DIR}; \ export MINIFORGE_HOME=/workspace/miniforge3; \ export HOME=/workspace; \ cd /workspace/presto/scripts; \ diff --git a/presto/slurm/presto-nvl72/launch-run.sh b/presto/slurm/presto-nvl72/launch-run.sh index 92807312..4ce862ae 100755 --- a/presto/slurm/presto-nvl72/launch-run.sh +++ b/presto/slurm/presto-nvl72/launch-run.sh @@ -14,7 +14,7 @@ # [-i|--iterations ] [--cpu] [-g|--num-workers-per-node ] # [-w|--worker-image ] [-c|--coord-image ] # [-o|--output-path ] [-q|--queries ] -# [--disable-gds] [-m|--metrics] [-p|--profile] +# [--write-results-to-file] [--disable-gds] [-m|--metrics] [-p|--profile] # [additional sbatch options] # ============================================================================== @@ -42,6 +42,7 @@ SCRIPT_DIR="$PWD" ENABLE_GDS=1 ENABLE_METRICS=0 ENABLE_NSYS=0 +WRITE_RESULTS_TO_FILE=0 NSYS_WORKER_ID=0 QUERIES="" @@ -62,6 +63,7 @@ Options: -c, --coord-image Override coordinator image from cluster config -o, --output-path Copy results into this directory after the run -q, --queries Comma-separated query filter (e.g. "1,6,21") + --write-results-to-file Write distributed Parquet results with CTAS. Requires PRESTO_OUTPUT_DIR. --worker-env-file Override worker.env (default: ./worker.env) --cpu Use CPU partition/images (overrides cluster default) --gpu Use GPU partition/images (overrides cluster default) @@ -96,6 +98,7 @@ while [[ $# -gt 0 ]]; do --gpu) VARIANT_TYPE="gpu"; shift ;; --no-numa) USE_NUMA="0"; shift ;; --disable-gds) ENABLE_GDS=0; shift ;; + --write-results-to-file) WRITE_RESULTS_TO_FILE=1; shift ;; -m|--metrics) ENABLE_METRICS=1; shift ;; -p|--profile) ENABLE_NSYS=1; shift ;; -h|--help) usage; exit 0 ;; @@ -107,6 +110,16 @@ done [[ -z "${NODES_COUNT}" ]] && { echo "Error: -n|--nodes is required (see --help)" >&2; exit 1; } [[ -z "${SCALE_FACTOR}" ]] && { echo "Error: -s|--scale-factor is required (see --help)" >&2; exit 1; } +if [[ "${WRITE_RESULTS_TO_FILE}" == "1" ]]; then + [[ -n "${PRESTO_OUTPUT_DIR:-}" ]] || { echo "Error: PRESTO_OUTPUT_DIR is required with --write-results-to-file" >&2; exit 1; } + mkdir -p "${PRESTO_OUTPUT_DIR}" + PRESTO_OUTPUT_DIR="$(readlink -f "${PRESTO_OUTPUT_DIR}")" + [[ -d "${PRESTO_OUTPUT_DIR}" && -w "${PRESTO_OUTPUT_DIR}" ]] || { + echo "Error: PRESTO_OUTPUT_DIR must be a writable shared directory: ${PRESTO_OUTPUT_DIR}" >&2 + exit 1 + } +fi + # Clean up old output files — use rm -rf so subdirectories (e.g. query_results/) # are fully removed and stale benchmark_result.json cannot survive a cancelled run. rm -rf result_dir logs 2>/dev/null || true @@ -163,6 +176,10 @@ build_common_export_vars EXPORT_VARS+=",NUM_ITERATIONS=${NUM_ITERATIONS}" EXPORT_VARS+=",ENABLE_GDS=${ENABLE_GDS},ENABLE_METRICS=${ENABLE_METRICS}" EXPORT_VARS+=",ENABLE_NSYS=${ENABLE_NSYS},NSYS_WORKER_ID=${NSYS_WORKER_ID}" +EXPORT_VARS+=",WRITE_RESULTS_TO_FILE=${WRITE_RESULTS_TO_FILE}" +if [[ "${WRITE_RESULTS_TO_FILE}" == "1" ]]; then + EXPORT_VARS+=",PRESTO_OUTPUT_DIR=${PRESTO_OUTPUT_DIR}" +fi # Comma-separated query list can't ride EXPORT_VARS (comma is the separator); # export it so sbatch picks it up via the ALL inheritance. [[ -n "${QUERIES}" ]] && export QUERIES diff --git a/presto/testing/performance_benchmarks/common_fixtures.py b/presto/testing/performance_benchmarks/common_fixtures.py index f1a01a12..92935b9f 100644 --- a/presto/testing/performance_benchmarks/common_fixtures.py +++ b/presto/testing/performance_benchmarks/common_fixtures.py @@ -1,6 +1,11 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION. # SPDX-License-Identifier: Apache-2.0 +import os +import re +import shutil +from contextlib import suppress +from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path @@ -9,12 +14,89 @@ import pytest from common.testing.performance_benchmarks.benchmark_keys import BenchmarkKeys +from common.testing.performance_benchmarks.conftest import get_output_dir from common.testing.performance_benchmarks.profiler_utils import start_profiler, stop_profiler from ..integration_tests.analyze_tables import check_tables_analyzed from .metrics_collector import collect_metrics from .run_context import gather_run_context +CTAS_CATALOG = "hive_output" + + +@dataclass(frozen=True) +class CtasResults: + catalog: str + schema: str + + +def ctas_schema_name(tag): + if tag and not re.fullmatch(r"[A-Za-z0-9_]+", tag): + raise ValueError("CTAS result tags must contain only alphanumeric and underscore characters") + return f"benchmark_results_{tag.lower()}" if tag else "benchmark_results" + + +def strip_trailing_semicolon(query): + return query.rstrip().removesuffix(";").rstrip() + + +def ctas_table_name(query_id, iteration_num): + return query_id.lower() if iteration_num == 0 else f"{query_id.lower()}_iteration_{iteration_num + 1}" + + +def build_ctas_query(benchmark_type, query_id, query, catalog, schema, table): + return ( + f"--{benchmark_type}_{query_id}--\n" + f"CREATE TABLE {catalog}.{schema}.{table} WITH (format = 'PARQUET') AS\n" + f"{strip_trailing_semicolon(query)}" + ) + + +def _drop_results_schema(cursor, catalog, schema): + schemas = {row[0] for row in cursor.execute(f"SHOW SCHEMAS FROM {catalog}").fetchall()} + if schema not in schemas: + return + tables = cursor.execute(f"SHOW TABLES FROM {catalog}.{schema}").fetchall() + for (table,) in tables: + cursor.execute(f'DROP TABLE IF EXISTS {catalog}.{schema}."{table}"').fetchall() + cursor.execute(f"DROP SCHEMA {catalog}.{schema}").fetchall() + + +@pytest.fixture(scope="session") +def ctas_results(request): + if not request.config.getoption("--write-results-to-file"): + return None + + output_dir = os.environ.get("PRESTO_OUTPUT_DIR") + if not output_dir: + pytest.exit("PRESTO_OUTPUT_DIR must be set when --write-results-to-file is enabled.", returncode=2) + + tag = request.config.getoption("--tag") + schema = ctas_schema_name(tag) + host_results_dir = Path(output_dir).expanduser() / schema + + conn = prestodb.dbapi.connect( + host=request.config.getoption("--hostname"), + port=request.config.getoption("--port"), + user=request.config.getoption("--user"), + catalog="hive", + ) + cursor = conn.cursor() + try: + # Drop registered managed tables before removing any unregistered files + # that may have survived an interrupted run. + _drop_results_schema(cursor, CTAS_CATALOG, schema) + if host_results_dir.exists(): + shutil.rmtree(host_results_dir) + cursor.execute(f"CREATE SCHEMA {CTAS_CATALOG}.{schema}").fetchall() + except Exception as error: + pytest.exit(f"Failed to prepare CTAS result output: {error}", returncode=2) + finally: + cursor.close() + conn.close() + + return CtasResults(catalog=CTAS_CATALOG, schema=schema) + @pytest.fixture(scope="session", autouse=True) def run_context_collector(request): @@ -71,13 +153,13 @@ def presto_cursor(request): @pytest.fixture(scope="module") -def benchmark_query(request, presto_cursor, benchmark_queries, benchmark_result_collector): +def benchmark_query(request, presto_cursor, benchmark_queries, benchmark_result_collector, ctas_results): iterations = request.config.getoption("--iterations") profile = request.config.getoption("--profile") profile_script_path = request.config.getoption("--profile-script-path") metrics = request.config.getoption("--metrics") benchmark_type = request.node.obj.BENCHMARK_TYPE - bench_output_dir = request.config.getoption("--output-dir") + bench_output_dir = get_output_dir(request.config) hostname = request.config.getoption("--hostname") port = request.config.getoption("--port") @@ -100,6 +182,7 @@ def benchmark_query(request, presto_cursor, benchmark_queries, benchmark_result_ def benchmark_query_function(query_id): profile_output_file_path = None + scratch_table = None try: if profile: # Base path without .nsys-rep extension: {dir}/{query_id} @@ -107,13 +190,27 @@ def benchmark_query_function(query_id): start_profiler(profile_script_path, profile_output_file_path) result = [] for iteration_num in range(iterations): - cursor = presto_cursor.execute( - "--" + str(benchmark_type) + "_" + str(query_id) + "--" + "\n" + benchmark_queries[query_id] - ) - result.append(cursor.stats["elapsedTimeMillis"]) + if ctas_results is not None: + table = ctas_table_name(query_id, iteration_num) + scratch_table = table if iteration_num > 0 else None + query = build_ctas_query( + benchmark_type, + query_id, + benchmark_queries[query_id], + ctas_results.catalog, + ctas_results.schema, + table, + ) + else: + query = "--" + str(benchmark_type) + "_" + str(query_id) + "--" + "\n" + benchmark_queries[query_id] - # Save query results to Parquet (only on first iteration) - if iteration_num == 0: + cursor = presto_cursor.execute(query) + + if ctas_results is not None: + # CTAS returns only its update count to the client. Consuming it + # ensures all worker writes have completed before recording stats. + cursor.fetchall() + elif iteration_num == 0: rows = cursor.fetchall() columns = [desc[0] for desc in cursor.description] df = pd.DataFrame(rows, columns=columns) @@ -124,6 +221,8 @@ def benchmark_query_function(query_id): parquet_path = results_dir / f"{query_id.lower()}.parquet" df.to_parquet(parquet_path, index=False) + result.append(cursor.stats["elapsedTimeMillis"]) + # Collect metrics after each query iteration if enabled if metrics: presto_query_id = cursor._query.query_id @@ -135,12 +234,25 @@ def benchmark_query_function(query_id): port=port, output_dir=bench_output_dir, ) + + if scratch_table is not None: + presto_cursor.execute( + f"DROP TABLE IF EXISTS {ctas_results.catalog}.{ctas_results.schema}.{scratch_table}" + ).fetchall() + scratch_table = None raw_times_dict[query_id] = result except Exception as e: - failed_queries_dict[query_id] = f"{e.error_type}: {e.error_name}" + error_type = getattr(e, "error_type", type(e).__name__) + error_name = getattr(e, "error_name", str(e)) + failed_queries_dict[query_id] = f"{error_type}: {error_name}" raw_times_dict[query_id] = None raise finally: + if scratch_table is not None and ctas_results is not None: + with suppress(Exception): + presto_cursor.execute( + f"DROP TABLE IF EXISTS {ctas_results.catalog}.{ctas_results.schema}.{scratch_table}" + ).fetchall() if profile and profile_output_file_path is not None: stop_profiler(profile_script_path, profile_output_file_path) diff --git a/presto/testing/performance_benchmarks/conftest.py b/presto/testing/performance_benchmarks/conftest.py index e791501d..f160b7da 100644 --- a/presto/testing/performance_benchmarks/conftest.py +++ b/presto/testing/performance_benchmarks/conftest.py @@ -25,6 +25,7 @@ ) from .common_fixtures import ( benchmark_query, # noqa: F401 + ctas_results, # noqa: F401 presto_cursor, # noqa: F401 run_context_collector, # noqa: F401 verify_tables_analyzed, # noqa: F401 @@ -47,6 +48,7 @@ def pytest_addoption(parser): parser.addoption("--metrics", action="store_true", default=False) parser.addoption("--skip-drop-cache", action="store_true", default=False) parser.addoption("--skip-analyze-check", action="store_true", default=False) + parser.addoption("--write-results-to-file", action="store_true", default=False) def pytest_configure(config): diff --git a/tests/test_presto_ctas_results.py b/tests/test_presto_ctas_results.py new file mode 100644 index 00000000..a572a950 --- /dev/null +++ b/tests/test_presto_ctas_results.py @@ -0,0 +1,81 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from pathlib import Path + +import pandas as pd +import pytest + +from common.testing.validate_results import validate +from presto.testing.performance_benchmarks.common_fixtures import ( + _drop_results_schema, + build_ctas_query, + ctas_schema_name, + ctas_table_name, + strip_trailing_semicolon, +) + + +class FakeCursor: + def __init__(self, responses): + self.responses = responses + self.statements = [] + self.current = [] + + def execute(self, statement): + self.statements.append(statement) + self.current = self.responses.get(statement, []) + return self + + def fetchall(self): + return self.current + + +def test_ctas_names_and_sql(): + assert ctas_schema_name(None) == "benchmark_results" + assert ctas_schema_name("GPU_SF100") == "benchmark_results_gpu_sf100" + with pytest.raises(ValueError, match="alphanumeric"): + ctas_schema_name("../other") + assert ctas_table_name("Q7", 0) == "q7" + assert ctas_table_name("Q7", 2) == "q7_iteration_3" + assert strip_trailing_semicolon("SELECT 1;\n") == "SELECT 1" + assert build_ctas_query("tpch", "Q7", "SELECT 1;", "hive_output", "results", "q7") == ( + "--tpch_Q7--\n" + "CREATE TABLE hive_output.results.q7 WITH (format = 'PARQUET') AS\n" + "SELECT 1" + ) + + +def test_validator_reads_distributed_parquet_dataset(tmp_path: Path): + results_dir = tmp_path / "actual" + query_dir = results_dir / "q1" + query_dir.mkdir(parents=True) + pd.DataFrame({"value": [1, 2]}).to_parquet(query_dir / "part-00000.parquet", index=False) + + expected_dir = tmp_path / "expected" + expected_dir.mkdir() + pd.DataFrame({"value": [1, 2]}).to_parquet(expected_dir / "q01.parquet", index=False) + + result = validate(results_dir, expected_dir, {"Q1": "SELECT value FROM source"}) + + assert result["overall_status"] == "passed" + assert result["queries"]["q1"]["status"] == "passed" + + +def test_drop_results_schema_removes_managed_tables_first(): + cursor = FakeCursor( + { + "SHOW SCHEMAS FROM hive_output": [("benchmark_results",)], + "SHOW TABLES FROM hive_output.benchmark_results": [("q1",), ("q2_iteration_2",)], + } + ) + + _drop_results_schema(cursor, "hive_output", "benchmark_results") + + assert cursor.statements == [ + "SHOW SCHEMAS FROM hive_output", + "SHOW TABLES FROM hive_output.benchmark_results", + 'DROP TABLE IF EXISTS hive_output.benchmark_results."q1"', + 'DROP TABLE IF EXISTS hive_output.benchmark_results."q2_iteration_2"', + "DROP SCHEMA hive_output.benchmark_results", + ] From c9cca4a4c5ab10b40a8af7c930ef93f6f02544b9 Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Mon, 20 Jul 2026 09:40:35 -0400 Subject: [PATCH 02/12] Restore ordering for distributed CTAS results --- common/testing/result_comparison.py | 86 ++++++++++++++++++------ common/testing/result_comparison_test.py | 33 +++++++++ common/testing/validate_results.py | 8 ++- tests/test_presto_ctas_results.py | 23 ++++++- 4 files changed, 125 insertions(+), 25 deletions(-) diff --git a/common/testing/result_comparison.py b/common/testing/result_comparison.py index 336abc7e..34559291 100644 --- a/common/testing/result_comparison.py +++ b/common/testing/result_comparison.py @@ -34,22 +34,19 @@ # --------------------------------------------------------------------------- -def get_orderby_col_indices(query_sql: str, expected_col_names: list[str]) -> tuple[list[int], list[bool]]: - """ - Extract ORDER BY column positions and directions from SQL using sqlglot. - - Returns (indices, ascending) where ascending[i] is True iff indices[i] - is sorted ASC. Returns ([], []) when there is no ORDER BY, or when any - ORDER BY expression is too complex to map to a result column (CASE, - aggregate, etc.). - """ - expr = sqlglot.parse_one(query_sql) +def _get_orderby_sort_spec( + query_sql: str, + expected_col_names: list[str], +) -> tuple[list[int], list[bool], list[bool]]: + """Resolve ORDER BY expressions to positional columns and sort options.""" + expr = sqlglot.parse_one(query_sql, read="presto") order = next((e for e in expr.find_all(sqlglot.exp.Order)), None) if not order: - return [], [] + return [], [], [] sort_col_indices: list[int] = [] ascending: list[bool] = [] + nulls_first: list[bool] = [] for ordered in order.expressions: key = ordered.this @@ -76,10 +73,25 @@ def get_orderby_col_indices(query_sql: str, expected_col_names: list[str]) -> tu f"ORDER BY expression {key.sql()!r} couldn't be mapped to a result column; " "ORDER BY validation and tie-boundary handling will be skipped for this query." ) - return [], [] + return [], [], [] sort_col_indices.append(resolved_idx) ascending.append(not is_desc) + nulls_first.append(bool(ordered.args.get("nulls_first"))) + + return sort_col_indices, ascending, nulls_first + + +def get_orderby_col_indices(query_sql: str, expected_col_names: list[str]) -> tuple[list[int], list[bool]]: + """ + Extract ORDER BY column positions and directions from SQL using sqlglot. + + Returns (indices, ascending) where ascending[i] is True iff indices[i] + is sorted ASC. Returns ([], []) when there is no ORDER BY, or when any + ORDER BY expression is too complex to map to a result column (CASE, + aggregate, etc.). + """ + sort_col_indices, ascending, _ = _get_orderby_sort_spec(query_sql, expected_col_names) return sort_col_indices, ascending @@ -223,11 +235,10 @@ def _validate_orderby(df: pd.DataFrame, sort_col_indices: list[int], ascending: itself. Null handling: SQL leaves the position of NULLs in ORDER BY engine- - defined (Presto/Velox default to NULLS LAST for ASC, NULLS FIRST for - DESC, others differ), so adjacent pairs where either side is NULL are - not counted as violations. Two NULLs are treated as tied (do not break - the outer tie group); a NULL adjacent to a non-NULL value is treated as - a change for tie-group tracking only. + defined unless NULLS FIRST/LAST is explicit, so adjacent pairs where + either side is NULL are not counted as violations. Two NULLs are treated + as tied (do not break the outer tie group); a NULL adjacent to a non-NULL + value is treated as a change for tie-group tracking only. Raises AssertionError pointing at the first offending row. A no-op when sort_col_indices is empty or the frame has 0-1 rows. @@ -288,6 +299,32 @@ def _validate_orderby(df: pd.DataFrame, sort_col_indices: list[int], ascending: prior_changed = prior_changed | value_changed +def _restore_orderby( + df: pd.DataFrame, + sort_col_indices: list[int], + ascending: list[bool], + nulls_first: list[bool], +) -> pd.DataFrame: + """Reconstruct SQL result order after unordered distributed storage.""" + if not sort_col_indices or len(df) <= 1: + return df.reset_index(drop=True) + + # Apply stable sorts from the least-significant key to the most-significant + # key. Access columns positionally because SQL results may contain duplicate + # column labels. + row_order = np.arange(len(df)) + for col_idx, asc, null_first in reversed(list(zip(sort_col_indices, ascending, nulls_first))): + values = df.iloc[row_order, col_idx].reset_index(drop=True) + permutation = values.sort_values( + ascending=asc, + na_position="first" if null_first else "last", + kind="stable", + ).index.to_numpy() + row_order = row_order[permutation] + + return df.iloc[row_order].reset_index(drop=True) + + def _canonical_sort(df: pd.DataFrame) -> pd.DataFrame: """ Sort df by [all non-float columns, all float columns] for a deterministic, @@ -409,6 +446,7 @@ def compare_result_frames( expected: pd.DataFrame, query_sql: str, actual_column_types: list[str] | None = None, + actual_order_preserved: bool = True, ) -> None: """ Full comparison pipeline. Raises AssertionError on any mismatch. @@ -420,7 +458,8 @@ def compare_result_frames( 1. Validate column count and capture expected column names. 2. Normalize dtypes / value types. 3. Validate row count. - 4. Parse ORDER BY / LIMIT from SQL. + 4. Parse ORDER BY / LIMIT from SQL and reconstruct logical order when + the actual rows came from unordered distributed storage. 5. Validate each engine respected ORDER BY (per-frame, strict equality). 6. With LIMIT: peel the tied tail in engine order, then canonical-sort each piece. Compare front fully; compare tail's ORDER BY values only. @@ -443,9 +482,13 @@ def compare_result_frames( if len(actual) != len(expected): raise AssertionError(f"Row count mismatch: {len(actual)} (actual) vs {len(expected)} (expected)") - # 4. Parse ORDER BY / LIMIT - sort_col_indices, ascending = get_orderby_col_indices(query_sql, expected_col_names) + # 4. Parse ORDER BY / LIMIT. A distributed Parquet dataset has no physical + # row-order guarantee, so reconstruct its logical SQL order before applying + # the same ordered-result checks used for coordinator-returned results. + sort_col_indices, ascending, nulls_first = _get_orderby_sort_spec(query_sql, expected_col_names) limit = get_limit(query_sql) + if not actual_order_preserved: + actual = _restore_orderby(actual, sort_col_indices, ascending, nulls_first) # 5. Per-frame ORDER BY validation — engine order is intact here. _validate_orderby(actual, sort_col_indices, ascending) @@ -492,6 +535,7 @@ def validate_query_result( actual: pd.DataFrame, expected: pd.DataFrame, query_sql: str, + actual_order_preserved: bool = True, ) -> tuple[ValidationStatus, str | None]: """ Compare actual vs expected for one query. Returns (status, message). @@ -509,7 +553,7 @@ def validate_query_result( ) try: - compare_result_frames(actual, expected, query_sql) + compare_result_frames(actual, expected, query_sql, actual_order_preserved=actual_order_preserved) return "passed", None except Exception as e: return "failed", f"{type(e).__name__}: {e}"[:500] diff --git a/common/testing/result_comparison_test.py b/common/testing/result_comparison_test.py index be5ec902..61384ca7 100644 --- a/common/testing/result_comparison_test.py +++ b/common/testing/result_comparison_test.py @@ -12,8 +12,11 @@ from common.testing.result_comparison import ( _canonical_sort, _find_last_tie_start, + _get_orderby_sort_spec, _normalize_to_expected, + _restore_orderby, _validate_orderby, + compare_result_frames, ) # --------------------------------------------------------------------------- @@ -103,6 +106,36 @@ def test_validate_orderby_handles_duplicate_column_labels(): _validate_orderby(df, sort_col_indices=[0], ascending=[True]) +def test_restore_orderby_uses_mixed_directions_and_explicit_null_placement(): + df = pd.DataFrame( + { + "a": [2, 1, None, 1, 1], + "b": [4, 1, 9, None, 3], + "payload": ["a2", "a1-low", "null-a", "a1-null", "a1-high"], + } + ) + indices, ascending, nulls_first = _get_orderby_sort_spec( + "SELECT a, b, payload FROM source ORDER BY a ASC NULLS LAST, b DESC NULLS FIRST", + list(df.columns), + ) + + restored = _restore_orderby(df, indices, ascending, nulls_first) + + assert restored["payload"].tolist() == ["a1-null", "a1-high", "a1-low", "a2", "null-a"] + + +def test_compare_restores_unordered_limit_result_before_tie_handling(): + actual = pd.DataFrame({"score": [5.0, 10.0, 5.0], "payload": ["different-tie", "top", "left-tie"]}) + expected = pd.DataFrame({"score": [10.0, 5.0, 5.0], "payload": ["top", "left-tie", "right-tie"]}) + + compare_result_frames( + actual, + expected, + "SELECT score, payload FROM source ORDER BY score DESC LIMIT 3", + actual_order_preserved=False, + ) + + # --------------------------------------------------------------------------- # _validate_orderby null handling # --------------------------------------------------------------------------- diff --git a/common/testing/validate_results.py b/common/testing/validate_results.py index b3f18823..7a614c69 100644 --- a/common/testing/validate_results.py +++ b/common/testing/validate_results.py @@ -108,7 +108,13 @@ def validate( not_validated += 1 continue - status, msg = validate_query_result(query_id, actual, expected, query_sql) + status, msg = validate_query_result( + query_id, + actual, + expected, + query_sql, + actual_order_preserved=result_file.is_file(), + ) if status == "not-validated": query_results[query_id] = {"status": "not-validated", "message": msg} diff --git a/tests/test_presto_ctas_results.py b/tests/test_presto_ctas_results.py index a572a950..6245a0ee 100644 --- a/tests/test_presto_ctas_results.py +++ b/tests/test_presto_ctas_results.py @@ -50,18 +50,35 @@ def test_validator_reads_distributed_parquet_dataset(tmp_path: Path): results_dir = tmp_path / "actual" query_dir = results_dir / "q1" query_dir.mkdir(parents=True) - pd.DataFrame({"value": [1, 2]}).to_parquet(query_dir / "part-00000.parquet", index=False) + # Distributed writer files do not preserve the query's global row order. + pd.DataFrame({"value": [3, 4]}).to_parquet(query_dir / "part-00000.parquet", index=False) + pd.DataFrame({"value": [1, 2]}).to_parquet(query_dir / "part-00001.parquet", index=False) expected_dir = tmp_path / "expected" expected_dir.mkdir() - pd.DataFrame({"value": [1, 2]}).to_parquet(expected_dir / "q01.parquet", index=False) + pd.DataFrame({"value": [1, 2, 3, 4]}).to_parquet(expected_dir / "q01.parquet", index=False) - result = validate(results_dir, expected_dir, {"Q1": "SELECT value FROM source"}) + result = validate(results_dir, expected_dir, {"Q1": "SELECT value FROM source ORDER BY value"}) assert result["overall_status"] == "passed" assert result["queries"]["q1"]["status"] == "passed" +def test_validator_keeps_single_file_result_order_sensitive(tmp_path: Path): + results_dir = tmp_path / "actual" + results_dir.mkdir() + pd.DataFrame({"value": [2, 1]}).to_parquet(results_dir / "q1.parquet", index=False) + + expected_dir = tmp_path / "expected" + expected_dir.mkdir() + pd.DataFrame({"value": [1, 2]}).to_parquet(expected_dir / "q01.parquet", index=False) + + result = validate(results_dir, expected_dir, {"Q1": "SELECT value FROM source ORDER BY value"}) + + assert result["overall_status"] == "failed" + assert "Engine violated ORDER BY" in result["queries"]["q1"]["message"] + + def test_drop_results_schema_removes_managed_tables_first(): cursor = FakeCursor( { From caaafec513d6f70471622952d0fa633c8b270c29 Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Mon, 20 Jul 2026 09:45:03 -0400 Subject: [PATCH 03/12] Reject duplicate benchmark result representations --- common/testing/validate_results.py | 14 +++++++++++++- tests/test_presto_ctas_results.py | 21 +++++++++++++++++++++ 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/common/testing/validate_results.py b/common/testing/validate_results.py index 7a614c69..d7f2cb8d 100644 --- a/common/testing/validate_results.py +++ b/common/testing/validate_results.py @@ -65,8 +65,20 @@ def validate( print(f"No result Parquet files or datasets found in {results_dir}", file=sys.stderr) return {"overall_status": "not-validated", "queries": {}} + results_by_query: dict[str, list[Path]] = {} for result_file in result_files: - query_id = result_file.stem # e.g. "q1" for q1 or q1.parquet + results_by_query.setdefault(result_file.stem, []).append(result_file) + + for query_id, query_paths in sorted(results_by_query.items(), key=lambda item: int(item[0].lstrip("q"))): + if len(query_paths) != 1: + names = ", ".join(sorted(path.name for path in query_paths)) + msg = f"multiple result representations found: {names}" + print(f"[Validation] {query_id.upper():4s}: FAIL {msg}") + query_results[query_id] = {"status": "failed", "message": msg} + failed += 1 + continue + + result_file = query_paths[0] q_num = int(query_id.lstrip("q")) # Accepted naming conventions for expected files (tried in order): diff --git a/tests/test_presto_ctas_results.py b/tests/test_presto_ctas_results.py index 6245a0ee..5cce23ad 100644 --- a/tests/test_presto_ctas_results.py +++ b/tests/test_presto_ctas_results.py @@ -79,6 +79,27 @@ def test_validator_keeps_single_file_result_order_sensitive(tmp_path: Path): assert "Engine violated ORDER BY" in result["queries"]["q1"]["message"] +@pytest.mark.parametrize("query_numbers", [None, [1]]) +def test_validator_rejects_file_and_dataset_for_same_query(tmp_path: Path, query_numbers): + results_dir = tmp_path / "actual" + query_dir = results_dir / "q1" + query_dir.mkdir(parents=True) + pd.DataFrame({"value": [1]}).to_parquet(query_dir / "part-00000.parquet", index=False) + pd.DataFrame({"value": [1]}).to_parquet(results_dir / "q1.parquet", index=False) + + expected_dir = tmp_path / "expected" + expected_dir.mkdir() + pd.DataFrame({"value": [1]}).to_parquet(expected_dir / "q01.parquet", index=False) + + result = validate(results_dir, expected_dir, {"Q1": "SELECT value FROM source"}, query_numbers=query_numbers) + + assert result["overall_status"] == "failed" + assert result["queries"]["q1"] == { + "status": "failed", + "message": "multiple result representations found: q1, q1.parquet", + } + + def test_drop_results_schema_removes_managed_tables_first(): cursor = FakeCursor( { From a8368aae18c3ae0c00b1a423604f9059d97e63d1 Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Mon, 20 Jul 2026 09:50:44 -0400 Subject: [PATCH 04/12] Allow empty distributed CTAS results --- presto/scripts/run_benchmark.sh | 15 +++++++++------ tests/test_presto_ctas_results.py | 16 ++++++++++++++++ 2 files changed, 25 insertions(+), 6 deletions(-) diff --git a/presto/scripts/run_benchmark.sh b/presto/scripts/run_benchmark.sh index 830ac987..f698220e 100755 --- a/presto/scripts/run_benchmark.sh +++ b/presto/scripts/run_benchmark.sh @@ -364,10 +364,12 @@ normalize_ctas_result_permissions() { fi # Slurm and non-remapped local files are normally owned by the invoking user. + find "${host_results_dir}" -type d -exec chmod a+rx {} + 2>/dev/null || true find "${host_results_dir}" -type f -name '*.parquet' -exec chmod a+r {} + 2>/dev/null || true - local parquet_file - parquet_file="$(find "${host_results_dir}" -type f -name '*.parquet' -print -quit 2>/dev/null || true)" - if [[ -n "${parquet_file}" && -z "$(find "${host_results_dir}" -type f -name '*.parquet' ! -readable -print -quit 2>/dev/null || true)" ]]; then + local committed_table + committed_table="$(find "${host_results_dir}" -mindepth 1 -maxdepth 1 -type d -name 'q[0-9]*' \ + -exec test -f '{}/.prestoSchema' \; -print -quit 2>/dev/null || true)" + if [[ -n "${committed_table}" && -z "$(find "${host_results_dir}" -type f -name '*.parquet' ! -readable -print -quit 2>/dev/null || true)" ]]; then return 0 fi @@ -385,9 +387,10 @@ normalize_ctas_result_permissions() { done < <(docker ps --format '{{.Names}}' | grep -E '^presto-.*worker') fi - parquet_file="$(find "${host_results_dir}" -type f -name '*.parquet' -print -quit 2>/dev/null || true)" - if [[ -z "${parquet_file}" ]]; then - echo "Error: No CTAS Parquet result files found in ${host_results_dir}" >&2 + committed_table="$(find "${host_results_dir}" -mindepth 1 -maxdepth 1 -type d -name 'q[0-9]*' \ + -exec test -f '{}/.prestoSchema' \; -print -quit 2>/dev/null || true)" + if [[ -z "${committed_table}" ]]; then + echo "Error: No committed CTAS result tables found in ${host_results_dir}" >&2 return 1 fi if [[ -n "$(find "${host_results_dir}" -type f -name '*.parquet' ! -readable -print -quit 2>/dev/null || true)" ]]; then diff --git a/tests/test_presto_ctas_results.py b/tests/test_presto_ctas_results.py index 5cce23ad..323d0d0f 100644 --- a/tests/test_presto_ctas_results.py +++ b/tests/test_presto_ctas_results.py @@ -64,6 +64,22 @@ def test_validator_reads_distributed_parquet_dataset(tmp_path: Path): assert result["queries"]["q1"]["status"] == "passed" +def test_validator_accepts_empty_distributed_q15_dataset(tmp_path: Path): + results_dir = tmp_path / "actual" + query_dir = results_dir / "q15" + query_dir.mkdir(parents=True) + (query_dir / ".prestoSchema").write_text("{}") + + expected_dir = tmp_path / "expected" + expected_dir.mkdir() + pd.DataFrame({"supplier_no": [1]}).to_parquet(expected_dir / "q15.parquet", index=False) + + result = validate(results_dir, expected_dir, {"Q15": "SELECT supplier_no FROM source"}) + + assert result["overall_status"] == "expected-failure" + assert result["queries"]["q15"]["status"] == "expected-failure" + + def test_validator_keeps_single_file_result_order_sensitive(tmp_path: Path): results_dir = tmp_path / "actual" results_dir.mkdir() From 97aa11e7364aa4b500f9327ff474c90a3e11210f Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Mon, 20 Jul 2026 11:36:18 -0400 Subject: [PATCH 05/12] Colocate CTAS unit tests --- .../testing/validate_results_test.py | 56 ----------------- .../performance_benchmark_fixtures_test.py | 61 +++++++++++++++++++ 2 files changed, 61 insertions(+), 56 deletions(-) rename tests/test_presto_ctas_results.py => common/testing/validate_results_test.py (63%) create mode 100644 presto/testing/performance_benchmark_fixtures_test.py diff --git a/tests/test_presto_ctas_results.py b/common/testing/validate_results_test.py similarity index 63% rename from tests/test_presto_ctas_results.py rename to common/testing/validate_results_test.py index 323d0d0f..03e02e57 100644 --- a/tests/test_presto_ctas_results.py +++ b/common/testing/validate_results_test.py @@ -7,43 +7,6 @@ import pytest from common.testing.validate_results import validate -from presto.testing.performance_benchmarks.common_fixtures import ( - _drop_results_schema, - build_ctas_query, - ctas_schema_name, - ctas_table_name, - strip_trailing_semicolon, -) - - -class FakeCursor: - def __init__(self, responses): - self.responses = responses - self.statements = [] - self.current = [] - - def execute(self, statement): - self.statements.append(statement) - self.current = self.responses.get(statement, []) - return self - - def fetchall(self): - return self.current - - -def test_ctas_names_and_sql(): - assert ctas_schema_name(None) == "benchmark_results" - assert ctas_schema_name("GPU_SF100") == "benchmark_results_gpu_sf100" - with pytest.raises(ValueError, match="alphanumeric"): - ctas_schema_name("../other") - assert ctas_table_name("Q7", 0) == "q7" - assert ctas_table_name("Q7", 2) == "q7_iteration_3" - assert strip_trailing_semicolon("SELECT 1;\n") == "SELECT 1" - assert build_ctas_query("tpch", "Q7", "SELECT 1;", "hive_output", "results", "q7") == ( - "--tpch_Q7--\n" - "CREATE TABLE hive_output.results.q7 WITH (format = 'PARQUET') AS\n" - "SELECT 1" - ) def test_validator_reads_distributed_parquet_dataset(tmp_path: Path): @@ -114,22 +77,3 @@ def test_validator_rejects_file_and_dataset_for_same_query(tmp_path: Path, query "status": "failed", "message": "multiple result representations found: q1, q1.parquet", } - - -def test_drop_results_schema_removes_managed_tables_first(): - cursor = FakeCursor( - { - "SHOW SCHEMAS FROM hive_output": [("benchmark_results",)], - "SHOW TABLES FROM hive_output.benchmark_results": [("q1",), ("q2_iteration_2",)], - } - ) - - _drop_results_schema(cursor, "hive_output", "benchmark_results") - - assert cursor.statements == [ - "SHOW SCHEMAS FROM hive_output", - "SHOW TABLES FROM hive_output.benchmark_results", - 'DROP TABLE IF EXISTS hive_output.benchmark_results."q1"', - 'DROP TABLE IF EXISTS hive_output.benchmark_results."q2_iteration_2"', - "DROP SCHEMA hive_output.benchmark_results", - ] diff --git a/presto/testing/performance_benchmark_fixtures_test.py b/presto/testing/performance_benchmark_fixtures_test.py new file mode 100644 index 00000000..e2fc1d00 --- /dev/null +++ b/presto/testing/performance_benchmark_fixtures_test.py @@ -0,0 +1,61 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import pytest + +from presto.testing.performance_benchmarks.common_fixtures import ( + _drop_results_schema, + build_ctas_query, + ctas_schema_name, + ctas_table_name, + strip_trailing_semicolon, +) + + +class FakeCursor: + def __init__(self, responses): + self.responses = responses + self.statements = [] + self.current = [] + + def execute(self, statement): + self.statements.append(statement) + self.current = self.responses.get(statement, []) + return self + + def fetchall(self): + return self.current + + +def test_ctas_names_and_sql(): + assert ctas_schema_name(None) == "benchmark_results" + assert ctas_schema_name("GPU_SF100") == "benchmark_results_gpu_sf100" + with pytest.raises(ValueError, match="alphanumeric"): + ctas_schema_name("../other") + assert ctas_table_name("Q7", 0) == "q7" + assert ctas_table_name("Q7", 2) == "q7_iteration_3" + assert strip_trailing_semicolon("SELECT 1;\n") == "SELECT 1" + assert build_ctas_query("tpch", "Q7", "SELECT 1;", "hive_output", "results", "q7") == ( + "--tpch_Q7--\n" + "CREATE TABLE hive_output.results.q7 WITH (format = 'PARQUET') AS\n" + "SELECT 1" + ) + + +def test_drop_results_schema_removes_managed_tables_first(): + cursor = FakeCursor( + { + "SHOW SCHEMAS FROM hive_output": [("benchmark_results",)], + "SHOW TABLES FROM hive_output.benchmark_results": [("q1",), ("q2_iteration_2",)], + } + ) + + _drop_results_schema(cursor, "hive_output", "benchmark_results") + + assert cursor.statements == [ + "SHOW SCHEMAS FROM hive_output", + "SHOW TABLES FROM hive_output.benchmark_results", + 'DROP TABLE IF EXISTS hive_output.benchmark_results."q1"', + 'DROP TABLE IF EXISTS hive_output.benchmark_results."q2_iteration_2"', + "DROP SCHEMA hive_output.benchmark_results", + ] From 66614f9e03ee54f180b08ee94e77a61b48464242 Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Mon, 20 Jul 2026 15:09:21 -0400 Subject: [PATCH 06/12] Alias unnamed CTAS output columns --- .../performance_benchmark_fixtures_test.py | 43 +++++++++++++++- .../performance_benchmarks/common_fixtures.py | 49 ++++++++++++++++++- 2 files changed, 90 insertions(+), 2 deletions(-) diff --git a/presto/testing/performance_benchmark_fixtures_test.py b/presto/testing/performance_benchmark_fixtures_test.py index e2fc1d00..51c3d899 100644 --- a/presto/testing/performance_benchmark_fixtures_test.py +++ b/presto/testing/performance_benchmark_fixtures_test.py @@ -1,10 +1,14 @@ # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import json +from pathlib import Path + import pytest from presto.testing.performance_benchmarks.common_fixtures import ( _drop_results_schema, + alias_ctas_output_expressions, build_ctas_query, ctas_schema_name, ctas_table_name, @@ -38,10 +42,47 @@ def test_ctas_names_and_sql(): assert build_ctas_query("tpch", "Q7", "SELECT 1;", "hive_output", "results", "q7") == ( "--tpch_Q7--\n" "CREATE TABLE hive_output.results.q7 WITH (format = 'PARQUET') AS\n" - "SELECT 1" + "SELECT 1 AS result_column_1" + ) + + +def test_ctas_aliases_unnamed_and_duplicate_output_expressions(): + query = ( + "SELECT c_name, sum(l_quantity), c_name, count(*) AS row_count " + "FROM customer, lineitem GROUP BY c_name" + ) + + rewritten = alias_ctas_output_expressions(query) + + assert rewritten == ( + "SELECT c_name, SUM(l_quantity) AS result_column_2, " + "c_name AS result_column_3, COUNT(*) AS row_count " + "FROM customer, lineitem GROUP BY c_name" ) +def test_ctas_alias_rewrite_leaves_stars_for_presto_to_expand(): + assert alias_ctas_output_expressions("SELECT nested.* FROM (SELECT 1 AS value) nested") == ( + "SELECT nested.* FROM (SELECT 1 AS value) nested" + ) + + +@pytest.mark.parametrize("benchmark_type", ["tpch", "tpcds"]) +def test_ctas_alias_rewrite_parses_all_builtin_queries(benchmark_type): + repo_root = Path(__file__).parents[2] + query_path = repo_root / "common" / "testing" / "queries" / benchmark_type / "queries.json" + queries = json.loads(query_path.read_text()) + + rewritten = { + query_id: alias_ctas_output_expressions(query.replace("{SF_FRACTION}", "0.0001")) + for query_id, query in queries.items() + } + + assert rewritten.keys() == queries.keys() + if benchmark_type == "tpch": + assert "SUM(l_quantity) AS result_column_6" in rewritten["Q18"] + + def test_drop_results_schema_removes_managed_tables_first(): cursor = FakeCursor( { diff --git a/presto/testing/performance_benchmarks/common_fixtures.py b/presto/testing/performance_benchmarks/common_fixtures.py index 92935b9f..4c70f6bf 100644 --- a/presto/testing/performance_benchmarks/common_fixtures.py +++ b/presto/testing/performance_benchmarks/common_fixtures.py @@ -12,6 +12,8 @@ import pandas as pd import prestodb import pytest +import sqlglot +from sqlglot import exp from common.testing.performance_benchmarks.benchmark_keys import BenchmarkKeys from common.testing.performance_benchmarks.conftest import get_output_dir @@ -44,11 +46,56 @@ def ctas_table_name(query_id, iteration_num): return query_id.lower() if iteration_num == 0 else f"{query_id.lower()}_iteration_{iteration_num + 1}" +def alias_ctas_output_expressions(query): + """Give every explicit CTAS output expression a unique column name.""" + query = strip_trailing_semicolon(query) + parsed = sqlglot.parse_one(query, read="presto") + select = next(parsed.find_all(exp.Select)) + expressions = list(select.expressions) + + reserved_names = { + expression.alias_or_name.casefold() + for expression in expressions + if isinstance(expression, (exp.Alias, exp.Column)) and not expression.is_star + } + used_names = set() + changed = False + + for position, expression in enumerate(expressions, start=1): + if expression.is_star: + continue + + name = expression.alias_or_name if isinstance(expression, (exp.Alias, exp.Column)) else "" + normalized_name = name.casefold() + if name and normalized_name not in used_names: + used_names.add(normalized_name) + continue + + base_alias = f"result_column_{position}" + alias = base_alias + suffix = 2 + while alias.casefold() in reserved_names or alias.casefold() in used_names: + alias = f"{base_alias}_{suffix}" + suffix += 1 + + if isinstance(expression, exp.Alias): + expression.set("alias", exp.to_identifier(alias)) + else: + expressions[position - 1] = exp.alias_(expression, alias, copy=False) + used_names.add(alias.casefold()) + changed = True + + if not changed: + return query + select.set("expressions", expressions) + return parsed.sql(dialect="presto") + + def build_ctas_query(benchmark_type, query_id, query, catalog, schema, table): return ( f"--{benchmark_type}_{query_id}--\n" f"CREATE TABLE {catalog}.{schema}.{table} WITH (format = 'PARQUET') AS\n" - f"{strip_trailing_semicolon(query)}" + f"{alias_ctas_output_expressions(query)}" ) From 132e13d1344f34b67be9b5645bc308ae85fb4d4b Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Tue, 21 Jul 2026 10:16:09 -0400 Subject: [PATCH 07/12] Mount expected benchmark results in Slurm runs --- presto/slurm/presto-nvl72/README.md | 13 ++++++++++++ .../presto-nvl72/cluster_config.env.example | 5 +++++ presto/slurm/presto-nvl72/functions.sh | 21 ++++++++++++++++--- presto/slurm/presto-nvl72/launch-run.sh | 14 +++++++++++-- 4 files changed, 48 insertions(+), 5 deletions(-) diff --git a/presto/slurm/presto-nvl72/README.md b/presto/slurm/presto-nvl72/README.md index c799e801..fda4a513 100644 --- a/presto/slurm/presto-nvl72/README.md +++ b/presto/slurm/presto-nvl72/README.md @@ -115,6 +115,18 @@ profiling, metrics, …). In CTAS mode, benchmark reports remain in `result_dir/`, while `benchmark_results_/qN/` Parquet datasets are written beneath `PRESTO_OUTPUT_DIR` (`benchmark_results/qN/` without a tag). +To validate benchmark output, set the host-side reference directory in +`~/.cluster_config.env`: + +```bash +PRESTO_EXPECTED_RESULTS_DIR=/shared/reference-results/tpch-sf3000 +``` + +When the directory exists, `launch-run.sh` mounts it read-only at +`/var/lib/presto/expected-results` in the CLI container. If it does not exist, +the run continues without validation. The directory may be outside `VT_ROOT`; +no `CLUSTER_EXTRA_MOUNTS` entry is needed. + **Prerequisites:** worker + coord images on disk; data from step 1; analyzed metastore from step 2 (either local or shared). @@ -159,6 +171,7 @@ All cluster-specific values come from `~/.cluster_config.env`. See | `CLUSTER_GPU_DEFAULT_COORD_IMAGE` / `CLUSTER_CPU_DEFAULT_COORD_IMAGE` | Default coordinator image name | | `DATA` | TPC-H parquet data root (parent of `tpch-rs-/`) | | `IMAGE_DIR` | Directory containing `.sqsh` container images | +| `PRESTO_EXPECTED_RESULTS_DIR` | Optional host directory of expected Parquet results, mounted read-only for validation | Any variable can also be exported in your shell before invoking a launcher to override the config for a single run. diff --git a/presto/slurm/presto-nvl72/cluster_config.env.example b/presto/slurm/presto-nvl72/cluster_config.env.example index 11161f47..e485f6d7 100644 --- a/presto/slurm/presto-nvl72/cluster_config.env.example +++ b/presto/slurm/presto-nvl72/cluster_config.env.example @@ -29,6 +29,11 @@ DATA=/path/to/tpch/data # Directory containing .sqsh container images. IMAGE_DIR=/path/to/images +# Optional directory containing qN.parquet files (or qN/ Parquet datasets) +# used to validate benchmark results. This is a host path on shared storage; +# launch-run.sh mounts it read-only into the Presto CLI container. +# PRESTO_EXPECTED_RESULTS_DIR=/path/to/expected-results + # Root for shared pre-analyzed Hive metastore snapshots (shared across users). # Unset (comment out) to disable metastore sharing and always run ANALYZE locally. HIVE_METASTORE_SHARED_ROOT=/path/to/shared/hive_metadata diff --git a/presto/slurm/presto-nvl72/functions.sh b/presto/slurm/presto-nvl72/functions.sh index 39d0b673..99412e1c 100644 --- a/presto/slurm/presto-nvl72/functions.sh +++ b/presto/slurm/presto-nvl72/functions.sh @@ -3,6 +3,7 @@ # SPDX-License-Identifier: Apache-2.0 CTAS_CONTAINER_OUTPUT_DIR=/var/lib/presto/data/hive/benchmark_output +EXPECTED_RESULTS_CONTAINER_DIR=/var/lib/presto/expected-results ctas_output_mount_arg() { if [[ "${WRITE_RESULTS_TO_FILE:-0}" == "1" ]]; then @@ -12,6 +13,13 @@ ctas_output_mount_arg() { return 0 } +expected_results_mount_arg() { + if [[ -n "${PRESTO_EXPECTED_RESULTS_DIR:-}" && -d "${PRESTO_EXPECTED_RESULTS_DIR}" ]]; then + printf ',%s:%s:ro' "${PRESTO_EXPECTED_RESULTS_DIR}" "${EXPECTED_RESULTS_CONTAINER_DIR}" + fi + return 0 +} + # Echo the ",:" bind-mount fragment for the host's # miniforge3 install if it exists, else nothing. Used by every container # that needs to run Python via init_python_virtual_env / run_py_script.sh: @@ -127,6 +135,7 @@ function run_coord_image { local extra_mounts extra_mounts="$(miniforge_mount_arg)" extra_mounts+="$(ctas_output_mount_arg)" + [[ "${type}" == "cli" ]] && extra_mounts+="$(expected_results_mount_arg)" if [[ -n "${CLUSTER_EXTRA_MOUNTS:-}" ]]; then extra_mounts="${extra_mounts},${CLUSTER_EXTRA_MOUNTS}" fi @@ -544,6 +553,14 @@ function run_queries { # (VT_ROOT is bind-mounted as /workspace) so the path stays correct even # if this directory is renamed away from `presto-nvl72`. local container_script_dir="${SCRIPT_DIR/${VT_ROOT}//workspace}" + local container_expected_results_dir="" + if [[ -n "${PRESTO_EXPECTED_RESULTS_DIR:-}" ]]; then + if [[ -d "${PRESTO_EXPECTED_RESULTS_DIR}" ]]; then + container_expected_results_dir="${EXPECTED_RESULTS_CONTAINER_DIR}" + else + echo "[Validation] Warning: PRESTO_EXPECTED_RESULTS_DIR not found: ${PRESTO_EXPECTED_RESULTS_DIR}; validation skipped." + fi + fi local extra_args=() [[ "${ENABLE_METRICS}" == "1" ]] && extra_args+=("-m") @@ -575,9 +592,6 @@ function run_queries { chmod +x "${jq_cache}/jq" fi - # Result validation is intentionally not wired here yet: that belongs - # to PR #275 (upstream validate_results.py) and will be hooked up - # after the PR merges and this branch is rebased. # Cache-drop is skipped because it requires docker (not available on # the cluster). run_coord_image "export PATH=/workspace/.cache/bin:\$PATH; \ @@ -585,6 +599,7 @@ function run_queries { export HOSTNAME=$COORD; \ export PRESTO_DATA_DIR=/var/lib/presto/data/hive/data/user_data; \ export PRESTO_OUTPUT_DIR=${CTAS_CONTAINER_OUTPUT_DIR}; \ + export PRESTO_EXPECTED_RESULTS_DIR=${container_expected_results_dir}; \ export MINIFORGE_HOME=/workspace/miniforge3; \ export HOME=/workspace; \ cd /workspace/presto/scripts; \ diff --git a/presto/slurm/presto-nvl72/launch-run.sh b/presto/slurm/presto-nvl72/launch-run.sh index 4ce862ae..2e6d0cca 100755 --- a/presto/slurm/presto-nvl72/launch-run.sh +++ b/presto/slurm/presto-nvl72/launch-run.sh @@ -77,8 +77,9 @@ Options: Any arguments after -- are passed directly to sbatch. Cluster config (~/.cluster_config.env or \$CLUSTER_CONFIG) supplies partition, -account, time limits, image names, and per-variant defaults. See -cluster_config.env.example. +account, time limits, image names, per-variant defaults, and the optional +PRESTO_EXPECTED_RESULTS_DIR. Expected results are mounted read-only into the +CLI container. See cluster_config.env.example. EOF } @@ -157,6 +158,12 @@ preflight_image "${COORD_IMAGE}" \ "Pull it (see ./pull_ghcr_image.sh) or override with -c " preflight_dir "${DATA}/tpch-rs-${SCALE_FACTOR}" "TPC-H SF${SCALE_FACTOR} data" \ "./launch-gen-data.sh -s ${SCALE_FACTOR} -o ${DATA}/tpch-rs-${SCALE_FACTOR}" +if [[ -n "${PRESTO_EXPECTED_RESULTS_DIR:-}" ]]; then + [[ "${PRESTO_EXPECTED_RESULTS_DIR}" == /* ]] || { + echo "Error: PRESTO_EXPECTED_RESULTS_DIR must be an absolute host path: ${PRESTO_EXPECTED_RESULTS_DIR}" >&2 + exit 1 + } +fi preflight_metastore "${SCALE_FACTOR}" "${ANALYZE_HINT}" # Submit job (include nodes/SF/iterations in file names) @@ -180,6 +187,9 @@ EXPORT_VARS+=",WRITE_RESULTS_TO_FILE=${WRITE_RESULTS_TO_FILE}" if [[ "${WRITE_RESULTS_TO_FILE}" == "1" ]]; then EXPORT_VARS+=",PRESTO_OUTPUT_DIR=${PRESTO_OUTPUT_DIR}" fi +if [[ -n "${PRESTO_EXPECTED_RESULTS_DIR:-}" ]]; then + EXPORT_VARS+=",PRESTO_EXPECTED_RESULTS_DIR=${PRESTO_EXPECTED_RESULTS_DIR}" +fi # Comma-separated query list can't ride EXPORT_VARS (comma is the separator); # export it so sbatch picks it up via the ALL inheritance. [[ -n "${QUERIES}" ]] && export QUERIES From 048d595d6e8d015893cb4f39e492ac4891531a70 Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Tue, 28 Jul 2026 10:23:43 -0700 Subject: [PATCH 08/12] Address CTAS benchmark review feedback --- common/testing/result_comparison_test.py | 22 ++++++++++++--- common/testing/validate_results.py | 6 ++--- presto/scripts/run_benchmark.sh | 18 ++++++++++--- .../performance_benchmark_fixtures_test.py | 9 ++----- .../performance_benchmarks/common_fixtures.py | 27 ++++++++++--------- 5 files changed, 53 insertions(+), 29 deletions(-) diff --git a/common/testing/result_comparison_test.py b/common/testing/result_comparison_test.py index 61384ca7..b11ac2e7 100644 --- a/common/testing/result_comparison_test.py +++ b/common/testing/result_comparison_test.py @@ -124,6 +124,20 @@ def test_restore_orderby_uses_mixed_directions_and_explicit_null_placement(): assert restored["payload"].tolist() == ["a1-null", "a1-high", "a1-low", "a2", "null-a"] +def test_restore_orderby_defaults_to_presto_nulls_last_for_desc(): + df = pd.DataFrame({"score": [None, 1, 2], "payload": ["null", "low", "high"]}) + indices, ascending, nulls_first = _get_orderby_sort_spec( + "SELECT score, payload FROM source ORDER BY score DESC", + list(df.columns), + ) + + restored = _restore_orderby(df, indices, ascending, nulls_first) + + assert ascending == [False] + assert nulls_first == [False] + assert restored["payload"].tolist() == ["high", "low", "null"] + + def test_compare_restores_unordered_limit_result_before_tie_handling(): actual = pd.DataFrame({"score": [5.0, 10.0, 5.0], "payload": ["different-tie", "top", "left-tie"]}) expected = pd.DataFrame({"score": [10.0, 5.0, 5.0], "payload": ["top", "left-tie", "right-tie"]}) @@ -156,15 +170,15 @@ def test_validate_orderby_object_column_all_none_does_not_raise(): def test_validate_orderby_object_column_null_then_value_does_not_raise(): - # NULLS-FIRST style ordering (e.g. Presto DESC default): nulls then ASC - # of non-null values should pass. + # Explicit NULLS-FIRST style ordering: nulls then ASC of non-null values + # should pass. df = pd.DataFrame({0: [None, None, "a", "b", "c"]}, dtype=object) _validate_orderby(df, sort_col_indices=[0], ascending=[True]) def test_validate_orderby_object_column_value_then_null_does_not_raise(): - # NULLS-LAST style ordering (e.g. Presto ASC default): non-null ASC - # values then nulls at the tail should pass. + # Presto's default NULLS-LAST ordering: non-null values then nulls at the + # tail should pass regardless of sort direction. df = pd.DataFrame({0: ["a", "b", "c", None, None]}, dtype=object) _validate_orderby(df, sort_col_indices=[0], ascending=[True]) diff --git a/common/testing/validate_results.py b/common/testing/validate_results.py index d7f2cb8d..2bbd8281 100644 --- a/common/testing/validate_results.py +++ b/common/testing/validate_results.py @@ -57,9 +57,7 @@ def validate( path for path in (results_dir / f"q{q_num}.parquet", results_dir / f"q{q_num}") if path.exists() ) else: - result_files = sorted( - path for path in results_dir.iterdir() if re.fullmatch(r"q\d+(?:\.parquet)?", path.name) - ) + result_files = sorted(path for path in results_dir.iterdir() if re.fullmatch(r"q\d+(?:\.parquet)?", path.name)) if not result_files: print(f"No result Parquet files or datasets found in {results_dir}", file=sys.stderr) @@ -103,6 +101,8 @@ def validate( failed += 1 continue + # PyArrow treats a CTAS dataset containing only `.prestoSchema` as an + # empty, schema-less frame; the empty-result handling below covers it. actual = pd.read_parquet(result_file) expected = pd.read_parquet(expected_file) diff --git a/presto/scripts/run_benchmark.sh b/presto/scripts/run_benchmark.sh index f698220e..695dcafd 100755 --- a/presto/scripts/run_benchmark.sh +++ b/presto/scripts/run_benchmark.sh @@ -352,7 +352,8 @@ fi echo "Using PRESTO_IMAGE_TAG: $PRESTO_IMAGE_TAG" BENCHMARK_TEST_DIR=${TEST_DIR}/performance_benchmarks -pytest -q -s ${BENCHMARK_TEST_DIR}/${BENCHMARK_TYPE}_test.py ${PYTEST_ARGS[*]} +PYTEST_EXIT=0 +pytest -q -s ${BENCHMARK_TEST_DIR}/${BENCHMARK_TYPE}_test.py ${PYTEST_ARGS[*]} || PYTEST_EXIT=$? normalize_ctas_result_permissions() { local host_results_dir=$1 @@ -399,8 +400,9 @@ normalize_ctas_result_permissions() { fi } +NORMALIZE_EXIT=0 if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then - normalize_ctas_result_permissions "${CTAS_RESULTS_DIR}" + normalize_ctas_result_permissions "${CTAS_RESULTS_DIR}" || NORMALIZE_EXIT=$? fi # Snapshot logs and engine configs into the benchmark output directory so that @@ -432,6 +434,7 @@ VALIDATE_REQUIREMENTS="${SCRIPT_DIR}/../testing/requirements.txt" # Resolve reference results directory. # PRESTO_EXPECTED_RESULTS_DIR env var is the implicit fallback (warning if missing). # Explicit --reference-results-dir was already validated before the benchmark ran. +VALIDATION_EXIT=0 if [[ -n ${PRESTO_EXPECTED_RESULTS_DIR} && ! -d ${PRESTO_EXPECTED_RESULTS_DIR} ]]; then echo "[Validation] Warning: PRESTO_EXPECTED_RESULTS_DIR not found: ${PRESTO_EXPECTED_RESULTS_DIR}; validation skipped." else @@ -460,5 +463,14 @@ else "${SCRIPT_DIR}/../../scripts/run_py_script.sh" --quiet \ -p "${VALIDATE_SCRIPT}" \ -r "${VALIDATE_REQUIREMENTS}" \ - -- "${VALIDATE_ARGS[@]}" + -- "${VALIDATE_ARGS[@]}" || VALIDATION_EXIT=$? fi + +# Preserve the benchmark failure as the primary exit status while still +# attempting CTAS permission normalization, artifact collection, and validation. +if [[ ${PYTEST_EXIT} -ne 0 ]]; then + exit "${PYTEST_EXIT}" +elif [[ ${NORMALIZE_EXIT} -ne 0 ]]; then + exit "${NORMALIZE_EXIT}" +fi +exit "${VALIDATION_EXIT}" diff --git a/presto/testing/performance_benchmark_fixtures_test.py b/presto/testing/performance_benchmark_fixtures_test.py index 51c3d899..c486a914 100644 --- a/presto/testing/performance_benchmark_fixtures_test.py +++ b/presto/testing/performance_benchmark_fixtures_test.py @@ -40,17 +40,12 @@ def test_ctas_names_and_sql(): assert ctas_table_name("Q7", 2) == "q7_iteration_3" assert strip_trailing_semicolon("SELECT 1;\n") == "SELECT 1" assert build_ctas_query("tpch", "Q7", "SELECT 1;", "hive_output", "results", "q7") == ( - "--tpch_Q7--\n" - "CREATE TABLE hive_output.results.q7 WITH (format = 'PARQUET') AS\n" - "SELECT 1 AS result_column_1" + "--tpch_Q7--\nCREATE TABLE hive_output.results.q7 WITH (format = 'PARQUET') AS\nSELECT 1 AS result_column_1" ) def test_ctas_aliases_unnamed_and_duplicate_output_expressions(): - query = ( - "SELECT c_name, sum(l_quantity), c_name, count(*) AS row_count " - "FROM customer, lineitem GROUP BY c_name" - ) + query = "SELECT c_name, sum(l_quantity), c_name, count(*) AS row_count FROM customer, lineitem GROUP BY c_name" rewritten = alias_ctas_output_expressions(query) diff --git a/presto/testing/performance_benchmarks/common_fixtures.py b/presto/testing/performance_benchmarks/common_fixtures.py index 4c70f6bf..52fde623 100644 --- a/presto/testing/performance_benchmarks/common_fixtures.py +++ b/presto/testing/performance_benchmarks/common_fixtures.py @@ -257,18 +257,21 @@ def benchmark_query_function(query_id): # CTAS returns only its update count to the client. Consuming it # ensures all worker writes have completed before recording stats. cursor.fetchall() - elif iteration_num == 0: - rows = cursor.fetchall() - columns = [desc[0] for desc in cursor.description] - df = pd.DataFrame(rows, columns=columns) - - # Save to Parquet format to match expected results - results_dir = Path(f"{bench_output_dir}/query_results") - results_dir.mkdir(parents=True, exist_ok=True) - parquet_path = results_dir / f"{query_id.lower()}.parquet" - df.to_parquet(parquet_path, index=False) - - result.append(cursor.stats["elapsedTimeMillis"]) + result.append(cursor.stats["elapsedTimeMillis"]) + else: + # Preserve the historical non-CTAS timing point: record stats + # immediately after execute(), before consuming result pages. + result.append(cursor.stats["elapsedTimeMillis"]) + if iteration_num == 0: + rows = cursor.fetchall() + columns = [desc[0] for desc in cursor.description] + df = pd.DataFrame(rows, columns=columns) + + # Save to Parquet format to match expected results + results_dir = Path(f"{bench_output_dir}/query_results") + results_dir.mkdir(parents=True, exist_ok=True) + parquet_path = results_dir / f"{query_id.lower()}.parquet" + df.to_parquet(parquet_path, index=False) # Collect metrics after each query iteration if enabled if metrics: From 225bd6bc77065af2d8c21094651da06892a8d44c Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Thu, 30 Jul 2026 15:32:00 -0700 Subject: [PATCH 09/12] Normalize CTAS benchmark results --- common/testing/normalize_ctas_results.py | 117 ++++++++++++++++++ common/testing/normalize_ctas_results_test.py | 56 +++++++++ common/testing/result_comparison.py | 15 +-- common/testing/result_comparison_test.py | 13 -- common/testing/validate_results.py | 49 ++------ common/testing/validate_results_test.py | 47 +------ presto/README.md | 17 +-- .../etc_common/catalog/hive_output.properties | 4 +- presto/docker/docker-compose.common.yml | 2 +- presto/scripts/run_benchmark.sh | 84 +++++++------ presto/scripts/run_multiple_benchmarks.sh | 12 +- .../scripts/start_presto_helper_parse_args.sh | 14 +-- presto/slurm/presto-nvl72/README.md | 15 +-- .../presto-nvl72/cluster_config.env.example | 9 +- presto/slurm/presto-nvl72/functions.sh | 18 +-- presto/slurm/presto-nvl72/launch-run.sh | 31 +++-- .../performance_benchmark_fixtures_test.py | 34 ++--- .../performance_benchmarks/common_fixtures.py | 77 +++++------- .../performance_benchmarks/conftest.py | 2 +- 19 files changed, 347 insertions(+), 269 deletions(-) create mode 100644 common/testing/normalize_ctas_results.py create mode 100644 common/testing/normalize_ctas_results_test.py diff --git a/common/testing/normalize_ctas_results.py b/common/testing/normalize_ctas_results.py new file mode 100644 index 00000000..66973abc --- /dev/null +++ b/common/testing/normalize_ctas_results.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Convert temporary distributed CTAS tables into canonical qN.parquet files.""" + +import argparse +import os +import re +import shutil +import sys +from pathlib import Path + +import pandas as pd +import pyarrow.parquet as pq +import sqlglot +from sqlglot import exp + +# Allow importing from the repo root when this file is executed directly. +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +from common.testing.result_comparison import _get_orderby_sort_spec, _restore_orderby +from common.testing.test_utils import get_queries + + +def _query_directories(source_dir: Path) -> list[Path]: + return sorted( + (path for path in source_dir.iterdir() if path.is_dir() and re.fullmatch(r"q\d+", path.name)), + key=lambda path: int(path.name[1:]), + ) + + +def _source_output_names(query_sql: str, physical_names: list[str]) -> list[str]: + """Return names from an explicit SELECT projection for ORDER BY lookup.""" + parsed = sqlglot.parse_one(query_sql, read="presto") + select = next(parsed.find_all(exp.Select)) + has_wildcard = any( + isinstance(projection := expression.this if isinstance(expression, exp.Alias) else expression, exp.Star) + or (isinstance(projection, exp.Column) and isinstance(projection.this, exp.Star)) + for expression in select.expressions + ) + if len(select.expressions) != len(physical_names) or has_wildcard: + return physical_names + return [expression.alias_or_name or physical_names[position] for position, expression in enumerate(select.expressions)] + + +def _write_canonical_result(query_dir: Path, target: Path, query_sql: str | None) -> None: + parquet_parts = sorted(query_dir.rglob("*.parquet")) + sort_spec: tuple[list[int], list[bool], list[bool]] = ([], [], []) + + if parquet_parts and query_sql is not None: + column_names = _source_output_names(query_sql, pq.read_schema(parquet_parts[0]).names) + sort_spec = _get_orderby_sort_spec(query_sql, column_names) + + temporary = target.with_name(f".{target.name}.tmp-{os.getpid()}") + try: + # A single unordered part is already a canonical Parquet file. Copying + # it avoids decoding and rewriting large results. + if len(parquet_parts) == 1 and not sort_spec[0]: + shutil.copy2(parquet_parts[0], temporary) + else: + frame = pd.read_parquet(query_dir) + if sort_spec[0]: + frame = _restore_orderby(frame, *sort_spec) + frame.to_parquet(temporary, index=False) + os.replace(temporary, target) + finally: + temporary.unlink(missing_ok=True) + + +def normalize_ctas_results( + source_dir: Path, + output_dir: Path, + queries: dict[str, str], +) -> list[Path]: + """Convert each committed qN CTAS table into one canonical Parquet file.""" + query_dirs = _query_directories(source_dir) + if not query_dirs: + raise ValueError(f"No committed CTAS result tables found in {source_dir}") + + output_dir.mkdir(parents=True, exist_ok=True) + normalized: list[Path] = [] + for query_dir in query_dirs: + target = output_dir / f"{query_dir.name}.parquet" + _write_canonical_result(query_dir, target, queries.get(query_dir.name.upper())) + shutil.rmtree(query_dir) + normalized.append(target) + return normalized + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--source-dir", required=True, type=Path, help="Directory containing temporary qN CTAS tables.") + parser.add_argument( + "--output-dir", + required=True, + type=Path, + help="Canonical query_results directory that will receive qN.parquet files.", + ) + parser.add_argument("--benchmark-type", required=True, choices=["tpch", "tpcds"]) + parser.add_argument("--queries-file", default=None, help="Optional custom query-definition JSON file.") + return parser.parse_args() + + +if __name__ == "__main__": + args = parse_args() + try: + outputs = normalize_ctas_results( + args.source_dir, + args.output_dir, + get_queries(args.benchmark_type, args.queries_file), + ) + except (OSError, ValueError) as error: + print(f"Error: Failed to normalize CTAS results: {error}", file=sys.stderr) + sys.exit(1) + for output in outputs: + print(f"Normalized CTAS result: {output}") diff --git a/common/testing/normalize_ctas_results_test.py b/common/testing/normalize_ctas_results_test.py new file mode 100644 index 00000000..ac997d0e --- /dev/null +++ b/common/testing/normalize_ctas_results_test.py @@ -0,0 +1,56 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from pathlib import Path + +import pandas as pd + +from common.testing.normalize_ctas_results import normalize_ctas_results + + +def test_normalize_ctas_results_merges_parts_and_restores_order(tmp_path: Path): + scratch = tmp_path / "scratch" + query_dir = scratch / "q1" + query_dir.mkdir(parents=True) + # CTAS assigns positional storage names, while ORDER BY still refers to + # the original SELECT output name. + pd.DataFrame({"c1": [3, 1]}).to_parquet(query_dir / "part-00000.parquet", index=False) + pd.DataFrame({"c1": [4, 2]}).to_parquet(query_dir / "part-00001.parquet", index=False) + + output = tmp_path / "output" + normalized = normalize_ctas_results( + scratch, + output, + {"Q1": "SELECT value FROM source ORDER BY value"}, + ) + + assert normalized == [output / "q1.parquet"] + assert pd.read_parquet(normalized[0])["c1"].tolist() == [1, 2, 3, 4] + assert not query_dir.exists() + + +def test_normalize_ctas_results_promotes_single_unordered_part(tmp_path: Path): + scratch = tmp_path / "scratch" + query_dir = scratch / "q6" + query_dir.mkdir(parents=True) + expected = pd.DataFrame({"revenue": [42.0]}) + expected.to_parquet(query_dir / "part-00000.parquet", index=False) + + output = tmp_path / "output" + normalize_ctas_results(scratch, output, {"Q6": "SELECT revenue FROM source"}) + + pd.testing.assert_frame_equal(pd.read_parquet(output / "q6.parquet"), expected) + assert not query_dir.exists() + + +def test_normalize_ctas_results_writes_empty_result(tmp_path: Path): + scratch = tmp_path / "scratch" + query_dir = scratch / "q15" + query_dir.mkdir(parents=True) + (query_dir / ".prestoSchema").write_text("{}") + + output = tmp_path / "output" + normalize_ctas_results(scratch, output, {"Q15": "SELECT supplier_no FROM source"}) + + assert pd.read_parquet(output / "q15.parquet").empty + assert not query_dir.exists() diff --git a/common/testing/result_comparison.py b/common/testing/result_comparison.py index 34559291..053e801b 100644 --- a/common/testing/result_comparison.py +++ b/common/testing/result_comparison.py @@ -446,7 +446,6 @@ def compare_result_frames( expected: pd.DataFrame, query_sql: str, actual_column_types: list[str] | None = None, - actual_order_preserved: bool = True, ) -> None: """ Full comparison pipeline. Raises AssertionError on any mismatch. @@ -458,8 +457,7 @@ def compare_result_frames( 1. Validate column count and capture expected column names. 2. Normalize dtypes / value types. 3. Validate row count. - 4. Parse ORDER BY / LIMIT from SQL and reconstruct logical order when - the actual rows came from unordered distributed storage. + 4. Parse ORDER BY / LIMIT from SQL. 5. Validate each engine respected ORDER BY (per-frame, strict equality). 6. With LIMIT: peel the tied tail in engine order, then canonical-sort each piece. Compare front fully; compare tail's ORDER BY values only. @@ -482,13 +480,9 @@ def compare_result_frames( if len(actual) != len(expected): raise AssertionError(f"Row count mismatch: {len(actual)} (actual) vs {len(expected)} (expected)") - # 4. Parse ORDER BY / LIMIT. A distributed Parquet dataset has no physical - # row-order guarantee, so reconstruct its logical SQL order before applying - # the same ordered-result checks used for coordinator-returned results. - sort_col_indices, ascending, nulls_first = _get_orderby_sort_spec(query_sql, expected_col_names) + # 4. Parse ORDER BY / LIMIT. + sort_col_indices, ascending, _ = _get_orderby_sort_spec(query_sql, expected_col_names) limit = get_limit(query_sql) - if not actual_order_preserved: - actual = _restore_orderby(actual, sort_col_indices, ascending, nulls_first) # 5. Per-frame ORDER BY validation — engine order is intact here. _validate_orderby(actual, sort_col_indices, ascending) @@ -535,7 +529,6 @@ def validate_query_result( actual: pd.DataFrame, expected: pd.DataFrame, query_sql: str, - actual_order_preserved: bool = True, ) -> tuple[ValidationStatus, str | None]: """ Compare actual vs expected for one query. Returns (status, message). @@ -553,7 +546,7 @@ def validate_query_result( ) try: - compare_result_frames(actual, expected, query_sql, actual_order_preserved=actual_order_preserved) + compare_result_frames(actual, expected, query_sql) return "passed", None except Exception as e: return "failed", f"{type(e).__name__}: {e}"[:500] diff --git a/common/testing/result_comparison_test.py b/common/testing/result_comparison_test.py index b11ac2e7..3282ce19 100644 --- a/common/testing/result_comparison_test.py +++ b/common/testing/result_comparison_test.py @@ -16,7 +16,6 @@ _normalize_to_expected, _restore_orderby, _validate_orderby, - compare_result_frames, ) # --------------------------------------------------------------------------- @@ -138,18 +137,6 @@ def test_restore_orderby_defaults_to_presto_nulls_last_for_desc(): assert restored["payload"].tolist() == ["high", "low", "null"] -def test_compare_restores_unordered_limit_result_before_tie_handling(): - actual = pd.DataFrame({"score": [5.0, 10.0, 5.0], "payload": ["different-tie", "top", "left-tie"]}) - expected = pd.DataFrame({"score": [10.0, 5.0, 5.0], "payload": ["top", "left-tie", "right-tie"]}) - - compare_result_frames( - actual, - expected, - "SELECT score, payload FROM source ORDER BY score DESC LIMIT 3", - actual_order_preserved=False, - ) - - # --------------------------------------------------------------------------- # _validate_orderby null handling # --------------------------------------------------------------------------- diff --git a/common/testing/validate_results.py b/common/testing/validate_results.py index 2bbd8281..46f9b261 100644 --- a/common/testing/validate_results.py +++ b/common/testing/validate_results.py @@ -37,7 +37,7 @@ def validate( """Run validation and return a results dict. Args: - results_dir: Directory containing qN.parquet files or distributed qN datasets. + results_dir: Directory containing qN.parquet files. expected_dir: Directory containing expected parquet files. queries: Dict mapping query IDs (e.g. "Q1") to SQL strings. query_numbers: Optional list of query numbers to validate. When provided, @@ -51,32 +51,17 @@ def validate( passed = failed = not_validated = expected_failures = 0 if query_numbers is not None: - result_files = [] - for q_num in query_numbers: - result_files.extend( - path for path in (results_dir / f"q{q_num}.parquet", results_dir / f"q{q_num}") if path.exists() - ) + result_files = [results_dir / f"q{q_num}.parquet" for q_num in query_numbers] + result_files = [path for path in result_files if path.exists()] else: - result_files = sorted(path for path in results_dir.iterdir() if re.fullmatch(r"q\d+(?:\.parquet)?", path.name)) + result_files = sorted(path for path in results_dir.iterdir() if re.fullmatch(r"q\d+\.parquet", path.name)) if not result_files: - print(f"No result Parquet files or datasets found in {results_dir}", file=sys.stderr) + print(f"No result Parquet files found in {results_dir}", file=sys.stderr) return {"overall_status": "not-validated", "queries": {}} - results_by_query: dict[str, list[Path]] = {} - for result_file in result_files: - results_by_query.setdefault(result_file.stem, []).append(result_file) - - for query_id, query_paths in sorted(results_by_query.items(), key=lambda item: int(item[0].lstrip("q"))): - if len(query_paths) != 1: - names = ", ".join(sorted(path.name for path in query_paths)) - msg = f"multiple result representations found: {names}" - print(f"[Validation] {query_id.upper():4s}: FAIL {msg}") - query_results[query_id] = {"status": "failed", "message": msg} - failed += 1 - continue - - result_file = query_paths[0] + for result_file in sorted(result_files, key=lambda path: int(path.stem.lstrip("q"))): + query_id = result_file.stem q_num = int(query_id.lstrip("q")) # Accepted naming conventions for expected files (tried in order): @@ -101,8 +86,6 @@ def validate( failed += 1 continue - # PyArrow treats a CTAS dataset containing only `.prestoSchema` as an - # empty, schema-less frame; the empty-result handling below covers it. actual = pd.read_parquet(result_file) expected = pd.read_parquet(expected_file) @@ -120,13 +103,7 @@ def validate( not_validated += 1 continue - status, msg = validate_query_result( - query_id, - actual, - expected, - query_sql, - actual_order_preserved=result_file.is_file(), - ) + status, msg = validate_query_result(query_id, actual, expected, query_sql) if status == "not-validated": query_results[query_id] = {"status": "not-validated", "message": msg} @@ -168,7 +145,7 @@ def validate( def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( - description="Validate TPC-H/TPC-DS query results against expected Parquet files or datasets.", + description="Validate TPC-H/TPC-DS query results against expected Parquet files.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) @@ -177,12 +154,6 @@ def parse_args() -> argparse.Namespace: required=True, help="Benchmark output directory (same as pytest --output-dir).", ) - parser.add_argument( - "--actual-results-dir", - default=None, - help="Optional directory containing actual qN Parquet files or dataset directories. " - "Defaults to query_results beneath --output-dir/--tag.", - ) parser.add_argument( "--tag", default=None, @@ -223,7 +194,7 @@ def _write_not_validated(output_path: Path, reason: str) -> None: output_dir = Path(args.output_dir) report_dir = output_dir / args.tag if args.tag else output_dir - results_dir = Path(args.actual_results_dir) if args.actual_results_dir else report_dir / "query_results" + results_dir = report_dir / "query_results" output_path = report_dir / "validation_results.json" if not results_dir.is_dir(): diff --git a/common/testing/validate_results_test.py b/common/testing/validate_results_test.py index 03e02e57..40ed745b 100644 --- a/common/testing/validate_results_test.py +++ b/common/testing/validate_results_test.py @@ -4,34 +4,14 @@ from pathlib import Path import pandas as pd -import pytest from common.testing.validate_results import validate -def test_validator_reads_distributed_parquet_dataset(tmp_path: Path): +def test_validator_accepts_empty_q15_result(tmp_path: Path): results_dir = tmp_path / "actual" - query_dir = results_dir / "q1" - query_dir.mkdir(parents=True) - # Distributed writer files do not preserve the query's global row order. - pd.DataFrame({"value": [3, 4]}).to_parquet(query_dir / "part-00000.parquet", index=False) - pd.DataFrame({"value": [1, 2]}).to_parquet(query_dir / "part-00001.parquet", index=False) - - expected_dir = tmp_path / "expected" - expected_dir.mkdir() - pd.DataFrame({"value": [1, 2, 3, 4]}).to_parquet(expected_dir / "q01.parquet", index=False) - - result = validate(results_dir, expected_dir, {"Q1": "SELECT value FROM source ORDER BY value"}) - - assert result["overall_status"] == "passed" - assert result["queries"]["q1"]["status"] == "passed" - - -def test_validator_accepts_empty_distributed_q15_dataset(tmp_path: Path): - results_dir = tmp_path / "actual" - query_dir = results_dir / "q15" - query_dir.mkdir(parents=True) - (query_dir / ".prestoSchema").write_text("{}") + results_dir.mkdir() + pd.DataFrame().to_parquet(results_dir / "q15.parquet", index=False) expected_dir = tmp_path / "expected" expected_dir.mkdir() @@ -56,24 +36,3 @@ def test_validator_keeps_single_file_result_order_sensitive(tmp_path: Path): assert result["overall_status"] == "failed" assert "Engine violated ORDER BY" in result["queries"]["q1"]["message"] - - -@pytest.mark.parametrize("query_numbers", [None, [1]]) -def test_validator_rejects_file_and_dataset_for_same_query(tmp_path: Path, query_numbers): - results_dir = tmp_path / "actual" - query_dir = results_dir / "q1" - query_dir.mkdir(parents=True) - pd.DataFrame({"value": [1]}).to_parquet(query_dir / "part-00000.parquet", index=False) - pd.DataFrame({"value": [1]}).to_parquet(results_dir / "q1.parquet", index=False) - - expected_dir = tmp_path / "expected" - expected_dir.mkdir() - pd.DataFrame({"value": [1]}).to_parquet(expected_dir / "q01.parquet", index=False) - - result = validate(results_dir, expected_dir, {"Q1": "SELECT value FROM source"}, query_numbers=query_numbers) - - assert result["overall_status"] == "failed" - assert result["queries"]["q1"] == { - "status": "failed", - "message": "multiple result representations found: q1, q1.parquet", - } diff --git a/presto/README.md b/presto/README.md index 3d9e5594..746edd5d 100644 --- a/presto/README.md +++ b/presto/README.md @@ -51,12 +51,12 @@ All three repositories must be checked out as sibling directories. **Important:* export PRESTO_DATA_DIR=/path/to/your/benchmark/data ``` - Benchmark source data is mounted read-only. To have workers write large query - results directly as distributed Parquet datasets, also select a separate - writable directory before starting the cluster: + Benchmark source data is mounted read-only. To run large-result queries as + distributed CTAS writes, also select a separate writable scratch directory + before starting the cluster: ```bash - export PRESTO_OUTPUT_DIR=/path/to/writable/benchmark-results + export PRESTO_CTAS_SCRATCH_DIR=/path/to/writable/benchmark-results ``` > **Tip:** Add this export to your `~/.bashrc` to avoid setting it each time. @@ -104,12 +104,13 @@ pytest tpch_test.py ``` > This step is necessary because aggregation for statistics collection is not yet supported on GPU. Collecting statistics improves performance and reduces OOM errors for GPU execution. -4. Run a benchmark. Add `--write-results-to-file` to write distributed - `benchmark_results_/qN/` Parquet datasets beneath `PRESTO_OUTPUT_DIR` - (`benchmark_results/qN/` when no tag is supplied): +4. Run a benchmark. Add `--run-as-ctas-queries` to keep distributed result + transfer out of the measured query. Workers first write beneath + `PRESTO_CTAS_SCRATCH_DIR`; after timing, the script combines each result into + the same `query_results/qN.parquet` layout used by the default mode: ```bash - ./run_benchmark.sh -b tpch -s bench_sf100 --write-results-to-file + ./run_benchmark.sh -b tpch -s bench_sf100 --run-as-ctas-queries ``` 5. For the standard coordinator-returned result mode, run: diff --git a/presto/docker/config/template/etc_common/catalog/hive_output.properties b/presto/docker/config/template/etc_common/catalog/hive_output.properties index c779c99c..e26ab90f 100644 --- a/presto/docker/config/template/etc_common/catalog/hive_output.properties +++ b/presto/docker/config/template/etc_common/catalog/hive_output.properties @@ -1,4 +1,4 @@ -# Hive catalog dedicated to benchmark CTAS output. Its file metastore root is +# Hive catalog dedicated to temporary benchmark CTAS output. Its file metastore root is # mounted separately from PRESTO_DATA_DIR so source data can remain read-only. connector.name=hive-hadoop2 hive.metastore=file @@ -6,5 +6,5 @@ hive.metastore.catalog.dir=file:/var/lib/presto/data/hive/benchmark_output hive.allow-drop-table=true # Native workers otherwise stage writes under their container-private /tmp, # which prevents the coordinator from committing the files. Write directly to -# the shared result mount; interrupted runs are cleaned before the next run. +# the shared scratch mount; interrupted runs are cleaned before the next run. hive.temporary-staging-directory-enabled=false diff --git a/presto/docker/docker-compose.common.yml b/presto/docker/docker-compose.common.yml index cb964c65..68ef8c71 100644 --- a/presto/docker/docker-compose.common.yml +++ b/presto/docker/docker-compose.common.yml @@ -4,7 +4,7 @@ services: - ./.hive_metastore:/var/lib/presto/data/hive/metastore - ../../common/testing/integration_tests/data:/var/lib/presto/data/hive/data/integration_test - ${PRESTO_DATA_DIR:-/dev/null}:/var/lib/presto/data/hive/data/user_data:ro - - ${PRESTO_OUTPUT_DIR:-/dev/null}:/var/lib/presto/data/hive/benchmark_output + - ${PRESTO_CTAS_SCRATCH_DIR:-/dev/null}:/var/lib/presto/data/hive/benchmark_output - ${LOGS_DIR:-/dev/null}:/opt/presto-server/logs - ./launch_presto_servers.sh:/opt/launch_presto_servers.sh:ro diff --git a/presto/scripts/run_benchmark.sh b/presto/scripts/run_benchmark.sh index 695dcafd..db141dcc 100755 --- a/presto/scripts/run_benchmark.sh +++ b/presto/scripts/run_benchmark.sh @@ -43,8 +43,8 @@ OPTIONS: --profile-script-path Path to a custom profiler functions script. Defaults to ./profiler_functions.sh. --skip-drop-cache Skip dropping system caches before each benchmark query (dropped by default). --skip-analyze-check Skip checking that ANALYZE TABLE has been run on all tables (checked by default). - --write-results-to-file Write query results with distributed Hive CTAS operations instead of returning them - through the coordinator. PRESTO_OUTPUT_DIR must be set and mounted when the cluster starts. + --run-as-ctas-queries Run queries as distributed Hive CTAS operations instead of returning results + through the coordinator. PRESTO_CTAS_SCRATCH_DIR must be set and mounted when the cluster starts. -m, --metrics Collect detailed metrics from Presto REST API after each query. Metrics are stored in query-specific directories. --reference-results-dir Path to a directory containing reference (expected) parquet files. @@ -59,8 +59,9 @@ OPTIONS: Set to the full host path of the expected results directory (e.g. PRESTO_EXPECTED_RESULTS_DIR=/data/sf100_expected). Warning (not error) if set but the directory does not exist. - PRESTO_OUTPUT_DIR Writable host directory used for distributed CTAS result datasets. This must be set - before both cluster startup and this script when --write-results-to-file is enabled. + PRESTO_CTAS_SCRATCH_DIR + Writable host scratch directory used for temporary distributed CTAS results. This must be set + before both cluster startup and this script when --run-as-ctas-queries is enabled. -v, --verbose Print debug logs for worker/engine detection (e.g. node URIs, cluster-tag, GPU model). Use when engine is misdetected or the run fails. @@ -72,7 +73,7 @@ EXAMPLES: $0 -b tpch -s bench_sf100 -t gh200_cpu_sf100 $0 -b tpch -s bench_sf100 --profile $0 -b tpch -s bench_sf100 --metrics - PRESTO_OUTPUT_DIR=/results $0 -b tpch -s bench_sf100 --write-results-to-file + PRESTO_CTAS_SCRATCH_DIR=/results $0 -b tpch -s bench_sf100 --run-as-ctas-queries $0 -b tpch -s bench_sf100 --verbose EOF @@ -196,8 +197,8 @@ parse_args() { SKIP_ANALYZE_CHECK=true shift ;; - --write-results-to-file) - WRITE_RESULTS_TO_FILE=true + --run-as-ctas-queries) + RUN_AS_CTAS_QUERIES=true shift ;; -m|--metrics) @@ -241,16 +242,16 @@ if [[ -z ${SCHEMA_NAME} ]]; then exit 1 fi -if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then - if [[ -z "${PRESTO_OUTPUT_DIR}" ]]; then - echo "Error: PRESTO_OUTPUT_DIR must be set when --write-results-to-file is enabled." >&2 +if [[ "${RUN_AS_CTAS_QUERIES}" == "true" ]]; then + if [[ -z "${PRESTO_CTAS_SCRATCH_DIR}" ]]; then + echo "Error: PRESTO_CTAS_SCRATCH_DIR must be set when --run-as-ctas-queries is enabled." >&2 exit 1 fi - mkdir -p "${PRESTO_OUTPUT_DIR}" - PRESTO_OUTPUT_DIR="$(readlink -f "${PRESTO_OUTPUT_DIR}")" - export PRESTO_OUTPUT_DIR - if [[ ! -d "${PRESTO_OUTPUT_DIR}" || ! -w "${PRESTO_OUTPUT_DIR}" ]]; then - echo "Error: PRESTO_OUTPUT_DIR must be a writable directory: ${PRESTO_OUTPUT_DIR}" >&2 + mkdir -p "${PRESTO_CTAS_SCRATCH_DIR}" + PRESTO_CTAS_SCRATCH_DIR="$(readlink -f "${PRESTO_CTAS_SCRATCH_DIR}")" + export PRESTO_CTAS_SCRATCH_DIR + if [[ ! -d "${PRESTO_CTAS_SCRATCH_DIR}" || ! -w "${PRESTO_CTAS_SCRATCH_DIR}" ]]; then + echo "Error: PRESTO_CTAS_SCRATCH_DIR must be a writable directory: ${PRESTO_CTAS_SCRATCH_DIR}" >&2 exit 1 fi if [[ -n ${TAG} ]]; then @@ -258,7 +259,7 @@ if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then else CTAS_RESULTS_SCHEMA="benchmark_results" fi - CTAS_RESULTS_DIR="${PRESTO_OUTPUT_DIR}/${CTAS_RESULTS_SCHEMA}" + CTAS_SCRATCH_RESULTS_DIR="${PRESTO_CTAS_SCRATCH_DIR}/${CTAS_RESULTS_SCHEMA}" fi # Fail fast if an explicit --reference-results-dir was given but doesn't exist. @@ -328,8 +329,8 @@ if [[ "${SKIP_ANALYZE_CHECK}" == "true" ]]; then PYTEST_ARGS+=("--skip-analyze-check") fi -if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then - PYTEST_ARGS+=("--write-results-to-file") +if [[ "${RUN_AS_CTAS_QUERIES}" == "true" ]]; then + PYTEST_ARGS+=("--run-as-ctas-queries") fi source "${SCRIPT_DIR}/../../scripts/py_env_functions.sh" @@ -355,7 +356,7 @@ BENCHMARK_TEST_DIR=${TEST_DIR}/performance_benchmarks PYTEST_EXIT=0 pytest -q -s ${BENCHMARK_TEST_DIR}/${BENCHMARK_TYPE}_test.py ${PYTEST_ARGS[*]} || PYTEST_EXIT=$? -normalize_ctas_result_permissions() { +ensure_ctas_scratch_readable() { local host_results_dir=$1 local container_results_dir="/var/lib/presto/data/hive/benchmark_output/${CTAS_RESULTS_SCHEMA}" @@ -400,9 +401,9 @@ normalize_ctas_result_permissions() { fi } -NORMALIZE_EXIT=0 -if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then - normalize_ctas_result_permissions "${CTAS_RESULTS_DIR}" || NORMALIZE_EXIT=$? +CTAS_PERMISSION_EXIT=0 +if [[ "${RUN_AS_CTAS_QUERIES}" == "true" ]]; then + ensure_ctas_scratch_readable "${CTAS_SCRATCH_RESULTS_DIR}" || CTAS_PERMISSION_EXIT=$? fi # Snapshot logs and engine configs into the benchmark output directory so that @@ -415,6 +416,21 @@ else EFFECTIVE_BENCHMARK_DIR="${EFFECTIVE_OUTPUT_DIR}" fi +CTAS_NORMALIZATION_EXIT=0 +if [[ "${RUN_AS_CTAS_QUERIES}" == "true" && ${CTAS_PERMISSION_EXIT} -eq 0 ]]; then + NORMALIZE_ARGS=( + --source-dir "${CTAS_SCRATCH_RESULTS_DIR}" + --output-dir "${EFFECTIVE_BENCHMARK_DIR}/query_results" + --benchmark-type "${BENCHMARK_TYPE}" + ) + if [[ -n ${QUERIES_FILE} ]]; then + NORMALIZE_ARGS+=(--queries-file "${QUERIES_FILE}") + fi + echo "[CTAS] Normalizing temporary results into ${EFFECTIVE_BENCHMARK_DIR}/query_results" + python "${SCRIPT_DIR}/../../common/testing/normalize_ctas_results.py" \ + "${NORMALIZE_ARGS[@]}" || CTAS_NORMALIZATION_EXIT=$? +fi + if [[ -d "${LOGS_DIR}" ]]; then mkdir -p "${EFFECTIVE_BENCHMARK_DIR}/logs" cp -r "${LOGS_DIR}/." "${EFFECTIVE_BENCHMARK_DIR}/logs/" @@ -435,7 +451,9 @@ VALIDATE_REQUIREMENTS="${SCRIPT_DIR}/../testing/requirements.txt" # PRESTO_EXPECTED_RESULTS_DIR env var is the implicit fallback (warning if missing). # Explicit --reference-results-dir was already validated before the benchmark ran. VALIDATION_EXIT=0 -if [[ -n ${PRESTO_EXPECTED_RESULTS_DIR} && ! -d ${PRESTO_EXPECTED_RESULTS_DIR} ]]; then +if [[ "${RUN_AS_CTAS_QUERIES}" == "true" && ( ${CTAS_PERMISSION_EXIT} -ne 0 || ${CTAS_NORMALIZATION_EXIT} -ne 0 ) ]]; then + echo "[Validation] Skipped because CTAS result normalization failed." +elif [[ -n ${PRESTO_EXPECTED_RESULTS_DIR} && ! -d ${PRESTO_EXPECTED_RESULTS_DIR} ]]; then echo "[Validation] Warning: PRESTO_EXPECTED_RESULTS_DIR not found: ${PRESTO_EXPECTED_RESULTS_DIR}; validation skipped." else VALIDATE_ARGS=(--output-dir "${OUTPUT_DIR:-$(pwd)/benchmark_output}" --benchmark-type "${BENCHMARK_TYPE}") @@ -448,17 +466,7 @@ else if [[ -n ${QUERIES} ]]; then VALIDATE_ARGS+=(--queries "${QUERIES}") fi - if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then - VALIDATE_ARGS+=(--actual-results-dir "${CTAS_RESULTS_DIR}") - fi - - ACTUAL_OUTPUT_DIR="${OUTPUT_DIR:-$(pwd)/benchmark_output}" - [[ -n ${TAG} ]] && ACTUAL_OUTPUT_DIR="${ACTUAL_OUTPUT_DIR}/${TAG}" - if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then - ACTUAL_RESULTS_DIR="${CTAS_RESULTS_DIR}" - else - ACTUAL_RESULTS_DIR="${ACTUAL_OUTPUT_DIR}/query_results" - fi + ACTUAL_RESULTS_DIR="${EFFECTIVE_BENCHMARK_DIR}/query_results" echo "[Validation] Running validation: ${ACTUAL_RESULTS_DIR} vs ${PRESTO_EXPECTED_RESULTS_DIR:-}" "${SCRIPT_DIR}/../../scripts/run_py_script.sh" --quiet \ -p "${VALIDATE_SCRIPT}" \ @@ -467,10 +475,12 @@ else fi # Preserve the benchmark failure as the primary exit status while still -# attempting CTAS permission normalization, artifact collection, and validation. +# attempting CTAS normalization, artifact collection, and validation. if [[ ${PYTEST_EXIT} -ne 0 ]]; then exit "${PYTEST_EXIT}" -elif [[ ${NORMALIZE_EXIT} -ne 0 ]]; then - exit "${NORMALIZE_EXIT}" +elif [[ ${CTAS_PERMISSION_EXIT} -ne 0 ]]; then + exit "${CTAS_PERMISSION_EXIT}" +elif [[ ${CTAS_NORMALIZATION_EXIT} -ne 0 ]]; then + exit "${CTAS_NORMALIZATION_EXIT}" fi exit "${VALIDATION_EXIT}" diff --git a/presto/scripts/run_multiple_benchmarks.sh b/presto/scripts/run_multiple_benchmarks.sh index ead5babe..ec17291d 100755 --- a/presto/scripts/run_multiple_benchmarks.sh +++ b/presto/scripts/run_multiple_benchmarks.sh @@ -7,7 +7,7 @@ KVIKIO_ARRAY=(8) DRIVERS_ARRAY=(2) WORKERS_ARRAY=(1) SCHEMA_ARRAY=() -WRITE_RESULTS_TO_FILE=false +RUN_AS_CTAS_QUERIES=false parse_args() { while [[ $# -gt 0 ]]; do case $1 in @@ -56,8 +56,8 @@ parse_args() { exit 1 fi ;; - --write-results-to-file) - WRITE_RESULTS_TO_FILE=true + --run-as-ctas-queries) + RUN_AS_CTAS_QUERIES=true shift ;; *) @@ -81,13 +81,13 @@ if [[ -z ${PRESTO_DATA_DIR} ]]; then exit 1 fi -if [[ "${WRITE_RESULTS_TO_FILE}" == "true" && -z "${PRESTO_OUTPUT_DIR:-}" ]]; then - echo "Error: PRESTO_OUTPUT_DIR is required with --write-results-to-file." +if [[ "${RUN_AS_CTAS_QUERIES}" == "true" && -z "${PRESTO_CTAS_SCRATCH_DIR:-}" ]]; then + echo "Error: PRESTO_CTAS_SCRATCH_DIR is required with --run-as-ctas-queries." exit 1 fi BENCHMARK_OUTPUT_ARGS=() -[[ "${WRITE_RESULTS_TO_FILE}" == "true" ]] && BENCHMARK_OUTPUT_ARGS+=(--write-results-to-file) +[[ "${RUN_AS_CTAS_QUERIES}" == "true" ]] && BENCHMARK_OUTPUT_ARGS+=(--run-as-ctas-queries) for schema in "${SCHEMA_ARRAY[@]}"; do for kvikio in "${KVIKIO_ARRAY[@]}"; do diff --git a/presto/scripts/start_presto_helper_parse_args.sh b/presto/scripts/start_presto_helper_parse_args.sh index 41b5812c..819c8cd3 100644 --- a/presto/scripts/start_presto_helper_parse_args.sh +++ b/presto/scripts/start_presto_helper_parse_args.sh @@ -44,7 +44,7 @@ OPTIONS: ENVIRONMENT VARIABLES: SCCACHE_AUTH_DIR Directory containing sccache auth files (default: ~/.sccache-auth/). - PRESTO_OUTPUT_DIR Optional writable host directory mounted for distributed CTAS benchmark results. + PRESTO_CTAS_SCRATCH_DIR Optional writable host scratch directory mounted for temporary CTAS benchmark results. EXAMPLES: $SCRIPT_NAME --no-cache @@ -218,14 +218,14 @@ parse_args() { parse_args "$@" -if [[ -n "${PRESTO_OUTPUT_DIR:-}" ]]; then - mkdir -p "${PRESTO_OUTPUT_DIR}" - PRESTO_OUTPUT_DIR="$(readlink -f "${PRESTO_OUTPUT_DIR}")" - if [[ ! -d "${PRESTO_OUTPUT_DIR}" || ! -w "${PRESTO_OUTPUT_DIR}" ]]; then - echo "Error: PRESTO_OUTPUT_DIR must be a writable directory: ${PRESTO_OUTPUT_DIR}" >&2 +if [[ -n "${PRESTO_CTAS_SCRATCH_DIR:-}" ]]; then + mkdir -p "${PRESTO_CTAS_SCRATCH_DIR}" + PRESTO_CTAS_SCRATCH_DIR="$(readlink -f "${PRESTO_CTAS_SCRATCH_DIR}")" + if [[ ! -d "${PRESTO_CTAS_SCRATCH_DIR}" || ! -w "${PRESTO_CTAS_SCRATCH_DIR}" ]]; then + echo "Error: PRESTO_CTAS_SCRATCH_DIR must be a writable directory: ${PRESTO_CTAS_SCRATCH_DIR}" >&2 exit 1 fi - export PRESTO_OUTPUT_DIR + export PRESTO_CTAS_SCRATCH_DIR fi if [[ -n ${BUILD_TARGET} && ! ${BUILD_TARGET} =~ ^(coordinator|c|worker|w|all|a)$ ]]; then diff --git a/presto/slurm/presto-nvl72/README.md b/presto/slurm/presto-nvl72/README.md index fda4a513..d4fa0539 100644 --- a/presto/slurm/presto-nvl72/README.md +++ b/presto/slurm/presto-nvl72/README.md @@ -103,17 +103,18 @@ This step only needs to repeat when the worker image's stat tracking changes # Profile queries 5 and 6 on worker 2 ./launch-run.sh -n 8 -s 3000 -p --nsys-worker-id 2 -q 5,6 -# Have workers write distributed Parquet results to separate shared storage -PRESTO_OUTPUT_DIR=/shared/writable/presto-results \ - ./launch-run.sh -n 8 -s 3000 --write-results-to-file +# Use separate shared scratch storage for distributed CTAS writes +PRESTO_CTAS_SCRATCH_DIR=/shared/writable/presto-results \ + ./launch-run.sh -n 8 -s 3000 --run-as-ctas-queries ``` Submits a benchmark sbatch job, polls until completion, and prints a summary. Results land under `result_dir/`. See `./launch-run.sh --help` for the full -flag list (queries filter, output path, CTAS result writing, GDS toggle, -profiling, metrics, …). In CTAS mode, benchmark reports remain in -`result_dir/`, while `benchmark_results_/qN/` Parquet datasets are written -beneath `PRESTO_OUTPUT_DIR` (`benchmark_results/qN/` without a tag). +flag list (queries filter, output path, CTAS execution, GDS toggle, profiling, +metrics, …). In CTAS mode, workers write temporary distributed results beneath +`PRESTO_CTAS_SCRATCH_DIR`. After query timing, those parts are combined into +the normal `result_dir/query_results/qN.parquet` files, alongside the benchmark +reports. To validate benchmark output, set the host-side reference directory in `~/.cluster_config.env`: diff --git a/presto/slurm/presto-nvl72/cluster_config.env.example b/presto/slurm/presto-nvl72/cluster_config.env.example index e485f6d7..ef3a68c5 100644 --- a/presto/slurm/presto-nvl72/cluster_config.env.example +++ b/presto/slurm/presto-nvl72/cluster_config.env.example @@ -29,11 +29,16 @@ DATA=/path/to/tpch/data # Directory containing .sqsh container images. IMAGE_DIR=/path/to/images -# Optional directory containing qN.parquet files (or qN/ Parquet datasets) -# used to validate benchmark results. This is a host path on shared storage; +# Optional directory containing qN.parquet files used to validate benchmark +# results. This is a host path on shared storage; # launch-run.sh mounts it read-only into the Presto CLI container. # PRESTO_EXPECTED_RESULTS_DIR=/path/to/expected-results +# Writable shared scratch directory used only with --run-as-ctas-queries. +# Workers write temporary distributed results here; the CLI container combines +# them into the normal result_dir/query_results/qN.parquet files after timing. +# PRESTO_CTAS_SCRATCH_DIR=/path/to/writable/ctas-scratch + # Root for shared pre-analyzed Hive metastore snapshots (shared across users). # Unset (comment out) to disable metastore sharing and always run ANALYZE locally. HIVE_METASTORE_SHARED_ROOT=/path/to/shared/hive_metadata diff --git a/presto/slurm/presto-nvl72/functions.sh b/presto/slurm/presto-nvl72/functions.sh index 99412e1c..694ed7a3 100644 --- a/presto/slurm/presto-nvl72/functions.sh +++ b/presto/slurm/presto-nvl72/functions.sh @@ -2,13 +2,13 @@ # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -CTAS_CONTAINER_OUTPUT_DIR=/var/lib/presto/data/hive/benchmark_output +CTAS_CONTAINER_SCRATCH_DIR=/var/lib/presto/data/hive/benchmark_output EXPECTED_RESULTS_CONTAINER_DIR=/var/lib/presto/expected-results -ctas_output_mount_arg() { - if [[ "${WRITE_RESULTS_TO_FILE:-0}" == "1" ]]; then - validate_environment_preconditions PRESTO_OUTPUT_DIR - printf ',%s:%s' "${PRESTO_OUTPUT_DIR}" "${CTAS_CONTAINER_OUTPUT_DIR}" +ctas_scratch_mount_arg() { + if [[ "${RUN_AS_CTAS_QUERIES:-0}" == "1" ]]; then + validate_environment_preconditions PRESTO_CTAS_SCRATCH_DIR + printf ',%s:%s' "${PRESTO_CTAS_SCRATCH_DIR}" "${CTAS_CONTAINER_SCRATCH_DIR}" fi return 0 } @@ -134,7 +134,7 @@ function run_coord_image { local extra_mounts extra_mounts="$(miniforge_mount_arg)" - extra_mounts+="$(ctas_output_mount_arg)" + extra_mounts+="$(ctas_scratch_mount_arg)" [[ "${type}" == "cli" ]] && extra_mounts+="$(expected_results_mount_arg)" if [[ -n "${CLUSTER_EXTRA_MOUNTS:-}" ]]; then extra_mounts="${extra_mounts},${CLUSTER_EXTRA_MOUNTS}" @@ -312,7 +312,7 @@ function run_worker { driver_mounts="${driver_mounts},${CLUSTER_LIBNVIDIA_ML_HOST_PATH}:${CLUSTER_LIBNVIDIA_ML_CONTAINER_PATH}" fi local worker_extra_mounts="" - worker_extra_mounts="$(ctas_output_mount_arg)" + worker_extra_mounts="$(ctas_scratch_mount_arg)" if [[ -n "${CLUSTER_EXTRA_MOUNTS:-}" ]]; then worker_extra_mounts+=",${CLUSTER_EXTRA_MOUNTS}" fi @@ -566,7 +566,7 @@ function run_queries { [[ "${ENABLE_METRICS}" == "1" ]] && extra_args+=("-m") [[ "${ENABLE_NSYS}" == "1" ]] && extra_args+=("-p" "--profile-script-path" "${container_script_dir}/profiler_functions.sh") [[ -n "${QUERIES:-}" ]] && extra_args+=("-q" "${QUERIES}") - [[ "${WRITE_RESULTS_TO_FILE:-0}" == "1" ]] && extra_args+=("--write-results-to-file") + [[ "${RUN_AS_CTAS_QUERIES:-0}" == "1" ]] && extra_args+=("--run-as-ctas-queries") source "${SCRIPT_DIR}/defaults.env" @@ -598,7 +598,7 @@ function run_queries { export PORT=$PORT; \ export HOSTNAME=$COORD; \ export PRESTO_DATA_DIR=/var/lib/presto/data/hive/data/user_data; \ - export PRESTO_OUTPUT_DIR=${CTAS_CONTAINER_OUTPUT_DIR}; \ + export PRESTO_CTAS_SCRATCH_DIR=${CTAS_CONTAINER_SCRATCH_DIR}; \ export PRESTO_EXPECTED_RESULTS_DIR=${container_expected_results_dir}; \ export MINIFORGE_HOME=/workspace/miniforge3; \ export HOME=/workspace; \ diff --git a/presto/slurm/presto-nvl72/launch-run.sh b/presto/slurm/presto-nvl72/launch-run.sh index 2e6d0cca..64f375a8 100755 --- a/presto/slurm/presto-nvl72/launch-run.sh +++ b/presto/slurm/presto-nvl72/launch-run.sh @@ -14,7 +14,7 @@ # [-i|--iterations ] [--cpu] [-g|--num-workers-per-node ] # [-w|--worker-image ] [-c|--coord-image ] # [-o|--output-path ] [-q|--queries ] -# [--write-results-to-file] [--disable-gds] [-m|--metrics] [-p|--profile] +# [--run-as-ctas-queries] [--disable-gds] [-m|--metrics] [-p|--profile] # [additional sbatch options] # ============================================================================== @@ -42,7 +42,7 @@ SCRIPT_DIR="$PWD" ENABLE_GDS=1 ENABLE_METRICS=0 ENABLE_NSYS=0 -WRITE_RESULTS_TO_FILE=0 +RUN_AS_CTAS_QUERIES=0 NSYS_WORKER_ID=0 QUERIES="" @@ -63,7 +63,7 @@ Options: -c, --coord-image Override coordinator image from cluster config -o, --output-path Copy results into this directory after the run -q, --queries Comma-separated query filter (e.g. "1,6,21") - --write-results-to-file Write distributed Parquet results with CTAS. Requires PRESTO_OUTPUT_DIR. + --run-as-ctas-queries Run queries as distributed CTAS writes. Requires PRESTO_CTAS_SCRATCH_DIR. --worker-env-file Override worker.env (default: ./worker.env) --cpu Use CPU partition/images (overrides cluster default) --gpu Use GPU partition/images (overrides cluster default) @@ -77,9 +77,8 @@ Options: Any arguments after -- are passed directly to sbatch. Cluster config (~/.cluster_config.env or \$CLUSTER_CONFIG) supplies partition, -account, time limits, image names, per-variant defaults, and the optional -PRESTO_EXPECTED_RESULTS_DIR. Expected results are mounted read-only into the -CLI container. See cluster_config.env.example. +account, time limits, image names, per-variant defaults, and optional CTAS +scratch and expected-result directories. See cluster_config.env.example. EOF } @@ -99,7 +98,7 @@ while [[ $# -gt 0 ]]; do --gpu) VARIANT_TYPE="gpu"; shift ;; --no-numa) USE_NUMA="0"; shift ;; --disable-gds) ENABLE_GDS=0; shift ;; - --write-results-to-file) WRITE_RESULTS_TO_FILE=1; shift ;; + --run-as-ctas-queries) RUN_AS_CTAS_QUERIES=1; shift ;; -m|--metrics) ENABLE_METRICS=1; shift ;; -p|--profile) ENABLE_NSYS=1; shift ;; -h|--help) usage; exit 0 ;; @@ -111,12 +110,12 @@ done [[ -z "${NODES_COUNT}" ]] && { echo "Error: -n|--nodes is required (see --help)" >&2; exit 1; } [[ -z "${SCALE_FACTOR}" ]] && { echo "Error: -s|--scale-factor is required (see --help)" >&2; exit 1; } -if [[ "${WRITE_RESULTS_TO_FILE}" == "1" ]]; then - [[ -n "${PRESTO_OUTPUT_DIR:-}" ]] || { echo "Error: PRESTO_OUTPUT_DIR is required with --write-results-to-file" >&2; exit 1; } - mkdir -p "${PRESTO_OUTPUT_DIR}" - PRESTO_OUTPUT_DIR="$(readlink -f "${PRESTO_OUTPUT_DIR}")" - [[ -d "${PRESTO_OUTPUT_DIR}" && -w "${PRESTO_OUTPUT_DIR}" ]] || { - echo "Error: PRESTO_OUTPUT_DIR must be a writable shared directory: ${PRESTO_OUTPUT_DIR}" >&2 +if [[ "${RUN_AS_CTAS_QUERIES}" == "1" ]]; then + [[ -n "${PRESTO_CTAS_SCRATCH_DIR:-}" ]] || { echo "Error: PRESTO_CTAS_SCRATCH_DIR is required with --run-as-ctas-queries" >&2; exit 1; } + mkdir -p "${PRESTO_CTAS_SCRATCH_DIR}" + PRESTO_CTAS_SCRATCH_DIR="$(readlink -f "${PRESTO_CTAS_SCRATCH_DIR}")" + [[ -d "${PRESTO_CTAS_SCRATCH_DIR}" && -w "${PRESTO_CTAS_SCRATCH_DIR}" ]] || { + echo "Error: PRESTO_CTAS_SCRATCH_DIR must be a writable shared directory: ${PRESTO_CTAS_SCRATCH_DIR}" >&2 exit 1 } fi @@ -183,9 +182,9 @@ build_common_export_vars EXPORT_VARS+=",NUM_ITERATIONS=${NUM_ITERATIONS}" EXPORT_VARS+=",ENABLE_GDS=${ENABLE_GDS},ENABLE_METRICS=${ENABLE_METRICS}" EXPORT_VARS+=",ENABLE_NSYS=${ENABLE_NSYS},NSYS_WORKER_ID=${NSYS_WORKER_ID}" -EXPORT_VARS+=",WRITE_RESULTS_TO_FILE=${WRITE_RESULTS_TO_FILE}" -if [[ "${WRITE_RESULTS_TO_FILE}" == "1" ]]; then - EXPORT_VARS+=",PRESTO_OUTPUT_DIR=${PRESTO_OUTPUT_DIR}" +EXPORT_VARS+=",RUN_AS_CTAS_QUERIES=${RUN_AS_CTAS_QUERIES}" +if [[ "${RUN_AS_CTAS_QUERIES}" == "1" ]]; then + EXPORT_VARS+=",PRESTO_CTAS_SCRATCH_DIR=${PRESTO_CTAS_SCRATCH_DIR}" fi if [[ -n "${PRESTO_EXPECTED_RESULTS_DIR:-}" ]]; then EXPORT_VARS+=",PRESTO_EXPECTED_RESULTS_DIR=${PRESTO_EXPECTED_RESULTS_DIR}" diff --git a/presto/testing/performance_benchmark_fixtures_test.py b/presto/testing/performance_benchmark_fixtures_test.py index c486a914..6a79089d 100644 --- a/presto/testing/performance_benchmark_fixtures_test.py +++ b/presto/testing/performance_benchmark_fixtures_test.py @@ -8,8 +8,8 @@ from presto.testing.performance_benchmarks.common_fixtures import ( _drop_results_schema, - alias_ctas_output_expressions, build_ctas_query, + ctas_output_column_aliases, ctas_schema_name, ctas_table_name, strip_trailing_semicolon, @@ -40,42 +40,42 @@ def test_ctas_names_and_sql(): assert ctas_table_name("Q7", 2) == "q7_iteration_3" assert strip_trailing_semicolon("SELECT 1;\n") == "SELECT 1" assert build_ctas_query("tpch", "Q7", "SELECT 1;", "hive_output", "results", "q7") == ( - "--tpch_Q7--\nCREATE TABLE hive_output.results.q7 WITH (format = 'PARQUET') AS\nSELECT 1 AS result_column_1" + "--tpch_Q7--\nCREATE TABLE hive_output.results.q7 (c1) WITH (format = 'PARQUET') AS\nSELECT 1" ) -def test_ctas_aliases_unnamed_and_duplicate_output_expressions(): +def test_ctas_uses_positional_aliases_for_explicit_output_expressions(): query = "SELECT c_name, sum(l_quantity), c_name, count(*) AS row_count FROM customer, lineitem GROUP BY c_name" - rewritten = alias_ctas_output_expressions(query) - - assert rewritten == ( - "SELECT c_name, SUM(l_quantity) AS result_column_2, " - "c_name AS result_column_3, COUNT(*) AS row_count " - "FROM customer, lineitem GROUP BY c_name" + assert ctas_output_column_aliases(query) == ["c1", "c2", "c3", "c4"] + assert build_ctas_query("tpch", "Q18", query, "hive_output", "results", "q18").endswith( + "(c1, c2, c3, c4) WITH (format = 'PARQUET') AS\n" + query ) -def test_ctas_alias_rewrite_leaves_stars_for_presto_to_expand(): - assert alias_ctas_output_expressions("SELECT nested.* FROM (SELECT 1 AS value) nested") == ( - "SELECT nested.* FROM (SELECT 1 AS value) nested" +def test_ctas_alias_list_leaves_stars_for_presto_to_expand(): + query = "SELECT nested.* FROM (SELECT 1 AS value) nested" + + assert ctas_output_column_aliases(query) is None + assert build_ctas_query("tpcds", "Q88", query, "hive_output", "results", "q88") == ( + "--tpcds_Q88--\nCREATE TABLE hive_output.results.q88 WITH (format = 'PARQUET') AS\n" + query ) @pytest.mark.parametrize("benchmark_type", ["tpch", "tpcds"]) -def test_ctas_alias_rewrite_parses_all_builtin_queries(benchmark_type): +def test_ctas_alias_list_parses_all_builtin_queries(benchmark_type): repo_root = Path(__file__).parents[2] query_path = repo_root / "common" / "testing" / "queries" / benchmark_type / "queries.json" queries = json.loads(query_path.read_text()) - rewritten = { - query_id: alias_ctas_output_expressions(query.replace("{SF_FRACTION}", "0.0001")) + aliases = { + query_id: ctas_output_column_aliases(query.replace("{SF_FRACTION}", "0.0001")) for query_id, query in queries.items() } - assert rewritten.keys() == queries.keys() + assert aliases.keys() == queries.keys() if benchmark_type == "tpch": - assert "SUM(l_quantity) AS result_column_6" in rewritten["Q18"] + assert aliases["Q18"] == ["c1", "c2", "c3", "c4", "c5", "c6"] def test_drop_results_schema_removes_managed_tables_first(): diff --git a/presto/testing/performance_benchmarks/common_fixtures.py b/presto/testing/performance_benchmarks/common_fixtures.py index 52fde623..58758e6e 100644 --- a/presto/testing/performance_benchmarks/common_fixtures.py +++ b/presto/testing/performance_benchmarks/common_fixtures.py @@ -33,6 +33,7 @@ class CtasResults: def ctas_schema_name(tag): + """Return the isolated scratch schema for this benchmark tag.""" if tag and not re.fullmatch(r"[A-Za-z0-9_]+", tag): raise ValueError("CTAS result tags must contain only alphanumeric and underscore characters") return f"benchmark_results_{tag.lower()}" if tag else "benchmark_results" @@ -46,56 +47,31 @@ def ctas_table_name(query_id, iteration_num): return query_id.lower() if iteration_num == 0 else f"{query_id.lower()}_iteration_{iteration_num + 1}" -def alias_ctas_output_expressions(query): - """Give every explicit CTAS output expression a unique column name.""" - query = strip_trailing_semicolon(query) - parsed = sqlglot.parse_one(query, read="presto") - select = next(parsed.find_all(exp.Select)) - expressions = list(select.expressions) +def ctas_output_column_aliases(query): + """Return unique CTAS aliases without modifying the original SELECT. - reserved_names = { - expression.alias_or_name.casefold() - for expression in expressions - if isinstance(expression, (exp.Alias, exp.Column)) and not expression.is_star - } - used_names = set() - changed = False - - for position, expression in enumerate(expressions, start=1): - if expression.is_star: - continue - - name = expression.alias_or_name if isinstance(expression, (exp.Alias, exp.Column)) else "" - normalized_name = name.casefold() - if name and normalized_name not in used_names: - used_names.add(normalized_name) - continue - - base_alias = f"result_column_{position}" - alias = base_alias - suffix = 2 - while alias.casefold() in reserved_names or alias.casefold() in used_names: - alias = f"{base_alias}_{suffix}" - suffix += 1 - - if isinstance(expression, exp.Alias): - expression.set("alias", exp.to_identifier(alias)) - else: - expressions[position - 1] = exp.alias_(expression, alias, copy=False) - used_names.add(alias.casefold()) - changed = True - - if not changed: - return query - select.set("expressions", expressions) - return parsed.sql(dialect="presto") + A wildcard's expanded column count depends on source schema metadata, so + wildcard projections keep their original output names instead. + """ + parsed = sqlglot.parse_one(strip_trailing_semicolon(query), read="presto") + select = next(parsed.find_all(exp.Select)) + if any( + isinstance(projection := expression.this if isinstance(expression, exp.Alias) else expression, exp.Star) + or (isinstance(projection, exp.Column) and isinstance(projection.this, exp.Star)) + for expression in select.expressions + ): + return None + return [f"c{position}" for position in range(1, len(select.expressions) + 1)] def build_ctas_query(benchmark_type, query_id, query, catalog, schema, table): + query = strip_trailing_semicolon(query) + aliases = ctas_output_column_aliases(query) + column_aliases = f" ({', '.join(aliases)})" if aliases else "" return ( f"--{benchmark_type}_{query_id}--\n" - f"CREATE TABLE {catalog}.{schema}.{table} WITH (format = 'PARQUET') AS\n" - f"{alias_ctas_output_expressions(query)}" + f"CREATE TABLE {catalog}.{schema}.{table}{column_aliases} WITH (format = 'PARQUET') AS\n" + f"{query}" ) @@ -111,16 +87,16 @@ def _drop_results_schema(cursor, catalog, schema): @pytest.fixture(scope="session") def ctas_results(request): - if not request.config.getoption("--write-results-to-file"): + if not request.config.getoption("--run-as-ctas-queries"): return None - output_dir = os.environ.get("PRESTO_OUTPUT_DIR") - if not output_dir: - pytest.exit("PRESTO_OUTPUT_DIR must be set when --write-results-to-file is enabled.", returncode=2) + scratch_dir = os.environ.get("PRESTO_CTAS_SCRATCH_DIR") + if not scratch_dir: + pytest.exit("PRESTO_CTAS_SCRATCH_DIR must be set when --run-as-ctas-queries is enabled.", returncode=2) tag = request.config.getoption("--tag") schema = ctas_schema_name(tag) - host_results_dir = Path(output_dir).expanduser() / schema + host_results_dir = Path(scratch_dir).expanduser() / schema conn = prestodb.dbapi.connect( host=request.config.getoption("--hostname"), @@ -238,6 +214,9 @@ def benchmark_query_function(query_id): result = [] for iteration_num in range(iterations): if ctas_results is not None: + # Retain the first iteration as the query result. Later + # iterations use short-lived tables so every measured + # execution performs the same CTAS work. table = ctas_table_name(query_id, iteration_num) scratch_table = table if iteration_num > 0 else None query = build_ctas_query( diff --git a/presto/testing/performance_benchmarks/conftest.py b/presto/testing/performance_benchmarks/conftest.py index f160b7da..383e9aa1 100644 --- a/presto/testing/performance_benchmarks/conftest.py +++ b/presto/testing/performance_benchmarks/conftest.py @@ -48,7 +48,7 @@ def pytest_addoption(parser): parser.addoption("--metrics", action="store_true", default=False) parser.addoption("--skip-drop-cache", action="store_true", default=False) parser.addoption("--skip-analyze-check", action="store_true", default=False) - parser.addoption("--write-results-to-file", action="store_true", default=False) + parser.addoption("--run-as-ctas-queries", action="store_true", default=False) def pytest_configure(config): From fa33349714a6d9092e0cc3dfc4387482dd48597c Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Thu, 30 Jul 2026 15:43:55 -0700 Subject: [PATCH 10/12] Explain CTAS output name mapping --- common/testing/normalize_ctas_results.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/common/testing/normalize_ctas_results.py b/common/testing/normalize_ctas_results.py index 66973abc..d42d2389 100644 --- a/common/testing/normalize_ctas_results.py +++ b/common/testing/normalize_ctas_results.py @@ -32,6 +32,10 @@ def _query_directories(source_dir: Path) -> list[Path]: def _source_output_names(query_sql: str, physical_names: list[str]) -> list[str]: """Return names from an explicit SELECT projection for ORDER BY lookup.""" + # Explicit CTAS projections are stored with unique positional names (c1, + # c2, ...), but ORDER BY still refers to names from the original SELECT. + # Recover those logical names so ORDER BY expressions can be mapped to the + # correct physical column positions without changing the stored schema. parsed = sqlglot.parse_one(query_sql, read="presto") select = next(parsed.find_all(exp.Select)) has_wildcard = any( From 66424c412fe76fee8a6b156ed796ebfa975ee88c Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Thu, 30 Jul 2026 22:55:43 +0000 Subject: [PATCH 11/12] Address code review feedback on PR #373 Make get_orderby_sort_spec and restore_orderby public so they can be imported without crossing a private-symbol boundary. Remove redundant alphabetical sort in validate_results.py (numeric sort on iteration is the meaningful one). Add clarifying comment on scratch_table cleanup invariant in benchmark_query. Fix --run-as-ctas-queries alignment in run_benchmark.sh usage text. --- common/testing/normalize_ctas_results.py | 6 +++--- common/testing/result_comparison.py | 8 ++++---- common/testing/result_comparison_test.py | 12 ++++++------ common/testing/validate_results.py | 2 +- presto/scripts/run_benchmark.sh | 2 +- .../performance_benchmarks/common_fixtures.py | 3 +++ 6 files changed, 18 insertions(+), 15 deletions(-) diff --git a/common/testing/normalize_ctas_results.py b/common/testing/normalize_ctas_results.py index 66973abc..0a21483d 100644 --- a/common/testing/normalize_ctas_results.py +++ b/common/testing/normalize_ctas_results.py @@ -19,7 +19,7 @@ # Allow importing from the repo root when this file is executed directly. sys.path.insert(0, str(Path(__file__).parent.parent.parent)) -from common.testing.result_comparison import _get_orderby_sort_spec, _restore_orderby +from common.testing.result_comparison import get_orderby_sort_spec, restore_orderby from common.testing.test_utils import get_queries @@ -50,7 +50,7 @@ def _write_canonical_result(query_dir: Path, target: Path, query_sql: str | None if parquet_parts and query_sql is not None: column_names = _source_output_names(query_sql, pq.read_schema(parquet_parts[0]).names) - sort_spec = _get_orderby_sort_spec(query_sql, column_names) + sort_spec = get_orderby_sort_spec(query_sql, column_names) temporary = target.with_name(f".{target.name}.tmp-{os.getpid()}") try: @@ -61,7 +61,7 @@ def _write_canonical_result(query_dir: Path, target: Path, query_sql: str | None else: frame = pd.read_parquet(query_dir) if sort_spec[0]: - frame = _restore_orderby(frame, *sort_spec) + frame = restore_orderby(frame, *sort_spec) frame.to_parquet(temporary, index=False) os.replace(temporary, target) finally: diff --git a/common/testing/result_comparison.py b/common/testing/result_comparison.py index 053e801b..617ee275 100644 --- a/common/testing/result_comparison.py +++ b/common/testing/result_comparison.py @@ -34,7 +34,7 @@ # --------------------------------------------------------------------------- -def _get_orderby_sort_spec( +def get_orderby_sort_spec( query_sql: str, expected_col_names: list[str], ) -> tuple[list[int], list[bool], list[bool]]: @@ -91,7 +91,7 @@ def get_orderby_col_indices(query_sql: str, expected_col_names: list[str]) -> tu ORDER BY expression is too complex to map to a result column (CASE, aggregate, etc.). """ - sort_col_indices, ascending, _ = _get_orderby_sort_spec(query_sql, expected_col_names) + sort_col_indices, ascending, _ = get_orderby_sort_spec(query_sql, expected_col_names) return sort_col_indices, ascending @@ -299,7 +299,7 @@ def _validate_orderby(df: pd.DataFrame, sort_col_indices: list[int], ascending: prior_changed = prior_changed | value_changed -def _restore_orderby( +def restore_orderby( df: pd.DataFrame, sort_col_indices: list[int], ascending: list[bool], @@ -481,7 +481,7 @@ def compare_result_frames( raise AssertionError(f"Row count mismatch: {len(actual)} (actual) vs {len(expected)} (expected)") # 4. Parse ORDER BY / LIMIT. - sort_col_indices, ascending, _ = _get_orderby_sort_spec(query_sql, expected_col_names) + sort_col_indices, ascending, _ = get_orderby_sort_spec(query_sql, expected_col_names) limit = get_limit(query_sql) # 5. Per-frame ORDER BY validation — engine order is intact here. diff --git a/common/testing/result_comparison_test.py b/common/testing/result_comparison_test.py index 3282ce19..1f3d8371 100644 --- a/common/testing/result_comparison_test.py +++ b/common/testing/result_comparison_test.py @@ -12,9 +12,9 @@ from common.testing.result_comparison import ( _canonical_sort, _find_last_tie_start, - _get_orderby_sort_spec, _normalize_to_expected, - _restore_orderby, + get_orderby_sort_spec, + restore_orderby, _validate_orderby, ) @@ -113,24 +113,24 @@ def test_restore_orderby_uses_mixed_directions_and_explicit_null_placement(): "payload": ["a2", "a1-low", "null-a", "a1-null", "a1-high"], } ) - indices, ascending, nulls_first = _get_orderby_sort_spec( + indices, ascending, nulls_first = get_orderby_sort_spec( "SELECT a, b, payload FROM source ORDER BY a ASC NULLS LAST, b DESC NULLS FIRST", list(df.columns), ) - restored = _restore_orderby(df, indices, ascending, nulls_first) + restored = restore_orderby(df, indices, ascending, nulls_first) assert restored["payload"].tolist() == ["a1-null", "a1-high", "a1-low", "a2", "null-a"] def test_restore_orderby_defaults_to_presto_nulls_last_for_desc(): df = pd.DataFrame({"score": [None, 1, 2], "payload": ["null", "low", "high"]}) - indices, ascending, nulls_first = _get_orderby_sort_spec( + indices, ascending, nulls_first = get_orderby_sort_spec( "SELECT score, payload FROM source ORDER BY score DESC", list(df.columns), ) - restored = _restore_orderby(df, indices, ascending, nulls_first) + restored = restore_orderby(df, indices, ascending, nulls_first) assert ascending == [False] assert nulls_first == [False] diff --git a/common/testing/validate_results.py b/common/testing/validate_results.py index 46f9b261..91971ce2 100644 --- a/common/testing/validate_results.py +++ b/common/testing/validate_results.py @@ -54,7 +54,7 @@ def validate( result_files = [results_dir / f"q{q_num}.parquet" for q_num in query_numbers] result_files = [path for path in result_files if path.exists()] else: - result_files = sorted(path for path in results_dir.iterdir() if re.fullmatch(r"q\d+\.parquet", path.name)) + result_files = [path for path in results_dir.iterdir() if re.fullmatch(r"q\d+\.parquet", path.name)] if not result_files: print(f"No result Parquet files found in {results_dir}", file=sys.stderr) diff --git a/presto/scripts/run_benchmark.sh b/presto/scripts/run_benchmark.sh index db141dcc..87704173 100755 --- a/presto/scripts/run_benchmark.sh +++ b/presto/scripts/run_benchmark.sh @@ -43,7 +43,7 @@ OPTIONS: --profile-script-path Path to a custom profiler functions script. Defaults to ./profiler_functions.sh. --skip-drop-cache Skip dropping system caches before each benchmark query (dropped by default). --skip-analyze-check Skip checking that ANALYZE TABLE has been run on all tables (checked by default). - --run-as-ctas-queries Run queries as distributed Hive CTAS operations instead of returning results + --run-as-ctas-queries Run queries as distributed Hive CTAS operations instead of returning results through the coordinator. PRESTO_CTAS_SCRATCH_DIR must be set and mounted when the cluster starts. -m, --metrics Collect detailed metrics from Presto REST API after each query. Metrics are stored in query-specific directories. diff --git a/presto/testing/performance_benchmarks/common_fixtures.py b/presto/testing/performance_benchmarks/common_fixtures.py index 58758e6e..fb7c6cd0 100644 --- a/presto/testing/performance_benchmarks/common_fixtures.py +++ b/presto/testing/performance_benchmarks/common_fixtures.py @@ -277,6 +277,9 @@ def benchmark_query_function(query_id): raw_times_dict[query_id] = None raise finally: + # scratch_table is non-None here only when an exception cut the loop + # short before the in-loop DROP ran. The in-loop path resets it to + # None after a successful drop, so this path is the exception handler. if scratch_table is not None and ctas_results is not None: with suppress(Exception): presto_cursor.execute( From 9cd4dbfba17ab6613ed76f6ccd9665161de4d16a Mon Sep 17 00:00:00 2001 From: misiugodfrey Date: Thu, 30 Jul 2026 16:25:46 -0700 Subject: [PATCH 12/12] Use a fixed CTAS scratch schema --- presto/scripts/run_benchmark.sh | 7 ++----- .../testing/performance_benchmark_fixtures_test.py | 7 ++----- .../performance_benchmarks/common_fixtures.py | 14 ++++---------- 3 files changed, 8 insertions(+), 20 deletions(-) diff --git a/presto/scripts/run_benchmark.sh b/presto/scripts/run_benchmark.sh index 87704173..a76085b6 100755 --- a/presto/scripts/run_benchmark.sh +++ b/presto/scripts/run_benchmark.sh @@ -11,6 +11,8 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" # LOGS_DIR points to the directory where server log files (including nvidia-smi # output) are written so that run_context.py can parse GPU info. export LOGS_DIR="${LOGS_DIR:-${SCRIPT_DIR}/presto_logs}" +# CTAS scratch is reset for every run; --tag only organizes permanent output. +CTAS_RESULTS_SCHEMA="benchmark_results" source "${SCRIPT_DIR}/presto_connection_defaults.sh" @@ -254,11 +256,6 @@ if [[ "${RUN_AS_CTAS_QUERIES}" == "true" ]]; then echo "Error: PRESTO_CTAS_SCRATCH_DIR must be a writable directory: ${PRESTO_CTAS_SCRATCH_DIR}" >&2 exit 1 fi - if [[ -n ${TAG} ]]; then - CTAS_RESULTS_SCHEMA="benchmark_results_${TAG,,}" - else - CTAS_RESULTS_SCHEMA="benchmark_results" - fi CTAS_SCRATCH_RESULTS_DIR="${PRESTO_CTAS_SCRATCH_DIR}/${CTAS_RESULTS_SCHEMA}" fi diff --git a/presto/testing/performance_benchmark_fixtures_test.py b/presto/testing/performance_benchmark_fixtures_test.py index 6a79089d..9f297100 100644 --- a/presto/testing/performance_benchmark_fixtures_test.py +++ b/presto/testing/performance_benchmark_fixtures_test.py @@ -7,10 +7,10 @@ import pytest from presto.testing.performance_benchmarks.common_fixtures import ( + CTAS_SCHEMA, _drop_results_schema, build_ctas_query, ctas_output_column_aliases, - ctas_schema_name, ctas_table_name, strip_trailing_semicolon, ) @@ -32,10 +32,7 @@ def fetchall(self): def test_ctas_names_and_sql(): - assert ctas_schema_name(None) == "benchmark_results" - assert ctas_schema_name("GPU_SF100") == "benchmark_results_gpu_sf100" - with pytest.raises(ValueError, match="alphanumeric"): - ctas_schema_name("../other") + assert CTAS_SCHEMA == "benchmark_results" assert ctas_table_name("Q7", 0) == "q7" assert ctas_table_name("Q7", 2) == "q7_iteration_3" assert strip_trailing_semicolon("SELECT 1;\n") == "SELECT 1" diff --git a/presto/testing/performance_benchmarks/common_fixtures.py b/presto/testing/performance_benchmarks/common_fixtures.py index fb7c6cd0..e19dc5ab 100644 --- a/presto/testing/performance_benchmarks/common_fixtures.py +++ b/presto/testing/performance_benchmarks/common_fixtures.py @@ -2,7 +2,6 @@ # SPDX-License-Identifier: Apache-2.0 import os -import re import shutil from contextlib import suppress from dataclasses import dataclass @@ -24,6 +23,9 @@ from .run_context import gather_run_context CTAS_CATALOG = "hive_output" +# This schema is temporary and reset before every CTAS benchmark. The regular +# benchmark --tag applies only to permanent output artifacts. +CTAS_SCHEMA = "benchmark_results" @dataclass(frozen=True) @@ -32,13 +34,6 @@ class CtasResults: schema: str -def ctas_schema_name(tag): - """Return the isolated scratch schema for this benchmark tag.""" - if tag and not re.fullmatch(r"[A-Za-z0-9_]+", tag): - raise ValueError("CTAS result tags must contain only alphanumeric and underscore characters") - return f"benchmark_results_{tag.lower()}" if tag else "benchmark_results" - - def strip_trailing_semicolon(query): return query.rstrip().removesuffix(";").rstrip() @@ -94,8 +89,7 @@ def ctas_results(request): if not scratch_dir: pytest.exit("PRESTO_CTAS_SCRATCH_DIR must be set when --run-as-ctas-queries is enabled.", returncode=2) - tag = request.config.getoption("--tag") - schema = ctas_schema_name(tag) + schema = CTAS_SCHEMA host_results_dir = Path(scratch_dir).expanduser() / schema conn = prestodb.dbapi.connect(