Add distributed CTAS output for Presto benchmarks by misiugodfrey · Pull Request #373 · rapidsai/velox-testing

misiugodfrey · 2026-07-15T23:23:17Z

Summary

add an opt-in --run-as-ctas-queries mode that executes benchmark queries as Hive CTAS operations, avoiding result-set transfer through the coordinator during measured execution
use a worker-visible PRESTO_CTAS_SCRATCH_DIR for temporary distributed output while keeping source data read-only
after timing and profiling, normalize each retained CTAS result into the same --output-dir[/tag]/query_results/qN.parquet layout produced by the default benchmark path
merge multi-part Parquet output and reconstruct resolvable SQL ORDER BY clauses before validation
keep validation and result posting independent of whether a query ran normally or as CTAS
use native CTAS column aliases (CREATE TABLE ... (c1, c2, ...) AS <original SELECT>) without rewriting explicit SELECT expressions; wildcard projections retain their original output names
retain the first iteration as the query result and remove later-iteration scratch tables
update Docker and Slurm mounts, configuration examples, help text, documentation, and focused tests

Output layout

Both execution modes now produce:

<output-dir>[/<tag>]/
└── query_results/
    ├── q1.parquet
    ├── q2.parquet
    └── ...

In CTAS mode, workers first write temporary qN/ datasets beneath PRESTO_CTAS_SCRATCH_DIR. Normalization happens after the measured query execution and before validation.

Verification

Current revision

Earlier CTAS revision

Docker native-CPU TPC-H Q1 smoke with two CTAS iterations and separate read-only source/writable result mounts
Live TPC-H Q18 CTAS run
Docker multi-worker CTAS run producing multiple Parquet parts
Live empty-result CTAS query
Docker native-GPU worker path
Slurm deployment path
Full TPC-H query suite

copy-pr-bot · 2026-07-15T23:23:21Z

Auto-sync is disabled for draft pull requests in this repository. Workflows must be run manually.

Contributors can view more details about this message here.

misiugodfrey · 2026-07-15T23:25:25Z

    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:


Multiple workers can each write their own Parquet part file, so we now have the results as a directory.

Can we instead normalize the Parquet files and not have the validation logic be concerned with the details of how the result set is generated?

misiugodfrey · 2026-07-20T14:20:25Z

-    aggregate, etc.).
-    """
-    expr = sqlglot.parse_one(query_sql)
+def _get_orderby_sort_spec(


We were able to skip null handling when we had a single parquet file, but now since we are merging the files back based on order-by info, we need to know where nulls should go.

I believe this comment also applies here.

misiugodfrey · 2026-07-20T14:23:24Z

        prior_changed = prior_changed | value_changed


+def _restore_orderby(


To merge multiple parquet files we now need to know the hierarchy of the order by columns from the original sql query.

misiugodfrey · 2026-07-21T15:17:19Z

@devavret I ran some performance benchmarks using these changes to make sure I saw the same performance increase you detailed in the issue on Q11. I saw about a %25 reduction in time on this query, although this was running a 4-gpu tpch-1k run with an older image; so I'm not sure how much that differs from your original observations.
From the issue, it looks like we should be expecting much more of a gain, although I don't know exactly what context the POC was in, so let me know if further changes are needed to realize the full gain.

paul-aiyedun · 2026-07-23T03:08:38Z

    --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


Do we plan on running/publishing benchmarks for the CTAS queries?

That would be a question for @devavret . I believe we are allowed to, but I don't know if we have explicitly declared intentions.

Yes, that's the primary use case. The reason we want this CTAS option is to speed up two queries.

devavret · 2026-07-28T16:11:22Z

@devavret I ran some performance benchmarks using these changes to make sure I saw the same performance increase you detailed in the issue on Q11. I saw about a %25 reduction in time on this query, although this was running a 4-gpu tpch-1k run with an older image; so I'm not sure how much that differs from your original observations.
From the issue, it looks like we should be expecting much more of a gain, although I don't know exactly what context the POC was in, so let me know if further changes are needed to realize the full gain.

@misiugodfrey If you see Q11 going from 50s to 3s on sf30k 8 nodes then this PR is sufficient.

patdevinwilson

Solid feature overall — separate writable hive_output catalog + reconstructing ORDER BY for unordered Parquet datasets is the right shape. Blocking items: style CI (ruff format) is red, and I think implicit NULLS placement for DESC is wrong in _restore_orderby. Details inline.

patdevinwilson · 2026-07-28T16:27:28Z


        sort_col_indices.append(resolved_idx)
        ascending.append(not is_desc)
+        nulls_first.append(bool(ordered.args.get("nulls_first")))


bool(ordered.args.get("nulls_first")) treats unspecified NULLS as False (NULLS LAST). Presto's default is NULLS LAST for ASC but NULLS FIRST for DESC. So for ORDER BY score DESC (no explicit NULLS clause), restore will put nulls last while the engine put them first, and the subsequent _validate_orderby / LIMIT tie handling can disagree with expected.

Suggest something like:

nulls_first_arg = ordered.args.get("nulls_first") if nulls_first_arg is None: nulls_first.append(bool(is_desc)) # Presto default else: nulls_first.append(bool(nulls_first_arg))

And add a test for bare ORDER BY x DESC with nulls.

Are you sure Presto changes NULLS FIRST/LAST based on the order? I was under the impression it was always last unless explicitly changed and as far as I can tell the documentation supports this: https://prestodb.github.io/docs/current/sql/select.html#order-by-clause:~:text=The%20default%20null%20ordering%20is%20NULLS%20LAST%2C%20regardless%20of%20the%20ordering%20direction.

If I'm missing something (or looking at the wrong docs), please link me to the appropriate documentation and I will change it.

You're right — I was wrong here. Presto's docs are clear:

The default null ordering is NULLS LAST, regardless of the ordering direction.

(https://prestodb.github.io/docs/current/sql/select.html#order-by-clause)

I was mixing that up with other engines (e.g. Postgres / Spark defaults that do flip with ASC/DESC). For Presto, bool(ordered.args.get("nulls_first")) → False when unspecified correctly matches NULLS LAST. Thanks for the correction.

patdevinwilson · 2026-07-28T16:27:28Z

@@ -321,6 +354,55 @@ 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[*]}


set -e is on (line 6). If pytest fails, we never reach normalize_ctas_result_permissions or validation — and for CTAS that also means permission fixup / result posting from Slurm may see unreadable files.

Same pattern as #382 would help:

pytest_exit=0 pytest ... || pytest_exit=$? # ... normalize + validate ... exit ${pytest_exit}

patdevinwilson · 2026-07-28T16:27:28Z

+# 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


Good call disabling temporary staging so native workers write straight to the shared mount. Worth confirming (maybe already did) that coordinator commit still works for multi-worker CTAS with this setting — if an interrupted run leaves partial qN/ dirs, the fixture's shutil.rmtree + DROP TABLE path looks covered.

patdevinwilson · 2026-07-28T16:27:28Z

      - ./.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


Making PRESTO_DATA_DIR :ro is a nice hardening win for the CTAS split. Just flagging: any existing workflow that still wrote into user_data (not via hive_output) will now fail — hopefully none remain.

I don't think any of our existing workflows should be writing to user_data. Our data-gen workflow acts outside of this container approach, so we should be good.

If anything was writing to the data like this, I think it would be considered a bug.

Makes sense — thanks for confirming. If something was writing through this mount, treating that as a bug (and failing loudly under :ro) is the right call. No change needed from me.

patdevinwilson · 2026-07-28T16:27:28Z

                    parquet_path = results_dir / f"{query_id.lower()}.parquet"
                    df.to_parquet(parquet_path, index=False)

+                result.append(cursor.stats["elapsedTimeMillis"])


Timing is now recorded after fetchall() for CTAS (good — waits for commit). For non-CTAS, elapsedTimeMillis is also after fetch, matching prior behavior for iteration 0 and changing later iterations (previously timed before fetch on iter>0). Intentional? If the goal is "time until results ready", this is better; if comparing to historical non-CTAS numbers for iter>0, they may shift slightly.

I'll preserve the original behaviour when we are not using CTAS.

patdevinwilson · 2026-07-28T16:27:28Z

+            actual,
+            expected,
+            query_sql,
+            actual_order_preserved=result_file.is_file(),


actual_order_preserved=result_file.is_file() is a clean way to distinguish single-file vs distributed dataset dirs. For an empty CTAS table that only has .prestoSchema (your q15 test), a brief comment near the read_parquet call about directory/empty-schema behavior would help future readers.

Also: CI is failing ruff format on this file and performance_benchmark_fixtures_test.py — a format pass should unblock merge.

paul-aiyedun

I think adding a CTAS execution mode in order to understand exchange overhead makes sense. However, I think there is some interface divergence in the current implementation.

paul-aiyedun · 2026-07-28T16:42:36Z

                            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


We already write outputs (including result sets) to --output-dir. I think adding another PRESTO_OUTPUT_DIR variable here can be confusing.

I can rename this to something separate PRESTO_CTAS_SCRATCH_DIR, but the problem is this directory needs to be mounted in the container when it starts - so using --output-dir isn't viable, since that variable isn't specified by the user until the containers are already running (at least in the docker path).

paul-aiyedun · 2026-07-28T16:44:44Z

    --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


Can we use a more direct parameter name e.g. --run-as-ctas-queries?

Sounds good.

paul-aiyedun · 2026-07-28T16:48:52Z

  exit 1
 fi

+if [[ "${WRITE_RESULTS_TO_FILE}" == "true" ]]; then


Similar to the above comment, we currently write to --output-dir, so why do we need special handling here and in normalize_ctas_result_permissions?

paul-aiyedun · 2026-07-28T16:52:13Z

    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:


Can we instead normalize the Parquet files and not have the validation logic be concerned with the details of how the result set is generated?

paul-aiyedun · 2026-07-28T16:53:41Z

-    aggregate, etc.).
-    """
-    expr = sqlglot.parse_one(query_sql)
+def _get_orderby_sort_spec(


I believe this comment also applies here.

paul-aiyedun · 2026-07-28T17:05:46Z

I don't think it is common to have tests for test fixtures?

These tests aren't really "testing the test", so much as testing the aliasing, naming, and schema logic that happen to be used in that fixture. I have found it useful to catch regressions with these - but perhaps we should organize them differently?

paul-aiyedun · 2026-07-28T17:23:05Z

+
+
+def ctas_schema_name(tag):
+    if tag and not re.fullmatch(r"[A-Za-z0-9_]+", tag):


What is the difference between the regular benchmark tag and this tag?

It's the same tag; it's just used here for separating potentially concurrent runs. Probably not a major concern, so I can remove it for simplicity.

paul-aiyedun · 2026-07-28T17:24:06Z

+    return query.rstrip().removesuffix(";").rstrip()
+
+
+def ctas_table_name(query_id, iteration_num):


Are we creating a new table for each iteration?

Yes. The first table is retained for normalization and validation and every other iteration creates and then drops a table.

paul-aiyedun · 2026-07-28T17:27:50Z

+    return query_id.lower() if iteration_num == 0 else f"{query_id.lower()}_iteration_{iteration_num + 1}"
+
+
+def alias_ctas_output_expressions(query):


Can you please expand on the reason for this? I would have thought that CTAS query construction would be more straightforward i.e. CREATE TABLE ... AS {original SELECT query}.

That was the original approach, but some of the queries contained result sets with unnamed expressions (legal) but can't be expressed in CTAS because it needs unique column names.

For instance, TPCH Q18 has:

SELECT c_name, c_custkey, o_orderkey, o_orderdate, o_totalprice, sum(l_quantity)

And that sum(l_quantity) needs an explicit column name. This function is handling that.

Got it. Do we care about the specific column names? If not, I think we can simplify this logic by changing the CREATE TABLE clause i.e. CREATE TABLE {table name} (c1, c2, c3, ...) AS {original SELECT query}. This also reduces the chances of unexpected changes to the SELECT query.

paul-aiyedun · 2026-07-28T17:29:45Z

How do the settings here interact with etc_worker/catalog/hive.properties and etc_coordinator/catalog/hive.properties?

It's a separate catalog and should not interact with them at all beyond inheriting some shared settings.

misiugodfrey · 2026-07-30T22:35:03Z

I've updated the PR to have the CTAS results stored in a scratch location and then normalize them to a single parquet file (if multiple files were created). This should keep the rest of the verification unchanged, and adds a separate "result normalization" step that should run rarely (we should not often create multiple result files).

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.

…x-testing into misiug/writeToFile

copy-pr-bot · 2026-07-30T22:56:59Z

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

misiugodfrey commented Jul 20, 2026

View reviewed changes

misiugodfrey added 5 commits July 20, 2026 11:37

Add distributed CTAS benchmark output

6809036

Restore ordering for distributed CTAS results

c9cca4a

Reject duplicate benchmark result representations

caaafec

Allow empty distributed CTAS results

a8368aa

Colocate CTAS unit tests

97aa11e

misiugodfrey force-pushed the misiug/writeToFile branch from 68a0655 to 97aa11e Compare July 20, 2026 15:38

misiugodfrey added 2 commits July 20, 2026 15:09

Alias unnamed CTAS output columns

66614f9

Mount expected benchmark results in Slurm runs

132e13d

misiugodfrey marked this pull request as ready for review July 21, 2026 15:14

misiugodfrey requested a review from a team as a code owner July 21, 2026 15:14

misiugodfrey requested review from devavret and paul-aiyedun July 21, 2026 15:14

paul-aiyedun reviewed Jul 23, 2026

View reviewed changes

devavret approved these changes Jul 28, 2026

View reviewed changes

patdevinwilson requested changes Jul 28, 2026

View reviewed changes

patdevinwilson mentioned this pull request Jul 28, 2026

Add nightly Presto benchmark script for NVL72 Slurm cluster #382

Open

5 tasks

misiugodfrey added 2 commits July 28, 2026 10:23

Address CTAS benchmark review feedback

048d595

Merge remote-tracking branch 'origin/main' into misiug/writeToFile

25a9c1f

misiugodfrey requested a review from patdevinwilson July 28, 2026 17:28

paul-aiyedun reviewed Jul 28, 2026

View reviewed changes

Normalize CTAS benchmark results

225bd6b

misiugodfrey added 3 commits July 30, 2026 15:43

Explain CTAS output name mapping

fa33349

Merge branch 'misiug/writeToFile' of https://github.com/rapidsai/velo…

c65cea7

…x-testing into misiug/writeToFile

Use a fixed CTAS scratch schema

9cd4dbf

misiugodfrey requested a review from paul-aiyedun July 30, 2026 23:42

		prior_changed = prior_changed \| value_changed


		def _restore_orderby(

		@@ -321,6 +354,55 @@ 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[*]}



		def ctas_schema_name(tag):
		if tag and not re.fullmatch(r"[A-Za-z0-9_]+", tag):

		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 alias_ctas_output_expressions(query):

Uh oh!

Conversation

misiugodfrey commented Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Output layout

Verification

Current revision

Earlier CTAS revision

Uh oh!

copy-pr-bot Bot commented Jul 15, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

misiugodfrey commented Jul 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

devavret commented Jul 28, 2026

Uh oh!

patdevinwilson left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

paul-aiyedun left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

misiugodfrey commented Jul 15, 2026 •

edited

Loading

misiugodfrey commented Jul 21, 2026 •

edited

Loading

misiugodfrey Jul 30, 2026 •

edited

Loading