@@ -78,10 +78,13 @@ cargo run --bin analyze_trace -- /path/to/comet-event-trace.json
 
 The tool reads counter events from the trace log. Because tracing logs metrics per thread, `native_allocated`
 and `jemalloc_allocated` are process-wide values (the same global allocation reported from whichever thread
-logs it), while `thread_NNN_comet_memory_reserved` values are per-thread pool reservations that are summed to
-get the total tracked memory. The tool analyzes `native_allocated` when the trace contains it, since that
-counts only what Rust code holds from the allocator, and otherwise falls back to `jemalloc_allocated`. The
-output names the counter it used. A trace with neither counter is rejected.
+logs it). The total tracked memory comes from `comet_memory_reserved_total`, which counts each pool once
+process-wide. The per-thread `thread_NNN_comet_memory_reserved` values must not be summed to obtain it: a
+task-shared pool reports its full reservation on every thread that references it, so summing multiplies it by
+the thread count. Traces recorded before that counter existed are still analyzed from the per-thread sum, and
+the tool warns that the total over-counts. The tool analyzes `native_allocated` when the trace contains it,
+since that counts only what Rust code holds from the allocator, and otherwise falls back to
+`jemalloc_allocated`. The output names the counter it used. A trace with neither counter is rejected.
 
 Sample output:
 
@@ -121,10 +124,11 @@ not being tracked by the pool.
 
 ## Definition of Labels
 
-| Label                            | Meaning                                                                                                                                                              |
-| -------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| jvm_heap_used                    | JVM heap memory usage of live objects for the executor process                                                                                                       |
-| jemalloc_allocated               | Native memory usage for the executor process (requires `jemalloc` feature)                                                                                           |
-| native_allocated                 | Bytes handed out by the Rust global allocator, process-wide (requires `alloc-accounting` feature). Approximate to within 64 KiB of un-flushed delta per live thread. |
-| thread_NNN_comet_memory_reserved | Memory reserved by Comet's DataFusion memory pool (summed across all contexts on the thread). NNN is the Rust thread ID.                                             |
-| thread_NNN_comet_jvm_shuffle     | Off-heap memory allocated by Comet for columnar shuffle. NNN is the Rust thread ID.                                                                                  |
+| Label                            | Meaning                                                                                                                                                                                                                                       |
+| -------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| jvm_heap_used                    | JVM heap memory usage of live objects for the executor process                                                                                                                                                                                |
+| jemalloc_allocated               | Native memory usage for the executor process (requires `jemalloc` feature)                                                                                                                                                                    |
+| native_allocated                 | Bytes handed out by the Rust global allocator, process-wide (requires `alloc-accounting` feature). Approximate to within 64 KiB of un-flushed delta per live thread.                                                                          |
+| comet_memory_reserved_total      | Total memory reserved across every live Comet memory pool, process-wide. Counts each task-shared pool once, so unlike the per-thread counters it can be compared directly against an allocation counter.                                      |
+| thread_NNN_comet_memory_reserved | Memory reserved by Comet's DataFusion memory pool (summed across all contexts on the thread). NNN is the Rust thread ID. Do not sum these across threads: a task-shared pool reports its full reservation on every thread that references it. |
+| thread_NNN_comet_jvm_shuffle     | Off-heap memory allocated by Comet for columnar shuffle. NNN is the Rust thread ID.                                                                                                                                                           |
@@ -37,6 +37,14 @@ use std::{env, fs::File};
 /// analyzed.
 const ALLOCATED_COUNTERS: [&str; 2] = ["native_allocated", "jemalloc_allocated"];
 
+/// The process-wide total of Comet's memory pool reservations.
+///
+/// Preferred over summing the per-thread `thread_NNN_comet_memory_reserved` counters. Those report
+/// the full reservation of a task-shared pool once per thread that references it, so adding them
+/// across threads multiplies a shared pool by its thread count. A trace without this counter is
+/// still analyzed from the per-thread sum, with a warning, so older traces remain readable.
+const POOL_TOTAL_COUNTER: &str = "comet_memory_reserved_total";
+
 /// A single Chrome trace event (only the fields we care about).
 #[derive(Deserialize)]
 struct TraceEvent {
@@ -75,8 +83,11 @@ fn main() {
     let mut source: Option<usize> = None;
     // Latest allocated value (global, not per-thread)
     let mut latest_allocated: u64 = 0;
-    // Per-thread pool reservations: thread_NNN -> bytes
+    // Per-thread pool reservations: thread_NNN -> bytes. Only used when the trace lacks the
+    // process-wide total, since summing these over-counts task-shared pools.
     let mut pool_by_thread: HashMap<String, u64> = HashMap::new();
+    // Process-wide pool total, when the trace carries it.
+    let mut pool_total_counter: Option<u64> = None;
     // Points where allocated exceeded pool total
     let mut violations: Vec<MemorySnapshot> = Vec::new();
     // Track peak values
@@ -146,6 +157,12 @@ fn main() {
                     peak_allocated = latest_allocated;
                 }
             }
+        } else if event.name == POOL_TOTAL_COUNTER {
+            // Must be matched before the per-thread branch below, whose `contains` check would
+            // otherwise also match this name and fold the process-wide total into the map.
+            if let Some(val) = event.args.get(&event.name) {
+                pool_total_counter = Some(val.as_u64().unwrap_or(0));
+            }
         } else if event.name.contains("comet_memory_reserved") {
             // Name format: thread_NNN_comet_memory_reserved
             let thread_key = event.name.clone();
@@ -161,12 +178,13 @@ fn main() {
         // After each allocated or pool update, check the current state. A comparison needs one
         // sample of each side: an observed zero reservation is a real value that allocation can
         // exceed, so only the absence of any pool sample defers the check.
-        let pool_total: u64 = pool_by_thread.values().sum();
+        let pool_total: u64 = pool_total_counter.unwrap_or_else(|| pool_by_thread.values().sum());
         if pool_total > peak_pool_total {
             peak_pool_total = pool_total;
         }
 
-        if source.is_some() && !pool_by_thread.is_empty() && latest_allocated > pool_total {
+        let have_pool_sample = pool_total_counter.is_some() || !pool_by_thread.is_empty();
+        if source.is_some() && have_pool_sample && latest_allocated > pool_total {
             let excess = latest_allocated - pool_total;
             if excess > peak_excess {
                 peak_excess = excess;
@@ -198,7 +216,19 @@ fn main() {
     println!("=== Comet Trace Memory Analysis ===\n");
     println!("Counter events parsed: {counter_events}");
     println!("Allocation counter:    {source}");
-    println!("Threads with memory pools: {}", pool_by_thread.len());
+    if pool_total_counter.is_some() {
+        println!("Pool total source:     {POOL_TOTAL_COUNTER} (process-wide)");
+    } else {
+        println!(
+            "Pool total source:     sum of {} per-thread counters",
+            pool_by_thread.len()
+        );
+        println!(
+            "WARNING: this trace predates {POOL_TOTAL_COUNTER}. Summing the per-thread counters\n\
+             over-counts: a task-shared pool reports its full reservation on every thread that\n\
+             references it, so the totals below are upper bounds and the excess is understated."
+        );
+    }
     println!("Peak {source}:   {}", format_bytes(peak_allocated));
     println!(
         "Peak pool total:           {}",
@@ -210,7 +240,7 @@ fn main() {
     );
     println!();
 
-    if pool_by_thread.is_empty() {
+    if pool_total_counter.is_none() && pool_by_thread.is_empty() {
         println!(
             "No pool reservation samples in the trace, so there is nothing to compare against."
         );
@@ -245,5 +275,8 @@ fn main() {
     for (thread, bytes) in &threads {
         println!("  {thread}: {}", format_bytes(**bytes));
     }
-    println!("\n  Total: {}", format_bytes(pool_by_thread.values().sum()));
+    println!(
+        "\n  Total: {}",
+        format_bytes(pool_total_counter.unwrap_or_else(|| pool_by_thread.values().sum()))
+    );
 }
@@ -1107,6 +1107,13 @@ pub unsafe extern "system" fn Java_org_apache_comet_Native_executePlan(
                 &exec_context.tracing_memory_metric_name,
                 total_reserved_for_thread(exec_context.rust_thread_id) as u64,
             );
+            // Process-wide total, emitted here so it is true at the same instant as the
+            // allocation counter above. The per-thread counter cannot be summed across threads:
+            // it reports a shared pool's full reservation once per referencing thread.
+            log_memory_usage(
+                "comet_memory_reserved_total",
+                crate::execution::memory_pools::total_reserved_across_tasks() as u64,
+            );
         }
 
         result
 
@@ -94,6 +94,22 @@ impl Drop for TaskSharedMemoryPool {
     }
 }
 
+/// Total bytes reserved across every live task-shared pool, process-wide.
+///
+/// Each registry entry is one task attempt's pool, so summing the entries counts each pool exactly
+/// once. This is deliberately not the sum of the per-thread `thread_NNN_comet_memory_reserved`
+/// tracing counters: each of those reports the full reservation of a shared pool once per thread
+/// that references it, so adding them across threads multiplies a task-shared pool by its thread
+/// count.
+pub(crate) fn total_reserved_across_tasks() -> usize {
+    TASK_SHARED_MEMORY_POOLS
+        .lock()
+        .values()
+        .filter_map(Weak::upgrade)
+        .map(|pool| pool.reserved())
+        .sum()
+}
+
 /// Returns the memory pool shared by every native plan in `task_attempt_id`, creating it with
 /// `create` if no live pool exists for the task. The returned `Arc` is the RAII handle: the pool
 /// stays registered until the last reference to it drops.
@@ -121,6 +137,7 @@ pub(crate) fn acquire_task_shared_pool(
 mod tests {
     use super::*;
     use datafusion::execution::memory_pool::UnboundedMemoryPool;
+    use std::sync::{Mutex, MutexGuard};
 
     /// Tests share the process-wide pool map, so each uses its own task attempt id.
     fn acquire(task_attempt_id: i64) -> Arc<dyn MemoryPool> {
@@ -133,6 +150,64 @@ mod tests {
             .contains_key(&task_attempt_id)
     }
 
+    /// `TASK_SHARED_MEMORY_POOLS` is process-wide and the crate's tests run in parallel, so any
+    /// test that reserves memory perturbs another's view of the total. The two tests below take
+    /// this lock so their deltas are exact; without it they observe each other's reservations.
+    static SERIAL: Mutex<()> = Mutex::new(());
+
+    fn serial() -> MutexGuard<'static, ()> {
+        SERIAL
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    /// The property the per-thread tracing counters get wrong: a task-shared pool contributes its
+    /// reservation once, however many execution contexts hold it.
+    #[test]
+    fn a_task_shared_pool_is_counted_once_however_many_contexts_hold_it() {
+        let _guard = serial();
+        let before = total_reserved_across_tasks();
+
+        let first = acquire(-1010);
+        let reservation = MemoryConsumer::new("counted").register(&first);
+        reservation.try_grow(4096).unwrap();
+        let with_one_context = total_reserved_across_tasks();
+        assert_eq!(
+            with_one_context - before,
+            4096,
+            "a live pool's reservation must appear in the total"
+        );
+
+        // A second execution context in the same task resolves to the same pool. The total must
+        // not double: summing the per-thread counters instead is exactly what over-counts.
+        let second = acquire(-1010);
+        assert!(Arc::ptr_eq(&first, &second));
+        assert_eq!(
+            total_reserved_across_tasks(),
+            with_one_context,
+            "a shared pool must be counted once, not once per holder"
+        );
+
+        drop(reservation);
+    }
+
+    #[test]
+    fn released_pools_leave_the_total() {
+        let _guard = serial();
+        let before = total_reserved_across_tasks();
+        {
+            let pool = acquire(-1011);
+            let reservation = MemoryConsumer::new("transient").register(&pool);
+            reservation.try_grow(8192).unwrap();
+            assert_eq!(total_reserved_across_tasks() - before, 8192);
+        }
+        assert_eq!(
+            total_reserved_across_tasks(),
+            before,
+            "dropping the last reference must remove the pool from the total"
+        );
+    }
+
     #[test]
     fn plans_in_the_same_task_share_one_pool() {
         let first = acquire(-1001);