Operator Merging

Here are examples of increased resource use when processing crosses operator boundaries:

• Performance counters to support rate monitoring
• Tracking what operator is being run in which thread to give better error messages
• Checking if debugging is active, to support dynamic debugger attach
• Checking the logging level to provide auto port-level logging
• Final punctuation logic to support self-terminating applications
• Virtual functions (need for binary compatibility)
• Indirection to support operator customization (onTuple clauses)

Here is a simple example of merging a filter operation into an upstream operator:

stream<someStreamType> tgt = Custom(src) {
  logic onTuple src: {
    v = v + 1;    // update v attribute of src stream
    submit (src,tgt);    // emit resulting stream
  }
}

stream<someStreamType> filtered = Filter(tgt) {
  param filter :  q > 10;     // q is an attribute of tgt stream
}

The merged operator becomes:

stream<someStreamType> filtered = Custom(src) {
  logic onTuple src: {       
    if (q > 10)  {
      v =  v + 1;    // update v attribute of src stream
      submit (src,filtered);    // emit resulting stream
    }
  } 
}