diff --git a/crates/adapters/src/static_compile/seroutput.rs b/crates/adapters/src/static_compile/seroutput.rs
index 052ad66ba82..2fca3090e54 100644
--- a/crates/adapters/src/static_compile/seroutput.rs
+++ b/crates/adapters/src/static_compile/seroutput.rs
@@ -541,7 +541,10 @@ where
     }
 
     fn into_trace(self: Arc<Self>) -> Box<dyn SerTrace> {
-        let mut spine = TypedBatch::new(DynSpine::<B::Inner>::new(&B::factories()));
+        let mut spine = TypedBatch::new(DynSpine::<B::Inner>::new(
+            &B::factories(),
+            Arc::new(String::from("SerTrace")),
+        ));
         TOKIO.block_on(spine.insert(Arc::unwrap_or_clone(self).batch.into_inner()));
         Box::new(SerBatchImpl::<Spine<B>, KD, VD>::new(spine))
     }
diff --git a/crates/dbsp/Cargo.toml b/crates/dbsp/Cargo.toml
index 74450c1f1bd..76785dab65d 100644
--- a/crates/dbsp/Cargo.toml
+++ b/crates/dbsp/Cargo.toml
@@ -46,11 +46,7 @@ arc-swap = { workspace = true }
 mimalloc-rust-sys = { workspace = true }
 rand = { workspace = true }
 rkyv = { workspace = true, features = ["std", "size_64", "validation", "uuid"] }
-size-of = { workspace = true, features = [
-    "hashbrown",
-    "xxhash-xxh3",
-    "arcstr"
-] }
+size-of = { workspace = true, features = [ "hashbrown", "xxhash-xxh3" ] }
 tokio-util = { workspace = true }
 futures = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread"] }
diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs
index 83b50eae080..d4dc2c6012c 100644
--- a/crates/dbsp/src/circuit/circuit_builder.rs
+++ b/crates/dbsp/src/circuit/circuit_builder.rs
@@ -1196,7 +1196,7 @@ impl NodeId {
         self.0
     }
 
-    pub(super) fn root() -> Self {
+    pub fn root() -> Self {
         Self(0)
     }
 }
diff --git a/crates/dbsp/src/circuit/operator_traits.rs b/crates/dbsp/src/circuit/operator_traits.rs
index da57e96ef9e..492b142a211 100644
--- a/crates/dbsp/src/circuit/operator_traits.rs
+++ b/crates/dbsp/src/circuit/operator_traits.rs
@@ -5,6 +5,7 @@
 
 #![allow(async_fn_in_trait)]
 
+use arc_swap::ArcSwap;
 use feldera_storage::{FileCommitter, StoragePath};
 
 use crate::Error;
@@ -16,6 +17,7 @@ use crate::{
     trace::cursor::Position,
 };
 use std::borrow::Cow;
+use std::fmt::Display;
 use std::sync::Arc;
 
 use super::GlobalNodeId;
@@ -628,3 +630,73 @@ pub trait ImportOperator<I, O>: Operator {
         OwnershipPreference::INDIFFERENT
     }
 }
+
+/// The name of an operator, primarily for use in CPU profiles.
+///
+/// An operator doesn't initially know its global node ID, but the global node
+/// ID is useful for debugging and profiling.  This type allows the name to
+/// initially omit the ID but adds it when it becomes available.
+pub struct OperatorName(ArcSwap<String>);
+
+impl OperatorName {
+    /// Creates a new `OperatorName` that doesn't initially know the global node
+    /// ID, since an operator doesn't know this until [Operator::init] is
+    /// called.
+    ///
+    /// If the name is fetched, with [OperatorName::get], before
+    /// [OperatorName::init], then it will just returned as simply `(base)`.
+    pub fn new(base: &str) -> Self {
+        Self(ArcSwap::new(Arc::new(format!("({base})"))))
+    }
+
+    /// Updates the name to include `global_id`.  This should normally be called
+    /// once, from the implementation of [Operator::init].  If it is called more
+    /// than once then later calls are ignored.
+    ///
+    /// After this function is called, the name will have the format `base
+    /// nodeid` where `nodeid` is in the format used by the circuit profiler so
+    /// that it's easy to find by searching.
+    pub fn init(&self, global_id: &GlobalNodeId) {
+        let s = self.0.load();
+        if let Some(rest) = s.strip_prefix('(')
+            && let Some(base) = rest.strip_suffix(')')
+        {
+            self.0
+                .store(Arc::new(format!("{base} {}", global_id.node_identifier())));
+        }
+    }
+
+    /// Returns a copy of the operator name string.
+    pub fn get(&self) -> Arc<String> {
+        self.0.load_full()
+    }
+}
+
+impl Display for OperatorName {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.0.load())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::circuit::{GlobalNodeId, NodeId, operator_traits::OperatorName};
+
+    #[test]
+    fn operator_name() {
+        let name = OperatorName::new("Name");
+        assert_eq!(name.to_string(), "(Name)");
+        name.init(&GlobalNodeId::from_path(&[
+            NodeId::new(1),
+            NodeId::new(2),
+            NodeId::new(3),
+        ]));
+        assert_eq!(name.to_string(), "Name nn1_n2_n3");
+        name.init(&GlobalNodeId::from_path(&[
+            NodeId::new(4),
+            NodeId::new(5),
+            NodeId::new(6),
+        ]));
+        assert_eq!(name.to_string(), "Name nn1_n2_n3");
+    }
+}
diff --git a/crates/dbsp/src/operator.rs b/crates/dbsp/src/operator.rs
index 3a15a33b6bb..dc7f013f15c 100644
--- a/crates/dbsp/src/operator.rs
+++ b/crates/dbsp/src/operator.rs
@@ -58,8 +58,9 @@ pub mod star_join_macros;
 pub mod time_series;
 mod trace;
 
+use std::fmt::Display;
+
 use crate::Error;
-use crate::circuit::GlobalNodeId;
 use crate::storage::backend::StorageError;
 
 pub use self::csv::CsvSource;
@@ -96,9 +97,6 @@ pub use transaction_z1::TransactionZ1;
 pub use z1::{DelayedFeedback, DelayedNestedFeedback, Z1, Z1Nested};
 
 /// Returns a `NoPersistentId` error if `persistent_id` is `None`.
-fn require_persistent_id<'a>(
-    persistent_id: Option<&'a str>,
-    global_id: &GlobalNodeId,
-) -> Result<&'a str, Error> {
-    persistent_id.ok_or_else(|| Error::Storage(StorageError::NoPersistentId(global_id.to_string())))
+fn require_persistent_id(persistent_id: Option<&str>, id: impl Display) -> Result<&str, Error> {
+    persistent_id.ok_or_else(|| Error::Storage(StorageError::NoPersistentId(id.to_string())))
 }
diff --git a/crates/dbsp/src/operator/communication/exchange.rs b/crates/dbsp/src/operator/communication/exchange.rs
index 5350b0fe05d..29af8d87170 100644
--- a/crates/dbsp/src/operator/communication/exchange.rs
+++ b/crates/dbsp/src/operator/communication/exchange.rs
@@ -14,7 +14,7 @@ use crate::{
             EXCHANGE_WAIT_TIME_SECONDS, INPUT_BATCHES_STATS, MetaItem, OUTPUT_BATCHES_STATS,
             OperatorLocation, OperatorMeta,
         },
-        operator_traits::{Operator, SinkOperator, SourceOperator},
+        operator_traits::{Operator, OperatorName, SinkOperator, SourceOperator},
         runtime::{WorkerLocation, WorkerLocations},
         tokio::TOKIO,
     },
@@ -40,7 +40,7 @@ use std::{
     ops::Range,
     pin::Pin,
     sync::{
-        Arc, Mutex, MutexGuard, OnceLock, RwLock,
+        Arc, Mutex, MutexGuard, RwLock,
         atomic::{AtomicIsize, AtomicPtr, AtomicU64, AtomicUsize, Ordering},
     },
     time::{Duration, Instant, SystemTime},
@@ -331,7 +331,7 @@ impl ExchangeClient {
 pub type ExchangeId = u32;
 
 pub trait ExchangeDelivery {
-    fn name(&self) -> &str;
+    fn name(&self) -> Arc<String>;
 
     fn received<'a>(
         &'a self,
@@ -618,7 +618,7 @@ pub(crate) struct Exchange<T> {
     exchange_id: ExchangeId,
 
     /// Identifies the `ExchangeReceiver` operator for use in profile data.
-    receiver_global_node_id: OnceLock<Arc<String>>,
+    name: OperatorName,
 
     /// The number of communicating peers.
     npeers: usize,
@@ -756,7 +756,7 @@ where
 
         let exchange = Arc::new(Self {
             exchange_id,
-            receiver_global_node_id: Default::default(),
+            name: OperatorName::new("ExchangeReceiver"),
             npeers,
             local_workers: layout.local_workers(),
             clients,
@@ -1026,8 +1026,8 @@ impl<T> ExchangeDelivery for Exchange<T>
 where
     T: Clone + Debug + Send + 'static,
 {
-    fn name(&self) -> &str {
-        self.receiver_global_node_id.get().unwrap()
+    fn name(&self) -> Arc<String> {
+        self.name.get()
     }
 
     fn received<'a>(
@@ -1431,10 +1431,7 @@ where
     }
 
     fn init(&mut self, global_id: &GlobalNodeId) {
-        let _ = self.exchange.receiver_global_node_id.set(Arc::new(format!(
-            "ExchangeReceiver {}",
-            global_id.node_identifier()
-        )));
+        self.exchange.name.init(global_id);
     }
 
     fn location(&self) -> OperatorLocation {
diff --git a/crates/dbsp/src/operator/dynamic/accumulate_trace.rs b/crates/dbsp/src/operator/dynamic/accumulate_trace.rs
index 0fef3ebbf65..f544dda92ff 100644
--- a/crates/dbsp/src/operator/dynamic/accumulate_trace.rs
+++ b/crates/dbsp/src/operator/dynamic/accumulate_trace.rs
@@ -1,6 +1,6 @@
 use crate::circuit::circuit_builder::{StreamId, register_replay_stream};
 use crate::circuit::metadata::{INPUT_RECORDS_COUNT, MEMORY_ALLOCATIONS_COUNT, RETAINMENT_BOUNDS};
-use crate::circuit::splitter_output_chunk_size;
+use crate::circuit::{NodeId, splitter_output_chunk_size};
 use crate::dynamic::{Factory, Weight, WeightTrait};
 use crate::operator::dynamic::trace::{DelayedTraceId, TraceBounds};
 use crate::operator::{TraceBound, require_persistent_id};
@@ -876,7 +876,8 @@ impl<T: Trace> ReplayState<T> {
 
 pub struct AccumulateZ1Trace<C: Circuit, B: Batch, T: Trace> {
     // For error reporting.
-    global_id: GlobalNodeId,
+    local_id: NodeId,
+    name: OperatorName,
     time: T::Time,
     trace: Option<T>,
     replay_state: Option<ReplayState<T>>,
@@ -909,7 +910,8 @@ where
         bounds: TraceBounds<T::Key, T::Val>,
     ) -> Self {
         Self {
-            global_id: GlobalNodeId::root(),
+            local_id: NodeId::root(),
+            name: OperatorName::new("AccumulateZ1Trace"),
             time: <T::Time as Timestamp>::clock_start(),
             trace: None,
             replay_state: None,
@@ -952,11 +954,8 @@ where
     /// replay, `replay_stream` is active while the operator that normally write to
     /// stream is disabled.
     pub fn prepare_replay_stream(&mut self, stream: &Stream<C, B>) -> Stream<C, B> {
-        let replay_stream = Stream::with_value(
-            stream.circuit().clone(),
-            self.global_id.local_node_id().unwrap(),
-            stream.value(),
-        );
+        let replay_stream =
+            Stream::with_value(stream.circuit().clone(), self.local_id, stream.value());
 
         self.delta_stream = Some(replay_stream.clone());
         replay_stream
@@ -977,7 +976,7 @@ where
         self.dirty[scope as usize] = false;
 
         if scope == 0 && self.trace.is_none() {
-            self.trace = Some(T::new(&self.trace_factories));
+            self.trace = Some(T::new(&self.trace_factories, self.name.get()));
         }
     }
 
@@ -992,7 +991,8 @@ where
     }
 
     fn init(&mut self, global_id: &GlobalNodeId) {
-        self.global_id = global_id.clone();
+        self.name.init(global_id);
+        self.local_id = global_id.local_node_id().unwrap();
     }
 
     fn metadata(&self, meta: &mut OperatorMeta) {
@@ -1032,7 +1032,7 @@ where
         pid: Option<&str>,
         files: &mut Vec<Arc<dyn FileCommitter>>,
     ) -> Result<(), Error> {
-        let pid = require_persistent_id(pid, &self.global_id)?;
+        let pid = require_persistent_id(pid, &self.name)?;
         self.trace
             .as_mut()
             .map(|trace| trace.save(base, pid, files))
@@ -1040,7 +1040,7 @@ where
     }
 
     fn restore(&mut self, base: &StoragePath, pid: Option<&str>) -> Result<(), Error> {
-        let pid = require_persistent_id(pid, &self.global_id)?;
+        let pid = require_persistent_id(pid, &self.name)?;
 
         self.trace
             .as_mut()
@@ -1050,7 +1050,7 @@ where
 
     fn clear_state(&mut self) -> Result<(), Error> {
         // println!("AccumulateZ1Trace-{}::clear_state", &self.global_id);
-        self.trace = Some(T::new(&self.trace_factories));
+        self.trace = Some(T::new(&self.trace_factories, self.name.get()));
         self.replay_state = None;
         self.dirty = vec![false; self.root_scope as usize + 1];
 
@@ -1071,7 +1071,7 @@ where
                 .trace
                 .take()
                 .expect("AccumulateZ1Trace::start_replay: no trace");
-            self.trace = Some(T::new(&self.trace_factories));
+            self.trace = Some(T::new(&self.trace_factories, self.name.get()));
 
             //println!("AccumulateZ1Trace-{}::initializing replay_state", &self.global_id);
 
@@ -1239,7 +1239,7 @@ where
     }
 }
 
-use crate::circuit::operator_traits::UnaryOperator;
+use crate::circuit::operator_traits::{OperatorName, UnaryOperator};
 
 pub struct AccumulateDelayTrace<B>
 where
diff --git a/crates/dbsp/src/operator/dynamic/accumulator.rs b/crates/dbsp/src/operator/dynamic/accumulator.rs
index 90fa0540c8b..1e387460726 100644
--- a/crates/dbsp/src/operator/dynamic/accumulator.rs
+++ b/crates/dbsp/src/operator/dynamic/accumulator.rs
@@ -13,14 +13,14 @@ use typedmap::TypedMapKey;
 use crate::{
     Circuit, Error, NumEntries, Runtime, Scope, Stream,
     circuit::{
-        LocalStoreMarker,
+        GlobalNodeId, LocalStoreMarker,
         circuit_builder::StreamId,
         metadata::{
             ALLOCATED_MEMORY_BYTES, BatchSizeStats, INPUT_BATCHES_STATS, MEMORY_ALLOCATIONS_COUNT,
             MetaItem, OUTPUT_BATCHES_STATS, OperatorLocation, OperatorMeta, SHARED_MEMORY_BYTES,
             STATE_RECORDS_COUNT, USED_MEMORY_BYTES,
         },
-        operator_traits::{Operator, UnaryOperator},
+        operator_traits::{Operator, OperatorName, UnaryOperator},
     },
     circuit_cache_key,
     trace::{Batch, BatchReader, Spine, Trace},
@@ -83,6 +83,7 @@ where
     B: Batch,
 {
     factories: B::Factories,
+    name: OperatorName,
     state: Spine<B>,
     flush: bool,
     location: &'static Location<'static>,
@@ -135,9 +136,11 @@ where
             }
         };
 
+        let name = OperatorName::new("Accumulator");
         Self {
             factories: factories.clone(),
-            state: Spine::new(factories),
+            state: Spine::new(factories, name.get()),
+            name,
             flush: false,
             location,
             input_batch_stats: BatchSizeStats::new(),
@@ -156,6 +159,11 @@ where
         Cow::Borrowed("Accumulator")
     }
 
+    fn init(&mut self, global_id: &GlobalNodeId) {
+        self.name.init(global_id);
+        self.state.set_name(self.name.get());
+    }
+
     fn location(&self) -> OperatorLocation {
         Some(self.location)
     }
@@ -193,7 +201,7 @@ where
 
     /// Clear the operator's state.
     fn clear_state(&mut self) -> Result<(), Error> {
-        self.state = Spine::new(&self.factories);
+        self.state = Spine::new(&self.factories, self.name.get());
         self.flush = false;
         Ok(())
     }
@@ -239,7 +247,7 @@ where
             self.flush = false;
             self.enabled_during_current_transaction = None;
 
-            let mut spine = Spine::<B>::new(&self.factories);
+            let mut spine = Spine::<B>::new(&self.factories, self.name.get());
             std::mem::swap(&mut self.state, &mut spine);
 
             self.output_batch_stats.add_batch(spine.len());
diff --git a/crates/dbsp/src/operator/dynamic/balance/rebalancing_accumulator.rs b/crates/dbsp/src/operator/dynamic/balance/rebalancing_accumulator.rs
index 1c1ae1aa87b..1e8ef0f8a32 100644
--- a/crates/dbsp/src/operator/dynamic/balance/rebalancing_accumulator.rs
+++ b/crates/dbsp/src/operator/dynamic/balance/rebalancing_accumulator.rs
@@ -5,6 +5,7 @@ use size_of::SizeOf;
 use crate::{
     Circuit, Error, NumEntries, Scope, Stream,
     circuit::{
+        GlobalNodeId,
         checkpointer::EmptyCheckpoint,
         circuit_builder::RefStreamValue,
         metadata::{
@@ -12,7 +13,7 @@ use crate::{
             MetaItem, OUTPUT_BATCHES_STATS, OperatorLocation, OperatorMeta, SHARED_MEMORY_BYTES,
             STATE_RECORDS_COUNT, USED_MEMORY_BYTES,
         },
-        operator_traits::{Operator, UnaryOperator},
+        operator_traits::{Operator, OperatorName, UnaryOperator},
     },
     trace::{Batch, BatchReader, Spine, Trace},
 };
@@ -51,6 +52,7 @@ where
     B: Batch,
 {
     factories: B::Factories,
+    name: OperatorName,
     state: Spine<B>,
     flush: bool,
     location: &'static Location<'static>,
@@ -69,9 +71,11 @@ where
     B: Batch,
 {
     pub fn new(factories: &B::Factories, location: &'static Location<'static>) -> Self {
+        let name = OperatorName::new("RebalancingAccumulator");
         Self {
             factories: factories.clone(),
-            state: Spine::new(factories),
+            state: Spine::new(factories, name.get()),
+            name,
             flush: false,
             location,
             input_batch_stats: BatchSizeStats::new(),
@@ -81,7 +85,7 @@ where
     }
 
     pub fn clear_state(&mut self) {
-        self.state = Spine::new(&self.factories);
+        self.state = Spine::new(&self.factories, self.name.get());
         self.flush = false;
     }
 
@@ -111,7 +115,14 @@ where
     B: Batch,
 {
     fn name(&self) -> std::borrow::Cow<'static, str> {
-        Cow::Borrowed("BalancingAccumulator")
+        Cow::Borrowed("Balancing Accumulator")
+    }
+
+    fn init(&mut self, global_id: &GlobalNodeId) {
+        let mut inner = self.0.borrow_mut();
+        inner.name.init(global_id);
+        let name = inner.name.get();
+        inner.state.set_name(name);
     }
 
     fn location(&self) -> OperatorLocation {
@@ -151,9 +162,10 @@ where
 
     /// Clear the operator's state.
     fn clear_state(&mut self) -> Result<(), Error> {
-        let state = Spine::new(&self.0.borrow().factories);
-        self.0.borrow_mut().state = state;
-        self.0.borrow_mut().flush = false;
+        let mut inner = self.0.borrow_mut();
+        let state = Spine::new(&inner.factories, inner.name.get());
+        inner.state = state;
+        inner.flush = false;
         Ok(())
     }
 
@@ -186,7 +198,7 @@ where
         let result = if inner.flush {
             inner.flush = false;
 
-            let mut spine = Spine::<B>::new(&inner.factories);
+            let mut spine = Spine::<B>::new(&inner.factories, inner.name.get());
             std::mem::swap(&mut inner.state, &mut spine);
 
             inner.output_batch_stats.add_batch(spine.len());
@@ -220,7 +232,7 @@ where
         let result = if inner.flush {
             inner.flush = false;
 
-            let mut spine = Spine::<B>::new(&inner.factories);
+            let mut spine = Spine::<B>::new(&inner.factories, inner.name.get());
             std::mem::swap(&mut inner.state, &mut spine);
 
             inner.output_batch_stats.add_batch(spine.len());
diff --git a/crates/dbsp/src/operator/dynamic/group/test.rs b/crates/dbsp/src/operator/dynamic/group/test.rs
index d64d7fabd63..0c4f9326c45 100644
--- a/crates/dbsp/src/operator/dynamic/group/test.rs
+++ b/crates/dbsp/src/operator/dynamic/group/test.rs
@@ -1,6 +1,6 @@
 #![allow(clippy::type_complexity)]
 
-use std::cmp::Ordering;
+use std::{cmp::Ordering, sync::Arc};
 
 use crate::{
     DBData, DynZWeight, RootCircuit, Runtime, ZWeight,
@@ -380,7 +380,7 @@ fn lead_test(trace: Vec<Vec<(i32, i32, ZWeight)>>, transaction: bool) {
     )
     .unwrap();
 
-    let mut ref_trace = TestBatch::new(&TestBatchFactories::new());
+    let mut ref_trace = TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
     if transaction {
         dbsp.start_transaction().unwrap();
@@ -435,7 +435,7 @@ fn lag_test(trace: Vec<Vec<(i32, i32, ZWeight)>>, transaction: bool) {
     )
     .unwrap();
 
-    let mut ref_trace = TestBatch::new(&TestBatchFactories::new());
+    let mut ref_trace = TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
     if transaction {
         dbsp.start_transaction().unwrap();
@@ -821,7 +821,7 @@ fn test_topk(trace: Vec<Vec<(i32, i32, ZWeight)>>, transaction: bool) {
     )
     .unwrap();
 
-    let mut ref_trace = TestBatch::new(&TestBatchFactories::new());
+    let mut ref_trace = TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
     if transaction {
         dbsp.start_transaction().unwrap();
diff --git a/crates/dbsp/src/operator/dynamic/join.rs b/crates/dbsp/src/operator/dynamic/join.rs
index 25dec2cf238..017fd01530e 100644
--- a/crates/dbsp/src/operator/dynamic/join.rs
+++ b/crates/dbsp/src/operator/dynamic/join.rs
@@ -4,6 +4,7 @@ use crate::circuit::metadata::{
     BatchSizeStats, COMPUTED_OUTPUT_RECORDS_COUNT, LEFT_INPUT_RECORDS_COUNT, OUTPUT_BATCHES_STATS,
     OUTPUT_REDUNDANCY_PERCENT, RIGHT_INPUT_INTEGRAL_RECORDS_COUNT,
 };
+use crate::circuit::operator_traits::OperatorName;
 use crate::circuit::splitter_output_chunk_size;
 use crate::dynamic::DynData;
 use crate::operator::async_stream_operators::{StreamingBinaryOperator, StreamingBinaryWrapper};
@@ -1262,6 +1263,7 @@ where
     T: ZBatchReader,
     Z: IndexedZSet,
 {
+    name: OperatorName,
     right_factories: T::Factories,
     output_factories: Z::Factories,
     timed_item_factory:
@@ -1322,6 +1324,7 @@ where
         clock: Clk,
     ) -> Self {
         Self {
+            name: OperatorName::new("JoinTrace"),
             right_factories: right_factories.clone(),
             output_factories: output_factories.clone(),
             timed_item_factory,
@@ -1353,6 +1356,10 @@ where
         Cow::Borrowed("JoinTrace")
     }
 
+    fn init(&mut self, global_id: &crate::circuit::GlobalNodeId) {
+        self.name.init(global_id);
+    }
+
     fn location(&self) -> OperatorLocation {
         Some(self.location)
     }
@@ -1685,7 +1692,7 @@ where
                     start += run_length;
 
                     if let Entry::Vacant(vacant) = self.future_outputs.borrow_mut().entry(batch_time) {
-                        let mut spine = <Spine<Z> as Trace>::new(&self.output_factories);
+                        let mut spine = <Spine<Z> as Trace>::new(&self.output_factories, self.name.get());
                         spine.insert(Z::dyn_from_tuples(&self.output_factories, (), &mut batch)).await;
                         vacant.insert(spine);
                     };
diff --git a/crates/dbsp/src/operator/dynamic/multijoin/match_keys.rs b/crates/dbsp/src/operator/dynamic/multijoin/match_keys.rs
index ce2414876de..3ad79e60d59 100644
--- a/crates/dbsp/src/operator/dynamic/multijoin/match_keys.rs
+++ b/crates/dbsp/src/operator/dynamic/multijoin/match_keys.rs
@@ -240,9 +240,7 @@ where
     }
 
     pub fn build(self) -> Match<C, I, O, F> {
-        let id = self.global_id.local_node_id().unwrap();
         Match {
-            global_id: self.global_id,
             labels: BTreeMap::new(),
             prefix: None,
             prefix_stream: self.prefix_stream,
@@ -252,11 +250,12 @@ where
             flush: Cell::new(false),
             async_stream: None,
             inner: Rc::new(MatchInternal::new(
-                id,
+                &self.global_id,
                 self.factories,
                 self.join_func,
                 self.circuit,
             )),
+            global_id: self.global_id,
         }
     }
 }
@@ -424,9 +423,8 @@ where
     O: IndexedZSet,
     F: MatchFunc<C, I, O::Key, O::Val>,
 {
-    // Used in debug prints.
-    #[allow(dead_code)]
-    id: NodeId,
+    /// For profiling and debug prints.
+    name: Arc<String>,
     key_factory: &'static dyn Factory<I::Key>,
     output_factories: O::Factories,
     timed_item_factory:
@@ -455,9 +453,14 @@ where
     O: IndexedZSet,
     F: MatchFunc<C, I, O::Key, O::Val>,
 {
-    fn new(id: NodeId, factories: MatchFactories<I, C::Time, O>, join_func: F, circuit: C) -> Self {
+    fn new(
+        global_id: &GlobalNodeId,
+        factories: MatchFactories<I, C::Time, O>,
+        join_func: F,
+        circuit: C,
+    ) -> Self {
         Self {
-            id,
+            name: Arc::new(format!("Match {}", global_id.node_identifier())),
             key_factory: factories.prefix_factories.key_factory(),
             output_factories: factories.output_factories,
             timed_item_factory: factories.timed_item_factory,
@@ -468,7 +471,7 @@ where
             future_outputs: RefCell::new(HashMap::new()),
             stats: RefCell::new(MatchStats::new()),
             circuit: circuit.clone(),
-            output_stream: Stream::new(circuit, id),
+            output_stream: Stream::new(circuit, global_id.local_node_id().unwrap()),
             phantom: PhantomData,
         }
     }
@@ -556,7 +559,7 @@ where
                     start += run_length;
 
                     if let Entry::Vacant(vacant) = self.future_outputs.borrow_mut().entry(batch_time) {
-                        let mut spine = <Spine<O> as Trace>::new(&self.output_factories);
+                        let mut spine = <Spine<O> as Trace>::new(&self.output_factories, self.name.clone());
                         spine.insert(O::dyn_from_tuples(&self.output_factories, (), &mut batch)).await;
                         vacant.insert(spine);
                     }
diff --git a/crates/dbsp/src/operator/dynamic/neighborhood.rs b/crates/dbsp/src/operator/dynamic/neighborhood.rs
index ebbcb9e28c3..94be16d53d2 100644
--- a/crates/dbsp/src/operator/dynamic/neighborhood.rs
+++ b/crates/dbsp/src/operator/dynamic/neighborhood.rs
@@ -598,7 +598,10 @@ mod test {
     use anyhow::Result as AnyResult;
     use feldera_storage::tokio::TOKIO;
     use proptest::{collection::vec, prelude::*};
-    use std::cmp::{max, min};
+    use std::{
+        cmp::{max, min},
+        sync::Arc,
+    };
 
     impl TestBatch<DynData, DynData, (), DynZWeight> {
         fn neighborhood<K, V>(
@@ -652,7 +655,7 @@ mod test {
 
                 TestBatch::from_data(output.as_slice())
             } else {
-                TestBatch::new(&TestBatchFactories::new())
+                TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")))
             }
         }
     }
@@ -914,7 +917,7 @@ mod test {
             let (mut dbsp, (descr_handle, input_handle, output_handle)) =
                 Runtime::init_circuit(4, test_circuit).unwrap();
 
-            let mut ref_trace = TestBatch::new(&TestBatchFactories::new());
+            let mut ref_trace = TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
             for (batch, (start_key, start_val), before, after) in trace.into_iter() {
 
diff --git a/crates/dbsp/src/operator/dynamic/sample.rs b/crates/dbsp/src/operator/dynamic/sample.rs
index c5cdbe81f4b..ed5e94726fb 100644
--- a/crates/dbsp/src/operator/dynamic/sample.rs
+++ b/crates/dbsp/src/operator/dynamic/sample.rs
@@ -349,7 +349,7 @@ mod test {
     use anyhow::Result as AnyResult;
     use feldera_storage::tokio::TOKIO;
     use proptest::{collection::vec, prelude::*};
-    use std::{cmp::Ordering, collections::BTreeSet};
+    use std::{cmp::Ordering, collections::BTreeSet, sync::Arc};
 
     fn test_circuit(
         circuit: &mut RootCircuit,
@@ -454,7 +454,7 @@ mod test {
             let (mut dbsp, (sample_size_handle, input_handle, output_sample_handle, output_quantile_handle)) =
                 Runtime::init_circuit(4, test_circuit).unwrap();
 
-            let mut ref_trace: TestBatch<DynData/*<i32>*/, DynData/*<i32>*/, (), DynZWeight> = TestBatch::new(&TestBatchFactories::new());
+            let mut ref_trace: TestBatch<DynData/*<i32>*/, DynData/*<i32>*/, (), DynZWeight> = TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
             for (batch, sample_size) in trace.into_iter() {
 
diff --git a/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs b/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs
index ec7b5006ed7..e000e8762e5 100644
--- a/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs
+++ b/crates/dbsp/src/operator/dynamic/sharded_accumulator.rs
@@ -5,7 +5,7 @@ use std::{
     ops::Range,
     panic::Location,
     pin::Pin,
-    sync::{Arc, Mutex, MutexGuard, OnceLock},
+    sync::{Arc, Mutex, MutexGuard},
 };
 
 use feldera_samply::Span;
@@ -23,7 +23,7 @@ use crate::{
             MetaItem, OUTPUT_BATCHES_STATS, OperatorLocation, OperatorMeta, SHARED_MEMORY_BYTES,
             SPINE_COUNT, STATE_RECORDS_COUNT, USED_MEMORY_BYTES,
         },
-        operator_traits::{Operator, SinkOperator, SourceOperator},
+        operator_traits::{Operator, OperatorName, SinkOperator, SourceOperator},
     },
     circuit_cache_key,
     operator::{
@@ -97,7 +97,7 @@ struct ShardedAccumulator<B>
 where
     B: Batch,
 {
-    receiver_global_node_id: OnceLock<Arc<String>>,
+    name: OperatorName,
     exchange_id: ExchangeId,
 
     /// The number of communicating peers.
@@ -159,17 +159,26 @@ where
         let layout = runtime.layout();
         let npeers = layout.n_workers();
 
+        let name = OperatorName::new("ShardedAccumulatorReceiver");
         let exchange = Arc::new(Self {
             exchange_id,
-            receiver_global_node_id: Default::default(),
             npeers,
             local_workers: layout.local_workers(),
             factories: factories.clone(),
             clients,
             rxq: layout
                 .local_workers()
-                .map(|_| Mutex::new(Rxq::new(runtime, worker_index, factories, npeers)))
+                .map(|_| {
+                    Mutex::new(Rxq::new(
+                        runtime,
+                        worker_index,
+                        factories,
+                        npeers,
+                        name.get(),
+                    ))
+                })
                 .collect(),
+            name,
         });
 
         directory.insert(exchange_id, exchange.clone());
@@ -204,7 +213,7 @@ where
         }
     }
 
-    async fn send(self: &Arc<Self>, global_node_id: &Arc<String>, batch: B, flush: bool) {
+    async fn send(self: &Arc<Self>, name: Arc<String>, batch: B, flush: bool) {
         let sender = Runtime::worker_index();
 
         let runtime = Runtime::runtime().unwrap();
@@ -264,7 +273,7 @@ where
                         .collect_vec();
                     let this = self.clone();
                     if let Some(waiter) = this.clients.connect(receivers.start).await.send(
-                        global_node_id.clone(),
+                        name.clone(),
                         this.exchange_id,
                         sender,
                         items,
@@ -280,7 +289,7 @@ where
                 .with_category("Exchange")
                 .with_tooltip(|| {
                     format!(
-                        "{global_node_id} wait for batches to merge in {} receive queues (for workers {})",
+                        "{name} wait for batches to merge in {} receive queues (for workers {})",
                         local_waiters.len(),
                         local_waiters
                             .iter()
@@ -297,7 +306,7 @@ where
                 .with_category("Exchange")
                 .with_tooltip(|| {
                     format!(
-                        "{global_node_id} wait for {} to drain from {} tx buffers",
+                        "{name} wait for {} to drain from {} tx buffers",
                         HumanBytes::from(serialized_bytes),
                         remote_waiters.len()
                     )
@@ -312,14 +321,21 @@ where
         let receiver = Runtime::worker_index();
         self.rxq(receiver).receive()
     }
+
+    fn set_name(&self, global_id: &GlobalNodeId) {
+        self.name.init(global_id);
+        for rxq in &self.rxq {
+            rxq.lock().unwrap().set_name(self.name.get());
+        }
+    }
 }
 
 impl<B> ExchangeDelivery for ShardedAccumulator<B>
 where
     B: Batch<Time = ()>,
 {
-    fn name(&self) -> &str {
-        self.receiver_global_node_id.get().unwrap()
+    fn name(&self) -> Arc<String> {
+        self.name.get()
     }
 
     fn received<'a>(
@@ -364,6 +380,9 @@ where
     /// The number of entries that have been popped off `spines` and received by
     /// the circuit.
     n_received: usize,
+
+    /// Name for use in profiles.
+    name: Arc<String>,
 }
 
 /// A spine that a [ShardedAccumulatorReceiver] is building from batches
@@ -391,10 +410,11 @@ where
         runtime: &Runtime,
         worker_index: usize,
         factories: &B::Factories,
+        name: Arc<String>,
     ) -> Self {
         Self {
             n_unflushed: npeers,
-            spine: Spine::with_runtime(runtime.clone(), worker_index, factories),
+            spine: Spine::with_runtime(runtime.clone(), worker_index, factories, name),
         }
     }
 }
@@ -408,12 +428,20 @@ where
         worker_index: usize,
         factories: &B::Factories,
         npeers: usize,
+        name: Arc<String>,
     ) -> Self {
         Self {
             runtime: runtime.clone(),
             worker_index,
             npeers,
-            spines: VecDeque::from([RxqEntry::new(npeers, runtime, worker_index, factories)]),
+            spines: VecDeque::from([RxqEntry::new(
+                npeers,
+                runtime,
+                worker_index,
+                factories,
+                name.clone(),
+            )]),
+            name,
             n_flushes: repeat_n(0, npeers).collect(),
             n_received: 0,
         }
@@ -440,6 +468,7 @@ where
                     &self.runtime,
                     self.worker_index,
                     factories,
+                    self.name.clone(),
                 ));
             }
         }
@@ -455,6 +484,13 @@ where
                 entry.spine
             })
     }
+
+    fn set_name(&mut self, name: Arc<String>) {
+        for entry in &mut self.spines {
+            entry.spine.set_name(name.clone());
+        }
+        self.name = name;
+    }
 }
 
 struct ShardedAccumulatorSender<B>
@@ -462,7 +498,7 @@ where
     B: Batch,
 {
     location: OperatorLocation,
-    global_node_id: Arc<String>,
+    name: OperatorName,
     exchange: Arc<ShardedAccumulator<B>>,
 
     // Input batch sizes.
@@ -478,7 +514,7 @@ where
     fn new(location: OperatorLocation, exchange: Arc<ShardedAccumulator<B>>) -> Self {
         Self {
             location,
-            global_node_id: Arc::new(format!("ShardedAccumulatorSender {}", exchange.exchange_id)),
+            name: OperatorName::new("ShardedAccumulatorSender"),
             exchange,
             input_batch_stats: BatchSizeStats::new(),
             flushed: false,
@@ -495,10 +531,7 @@ where
     }
 
     fn init(&mut self, global_id: &GlobalNodeId) {
-        self.global_node_id = Arc::new(format!(
-            "ShardedAccumulatorSender {}",
-            global_id.node_identifier()
-        ));
+        self.name.init(global_id);
     }
 
     fn metadata(&self, meta: &mut OperatorMeta) {
@@ -534,7 +567,7 @@ where
     async fn eval_owned(&mut self, batch: B) {
         self.input_batch_stats.add_batch(batch.num_entries_deep());
         self.exchange
-            .send(&self.global_node_id, batch, self.flushed)
+            .send(self.name.get(), batch, self.flushed)
             .await;
         self.flushed = false;
     }
@@ -577,10 +610,7 @@ where
     }
 
     fn init(&mut self, global_id: &GlobalNodeId) {
-        let _ = self.exchange.receiver_global_node_id.set(Arc::new(format!(
-            "ShardedAccumulatorReceiver {}",
-            global_id.node_identifier()
-        )));
+        self.exchange.set_name(global_id);
     }
 
     fn location(&self) -> OperatorLocation {
diff --git a/crates/dbsp/src/operator/dynamic/trace.rs b/crates/dbsp/src/operator/dynamic/trace.rs
index ae3aadd102e..962ddc8cb18 100644
--- a/crates/dbsp/src/operator/dynamic/trace.rs
+++ b/crates/dbsp/src/operator/dynamic/trace.rs
@@ -1,6 +1,7 @@
 use crate::circuit::circuit_builder::{StreamId, register_replay_stream};
 use crate::circuit::metadata::{INPUT_RECORDS_COUNT, MEMORY_ALLOCATIONS_COUNT, RETAINMENT_BOUNDS};
-use crate::circuit::splitter_output_chunk_size;
+use crate::circuit::operator_traits::OperatorName;
+use crate::circuit::{NodeId, splitter_output_chunk_size};
 use crate::dynamic::{Factory, Weight, WeightTrait};
 use crate::operator::require_persistent_id;
 use crate::trace::spine_async::WithSnapshot;
@@ -945,7 +946,8 @@ impl<T: Trace> ReplayState<T> {
 
 pub struct Z1Trace<C: Circuit, B: Batch, T: Trace> {
     // For error reporting.
-    global_id: GlobalNodeId,
+    name: OperatorName,
+    local_id: NodeId,
     time: T::Time,
     trace: Option<T>,
     replay_state: Option<ReplayState<T>>,
@@ -980,7 +982,8 @@ where
         bounds: TraceBounds<T::Key, T::Val>,
     ) -> Self {
         Self {
-            global_id: GlobalNodeId::root(),
+            name: OperatorName::new("Z1Trace"),
+            local_id: NodeId::root(),
             time: <T::Time as Timestamp>::clock_start(),
             trace: None,
             replay_state: None,
@@ -1023,11 +1026,8 @@ where
     /// replay, `replay_stream` is active while the operator that normally write to
     /// stream is disabled.
     pub fn prepare_replay_stream(&mut self, stream: &Stream<C, B>) -> Stream<C, B> {
-        let replay_stream = Stream::with_value(
-            stream.circuit().clone(),
-            self.global_id.local_node_id().unwrap(),
-            stream.value(),
-        );
+        let replay_stream =
+            Stream::with_value(stream.circuit().clone(), self.local_id, stream.value());
 
         self.delta_stream = Some(replay_stream.clone());
         replay_stream
@@ -1048,7 +1048,7 @@ where
         self.dirty[scope as usize] = false;
 
         if scope == 0 && self.trace.is_none() {
-            self.trace = Some(T::new(&self.trace_factories));
+            self.trace = Some(T::new(&self.trace_factories, self.name.get()));
         }
     }
 
@@ -1063,7 +1063,8 @@ where
     }
 
     fn init(&mut self, global_id: &GlobalNodeId) {
-        self.global_id = global_id.clone();
+        self.name.init(global_id);
+        self.local_id = global_id.local_node_id().unwrap()
     }
 
     fn metadata(&self, meta: &mut OperatorMeta) {
@@ -1103,7 +1104,7 @@ where
         pid: Option<&str>,
         files: &mut Vec<Arc<dyn FileCommitter>>,
     ) -> Result<(), Error> {
-        let pid = require_persistent_id(pid, &self.global_id)?;
+        let pid = require_persistent_id(pid, &self.name)?;
         self.trace
             .as_mut()
             .map(|trace| trace.save(base, pid, files))
@@ -1111,7 +1112,7 @@ where
     }
 
     fn restore(&mut self, base: &StoragePath, pid: Option<&str>) -> Result<(), Error> {
-        let pid = require_persistent_id(pid, &self.global_id)?;
+        let pid = require_persistent_id(pid, &self.name)?;
 
         self.trace
             .as_mut()
@@ -1121,7 +1122,7 @@ where
 
     fn clear_state(&mut self) -> Result<(), Error> {
         // println!("Z1Trace-{}::clear_state", &self.global_id);
-        self.trace = Some(T::new(&self.trace_factories));
+        self.trace = Some(T::new(&self.trace_factories, self.name.get()));
         self.replay_state = None;
         self.dirty = vec![false; self.root_scope as usize + 1];
 
@@ -1139,7 +1140,7 @@ where
         self.replay_mode = true;
         if self.delta_stream.is_some() && self.replay_state.is_none() {
             let trace = self.trace.take().expect("Z1Trace::start_replay: no trace");
-            self.trace = Some(T::new(&self.trace_factories));
+            self.trace = Some(T::new(&self.trace_factories, self.name.get()));
 
             //println!("Z1Trace-{}::initializing replay_state", &self.global_id);
 
diff --git a/crates/dbsp/src/operator/non_incremental.rs b/crates/dbsp/src/operator/non_incremental.rs
index a22e8936b04..96cacaf4665 100644
--- a/crates/dbsp/src/operator/non_incremental.rs
+++ b/crates/dbsp/src/operator/non_incremental.rs
@@ -3,7 +3,7 @@ use crate::{
     circuit::{
         OwnershipPreference,
         circuit_builder::{CircuitBase, NonIterativeCircuit},
-        operator_traits::{ImportOperator, Operator},
+        operator_traits::{ImportOperator, Operator, OperatorName},
         runtime::Consensus,
         schedule::{CommitProgress, DynamicScheduler, Executor, Scheduler},
     },
@@ -173,6 +173,7 @@ where
     B: DynBatch,
 {
     factories: B::Factories,
+    name: OperatorName,
     spine: DynSpine<B>,
 }
 
@@ -181,9 +182,11 @@ where
     B: DynBatch,
 {
     pub fn new(factories: &B::Factories) -> Self {
+        let name = OperatorName::new("ImportAccumulator");
         Self {
             factories: factories.clone(),
-            spine: DynSpine::<B>::new(factories),
+            spine: DynSpine::<B>::new(factories, name.get()),
+            name,
         }
     }
 }
@@ -196,6 +199,11 @@ where
         Cow::Borrowed("ImportAccumulator")
     }
 
+    fn init(&mut self, global_id: &crate::circuit::GlobalNodeId) {
+        self.name.init(global_id);
+        self.spine.set_name(self.name.get());
+    }
+
     fn fixedpoint(&self, _scope: crate::circuit::Scope) -> bool {
         self.spine.is_empty()
     }
@@ -214,7 +222,7 @@ where
     }
 
     async fn eval(&mut self) -> DynSpine<B> {
-        let mut spine = DynSpine::<B>::new(&self.factories);
+        let mut spine = DynSpine::<B>::new(&self.factories, self.name.get());
         std::mem::swap(&mut self.spine, &mut spine);
         spine
     }
diff --git a/crates/dbsp/src/trace.rs b/crates/dbsp/src/trace.rs
index aa2a267b323..5900a37c3ed 100644
--- a/crates/dbsp/src/trace.rs
+++ b/crates/dbsp/src/trace.rs
@@ -232,8 +232,12 @@ pub trait Trace: BatchReader {
             Factories = Self::Factories,
         >;
 
-    /// Allocates a new empty trace.
-    fn new(factories: &Self::Factories) -> Self;
+    /// Allocates a new empty trace associated with `name`, which should
+    /// identify the operator or other origin of the trace.
+    fn new(factories: &Self::Factories, name: Arc<String>) -> Self;
+
+    /// Updates the name of the trace to `name`.
+    fn set_name(&mut self, name: Arc<String>);
 
     /// Sets a compaction frontier, i.e., a timestamp such that timestamps
     /// below the frontier are indistinguishable to DBSP, therefore any `ts`
diff --git a/crates/dbsp/src/trace/spine_async.rs b/crates/dbsp/src/trace/spine_async.rs
index 7cbe81145a3..a87ec98d221 100644
--- a/crates/dbsp/src/trace/spine_async.rs
+++ b/crates/dbsp/src/trace/spine_async.rs
@@ -343,6 +343,7 @@ where
 {
     #[size_of(skip)]
     factories: B::Factories,
+    name: Arc<String>,
     #[size_of(skip)]
     key_filter: Option<Filter<B::Key>>,
     #[size_of(skip)]
@@ -360,9 +361,10 @@ impl<B> SharedState<B>
 where
     B: Batch,
 {
-    pub fn new(factories: &B::Factories) -> Self {
+    pub fn new(factories: &B::Factories, name: Arc<String>) -> Self {
         Self {
             factories: factories.clone(),
+            name,
             key_filter: None,
             value_filter: None,
             frontier: B::Time::minimum(),
@@ -391,22 +393,13 @@ where
         self.slots[level].notify.notify_one();
     }
 
-    fn should_apply_backpressure(&self) -> bool {
-        const HIGH_THRESHOLD: usize = 128;
-        self.slots
-            .iter()
-            .map(|s| s.loose_batches.len())
-            .sum::<usize>()
-            >= HIGH_THRESHOLD
-    }
-
-    fn should_relieve_backpressure(&self) -> bool {
-        const LOWER_THRESHOLD: usize = 127;
-        self.slots
-            .iter()
-            .map(|s| s.loose_batches.len())
-            .sum::<usize>()
-            <= LOWER_THRESHOLD
+    fn batch_count(&self) -> BatchCount {
+        BatchCount(
+            self.slots
+                .iter()
+                .map(|s| s.loose_batches.len())
+                .sum::<usize>(),
+        )
     }
 
     fn get_filters(&self) -> (Option<Filter<B::Key>>, Option<GroupFilter<B::Val>>) {
@@ -475,7 +468,9 @@ where
             .with_start(start)
             .with_tooltip(|| {
                 format!(
-                    "Merged {} batches ({pre_len} -> {post_len}) in {n_steps} steps using {:.1} ms CPU",
+                    "{} in worker {} merged {} batches ({pre_len} -> {post_len}) in {n_steps} steps using {:.1} ms CPU",
+                    &self.name,
+                    Runtime::worker_index(),
                     batches.len(),
                     elapsed.as_secs_f64() * 1000.0
                 )
@@ -560,6 +555,21 @@ where
     }
 }
 
+#[derive(Copy, Clone, Debug)]
+struct BatchCount(usize);
+
+impl BatchCount {
+    const HIGH_THRESHOLD: usize = 128;
+
+    fn should_apply_backpressure(&self) -> bool {
+        self.0 >= Self::HIGH_THRESHOLD
+    }
+
+    fn should_relieve_backpressure(&self) -> bool {
+        !self.should_apply_backpressure()
+    }
+}
+
 /// A fully asynchronous merger.
 ///
 /// Merging has two benefits:
@@ -646,10 +656,15 @@ impl<B> AsyncMerger<B>
 where
     B: Batch,
 {
-    pub fn new(runtime: Runtime, worker_index: usize, factories: &B::Factories) -> Self {
+    pub fn new(
+        runtime: Runtime,
+        worker_index: usize,
+        factories: &B::Factories,
+        name: Arc<String>,
+    ) -> Self {
         let idle = Arc::new(Condvar::new());
         let no_backpressure = Arc::new(Notify::new());
-        let state = Arc::new(Mutex::new(SharedState::new(factories)));
+        let state = Arc::new(Mutex::new(SharedState::new(factories, name)));
 
         let max_level0_batch_size_records = max_level0_batch_size_records() as usize;
         assert!(
@@ -706,23 +721,45 @@ where
     /// Blocks until the number of batches in the spine has declined below the
     /// backpressure threshold.
     async fn backpressure_wait(&self) {
-        let start = Instant::now();
-        loop {
+        let start_time = Instant::now();
+
+        // Do an initial check of the batch count.  If it's already dropped,
+        // exit early, otherwise save the name and the initial number of
+        // batches for later recording.
+        let (name, initial_batches);
+        {
+            let state = self.state.lock().unwrap();
+            let batch_count = state.batch_count();
+            if batch_count.should_relieve_backpressure() {
+                return;
+            }
+            name = state.name.clone();
+            initial_batches = batch_count.0;
+        }
+
+        // Wait for the batch count to drop below the threshold.
+        let final_batches = loop {
             let notify = self.no_backpressure.notified();
             {
                 let mut state = self.state.lock().unwrap();
-                if state.should_relieve_backpressure() {
-                    state.spine_stats.backpressure_wait += start.elapsed();
-                    break;
+                let batch_count = state.batch_count();
+                if batch_count.should_relieve_backpressure() {
+                    state.spine_stats.backpressure_wait += start_time.elapsed();
+                    break batch_count.0;
                 }
             }
             notify.await;
-        }
+        };
+
+        // Record the wait.
         COMPACTION_STALL_TIME_NANOSECONDS
-            .fetch_add(start.elapsed().as_nanos() as u64, Ordering::Relaxed);
+            .fetch_add(start_time.elapsed().as_nanos() as u64, Ordering::Relaxed);
         Span::new("backpressure-wait")
             .with_category("Spine")
-            .with_start(start)
+            .with_start(start_time)
+            .with_tooltip(|| {
+                format!("{name} wait for drop from {initial_batches} to {final_batches} batches")
+            })
             .record();
     }
 
@@ -731,6 +768,7 @@ where
         self.state
             .lock()
             .unwrap()
+            .batch_count()
             .should_apply_backpressure()
             .then_some(notify)
     }
@@ -1152,7 +1190,7 @@ where
                                 self.no_backpressure.notify_waiters();
                                 break;
                             }
-                            if state.should_relieve_backpressure() {
+                            if state.batch_count().should_relieve_backpressure() {
                                 self.no_backpressure.notify_waiters();
                             }
                         }
@@ -1860,15 +1898,20 @@ where
 {
     type Batch = B;
 
-    fn new(factories: &B::Factories) -> Self {
+    fn new(factories: &B::Factories, name: Arc<String>) -> Self {
         Self::with_runtime(
             Runtime::runtime()
                 .expect("Attempting to create a spine merger outside of a DBSP runtime"),
             Runtime::worker_index(),
             factories,
+            name,
         )
     }
 
+    fn set_name(&mut self, name: Arc<String>) {
+        self.merger.state.lock().unwrap().name = name;
+    }
+
     fn set_frontier(&mut self, frontier: &B::Time) {
         self.merger.set_frontier(frontier)
     }
@@ -1903,6 +1946,7 @@ where
             if self
                 .merger
                 .add_batch(batch, false)
+                .batch_count()
                 .should_apply_backpressure()
             {
                 self.merger.backpressure_wait().await;
@@ -2099,13 +2143,18 @@ where
     ///
     /// In the common case, [Spine::new] uses the current thread's runtime and
     /// worker index.
-    pub fn with_runtime(runtime: Runtime, worker_index: usize, factories: &B::Factories) -> Self {
+    pub fn with_runtime(
+        runtime: Runtime,
+        worker_index: usize,
+        factories: &B::Factories,
+        name: Arc<String>,
+    ) -> Self {
         Spine {
             factories: factories.clone(),
             dirty: false,
             key_filter: None,
             value_filter: None,
-            merger: AsyncMerger::new(runtime, worker_index, factories),
+            merger: AsyncMerger::new(runtime, worker_index, factories, name),
         }
     }
 
@@ -2139,7 +2188,7 @@ where
                 .with_category("Spine")
                 .with_tooltip(|| {
                     format!(
-                        "Eagerly spilling {} batch with {} keys and {} values",
+                        "Eagerly spill {} batch with {} keys and {} values",
                         HumanBytes::from(batch.approximate_byte_size()),
                         batch.key_count(),
                         batch.len()
@@ -2188,6 +2237,7 @@ where
             self.dirty = true;
             self.merger
                 .add_batch(batch, false)
+                .batch_count()
                 .should_apply_backpressure()
         } else {
             false
diff --git a/crates/dbsp/src/trace/test.rs b/crates/dbsp/src/trace/test.rs
index c8ca68d74c0..7a25131d8c6 100644
--- a/crates/dbsp/src/trace/test.rs
+++ b/crates/dbsp/src/trace/test.rs
@@ -193,10 +193,10 @@ fn test_zset_spine<B: ZSet<Key = DynI32>>(
     batches: Vec<(Vec<Tup2<i32, ZWeight>>, i32)>,
     seed: u64,
 ) {
-    let mut trace: Spine<B> = Spine::new(factories);
+    let mut trace: Spine<B> = Spine::new(factories, Arc::new(String::from("Test")));
 
     let mut ref_trace: TestBatch<DynI32, DynUnit /* <()> */, (), DynZWeight> =
-        TestBatch::new(&TestBatchFactories::new());
+        TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
     let mut kbound = 0;
     for (tuples, bound) in batches.into_iter() {
@@ -241,10 +241,10 @@ fn test_indexed_zset_spine<B: IndexedZSet<Key = DynI32, Val = DynI32>>(
     batches: Vec<(Vec<Tup2<Tup2<i32, i32>, ZWeight>>, i32, i32)>,
     seed: u64,
 ) {
-    let mut trace: Spine<B> = Spine::new(factories);
+    let mut trace: Spine<B> = Spine::new(factories, Arc::new(String::from("Test")));
 
     let mut ref_trace: TestBatch<DynI32, DynI32, (), DynZWeight> =
-        TestBatch::new(&TestBatchFactories::new());
+        TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
     let mut bound = 0;
     let mut kbound = 0;
@@ -305,9 +305,9 @@ fn test_val_batch_trace_spine<B: ZBatch<Key = DynI32, Val = DynI32, Time = u32>>
 ) {
     // `trace1` uses `truncate_keys_below`.
     // `trace2` uses `retain_keys`.
-    let mut trace: Spine<B> = Spine::new(factories);
+    let mut trace: Spine<B> = Spine::new(factories, Arc::new(String::from("Test")));
     let mut ref_trace: TestBatch<DynI32, DynI32, u32, DynZWeight> =
-        TestBatch::new(&TestBatchFactories::new());
+        TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
     let mut bound = 0;
     let mut kbound = 0;
@@ -437,9 +437,9 @@ fn test_key_batch_spine<B: ZBatch<Key = DynI32, Val = DynUnit, Time = u32>>(
     batches: Vec<(Vec<Tup2<i32, ZWeight>>, i32)>,
     seed: u64,
 ) {
-    let mut trace: Spine<B> = Spine::new(factories);
+    let mut trace: Spine<B> = Spine::new(factories, Arc::new(String::from("Test")));
     let mut ref_trace: TestBatch<DynI32, DynUnit /* <()> */, u32, DynZWeight> =
-        TestBatch::new(&TestBatchFactories::new());
+        TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
     let mut kbound = 0;
     for (time, (tuples, bound)) in batches.into_iter().enumerate() {
@@ -786,7 +786,7 @@ proptest! {
         Runtime::run(CircuitConfig::with_workers(1), move |_parker| {
             let factories = <OrdIndexedZSetFactories<DynI32, DynI32>>::new::<i32, i32, ZWeight>();
 
-            let mut trace: Spine<OrdIndexedZSet<DynI32, DynI32>> = Spine::new(&factories);
+            let mut trace: Spine<OrdIndexedZSet<DynI32, DynI32>> = Spine::new(&factories, Arc::new(String::from("Test")));
 
             for (i, tuples) in batches.into_iter().enumerate() {
                 let mut erased_tuples = indexed_zset_tuples(tuples);
@@ -810,7 +810,7 @@ proptest! {
         Runtime::run(CircuitConfig::with_workers(1), move |_parker| {
             let factories = <OrdIndexedZSetFactories<DynI32, DynI32>>::new::<i32, i32, ZWeight>();
 
-            let mut trace: Spine<OrdIndexedZSet<DynI32, DynI32>> = Spine::new(&factories);
+            let mut trace: Spine<OrdIndexedZSet<DynI32, DynI32>> = Spine::new(&factories, Arc::new(String::from("Test")));
 
             for (i, tuples) in batches.into_iter().enumerate() {
                 let mut erased_tuples = indexed_zset_tuples(tuples);
@@ -873,8 +873,8 @@ proptest! {
         Runtime::run(CircuitConfig::with_workers(1), move |_parker| {
             let factories = <OrdIndexedZSetFactories<DynI32, DynI32>>::new::<i32, i32, ZWeight>();
 
-            let mut trace: Spine<OrdIndexedZSet<DynI32, DynI32>> = Spine::new(&factories);
-            let mut ref_trace: TestBatch<DynI32, DynI32, (), DynZWeight> = TestBatch::new(&TestBatchFactories::new());
+            let mut trace: Spine<OrdIndexedZSet<DynI32, DynI32>> = Spine::new(&factories, Arc::new(String::from("Test")));
+            let mut ref_trace: TestBatch<DynI32, DynI32, (), DynZWeight> = TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
             trace.retain_values(GroupFilter::Simple(Filter::new(Box::new(
                 move |val: &DynI32| *val.downcast_checked::<i32>() % 2 == 0,
@@ -911,8 +911,8 @@ proptest! {
         Runtime::run(CircuitConfig::with_workers(1), move |_parker| {
             let factories = <OrdIndexedZSetFactories<DynI32, DynI32>>::new::<i32, i32, ZWeight>();
 
-            let mut trace: Spine<OrdIndexedZSet<DynI32, DynI32>> = Spine::new(&factories);
-            let mut ref_trace: TestBatch<DynI32, DynI32, (), DynZWeight> = TestBatch::new(&TestBatchFactories::new());
+            let mut trace: Spine<OrdIndexedZSet<DynI32, DynI32>> = Spine::new(&factories, Arc::new(String::from("Test")));
+            let mut ref_trace: TestBatch<DynI32, DynI32, (), DynZWeight> = TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
             trace.retain_keys(Filter::new(Box::new(move |val| *val.downcast_checked::<i32>() % 2 == 0)));
             ref_trace.retain_keys(Filter::new(Box::new(move |val| *val.downcast_checked::<i32>() % 2 == 0)));
@@ -982,8 +982,8 @@ proptest! {
 
             // `trace1` uses `truncate_keys_below`.
             // `trace2` uses `retain_keys`.
-            let mut trace: Spine<OrdValBatch<DynI32, DynI32, u32, DynZWeight>> = Spine::new(&factories);
-            let mut ref_trace: TestBatch<DynI32, DynI32, u32, DynZWeight> = TestBatch::new(&TestBatchFactories::new());
+            let mut trace: Spine<OrdValBatch<DynI32, DynI32, u32, DynZWeight>> = Spine::new(&factories, Arc::new(String::from("Test")));
+            let mut ref_trace: TestBatch<DynI32, DynI32, u32, DynZWeight> = TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
             trace.retain_values(GroupFilter::Simple(Filter::new(Box::new(
                 move |val: &DynI32| *val.downcast_checked::<i32>() % 2 == 0,
@@ -1019,8 +1019,8 @@ proptest! {
 
             // `trace1` uses `truncate_keys_below`.
             // `trace2` uses `retain_keys`.
-            let mut trace: Spine<OrdValBatch<DynI32, DynI32, u32, DynZWeight>> = Spine::new(&factories);
-            let mut ref_trace: TestBatch<DynI32, DynI32, u32, DynZWeight> = TestBatch::new(&TestBatchFactories::new());
+            let mut trace: Spine<OrdValBatch<DynI32, DynI32, u32, DynZWeight>> = Spine::new(&factories, Arc::new(String::from("Test")));
+            let mut ref_trace: TestBatch<DynI32, DynI32, u32, DynZWeight> = TestBatch::new(&TestBatchFactories::new(), Arc::new(String::from("Test")));
 
             trace.retain_keys(Filter::new(Box::new(move |key| *key.downcast_checked::<i32>() % 2 == 0)));
             ref_trace.retain_keys(Filter::new(Box::new(move |key| *key.downcast_checked::<i32>() % 2 == 0)));
diff --git a/crates/dbsp/src/trace/test/test_batch.rs b/crates/dbsp/src/trace/test/test_batch.rs
index 8b6e38ca524..62cca06f11f 100644
--- a/crates/dbsp/src/trace/test/test_batch.rs
+++ b/crates/dbsp/src/trace/test/test_batch.rs
@@ -817,7 +817,7 @@ where
     fn new_batcher(factories: &TestBatchFactories<K, V, T, R>, time: T) -> Self {
         Self {
             time,
-            result: TestBatch::new(factories),
+            result: TestBatch::new(factories, Arc::new(String::from("Test"))),
         }
     }
 
@@ -881,7 +881,7 @@ where
         _location: Option<BatchLocation>,
     ) -> Self {
         Self {
-            result: TestBatch::new(factories),
+            result: TestBatch::new(factories, Arc::new(String::from("Test"))),
             time_diffs: Vec::new(),
             vals: BTreeMap::new(),
             num_keys: 0,
@@ -1314,7 +1314,7 @@ where
 {
     type Batch = Self;
 
-    fn new(_factories: &Self::Factories) -> Self {
+    fn new(_factories: &Self::Factories, _name: Arc<String>) -> Self {
         Self {
             data: BTreeMap::new(),
             lower_key_bound: None,
@@ -1323,6 +1323,8 @@ where
         }
     }
 
+    fn set_name(&mut self, _name: Arc<String>) {}
+
     fn set_frontier(&mut self, _frontier: &Self::Time) {
         // Ok to do nothing here, since frontiers are an optimization and are meant to be applied lazily during merging.
     }
diff --git a/crates/samply/src/lib.rs b/crates/samply/src/lib.rs
index 91f1cdd1592..7c12ae78cda 100644
--- a/crates/samply/src/lib.rs
+++ b/crates/samply/src/lib.rs
@@ -82,21 +82,22 @@ impl Default for AtomicOptionTimestamp {
 }
 
 impl AtomicOptionTimestamp {
-    fn new(value: Option<Timestamp>) -> Self {
-        Self(AtomicI64::new(
-            value.map_or(i64::MIN, |timestamp| timestamp.0),
-        ))
+    const fn new(value: Option<Timestamp>) -> Self {
+        Self(AtomicI64::new(match value {
+            Some(timestamp) => timestamp.0,
+            None => i64::MIN,
+        }))
     }
 
     fn load(&self) -> Option<Timestamp> {
-        let value = self.0.load(Ordering::Relaxed);
+        let value = self.0.load(Ordering::Acquire);
         (value != i64::MIN).then_some(Timestamp(value))
     }
 
     fn store(&self, value: Option<Timestamp>) {
         self.0.store(
             value.map_or(i64::MIN, |timestamp| timestamp.0),
-            Ordering::Relaxed,
+            Ordering::Release,
         )
     }
 }
@@ -149,6 +150,15 @@ impl Timestamp {
             Self(now.tv_sec() as i64 * 1_000_000_000 + now.tv_nsec() as i64)
         }
     }
+
+    /// Computes `self - other`, returning zero if `self < other`.
+    fn saturating_sub(self, other: Self) -> Duration {
+        if self.0 >= other.0 {
+            Duration::from_nanos(self.0.abs_diff(other.0))
+        } else {
+            Duration::ZERO
+        }
+    }
 }
 
 /// Nanoseconds since the Unix epoch.
@@ -603,6 +613,7 @@ static CAPTURE_MUTEX: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(())
 /// Only one `Capture` may exist at one time.
 pub struct Capture {
     _guard: tokio::sync::MutexGuard<'static, ()>,
+    start_time: Timestamp,
     memory: Option<JoinHandle<Vec<(Timestamp, usize)>>>,
     unparker: Unparker,
     request_exit: Arc<AtomicBool>,
@@ -613,6 +624,7 @@ pub struct Capture {
 
 impl Capture {
     fn new(params: CaptureOptions, guard: tokio::sync::MutexGuard<'static, ()>) -> Self {
+        let start = Timestamp::now();
         if let Some(memory_limit) = params.memory_limit {
             tracing::info!(
                 "marker capture limited to {}",
@@ -644,8 +656,10 @@ impl Capture {
                 .expect("should be able to start a capture thread")
         });
         FREE_BLOCKS.store(block_limit, Ordering::Relaxed);
+        MARKERS_EXHAUSTED.store(None);
         CAPTURING.store(true, Ordering::Release);
         Self {
+            start_time: start,
             block_limit,
             memory,
             unparker,
@@ -660,15 +674,16 @@ impl Capture {
     pub fn finish(mut self) -> Annotations {
         let end_time = Timestamp::now();
         CAPTURING.store(false, Ordering::Release);
-        let remaining_blocks = FREE_BLOCKS.load(Ordering::Relaxed);
-        if remaining_blocks <= 0 {
-            tracing::info!("marker capture exceeded the limit");
+        let markers_exhausted = MARKERS_EXHAUSTED.load();
+        let free_blocks = FREE_BLOCKS.load(Ordering::Relaxed);
+        let used =
+            HumanBytes::from((self.block_limit - free_blocks.max(0)) as usize * BYTES_PER_BLOCK);
+        if free_blocks < 0 {
         } else {
-            let used_bytes = (self.block_limit - remaining_blocks) as usize * BYTES_PER_BLOCK;
-            tracing::info!("marker capture used {}", HumanBytes::from(used_bytes));
+            tracing::info!("marker capture used {used}");
         }
 
-        let markers = self
+        let mut markers: HashMap<usize, (Option<String>, Blocks)> = self
             .all_threads()
             .into_iter()
             .map(|thread| {
@@ -683,6 +698,28 @@ impl Capture {
             })
             .collect();
 
+        if let Some(markers_exhausted) = markers_exhausted {
+            let elapsed = markers_exhausted.saturating_sub(self.start_time);
+            let tooltip = format!(
+                "marker capture exceeded the limit ({used}) after {:.1} s",
+                elapsed.as_secs_f64()
+            );
+            tracing::info!("{tooltip}");
+            let marker = Marker {
+                start: self.start_time,
+                end: MarkerEnd::At(markers_exhausted),
+                category: "profiling",
+                name: "Profiling",
+                tooltip,
+            };
+            markers
+                .entry(nix::unistd::getpid().as_raw() as usize)
+                .or_default()
+                .1
+                .0
+                .push(Block::new(marker));
+        }
+
         Annotations {
             end_time,
             markers,
@@ -1216,10 +1253,20 @@ impl ThreadMarkers {
 /// [ThreadMarkers] for every thread that has recorded a marker.
 static ALL_THREAD_MARKERS: std::sync::Mutex<Vec<ThreadMarkers>> = std::sync::Mutex::new(Vec::new());
 
+/// Number of blocks of capacity left for allocation before we stop allocating.
+///
+/// If this is below zero, then we ran out and tried to allocate more anyhow.
 static FREE_BLOCKS: AtomicI64 = AtomicI64::new(0);
+
+/// Number of [Marker]s we allocate in each [Block].
 const MARKERS_PER_BLOCK: usize = 32;
+
+/// Number of bytes we account for each [Block].
 const BYTES_PER_BLOCK: usize = MARKERS_PER_BLOCK * Span::BYTES;
 
+/// The time at which [FREE_BLOCKS] dropped below zero.
+static MARKERS_EXHAUSTED: AtomicOptionTimestamp = AtomicOptionTimestamp::new(None);
+
 struct Block(Vec<Marker>);
 impl Block {
     fn new(marker: Marker) -> Self {
@@ -1252,7 +1299,14 @@ impl Blocks {
         } else {
             match FREE_BLOCKS.fetch_sub(1, Ordering::Relaxed) {
                 1.. => self.0.push(Block::new(marker)),
-                0 => warn!("marker capture space exhausted"),
+                0 => {
+                    // Record when marker space was exhausted.  The combination
+                    // of `load` and `store` is not an atomic transaction, but
+                    // it's good enough.
+                    if MARKERS_EXHAUSTED.load().is_none() {
+                        MARKERS_EXHAUSTED.store(Some(Timestamp::now()));
+                    }
+                }
                 _ => (),
             }
         }
diff --git a/crates/sqllib/Cargo.toml b/crates/sqllib/Cargo.toml
index d1197ec8865..47dbf3de83f 100644
--- a/crates/sqllib/Cargo.toml
+++ b/crates/sqllib/Cargo.toml
@@ -43,7 +43,7 @@ ryu = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 # `serde-with-arbitrary-precision` is needed because we enable `arbitrary_precision` in `serde_json`.
 serde_json = { workspace = true, features = ["arbitrary_precision"] }
-size-of = { workspace = true }
+size-of = { workspace = true, features = ["arcstr"] }
 smallstr = { workspace = true }
 smallvec = { workspace = true }
 thiserror = { workspace = true }
diff --git a/crates/sqllib/src/string.rs b/crates/sqllib/src/string.rs
index 0080b42c757..68230e4d2ab 100644
--- a/crates/sqllib/src/string.rs
+++ b/crates/sqllib/src/string.rs
@@ -872,6 +872,7 @@ mod tests {
         let total_size = SizeOf::size_of(&s);
 
         // The exact size may depend on the architecture's pointer size.
+        dbg!(&total_size);
         assert!(total_size.total_bytes() > 26);
         assert!(total_size.shared_bytes() >= 26);
     }