Log operator summaries using clever approach

timely/examples/logging-send.rs

@@ -2,7 +2,7 @@ use std::time::Duration;
 
 use timely::dataflow::{InputHandle, ProbeHandle};
 use timely::dataflow::operators::{Input, Exchange, Probe};
-use timely::logging::{TimelyEventBuilder, TimelyProgressEventBuilder};
+use timely::logging::{TimelyEventBuilder, TimelyProgressEventBuilder, TimelySummaryEventBuilder};
 use timely::container::CapacityContainerBuilder;
 use timely::progress::reachability::logging::TrackerEventBuilder;
 
@@ -61,6 +61,17 @@ fn main() {
             }
         );
 
+        worker.log_register().insert::<TimelySummaryEventBuilder<usize>,_>("timely/summary/usize", |time, data|
+            if let Some(data) = data {
+                data.iter().for_each(|(_, x)| {
+                    println!("SUMMARY: {:?}", x);
+                })
+            }
+            else {
+                println!("SUMMARY: Flush {time:?}");
+            }
+        );
+
         // create a new input, exchange data, and inspect its output
         worker.dataflow(|scope| {
             scope

timely/src/dataflow/scopes/child.rs

@@ -132,8 +132,9 @@ where
 
         let type_name = std::any::type_name::<T2>();
         let progress_logging = self.log_register().get(&format!("timely/progress/{type_name}"));
+        let summary_logging = self.log_register().get(&format!("timely/summary/{type_name}"));
 
-        let subscope = RefCell::new(SubgraphBuilder::new_from(path, self.logging(), name));
+        let subscope = RefCell::new(SubgraphBuilder::new_from(path, self.logging(), summary_logging, name));
         let result = {
             let mut builder = Child {
                 subgraph: &subscope,

timely/src/logging.rs

@@ -10,6 +10,10 @@ pub type TimelyLogger = crate::logging_core::TypedLogger<TimelyEventBuilder, Tim
 pub type TimelyProgressEventBuilder<T> = CapacityContainerBuilder<Vec<(Duration, TimelyProgressEvent<T>)>>;
 /// Logger for timely dataflow progress events (the "timely/progress" log stream).
 pub type TimelyProgressLogger<T> = crate::logging_core::Logger<TimelyProgressEventBuilder<T>>;
+/// Container builder for timely dataflow progress events.
+pub type TimelySummaryEventBuilder<TS> = CapacityContainerBuilder<Vec<(Duration, OperatesSummaryEvent<TS>)>>;
+/// Logger for timely dataflow progress events (the "timely/progress" log stream).
+pub type TimelySummaryLogger<TS> = crate::logging_core::Logger<TimelySummaryEventBuilder<TS>>;
 
 use std::time::Duration;
 use columnar::Columnar;
@@ -65,6 +69,16 @@ pub struct OperatesEvent {
     pub name: String,
 }
 
+
+#[derive(Serialize, Deserialize, Columnar, Debug, Clone, Eq, PartialEq)]
+/// The summary of internal connectivity of an `Operate` implementor.
+pub struct OperatesSummaryEvent<TS> {
+    /// Worker-unique identifier for the operator.
+    pub id: usize,
+    /// Timestamp action summaries for (input, output) pairs.
+    pub summary: Vec<Vec<crate::progress::Antichain<TS>>>,
+}
+
 #[derive(Serialize, Deserialize, Columnar, Debug, Clone, Hash, Eq, PartialEq, Ord, PartialOrd)]
 /// The creation of a channel between operators.
 pub struct ChannelsEvent {

timely/src/progress/subgraph.rs

@@ -11,6 +11,7 @@ use std::collections::BinaryHeap;
 use std::cmp::Reverse;
 
 use crate::logging::{TimelyLogger as Logger, TimelyProgressEventBuilder};
+use crate::logging::TimelySummaryLogger as SummaryLogger;
 
 use crate::scheduling::Schedule;
 use crate::scheduling::activate::Activations;
@@ -63,6 +64,8 @@ where
 
     /// Logging handle
     logging: Option<Logger>,
+    /// Typed logging handle for operator summaries.
+    summary_logging: Option<SummaryLogger<TInner::Summary>>,
 }
 
 impl<TOuter, TInner> SubgraphBuilder<TOuter, TInner>
@@ -95,6 +98,7 @@ where
     pub fn new_from(
         path: Rc<[usize]>,
         logging: Option<Logger>,
+        summary_logging: Option<SummaryLogger<TInner::Summary>>,
         name: &str,
     )
         -> SubgraphBuilder<TOuter, TInner>
@@ -113,6 +117,7 @@ where
             input_messages: Vec::new(),
             output_capabilities: Vec::new(),
             logging,
+            summary_logging,
         }
     }
 
@@ -135,7 +140,7 @@ where
                 name: child.name().to_owned(),
             });
         }
-        self.children.push(PerOperatorState::new(child, index, identifier, self.logging.clone()))
+        self.children.push(PerOperatorState::new(child, index, identifier, self.logging.clone(), &mut self.summary_logging));
     }
 
     /// Now that initialization is complete, actually build a subgraph.
@@ -636,7 +641,8 @@ impl<T: Timestamp> PerOperatorState<T> {
         mut scope: Box<dyn Operate<T>>,
         index: usize,
         identifier: usize,
-        logging: Option<Logger>
+        logging: Option<Logger>,
+        summary_logging: &mut Option<SummaryLogger<T::Summary>>,
     ) -> PerOperatorState<T>
     {
         let local = scope.local();
@@ -646,6 +652,13 @@ impl<T: Timestamp> PerOperatorState<T> {
 
         let (internal_summary, shared_progress) = scope.get_internal_summary();
 
+        if let Some(l) = summary_logging {
+            l.log(crate::logging::OperatesSummaryEvent {
+                id: identifier,
+                summary: internal_summary.clone(),
+            })
+        }
+
         assert_eq!(
             internal_summary.len(),
             inputs,

timely/src/worker.rs

@@ -630,7 +630,8 @@ impl<A: Allocate> Worker<A> {
 
         let type_name = std::any::type_name::<T>();
         let progress_logging = self.logging.borrow_mut().get(&format!("timely/progress/{type_name}"));
-        let subscope = SubgraphBuilder::new_from(addr, logging.clone(), name);
+        let summary_logging = self.logging.borrow_mut().get(&format!("timely/summary/{type_name}"));
+        let subscope = SubgraphBuilder::new_from(addr, logging.clone(), summary_logging, name);
         let subscope = RefCell::new(subscope);
 
         let result = {