diff --git a/pinot/benchmark.sh b/pinot/benchmark.sh
index e5db3195ea..f0515a7577 100755
--- a/pinot/benchmark.sh
+++ b/pinot/benchmark.sh
@@ -4,6 +4,29 @@ export BENCH_DOWNLOAD_SCRIPT="download-hits-tsv"
 # inside one JVM and takes longer than the lib's 300 s default to be
 # query-ready on a cold instance. 900 s clears the observed cold start.
 export BENCH_CHECK_TIMEOUT=900
+# Pinot QuickStart's -dataDir persistence is broken in 1.5.0/1.5.1: the
+# second QuickStart invocation against an existing -dataDir always dies
+# with IllegalStateException from Quickstart.java's
+# Preconditions.checkState(quickstartRunnerDir.mkdirs()) — mkdirs() returns
+# false because the directory from the first run already exists (verified
+# by reproducing this locally in Docker and reading the Pinot source).
+# A stop/start cycle therefore always wipes the in-memory table/segment
+# registration and there is no supported way to bring the daemon back up
+# against its previous state directory.
+#
+# What *does* survive the cycle is the segment tar files QuickStart wrote
+# to a plain local directory (batch/hits/segments/, not process memory —
+# verified locally: kill -9 the JVM, restart it, the .tar.gz files are
+# still on disk). BENCH_DURABLE=no makes the shared driver re-run ./load
+# after every restart; our ./load exploits this by re-pushing those
+# existing tars straight to the controller's segment-upload API
+# (https://docs.pinot.apache.org/build-with-pinot/ingestion/batch-ingestion/segment-upload)
+# instead of re-parsing hits.tsv, so the "reload" on cold tries 2-129
+# (43 queries x 3 tries) takes seconds instead of the ~78 min a real
+# from-source reload would need. This keeps the real stop/start/
+# drop_caches cycle other systems get, rather than skipping it via
+# BENCH_RESTARTABLE=no.
+export BENCH_DURABLE=no
 # Skip the pre-snapshot ./stop+./start cycle: the loaded
 # state lives only in the daemon's process memory (in-process
 # DataFrame, JVM heap caches) and stopping wipes it. The
diff --git a/pinot/check b/pinot/check
index 3bfe104c3f..ac6b50fc09 100755
--- a/pinot/check
+++ b/pinot/check
@@ -1,10 +1,28 @@
 #!/bin/bash
 set -e
 
-# Pinot is responsive once both the controller and the broker accept queries.
-RES=$(curl -sf -o /dev/null -w '%{http_code}' \
+# The QuickStart process only registers its clean-shutdown hook after
+# ./start's launched process finishes starting every component AND
+# bootstrapping every sample table (including waiting for each table's
+# async minion segment-generation-and-push task to finish). If ./stop
+# sends SIGTERM any earlier than that, either shutdown-hook registration
+# itself throws (component startup still in flight) or an in-flight
+# minion task gets killed (bootstrap still in flight) -- either way the
+# process dies without a clean stop and corrupts the next ./start.
+#
+# Broker readiness alone is not a safe proxy for "fully done starting":
+# it comes up long before bootstrap finishes. The only reliable signal is
+# the "Quick start setup complete" line QuickStart itself prints right
+# after runner.bootstrapTable() + waitForBootstrapToComplete() return
+# (see pinot-tools Quickstart.java, method execute()). Since pinot.log is
+# appended across restarts, only look at what was appended after the
+# offset ./start recorded for this invocation.
+BROKER_RES=$(curl -sf -o /dev/null -w '%{http_code}' \
     -XPOST -H'Content-Type: application/json' \
     http://localhost:8000/query/sql/ \
-    -d '{"sql":"SELECT 1"}')
+    -d '{"sql":"SELECT 1"}' 2>/dev/null || echo "000")
 
-[ "$RES" = "200" ]
+OFFSET=$(cat pinot.log.offset 2>/dev/null || echo 0)
+SETUP_DONE=$(tail -c +"$((OFFSET + 1))" pinot.log 2>/dev/null | grep -c 'Quick start setup complete' || true)
+
+[ "$BROKER_RES" = "200" ] && [ "${SETUP_DONE:-0}" -ge 1 ]
diff --git a/pinot/load b/pinot/load
index 7229254546..da2c1b3a22 100755
--- a/pinot/load
+++ b/pinot/load
@@ -3,6 +3,7 @@ set -e
 
 PINOT_VERSION=1.5.1
 PINOT_DIR="apache-pinot-$PINOT_VERSION-bin"
+SEGMENTS_DIR="batch/hits/segments"
 
 # Wait for the controller's REST API to be live. `start` only waits for
 # the broker; the controller can lag, and AddTable below talks to the
@@ -32,15 +33,68 @@ for _ in $(seq 1 60); do
     sleep 1
 done
 
-# Register the schema and table. Previously trailed `|| true`, which
-# masked genuine failures (controller not up, bad JSON, etc.) and let
-# LaunchDataIngestionJob run against a missing table — the symptom
-# was a misleading "Failed to fetch hits/schema" deep in the ingestion
-# job. Fail loudly here.
-"./${PINOT_DIR}/bin/pinot-admin.sh" AddTable \
+# Register the schema and table. AddTable fails with "already exists" on
+# every call after the first (BENCH_DURABLE=no re-runs ./load on every
+# cold cycle; QuickStart's own restart wipes the table registration but
+# not the segment tar files on disk — see the fast path below). Treat
+# that specific failure as success; anything else should still fail loudly
+# as before.
+add_table_out=$("./${PINOT_DIR}/bin/pinot-admin.sh" AddTable \
     -tableConfigFile offline_table.json \
-    -schemaFile schema.json -exec
+    -schemaFile schema.json -exec 2>&1) && add_table_rc=0 || add_table_rc=$?
+if [ "$add_table_rc" -ne 0 ]; then
+    if ! printf '%s\n' "$add_table_out" | grep -q 'already exists'; then
+        printf '%s\n' "$add_table_out" >&2
+        exit "$add_table_rc"
+    fi
+fi
+
+# Fast path: segment tar files from a previous ./load already exist on
+# disk. QuickStart -type batch writes them under $SEGMENTS_DIR (a plain
+# local directory, not process memory) and a stop/start cycle wipes only
+# the in-memory table/segment registration, not these files (verified by
+# reproducing this locally in Docker: kill -9 the QuickStart JVM, restart
+# it, curl POST each existing .tar.gz straight to the controller's
+# /segments endpoint — official "tar push" upload API, see
+# https://docs.pinot.apache.org/build-with-pinot/ingestion/batch-ingestion/segment-upload
+# — and the table is fully queryable again, star-tree index included,
+# without re-parsing hits.tsv). Re-pushing ~100 existing tars takes
+# seconds; re-running LaunchDataIngestionJob from hits.tsv takes ~78 min.
+if [ -d "$SEGMENTS_DIR" ] && ls "$SEGMENTS_DIR"/*.tar.gz >/dev/null 2>&1; then
+    for f in "$SEGMENTS_DIR"/*.tar.gz; do
+        curl -sS -X POST -H 'Content-Type: multipart/form-data' \
+            -F "file=@${f}" \
+            'http://localhost:9000/segments?tableName=hits&tableType=OFFLINE' \
+            | grep -q 'Successfully uploaded' || {
+                echo "bench: failed to re-push ${f}" >&2
+                exit 1
+            }
+    done
+
+    # Re-pushed segments take a few seconds to show up in the external
+    # view (Helix propagation), during which queries return a transient
+    # "N segments unavailable" error. Poll until every expected segment
+    # is ONLINE instead of racing the very first query against it
+    # (observed locally: up to ~20s for 5 segments; sized generously
+    # for ~100).
+    expected=$(ls "$SEGMENTS_DIR"/*.tar.gz | wc -l | tr -d ' ')
+    for _ in $(seq 1 120); do
+        online=$(curl -sf http://localhost:9000/tables/hits/externalview 2>/dev/null \
+            | python3 -c 'import json,sys
+d=json.load(sys.stdin)
+ov=d.get("OFFLINE") or {}
+print(sum(1 for st in ov.values() if list(st.values()) == ["ONLINE"]))' \
+            2>/dev/null || echo 0)
+        if [ "$online" -ge "$expected" ]; then
+            break
+        fi
+        sleep 1
+    done
 
+    exit 0
+fi
+
+# Slow path (first ./load only): generate segments from hits.tsv.
 # Pinot was unable to load data as a single file without errors. Split.
 split -d --additional-suffix .tsv -n l/100 hits.tsv parts
 
@@ -51,7 +105,30 @@ sed parts93.tsv -e 's/"tatuirovarki_redmond/tatuirovarki_redmond/g' -i
 sed splitted.yaml -e "s|PWD_DIR_PLACEHOLDER|$PWD|g" -i
 sed local.yaml    -e "s|PWD_DIR_PLACEHOLDER|$PWD|g" -i
 
+# pinot-admin.sh's own default is -Xms4G with NO -Xmx (falls back to the
+# JVM's default of ~25% of physical RAM). On a real ~74GB/100-segment
+# ClickBench hits.tsv this OOMed mid-job around segment 60/100 on a
+# 30GB-RAM c6a.4xlarge (OutOfMemoryError in SegmentDictionaryCreator) —
+# and LaunchDataIngestionJob still exited 0, silently leaving only 60/100
+# segments on disk. Force a real heap ceiling sized for this instance.
+export JAVA_OPTS="-Xms4G -Xmx20G"
+
 "./${PINOT_DIR}/bin/pinot-admin.sh" LaunchDataIngestionJob -jobSpecFile splitted.yaml
 
+# Defense against the exact failure this instance just hit: Pinot's own
+# ingestion job runner can swallow a per-file OutOfMemoryError (or other
+# per-task exception) and still exit 0, leaving fewer segment tars on
+# disk than input splits. Fail loudly instead of letting a partial load
+# silently pass bench_load's >5GB data-size check downstream.
+actual_segments=$(ls "$SEGMENTS_DIR"/*.tar.gz 2>/dev/null | wc -l | tr -d ' ')
+expected_segments=$(ls parts*.tsv 2>/dev/null | wc -l | tr -d ' ')
+if [ "$expected_segments" -gt 0 ] && [ "$actual_segments" -ne "$expected_segments" ]; then
+    echo "bench: LaunchDataIngestionJob produced ${actual_segments}/${expected_segments} segments — partial load (likely OOM swallowed per-task)" >&2
+    exit 1
+fi
+
+# Keep hits.tsv's derived parts only as long as needed; the source file
+# itself and the split parts are no longer needed once segments exist in
+# $SEGMENTS_DIR (used by the fast path above on every later cold cycle).
 rm -f hits.tsv parts*.tsv
 sync
diff --git a/pinot/queries.sql b/pinot/queries.sql
index 31f65fc898..47114dee7f 100644
--- a/pinot/queries.sql
+++ b/pinot/queries.sql
@@ -16,7 +16,7 @@ SELECT SearchEngineID, SearchPhrase, COUNT(*) AS c FROM hits WHERE SearchPhrase
 SELECT UserID, COUNT(*) FROM hits GROUP BY UserID ORDER BY COUNT(*) DESC LIMIT 10;
 SELECT UserID, SearchPhrase, COUNT(*) FROM hits GROUP BY UserID, SearchPhrase ORDER BY COUNT(*) DESC LIMIT 10;
 SELECT UserID, SearchPhrase, COUNT(*) FROM hits GROUP BY UserID, SearchPhrase LIMIT 10;
-SELECT UserID, extract(minute FROM EventTime) AS m, SearchPhrase, COUNT(*) FROM hits GROUP BY UserID, m, SearchPhrase ORDER BY COUNT(*) DESC LIMIT 10;
+SELECT UserID, extract(minute FROM CAST(EventTime AS TIMESTAMP)) AS m, SearchPhrase, COUNT(*) FROM hits GROUP BY UserID, m, SearchPhrase ORDER BY COUNT(*) DESC LIMIT 10;
 SELECT UserID FROM hits WHERE UserID = 435090932899640449;
 SELECT COUNT(*) FROM hits WHERE URL LIKE '%google%';
 SELECT SearchPhrase, MIN(URL), COUNT(*) AS c FROM hits WHERE URL LIKE '%google%' AND SearchPhrase <> '' GROUP BY SearchPhrase ORDER BY c DESC LIMIT 10;
@@ -40,4 +40,4 @@ SELECT URL, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >
 SELECT TraficSourceID, SearchEngineID, AdvEngineID, CASE WHEN (SearchEngineID = 0 AND AdvEngineID = 0) THEN Referer ELSE '' END AS Src, URL AS Dst, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND IsRefresh = 0 GROUP BY TraficSourceID, SearchEngineID, AdvEngineID, Src, Dst ORDER BY PageViews DESC LIMIT 10 OFFSET 1000;
 SELECT URLHash, EventDate, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND IsRefresh = 0 AND TraficSourceID IN (-1, 6) AND RefererHash = 3594120000172545465 GROUP BY URLHash, EventDate ORDER BY PageViews DESC LIMIT 10 OFFSET 100;
 SELECT WindowClientWidth, WindowClientHeight, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-01' AND EventDate <= '2013-07-31' AND IsRefresh = 0 AND DontCountHits = 0 AND URLHash = 2868770270353813622 GROUP BY WindowClientWidth, WindowClientHeight ORDER BY PageViews DESC LIMIT 10 OFFSET 10000;
-SELECT DATE_TRUNC('minute', EventTime) AS M, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-14' AND EventDate <= '2013-07-15' AND IsRefresh = 0 AND DontCountHits = 0 GROUP BY DATE_TRUNC('minute', EventTime) ORDER BY DATE_TRUNC('minute', EventTime) LIMIT 10 OFFSET 1000;
+SELECT DATE_TRUNC('minute', CAST(EventTime AS TIMESTAMP)) AS M, COUNT(*) AS PageViews FROM hits WHERE CounterID = 62 AND EventDate >= '2013-07-14' AND EventDate <= '2013-07-15' AND IsRefresh = 0 AND DontCountHits = 0 GROUP BY DATE_TRUNC('minute', CAST(EventTime AS TIMESTAMP)) ORDER BY DATE_TRUNC('minute', CAST(EventTime AS TIMESTAMP)) LIMIT 10 OFFSET 1000;
diff --git a/pinot/query b/pinot/query
index 337792437a..d8b068644a 100755
--- a/pinot/query
+++ b/pinot/query
@@ -12,7 +12,7 @@ query=$(printf '%s' "$query" | tr -d ';')
 req=$(printf '%s' "$query" | python3 -c '
 import json, sys
 q = sys.stdin.read()
-print(json.dumps({"sql": q + "  option(timeoutMs=300000)"}))
+print(json.dumps({"sql": q + "  option(timeoutMs=300000,autoRewriteAggregationType=true)"}))
 ')
 
 resp=$(curl -sS -XPOST -H'Content-Type: application/json' \
diff --git a/pinot/start b/pinot/start
index bd189e8077..b18733d796 100755
--- a/pinot/start
+++ b/pinot/start
@@ -4,7 +4,9 @@ set -e
 PINOT_VERSION=1.5.1
 PINOT_DIR="apache-pinot-$PINOT_VERSION-bin"
 
-# Idempotent: if broker query endpoint is up, do nothing.
+# Idempotent: if broker query endpoint is up, do nothing (leave the
+# existing pinot.log offset marker as-is; ./check will find the earlier
+# "setup complete" marker in the untouched log).
 if curl -sf -o /dev/null -w '%{http_code}' \
     -XPOST -H'Content-Type: application/json' \
     http://localhost:8000/query/sql/ \
@@ -12,6 +14,20 @@ if curl -sf -o /dev/null -w '%{http_code}' \
     exit 0
 fi
 
+# Record the current end of pinot.log so ./check can look only at what
+# this invocation appends, not a "setup complete" line left over from an
+# earlier run of the log-appending (>>) QuickStart process.
+touch pinot.log
+wc -c < pinot.log | tr -d ' ' > pinot.log.offset
+
+# pinot-admin.sh's own default is -Xms4G with NO -Xmx (falls back to the
+# JVM's default of ~25% of physical RAM). On a 100-segment/~74GB ClickBench
+# run this OOMed mid-LaunchDataIngestionJob around segment 60/100 on a
+# 30GB-RAM c6a.4xlarge (observed: OutOfMemoryError in
+# SegmentDictionaryCreator, job silently reported success anyway).
+# Force a real heap ceiling sized for this instance.
+export JAVA_OPTS="-Xms4G -Xmx20G"
+
 nohup "./${PINOT_DIR}/bin/pinot-admin.sh" QuickStart -type batch \
     >> pinot.log 2>&1 < /dev/null &
 disown
diff --git a/pinot/stop b/pinot/stop
index 140a589693..cbfab09ec6 100755
--- a/pinot/stop
+++ b/pinot/stop
@@ -3,5 +3,26 @@
 pkill -f 'pinot-admin' 2>/dev/null || true
 pkill -f 'pinot.tools.admin' 2>/dev/null || true
 pkill -f 'org.apache.pinot' 2>/dev/null || true
-sleep 2
+
+# pkill only sends SIGTERM; the actual teardown (JVM shutdown hook
+# stopping controller/broker/server, then the embedded Zookeeper
+# releasing its port) is asynchronous and takes several seconds. Two
+# things were previously observed to still be alive well after a fixed
+# short sleep or even after the broker itself stopped responding:
+#   - the broker itself (was seen answering 200 for ~7-8s after SIGTERM)
+#   - the embedded Zookeeper's port 2123, which a fixed sleep or a
+#     broker-only check does not account for; the next ./start's fresh
+#     Zookeeper can then fail to bind/connect on the still-held port,
+#     surfacing as "ZkTimeoutException: Unable to connect to zookeeper
+#     server within timeout" in the next start's controller.
+# The only reliable signal that teardown is actually finished is that
+# the pinot-admin JVM process itself is gone. Poll for that (up to 30s)
+# instead of guessing a delay or checking a single port.
+for _ in $(seq 1 30); do
+    if ! pgrep -f 'org.apache.pinot.tools.admin.PinotAdministrator' > /dev/null 2>&1; then
+        break
+    fi
+    sleep 1
+done
+
 exit 0