diff --git a/benchmarks/src/bin/benchmark_runner.rs b/benchmarks/src/bin/benchmark_runner.rs
index c7a16086c967..377ea6beff96 100644
--- a/benchmarks/src/bin/benchmark_runner.rs
+++ b/benchmarks/src/bin/benchmark_runner.rs
@@ -28,7 +28,9 @@ use datafusion_benchmarks::sql_benchmark_runner::{
     load_benchmark_definitions, make_ctx, prepare_benchmark,
     run_criterion_benchmarks_impl, sort_benchmarks,
 };
-use datafusion_benchmarks::util::{BenchmarkRun, CommonOpt, print_memory_stats};
+use datafusion_benchmarks::util::{
+    BenchmarkRun, CommonOpt, print_memory_stats, take_iteration_peak,
+};
 use datafusion_common::instant::Instant;
 use datafusion_common::{DataFusionError, exec_datafusion_err};
 use datafusion_common_runtime::SpawnedTask;
@@ -312,7 +314,8 @@ async fn run_simple_benchmark(
 
         println!("{case_name} iteration {iteration}: {ms:.1} ms, {row_count} rows");
 
-        run.write_iter(elapsed, row_count);
+        let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+        run.write_iter(elapsed, row_count, pool_peak_bytes);
     }
 
     print_memory_stats(&*ctx.runtime_env().memory_pool);
diff --git a/benchmarks/src/bin/external_aggr.rs b/benchmarks/src/bin/external_aggr.rs
index 226a619192ac..d7a7f7081fb5 100644
--- a/benchmarks/src/bin/external_aggr.rs
+++ b/benchmarks/src/bin/external_aggr.rs
@@ -40,7 +40,7 @@ use datafusion::physical_plan::display::DisplayableExecutionPlan;
 use datafusion::physical_plan::{collect, displayable};
 use datafusion::prelude::*;
 use datafusion_benchmarks::util::{
-    BenchmarkRun, CommonOpt, PeakRecordingPool, QueryResult,
+    BenchmarkRun, CommonOpt, PeakRecordingPool, QueryResult, take_iteration_peak,
 };
 use datafusion_common::instant::Instant;
 use datafusion_common::utils::get_available_parallelism;
@@ -174,7 +174,11 @@ impl ExternalAggrConfig {
                 .benchmark_query(query_id, mem_limit, mem_pool_type, &mut benchmark_run)
                 .await?;
             for iter in query_results {
-                benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                benchmark_run.write_iter(
+                    iter.elapsed,
+                    iter.row_count,
+                    iter.pool_peak_bytes,
+                );
             }
         }
 
@@ -242,7 +246,12 @@ impl ExternalAggrConfig {
             println!(
                 "{query_name} iteration {i} took {ms:.1} ms and returned {row_count} rows"
             );
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+            query_results.push(QueryResult {
+                elapsed,
+                row_count,
+                pool_peak_bytes,
+            });
         }
 
         let avg = millis.iter().sum::<f64>() / millis.len() as f64;
diff --git a/benchmarks/src/cancellation.rs b/benchmarks/src/cancellation.rs
index d3da1b0e8362..20ec2bbe14b1 100644
--- a/benchmarks/src/cancellation.rs
+++ b/benchmarks/src/cancellation.rs
@@ -106,7 +106,8 @@ impl RunOpt {
             let elapsed = run_test(self.wait_time, Arc::clone(&store))?;
             let ms = elapsed.as_secs_f64() * 1000.0;
             println!("Iteration {i} cancelled in {ms} ms");
-            rundata.write_iter(elapsed, 0);
+            // No memory limit here, so no pool to read.
+            rundata.write_iter(elapsed, 0, None);
         }
 
         rundata.maybe_write_json(self.output_path.as_ref())?;
diff --git a/benchmarks/src/clickbench.rs b/benchmarks/src/clickbench.rs
index a2e65aa5618a..f836dd9bdbf1 100644
--- a/benchmarks/src/clickbench.rs
+++ b/benchmarks/src/clickbench.rs
@@ -19,7 +19,9 @@ use std::fs;
 use std::io::ErrorKind;
 use std::path::{Path, PathBuf};
 
-use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
+use crate::util::{
+    BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
+};
 use clap::Args;
 use datafusion::logical_expr::{ExplainFormat, ExplainOption};
 use datafusion::{
@@ -230,7 +232,11 @@ impl RunOpt {
             match query_run {
                 Ok(query_results) => {
                     for iter in query_results {
-                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                        benchmark_run.write_iter(
+                            iter.elapsed,
+                            iter.row_count,
+                            iter.pool_peak_bytes,
+                        );
                     }
                 }
                 Err(e) => {
@@ -264,7 +270,12 @@ impl RunOpt {
             println!(
                 "Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
             );
-            query_results.push(QueryResult { elapsed, row_count })
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+            query_results.push(QueryResult {
+                elapsed,
+                row_count,
+                pool_peak_bytes,
+            })
         }
         if self.common.debug {
             ctx.sql(sql)
diff --git a/benchmarks/src/dict.rs b/benchmarks/src/dict.rs
index e04b5f816adc..d79fb5f3cc6b 100644
--- a/benchmarks/src/dict.rs
+++ b/benchmarks/src/dict.rs
@@ -17,7 +17,7 @@
 
 use crate::util::BenchmarkRun;
 use crate::util::CommonOpt;
-use crate::util::QueryResult;
+use crate::util::{QueryResult, take_iteration_peak};
 use arrow::array::{ArrayRef, DictionaryArray, Int32Array, ListArray, StringArray};
 use arrow::buffer::OffsetBuffer;
 use arrow::datatypes::{DataType, Field, Int32Type, Schema};
@@ -343,7 +343,11 @@ impl RunOpt {
             match query_run {
                 Ok(query_results) => {
                     for iter in query_results {
-                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                        benchmark_run.write_iter(
+                            iter.elapsed,
+                            iter.row_count,
+                            iter.pool_peak_bytes,
+                        );
                     }
                 }
                 Err(e) => {
@@ -378,7 +382,12 @@ impl RunOpt {
                 "Query '{}' iteration {i} returned {row_count} rows in {elapsed:?}",
                 query.name
             );
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+            query_results.push(QueryResult {
+                elapsed,
+                row_count,
+                pool_peak_bytes,
+            });
         }
 
         Ok(query_results)
diff --git a/benchmarks/src/h2o.rs b/benchmarks/src/h2o.rs
index feb4bf2fa11c..f70a1eb4eef8 100644
--- a/benchmarks/src/h2o.rs
+++ b/benchmarks/src/h2o.rs
@@ -20,7 +20,7 @@
 //! - [H2O AI Benchmark](https://duckdb.org/2023/04/14/h2oai.html)
 //! - [Extended window function benchmark](https://duckdb.org/2024/06/26/benchmarks-over-time.html#window-functions-benchmark)
 
-use crate::util::{BenchmarkRun, CommonOpt, print_memory_stats};
+use crate::util::{BenchmarkRun, CommonOpt, print_memory_stats, take_iteration_peak};
 use clap::Args;
 use datafusion::logical_expr::{ExplainFormat, ExplainOption};
 use datafusion::{error::Result, prelude::SessionContext};
@@ -126,7 +126,8 @@ impl RunOpt {
                 println!(
                     "Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
                 );
-                benchmark_run.write_iter(elapsed, row_count);
+                let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+                benchmark_run.write_iter(elapsed, row_count, pool_peak_bytes);
             }
             let avg = millis.iter().sum::<f64>() / millis.len() as f64;
             println!("Query {query_id} avg time: {avg:.2} ms");
diff --git a/benchmarks/src/hj.rs b/benchmarks/src/hj.rs
index 4f97b24d0f02..aca60a222ad8 100644
--- a/benchmarks/src/hj.rs
+++ b/benchmarks/src/hj.rs
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::util::{BenchmarkRun, CommonOpt, QueryResult};
+use crate::util::{BenchmarkRun, CommonOpt, QueryResult, take_iteration_peak};
 use clap::Args;
 use datafusion::physical_plan::execute_stream;
 use datafusion::{error::Result, prelude::SessionContext};
@@ -600,7 +600,11 @@ impl RunOpt {
             match query_run {
                 Ok(query_results) => {
                     for iter in query_results {
-                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                        benchmark_run.write_iter(
+                            iter.elapsed,
+                            iter.row_count,
+                            iter.pool_peak_bytes,
+                        );
                     }
                 }
                 Err(e) => {
@@ -646,7 +650,15 @@ impl RunOpt {
                 "Query {query_name} iteration {i} returned {row_count} rows in {elapsed:?}"
             );
 
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+
+            query_results.push(QueryResult {
+                elapsed,
+
+                row_count,
+
+                pool_peak_bytes,
+            });
         }
 
         Ok(query_results)
diff --git a/benchmarks/src/imdb/run.rs b/benchmarks/src/imdb/run.rs
index 5822bbcb0d89..829246f819bb 100644
--- a/benchmarks/src/imdb/run.rs
+++ b/benchmarks/src/imdb/run.rs
@@ -22,7 +22,9 @@ use super::{
     IMDB_QUERY_END_ID, IMDB_QUERY_START_ID, IMDB_TABLES, get_imdb_table_schema,
     get_query_sql,
 };
-use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
+use crate::util::{
+    BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
+};
 
 use arrow::record_batch::RecordBatch;
 use arrow::util::pretty::{self, pretty_format_batches};
@@ -297,7 +299,11 @@ impl RunOpt {
             benchmark_run.start_new_case(&format!("Query {query_id}"));
             let query_run = self.benchmark_query(query_id, &mut benchmark_run).await?;
             for iter in query_run {
-                benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                benchmark_run.write_iter(
+                    iter.elapsed,
+                    iter.row_count,
+                    iter.pool_peak_bytes,
+                );
             }
         }
         benchmark_run.maybe_write_json(self.output_path.as_ref())?;
@@ -348,7 +354,12 @@ impl RunOpt {
             println!(
                 "Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
             );
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+            query_results.push(QueryResult {
+                elapsed,
+                row_count,
+                pool_peak_bytes,
+            });
         }
 
         let avg = millis.iter().sum::<f64>() / millis.len() as f64;
diff --git a/benchmarks/src/nlj.rs b/benchmarks/src/nlj.rs
index 485ee069d1bb..17a4a4b8382b 100644
--- a/benchmarks/src/nlj.rs
+++ b/benchmarks/src/nlj.rs
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::util::{BenchmarkRun, CommonOpt, QueryResult};
+use crate::util::{BenchmarkRun, CommonOpt, QueryResult, take_iteration_peak};
 use clap::Args;
 use datafusion::physical_plan::execute_stream;
 use datafusion::{error::Result, prelude::SessionContext};
@@ -221,7 +221,11 @@ impl RunOpt {
             match query_run {
                 Ok(query_results) => {
                     for iter in query_results {
-                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                        benchmark_run.write_iter(
+                            iter.elapsed,
+                            iter.row_count,
+                            iter.pool_peak_bytes,
+                        );
                     }
                 }
                 Err(e) => {
@@ -272,7 +276,15 @@ impl RunOpt {
                 "Query {query_name} iteration {i} returned {row_count} rows in {elapsed:?}"
             );
 
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+
+            query_results.push(QueryResult {
+                elapsed,
+
+                row_count,
+
+                pool_peak_bytes,
+            });
         }
 
         Ok(query_results)
diff --git a/benchmarks/src/smj.rs b/benchmarks/src/smj.rs
index 9282f72c2fab..9f1f03c0d40e 100644
--- a/benchmarks/src/smj.rs
+++ b/benchmarks/src/smj.rs
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::util::{BenchmarkRun, CommonOpt, QueryResult};
+use crate::util::{BenchmarkRun, CommonOpt, QueryResult, take_iteration_peak};
 use clap::Args;
 use datafusion::physical_plan::execute_stream;
 use datafusion::{error::Result, prelude::SessionContext};
@@ -563,7 +563,11 @@ impl RunOpt {
             match query_run {
                 Ok(query_results) => {
                     for iter in query_results {
-                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                        benchmark_run.write_iter(
+                            iter.elapsed,
+                            iter.row_count,
+                            iter.pool_peak_bytes,
+                        );
                     }
                 }
                 Err(e) => {
@@ -616,7 +620,15 @@ impl RunOpt {
                 "Query {query_name} iteration {i} returned {row_count} rows in {elapsed:?}"
             );
 
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+
+            query_results.push(QueryResult {
+                elapsed,
+
+                row_count,
+
+                pool_peak_bytes,
+            });
         }
 
         Ok(query_results)
diff --git a/benchmarks/src/sort_pushdown.rs b/benchmarks/src/sort_pushdown.rs
index 77f889e702e3..c8ecf4cd8d64 100644
--- a/benchmarks/src/sort_pushdown.rs
+++ b/benchmarks/src/sort_pushdown.rs
@@ -54,7 +54,9 @@ use datafusion::prelude::*;
 use datafusion_common::DEFAULT_PARQUET_EXTENSION;
 use datafusion_common::instant::Instant;
 
-use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
+use crate::util::{
+    BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
+};
 
 /// Default path to query files, relative to the benchmark root
 const SORT_PUSHDOWN_QUERY_DIR: &str = "queries/sort_pushdown";
@@ -141,7 +143,11 @@ impl RunOpt {
             match query_results {
                 Ok(query_results) => {
                     for iter in query_results {
-                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                        benchmark_run.write_iter(
+                            iter.elapsed,
+                            iter.row_count,
+                            iter.pool_peak_bytes,
+                        );
                     }
                 }
                 Err(e) => {
@@ -192,7 +198,12 @@ impl RunOpt {
             println!(
                 "Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
             );
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+            query_results.push(QueryResult {
+                elapsed,
+                row_count,
+                pool_peak_bytes,
+            });
         }
 
         let avg = millis.iter().sum::<f64>() / millis.len() as f64;
diff --git a/benchmarks/src/sort_tpch.rs b/benchmarks/src/sort_tpch.rs
index d5f81c04a3ba..a5e74806926a 100644
--- a/benchmarks/src/sort_tpch.rs
+++ b/benchmarks/src/sort_tpch.rs
@@ -40,7 +40,9 @@ use datafusion_common::DEFAULT_PARQUET_EXTENSION;
 use datafusion_common::instant::Instant;
 use datafusion_common::utils::get_available_parallelism;
 
-use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
+use crate::util::{
+    BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
+};
 
 #[derive(Debug, Args)]
 pub struct RunOpt {
@@ -191,7 +193,11 @@ impl RunOpt {
             match query_results {
                 Ok(query_results) => {
                     for iter in query_results {
-                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                        benchmark_run.write_iter(
+                            iter.elapsed,
+                            iter.row_count,
+                            iter.pool_peak_bytes,
+                        );
                     }
                 }
                 Err(e) => {
@@ -251,7 +257,12 @@ impl RunOpt {
             println!(
                 "Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
             );
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+            query_results.push(QueryResult {
+                elapsed,
+                row_count,
+                pool_peak_bytes,
+            });
         }
 
         let avg = millis.iter().sum::<f64>() / millis.len() as f64;
diff --git a/benchmarks/src/tpcds/run.rs b/benchmarks/src/tpcds/run.rs
index 3eaaf172c0f1..9f6d1f3a90ca 100644
--- a/benchmarks/src/tpcds/run.rs
+++ b/benchmarks/src/tpcds/run.rs
@@ -19,7 +19,9 @@ use std::fs;
 use std::path::PathBuf;
 use std::sync::Arc;
 
-use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
+use crate::util::{
+    BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
+};
 
 use arrow::datatypes::Schema;
 use arrow::record_batch::RecordBatch;
@@ -236,7 +238,11 @@ impl RunOpt {
             match query_run {
                 Ok(query_results) => {
                     for iter in query_results {
-                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                        benchmark_run.write_iter(
+                            iter.elapsed,
+                            iter.row_count,
+                            iter.pool_peak_bytes,
+                        );
                     }
                 }
                 Err(e) => {
@@ -284,7 +290,12 @@ impl RunOpt {
             println!(
                 "Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
             );
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+            query_results.push(QueryResult {
+                elapsed,
+                row_count,
+                pool_peak_bytes,
+            });
         }
 
         let avg = millis.iter().sum::<f64>() / millis.len() as f64;
diff --git a/benchmarks/src/tpch/run.rs b/benchmarks/src/tpch/run.rs
index 47edfbac4b5a..6f406cc54710 100644
--- a/benchmarks/src/tpch/run.rs
+++ b/benchmarks/src/tpch/run.rs
@@ -22,7 +22,9 @@ use super::{
     TPCH_QUERY_END_ID, TPCH_QUERY_START_ID, TPCH_TABLES, get_query_sql_for_scale_factor,
     get_tbl_tpch_table_schema, get_tpch_table_schema, table_constraints,
 };
-use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
+use crate::util::{
+    BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
+};
 
 use arrow::record_batch::RecordBatch;
 use arrow::util::pretty::{self, pretty_format_batches};
@@ -148,7 +150,11 @@ impl RunOpt {
             match query_run {
                 Ok(query_results) => {
                     for iter in query_results {
-                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
+                        benchmark_run.write_iter(
+                            iter.elapsed,
+                            iter.row_count,
+                            iter.pool_peak_bytes,
+                        );
                     }
                 }
                 Err(e) => {
@@ -202,7 +208,12 @@ impl RunOpt {
             println!(
                 "Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
             );
-            query_results.push(QueryResult { elapsed, row_count });
+            let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
+            query_results.push(QueryResult {
+                elapsed,
+                row_count,
+                pool_peak_bytes,
+            });
         }
 
         let avg = millis.iter().sum::<f64>() / millis.len() as f64;
diff --git a/benchmarks/src/util/mod.rs b/benchmarks/src/util/mod.rs
index 43855ea468ef..6968c2f0bc8d 100644
--- a/benchmarks/src/util/mod.rs
+++ b/benchmarks/src/util/mod.rs
@@ -25,4 +25,4 @@ mod run;
 pub use memory::print_memory_stats;
 pub use memory_pool::PeakRecordingPool;
 pub use options::CommonOpt;
-pub use run::{BenchQuery, BenchmarkRun, QueryResult};
+pub use run::{BenchQuery, BenchmarkRun, QueryResult, take_iteration_peak};
diff --git a/benchmarks/src/util/run.rs b/benchmarks/src/util/run.rs
index 6c63ceec6423..b211a92d00b4 100644
--- a/benchmarks/src/util/run.rs
+++ b/benchmarks/src/util/run.rs
@@ -98,6 +98,11 @@ pub struct BenchQuery {
     /// bytes. Recorded for failed queries too, since a query that ran out of
     /// memory is one whose peak is worth seeing.
     ///
+    /// Each iteration is measured on its own and the smallest reading wins, so
+    /// this is the peak of the least-contended pass rather than of the whole
+    /// run — see [`BenchmarkRun::write_iter`]. A failed query reports what it
+    /// held when it gave up instead, having completed no iteration.
+    ///
     /// `None` (and omitted from the JSON) only when the benchmark ran without a
     /// memory limit, since there is then no pool to record.
     ///
@@ -109,6 +114,30 @@ pub struct BenchQuery {
 pub struct QueryResult {
     pub elapsed: Duration,
     pub row_count: usize,
+    /// Peak [`MemoryPool`] reservation for *this* iteration alone, from
+    /// [`take_iteration_peak`]. `None` when the benchmark ran without a
+    /// recording pool.
+    ///
+    /// [`MemoryPool`]: datafusion::execution::memory_pool::MemoryPool
+    pub pool_peak_bytes: Option<usize>,
+}
+
+/// Read the peak reservation accumulated since the last call and start a fresh
+/// reading, so each iteration is measured on its own.
+///
+/// Benchmarks run every iteration of a query before reporting any of them, so
+/// the reading has to be taken here, at the end of an iteration, rather than
+/// where the results are later written out — by then the recorder holds the
+/// high-water mark of the whole set.
+///
+/// Returns `None` when the benchmark is running without a memory limit, since
+/// there is then no recording pool in front of the runtime's.
+pub fn take_iteration_peak(pool: &Arc<dyn MemoryPool>) -> Option<usize> {
+    PeakRecordingPool::from_pool(pool.as_ref()).map(|recorder| {
+        let peak = recorder.peak_reserved();
+        recorder.reset_peak();
+        peak
+    })
 }
 /// collects benchmark run data and then serializes it at the end
 pub struct BenchmarkRun {
@@ -179,16 +208,32 @@ impl BenchmarkRun {
             self.current_case = Some(0);
         }
     }
-    /// Write a new iteration to the current case
-    pub fn write_iter(&mut self, elapsed: Duration, row_count: usize) {
-        // The peak is not reset between iterations, so this ends up holding the
-        // largest reservation seen across all of them.
-        let pool_peak_bytes = self.peak_recorder().map(PeakRecordingPool::peak_reserved);
+    /// Write a new iteration to the current case.
+    ///
+    /// `pool_peak_bytes` is this iteration's own reading, from
+    /// [`take_iteration_peak`]. What reaches the results JSON is the
+    /// *smallest* across the query's iterations: a peak reservation is a max
+    /// over the operators holding memory at one moment, so it moves with how
+    /// the runtime happened to interleave partitions on that pass. Reducing
+    /// with `min` keeps the reading reproducible, where taking the largest —
+    /// as this did while the recorder ran unreset across every iteration —
+    /// lets one unlucky pass set the number for the whole query.
+    pub fn write_iter(
+        &mut self,
+        elapsed: Duration,
+        row_count: usize,
+        pool_peak_bytes: Option<usize>,
+    ) {
         if let Some(idx) = self.current_case {
             self.queries[idx]
                 .iterations
                 .push(QueryIter { elapsed, row_count });
-            self.queries[idx].pool_peak_bytes = pool_peak_bytes;
+            self.queries[idx].pool_peak_bytes =
+                match (self.queries[idx].pool_peak_bytes, pool_peak_bytes) {
+                    (Some(seen), Some(peak)) => Some(seen.min(peak)),
+                    (None, peak) => peak,
+                    (seen, None) => seen,
+                };
         } else {
             panic!("no cases existed yet");
         }
@@ -249,6 +294,16 @@ mod tests {
         ))))
     }
 
+    /// One iteration of a query: reserve, release, then take the reading —
+    /// the order a benchmark's iteration loop runs in.
+    fn run_iteration(run: &mut BenchmarkRun, pool: &Arc<dyn MemoryPool>, grow: usize) {
+        let reservation = MemoryConsumer::new("q").register(pool);
+        reservation.try_grow(grow).unwrap();
+        drop(reservation);
+        let peak = take_iteration_peak(pool);
+        run.write_iter(Duration::from_millis(1), 1, peak);
+    }
+
     #[test]
     fn each_case_reports_its_own_peak() {
         let pool = recording_pool(1024);
@@ -256,21 +311,69 @@ mod tests {
         run.set_memory_pool(&pool);
 
         run.start_new_case("q1");
-        let reservation = MemoryConsumer::new("q1").register(&pool);
-        reservation.try_grow(600).unwrap();
-        run.write_iter(Duration::from_millis(1), 1);
-        drop(reservation);
+        run_iteration(&mut run, &pool, 600);
 
         // The second case must not inherit the first case's high-water mark.
         run.start_new_case("q2");
-        let reservation = MemoryConsumer::new("q2").register(&pool);
-        reservation.try_grow(100).unwrap();
-        run.write_iter(Duration::from_millis(1), 1);
+        run_iteration(&mut run, &pool, 100);
 
         assert_eq!(run.queries[0].pool_peak_bytes, Some(600));
         assert_eq!(run.queries[1].pool_peak_bytes, Some(100));
     }
 
+    #[test]
+    fn the_quietest_iteration_sets_the_peak() {
+        let pool = recording_pool(1024);
+        let mut run = BenchmarkRun::new();
+        run.set_memory_pool(&pool);
+
+        run.start_new_case("q1");
+        // Three passes over the same query. The 900 is what a pass looks like
+        // when more operators happened to hold memory at the same moment.
+        for grow in [300, 900, 400] {
+            run_iteration(&mut run, &pool, grow);
+        }
+
+        // Not 900: one unlucky pass must not set the number for the query.
+        assert_eq!(run.queries[0].pool_peak_bytes, Some(300));
+    }
+
+    #[test]
+    fn an_iteration_does_not_inherit_the_previous_peak() {
+        let pool = recording_pool(1024);
+        let mut run = BenchmarkRun::new();
+        run.set_memory_pool(&pool);
+
+        run.start_new_case("q1");
+        run_iteration(&mut run, &pool, 800);
+        // Before per-iteration readings the recorder still held 800 here, so
+        // this pass reported 800 rather than its own 100.
+        run_iteration(&mut run, &pool, 100);
+
+        assert_eq!(run.queries[0].pool_peak_bytes, Some(100));
+    }
+
+    #[test]
+    fn an_iteration_reading_covers_only_that_iteration() {
+        let pool = recording_pool(1024);
+        let setup = MemoryConsumer::new("setup").register(&pool);
+        setup.try_grow(700).unwrap();
+
+        // Bytes held when a reading is taken stay in it: `reset_peak` rebases
+        // on what is reserved, not on zero, so data the benchmark loaded up
+        // front counts for every iteration that runs with it.
+        assert_eq!(take_iteration_peak(&pool), Some(700));
+        assert_eq!(take_iteration_peak(&pool), Some(700));
+
+        // Releasing it does not pull the current reading down — a high-water
+        // mark only falls at a reset, so the interval that began with those
+        // bytes held still reports them...
+        drop(setup);
+        assert_eq!(take_iteration_peak(&pool), Some(700));
+        // ...and only the interval after that is clear of them.
+        assert_eq!(take_iteration_peak(&pool), Some(0));
+    }
+
     #[test]
     fn a_later_pool_replaces_an_earlier_one() {
         let first = recording_pool(1024);
@@ -290,7 +393,8 @@ mod tests {
             .register(&second)
             .try_grow(100)
             .unwrap();
-        run.write_iter(Duration::from_millis(1), 1);
+        let peak = take_iteration_peak(&second);
+        run.write_iter(Duration::from_millis(1), 1, peak);
 
         assert_eq!(run.queries[0].pool_peak_bytes, Some(100));
     }
@@ -314,7 +418,7 @@ mod tests {
     fn the_peak_is_omitted_without_a_recording_pool() {
         let mut run = BenchmarkRun::new();
         run.start_new_case("q1");
-        run.write_iter(Duration::from_millis(1), 1);
+        run.write_iter(Duration::from_millis(1), 1, None);
 
         assert_eq!(run.queries[0].pool_peak_bytes, None);
         assert!(!run.to_json().contains("pool_peak_bytes"));