Merge pull request #678 from antiguru/notify_less

Disable notifications for some operators

Author: frankmcsherry

timely/src/dataflow/operators/branch.rs

@@ -40,6 +40,7 @@ impl<S: Scope, D: Data> Branch<S, D> for Stream<S, D> {
         condition: impl Fn(&S::Timestamp, &D) -> bool + 'static,
     ) -> (Stream<S, D>, Stream<S, D>) {
         let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope());
+        builder.set_notify(false);
 
         let mut input = builder.new_input(self, Pipeline);
         let (mut output1, stream1) = builder.new_output();
@@ -95,6 +96,7 @@ pub trait BranchWhen<T>: Sized {
 impl<S: Scope, C: Container + Data> BranchWhen<S::Timestamp> for StreamCore<S, C> {
     fn branch_when(&self, condition: impl Fn(&S::Timestamp) -> bool + 'static) -> (Self, Self) {
         let mut builder = OperatorBuilder::new("Branch".to_owned(), self.scope());
+        builder.set_notify(false);
 
         let mut input = builder.new_input(self, Pipeline);
         let (mut output1, stream1) = builder.new_output();

timely/src/dataflow/operators/core/concat.rs

@@ -71,6 +71,7 @@ impl<G: Scope, C: Container + Data> Concatenate<G, C> for G {
         // create an operator builder.
         use crate::dataflow::operators::generic::builder_rc::OperatorBuilder;
         let mut builder = OperatorBuilder::new("Concatenate".to_string(), self.clone());
+        builder.set_notify(false);
 
         // create new input handles for each input stream.
         let mut handles = sources.into_iter().map(|s| builder.new_input(&s, Pipeline)).collect::<Vec<_>>();

timely/src/dataflow/operators/core/feedback.rs

@@ -72,6 +72,7 @@ impl<G: Scope> Feedback<G> for G {
     fn feedback<C: Container + Data>(&mut self, summary: <G::Timestamp as Timestamp>::Summary) -> (Handle<G, C>, StreamCore<G, C>) {
 
         let mut builder = OperatorBuilder::new("Feedback".to_owned(), self.clone());
+        builder.set_notify(false);
         let (output, stream) = builder.new_output();
 
         (Handle { builder, summary, output }, stream)

timely/src/dataflow/operators/core/ok_err.rs

@@ -50,6 +50,7 @@ impl<S: Scope, C: Container + Data> OkErr<S, C> for StreamCore<S, C> {
         L: FnMut(C::Item<'_>) -> Result<D1,D2>+'static
     {
         let mut builder = OperatorBuilder::new("OkErr".to_owned(), self.scope());
+        builder.set_notify(false);
 
         let mut input = builder.new_input(self, Pipeline);
         let (mut output1, stream1) = builder.new_output();

timely/src/dataflow/operators/core/partition.rs

@@ -42,6 +42,7 @@ impl<G: Scope, C: Container + Data> Partition<G, C> for StreamCore<G, C> {
         F: FnMut(C::Item<'_>) -> (u64, D2) + 'static,
     {
         let mut builder = OperatorBuilder::new("Partition".to_owned(), self.scope());
+        builder.set_notify(false);
 
         let mut input = builder.new_input(self, Pipeline);
         let mut outputs = Vec::with_capacity(parts as usize);