Support optional path summaries for disconnected ports

Author: frankmcsherry

timely/src/dataflow/operators/capability.rs

@@ -246,11 +246,13 @@ pub struct InputCapability<T: Timestamp> {
 impl<T: Timestamp> CapabilityTrait<T> for InputCapability<T> {
     fn time(&self) -> &T { self.time() }
     fn valid_for_output(&self, query_buffer: &Rc<RefCell<ChangeBatch<T>>>) -> bool {
-        let borrow = self.summaries.borrow();
-        self.internal.borrow().iter().enumerate().any(|(index, rc)| {
-            // To be valid, the output buffer must match and the timestamp summary needs to be the default.
-            Rc::ptr_eq(rc, query_buffer) && borrow[index].len() == 1 && borrow[index][0] == Default::default()
-        })
+        let summaries_borrow = self.summaries.borrow();
+        let internal_borrow = self.internal.borrow();
+        // To be valid, the output buffer must match and the timestamp summary needs to be the default.
+        let result = summaries_borrow.iter_ports().any(|(port, path)| {
+            Rc::ptr_eq(&internal_borrow[port], query_buffer) && path.len() == 1 && path[0] == Default::default()
+        });
+        result
     }
 }
 
@@ -281,10 +283,15 @@ impl<T: Timestamp> InputCapability<T> {
     /// Delays capability for a specific output port.
     pub fn delayed_for_output(&self, new_time: &T, output_port: usize) -> Capability<T> {
         use crate::progress::timestamp::PathSummary;
-        if self.summaries.borrow()[output_port].iter().flat_map(|summary| summary.results_in(self.time())).any(|time| time.less_equal(new_time)) {
-            Capability::new(new_time.clone(), Rc::clone(&self.internal.borrow()[output_port]))
-        } else {
-            panic!("Attempted to delay to a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", new_time, self.summaries.borrow()[output_port], self.time());
+        if let Some(path) = self.summaries.borrow().get(output_port) {
+            if path.iter().flat_map(|summary| summary.results_in(self.time())).any(|time| time.less_equal(new_time)) {
+                Capability::new(new_time.clone(), Rc::clone(&self.internal.borrow()[output_port]))
+            } else {
+                panic!("Attempted to delay to a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", new_time, path, self.time());
+            }
+        }
+        else {
+            panic!("Attempted to delay a capability for a disconnected output");
         }
     }
 
@@ -305,11 +312,16 @@ impl<T: Timestamp> InputCapability<T> {
     pub fn retain_for_output(self, output_port: usize) -> Capability<T> {
         use crate::progress::timestamp::PathSummary;
         let self_time = self.time().clone();
-        if self.summaries.borrow()[output_port].iter().flat_map(|summary| summary.results_in(&self_time)).any(|time| time.less_equal(&self_time)) {
-            Capability::new(self_time, Rc::clone(&self.internal.borrow()[output_port]))
+        if let Some(path) = self.summaries.borrow().get(output_port) {
+            if path.iter().flat_map(|summary| summary.results_in(&self_time)).any(|time| time.less_equal(&self_time)) {
+                Capability::new(self_time, Rc::clone(&self.internal.borrow()[output_port]))
+            }
+            else {
+                panic!("Attempted to retain a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", self_time, path, self_time);
+            }
         }
         else {
-            panic!("Attempted to retain a time ({:?}) not greater or equal to the operators input-output summary ({:?}) applied to the capabilities time ({:?})", self_time, self.summaries.borrow()[output_port], self_time);
+            panic!("Attempted to retain a capability for a disconnected output");
         }
     }
 }

timely/src/progress/operate.rs

@@ -73,12 +73,6 @@ impl<TS> Default for PortConnectivity<TS> {
 }
 
 impl<TS> PortConnectivity<TS> {
-    /// Introduces default summaries for `0 .. count` ports.
-    pub fn default_for(count: usize) -> Self {
-        let mut list = Vec::with_capacity(count);
-        for _ in 0 .. count { list.push(Default::default()) }
-        Self { list }
-    }
     /// Ensures an entry exists at `index` and returns a mutable reference to it.
     fn ensure(&mut self, index: usize) -> &mut Antichain<TS> {
         while self.list.len() <= index { self.add_port(self.list.len(), Antichain::new()); }
@@ -105,6 +99,10 @@ impl<TS> PortConnectivity<TS> {
     pub fn max_port(&self) -> usize {
         self.list.len() - 1
     }
+    /// Returns the associated path summary, if it exists.
+    pub fn get(&self, index: usize) -> Option<&Antichain<TS>> {
+        self.list.get(index)
+    }
 }
 
 impl<TS> From<Vec<Antichain<TS>>> for PortConnectivity<TS> {
@@ -113,13 +111,6 @@ impl<TS> From<Vec<Antichain<TS>>> for PortConnectivity<TS> {
     }
 }
 
-impl<TS> std::ops::Index<usize> for PortConnectivity<TS> {
-    type Output = Antichain<TS>;
-    fn index(&self, index: usize) -> &Self::Output {
-        &self.list[index]
-    }
-}
-
 /// Progress information shared between parent and child.
 #[derive(Debug)]
 pub struct SharedProgress<T: Timestamp> {

timely/src/progress/reachability.rs

@@ -781,10 +781,12 @@ fn summarize_outputs<T: Timestamp>(
                     let antichains = results.entry(location).or_default();
 
                     // Combine each operator-internal summary to the output with `summary`.
-                    for operator_summary in summaries[output_port].elements().iter() {
-                        if let Some(combined) = operator_summary.followed_by(&summary) {
-                            if antichains.insert_ref(output, &combined) {
-                                worklist.push_back((location, output, combined));
+                    if let Some(connection) = summaries.get(output_port) {
+                        for operator_summary in connection.elements().iter() {
+                            if let Some(combined) = operator_summary.followed_by(&summary) {
+                                if antichains.insert_ref(output, &combined) {
+                                    worklist.push_back((location, output, combined));
+                                }
                             }
                         }
                     }

timely/src/progress/subgraph.rs

@@ -168,7 +168,7 @@ where
         let mut builder = reachability::Builder::new();
 
         // Child 0 has `inputs` outputs and `outputs` inputs, not yet connected.
-        let summary = (0..outputs).map(|_| PortConnectivity::default_for(inputs)).collect();
+        let summary = (0..outputs).map(|_| PortConnectivity::default()).collect();
         builder.add_node(0, outputs, inputs, summary);
         for (index, child) in self.children.iter().enumerate().skip(1) {
             builder.add_node(index, child.inputs, child.outputs, child.internal_summary.clone());
@@ -555,7 +555,7 @@ where
         // Note that we need to have `self.inputs()` elements in the summary
         // with each element containing `self.outputs()` antichains regardless
         // of how long `self.scope_summary` is
-        let mut internal_summary = vec![PortConnectivity::default_for(self.outputs); self.inputs()];
+        let mut internal_summary = vec![PortConnectivity::default(); self.inputs()];
         for (input_idx, input) in self.scope_summary.iter().enumerate() {
             for (output_idx, output) in input.iter_ports() {
                 for outer in output.elements().iter().cloned().map(TInner::summarize) {