feldera · blp · Jun 3, 2026 · May 28, 2026 · May 28, 2026 · May 29, 2026
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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,15 +1032,15 @@ 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))
             .unwrap_or(Ok(()))
     }
 
     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
@@ -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());
-Original file line number
+Diff line change
@@ Expand Up / @@ -1196,7 +1196,7 @@ impl NodeId { @@
             self.0
         }
-        pub(super) fn root() -> Self {
+        pub fn root() -> Self {
             Self(0)
         }
     }
@@ Expand Down @@