diff --git a/common/testing/normalize_ctas_results.py b/common/testing/normalize_ctas_results.py
new file mode 100644
index 00000000..41df8fa5
--- /dev/null
+++ b/common/testing/normalize_ctas_results.py
@@ -0,0 +1,121 @@
+#!/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."""
+    # 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(
+        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 336abc7e..617ee275 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,
@@ -443,8 +480,8 @@ 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.
+    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 be5ec902..1f3d8371 100644
--- a/common/testing/result_comparison_test.py
+++ b/common/testing/result_comparison_test.py
@@ -13,6 +13,8 @@
     _canonical_sort,
     _find_last_tie_start,
     _normalize_to_expected,
+    get_orderby_sort_spec,
+    restore_orderby,
     _validate_orderby,
 )
 
@@ -103,6 +105,38 @@ 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_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"]
+
+
 # ---------------------------------------------------------------------------
 # _validate_orderby null handling
 # ---------------------------------------------------------------------------
@@ -123,15 +157,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 a19fe143..91971ce2 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.
         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,17 @@ 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 = [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(results_dir.glob("q*.parquet"))
+        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)
+        print(f"No result Parquet files 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"
+    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):
@@ -143,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.",
+        description="Validate TPC-H/TPC-DS query results against expected Parquet files.",
         formatter_class=argparse.RawDescriptionHelpFormatter,
         epilog=__doc__,
     )
@@ -179,11 +181,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 +193,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 = 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 +219,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/common/testing/validate_results_test.py b/common/testing/validate_results_test.py
new file mode 100644
index 00000000..40ed745b
--- /dev/null
+++ b/common/testing/validate_results_test.py
@@ -0,0 +1,38 @@
+# 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.validate_results import validate
+
+
+def test_validator_accepts_empty_q15_result(tmp_path: Path):
+    results_dir = tmp_path / "actual"
+    results_dir.mkdir()
+    pd.DataFrame().to_parquet(results_dir / "q15.parquet", index=False)
+
+    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()
+    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"]
diff --git a/presto/README.md b/presto/README.md
index c744e2f8..746edd5d 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 run large-result queries as
+   distributed CTAS writes, also select a separate writable scratch directory
+   before starting the cluster:
+
+   ```bash
+   export PRESTO_CTAS_SCRATCH_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,19 @@ 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 `--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 --run-as-ctas-queries
+   ```
+
+5. For the standard coordinator-returned result mode, run:
    ```bash
    ./run_benchmark.sh --help
-   ./run_benchmark.sh --benchmark tpch --schema-name <your_schema>
+   ./run_benchmark.sh --benchmark-type tpch --schema-name <your_schema>
    ```
    > **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..e26ab90f
--- /dev/null
+++ b/presto/docker/config/template/etc_common/catalog/hive_output.properties
@@ -0,0 +1,10 @@
+# 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
+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 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 f7805412..68ef8c71 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_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 49fd7bae..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"
 
@@ -43,6 +45,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).
+    --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.
@@ -57,6 +61,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_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.
@@ -68,6 +75,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_CTAS_SCRATCH_DIR=/results $0 -b tpch -s bench_sf100 --run-as-ctas-queries
     $0 -b tpch -s bench_sf100 --verbose
 
 EOF
@@ -191,6 +199,10 @@ parse_args() {
         SKIP_ANALYZE_CHECK=true
         shift
         ;;
+      --run-as-ctas-queries)
+        RUN_AS_CTAS_QUERIES=true
+        shift
+        ;;
       -m|--metrics)
         METRICS=true
         shift
@@ -232,6 +244,21 @@ if [[ -z ${SCHEMA_NAME} ]]; then
   exit 1
 fi
 
+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_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
+  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.
 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 +326,10 @@ if [[ "${SKIP_ANALYZE_CHECK}" == "true" ]]; then
   PYTEST_ARGS+=("--skip-analyze-check")
 fi
 
+if [[ "${RUN_AS_CTAS_QUERIES}" == "true" ]]; then
+  PYTEST_ARGS+=("--run-as-ctas-queries")
+fi
+
 source "${SCRIPT_DIR}/../../scripts/py_env_functions.sh"
 
 trap delete_python_virtual_env EXIT
@@ -319,7 +350,58 @@ 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=$?
+
+ensure_ctas_scratch_readable() {
+  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 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 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
+
+  # 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
+
+  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
+    echo "Error: CTAS result files are not readable from the benchmark host: ${host_results_dir}" >&2
+    return 1
+  fi
+}
+
+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
 # post_results.py has self-contained, run-specific data even when multiple runs
@@ -331,6 +413,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/"
@@ -350,7 +447,10 @@ 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.
-if [[ -n ${PRESTO_EXPECTED_RESULTS_DIR} && ! -d ${PRESTO_EXPECTED_RESULTS_DIR} ]]; then
+VALIDATION_EXIT=0
+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}")
@@ -363,12 +463,21 @@ else
   if [[ -n ${QUERIES} ]]; then
     VALIDATE_ARGS+=(--queries "${QUERIES}")
   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:-<not set>}"
+  ACTUAL_RESULTS_DIR="${EFFECTIVE_BENCHMARK_DIR}/query_results"
+  echo "[Validation] Running validation: ${ACTUAL_RESULTS_DIR} vs ${PRESTO_EXPECTED_RESULTS_DIR:-<not set>}"
   "${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 normalization, artifact collection, and validation.
+if [[ ${PYTEST_EXIT} -ne 0 ]]; then
+  exit "${PYTEST_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 2243266b..ec17291d 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=()
+RUN_AS_CTAS_QUERIES=false
 parse_args() {
   while [[ $# -gt 0 ]]; do
       case $1 in
@@ -55,6 +56,10 @@ parse_args() {
                 exit 1
             fi
             ;;
+        --run-as-ctas-queries)
+            RUN_AS_CTAS_QUERIES=true
+            shift
+            ;;
         *)
             echo "Error: Unknown argument $1"
             print_help
@@ -76,13 +81,21 @@ if [[ -z ${PRESTO_DATA_DIR} ]]; then
     exit 1
 fi
 
+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=()
+[[ "${RUN_AS_CTAS_QUERIES}" == "true" ]] && BENCHMARK_OUTPUT_ARGS+=(--run-as-ctas-queries)
+
 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..819c8cd3 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_CTAS_SCRATCH_DIR    Optional writable host scratch directory mounted for temporary CTAS benchmark results.
 
 EXAMPLES:
     $SCRIPT_NAME --no-cache
@@ -217,6 +218,16 @@ parse_args() {
 
 parse_args "$@"
 
+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_CTAS_SCRATCH_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..d4fa0539 100644
--- a/presto/slurm/presto-nvl72/README.md
+++ b/presto/slurm/presto-nvl72/README.md
@@ -102,11 +102,31 @@ 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
+
+# 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, GDS toggle, profiling, metrics, …).
+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`:
+
+```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).
@@ -152,6 +172,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-<SF>/`) |
 | `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..ef3a68c5 100644
--- a/presto/slurm/presto-nvl72/cluster_config.env.example
+++ b/presto/slurm/presto-nvl72/cluster_config.env.example
@@ -29,6 +29,16 @@ DATA=/path/to/tpch/data
 # Directory containing .sqsh container images.
 IMAGE_DIR=/path/to/images
 
+# 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 86808ddd..694ed7a3 100644
--- a/presto/slurm/presto-nvl72/functions.sh
+++ b/presto/slurm/presto-nvl72/functions.sh
@@ -2,6 +2,24 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+CTAS_CONTAINER_SCRATCH_DIR=/var/lib/presto/data/hive/benchmark_output
+EXPECTED_RESULTS_CONTAINER_DIR=/var/lib/presto/expected-results
+
+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
+}
+
+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 ",<host_path>:<host_path>" 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 +134,8 @@ function run_coord_image {
 
     local extra_mounts
     extra_mounts="$(miniforge_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}"
     fi
@@ -134,7 +154,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 +168,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 +312,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_scratch_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 +364,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},\
@@ -532,11 +553,20 @@ 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")
     [[ "${ENABLE_NSYS}" == "1" ]] && extra_args+=("-p" "--profile-script-path" "${container_script_dir}/profiler_functions.sh")
     [[ -n "${QUERIES:-}" ]] && extra_args+=("-q" "${QUERIES}")
+    [[ "${RUN_AS_CTAS_QUERIES:-0}" == "1" ]] && extra_args+=("--run-as-ctas-queries")
 
     source "${SCRIPT_DIR}/defaults.env"
 
@@ -562,15 +592,14 @@ 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; \
     export PORT=$PORT; \
     export HOSTNAME=$COORD; \
     export PRESTO_DATA_DIR=/var/lib/presto/data/hive/data/user_data; \
+    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; \
     cd /workspace/presto/scripts; \
diff --git a/presto/slurm/presto-nvl72/launch-run.sh b/presto/slurm/presto-nvl72/launch-run.sh
index 92807312..64f375a8 100755
--- a/presto/slurm/presto-nvl72/launch-run.sh
+++ b/presto/slurm/presto-nvl72/launch-run.sh
@@ -14,7 +14,7 @@
 #                  [-i|--iterations <n>] [--cpu] [-g|--num-workers-per-node <n>]
 #                  [-w|--worker-image <name>] [-c|--coord-image <name>]
 #                  [-o|--output-path <dir>] [-q|--queries <filter>]
-#                  [--disable-gds] [-m|--metrics] [-p|--profile]
+#                  [--run-as-ctas-queries] [--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
+RUN_AS_CTAS_QUERIES=0
 NSYS_WORKER_ID=0
 QUERIES=""
 
@@ -62,6 +63,7 @@ Options:
   -c, --coord-image <name>     Override coordinator image from cluster config
   -o, --output-path <dir>      Copy results into this directory after the run
   -q, --queries <list>         Comma-separated query filter (e.g. "1,6,21")
+      --run-as-ctas-queries     Run queries as distributed CTAS writes. Requires PRESTO_CTAS_SCRATCH_DIR.
       --worker-env-file <path> Override worker.env (default: ./worker.env)
       --cpu                    Use CPU partition/images (overrides cluster default)
       --gpu                    Use GPU partition/images (overrides cluster default)
@@ -75,8 +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, and per-variant defaults. 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
 }
 
@@ -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 ;;
+        --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 ;;
@@ -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 [[ "${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
+
 # 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
@@ -144,6 +157,12 @@ preflight_image "${COORD_IMAGE}" \
     "Pull it (see ./pull_ghcr_image.sh) or override with -c <name>"
 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)
@@ -163,6 +182,13 @@ 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+=",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}"
+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_benchmark_fixtures_test.py b/presto/testing/performance_benchmark_fixtures_test.py
new file mode 100644
index 00000000..9f297100
--- /dev/null
+++ b/presto/testing/performance_benchmark_fixtures_test.py
@@ -0,0 +1,94 @@
+# 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 (
+    CTAS_SCHEMA,
+    _drop_results_schema,
+    build_ctas_query,
+    ctas_output_column_aliases,
+    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 == "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"
+    assert build_ctas_query("tpch", "Q7", "SELECT 1;", "hive_output", "results", "q7") == (
+        "--tpch_Q7--\nCREATE TABLE hive_output.results.q7 (c1) WITH (format = 'PARQUET') AS\nSELECT 1"
+    )
+
+
+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"
+
+    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_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_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())
+
+    aliases = {
+        query_id: ctas_output_column_aliases(query.replace("{SF_FRACTION}", "0.0001"))
+        for query_id, query in queries.items()
+    }
+
+    assert aliases.keys() == queries.keys()
+    if benchmark_type == "tpch":
+        assert aliases["Q18"] == ["c1", "c2", "c3", "c4", "c5", "c6"]
+
+
+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_benchmarks/common_fixtures.py b/presto/testing/performance_benchmarks/common_fixtures.py
index f1a01a12..e19dc5ab 100644
--- a/presto/testing/performance_benchmarks/common_fixtures.py
+++ b/presto/testing/performance_benchmarks/common_fixtures.py
@@ -1,20 +1,119 @@
 # SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
 # SPDX-License-Identifier: Apache-2.0
 
+import os
+import shutil
+from contextlib import suppress
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 
 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
 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"
+# 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)
+class CtasResults:
+    catalog: str
+    schema: str
+
+
+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 ctas_output_column_aliases(query):
+    """Return unique CTAS aliases without modifying the original SELECT.
+
+    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}{column_aliases} WITH (format = 'PARQUET') AS\n"
+        f"{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("--run-as-ctas-queries"):
+        return None
+
+    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)
+
+    schema = CTAS_SCHEMA
+    host_results_dir = Path(scratch_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 +170,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 +199,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,22 +207,44 @@ 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"])
-
-                # Save query results to Parquet (only on first iteration)
-                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)
+                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(
+                        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]
+
+                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()
+                    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:
@@ -135,12 +257,28 @@ 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:
+            # 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(
+                        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..383e9aa1 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("--run-as-ctas-queries", action="store_true", default=False)
 
 
 def pytest_configure(config):