code: private void emitWindowResult(W window) throws Exception {
  		BaseRow aggResult = windowFunction.getWindowAggregationResult(window);
  		if (sendRetraction) {
  			previousState.setCurrentNamespace(window);
  			BaseRow previousAggResult = previousState.value();
  
  			// has emitted result for the window
  			if (previousAggResult != null) {
  				// current agg is not equal to the previous emitted, should emit retract
  				if (!equaliser.equalsWithoutHeader(aggResult, previousAggResult)) {
  					reuseOutput.replace((BaseRow) getCurrentKey(), previousAggResult);
  					BaseRowUtil.setRetract(reuseOutput);
  					// send retraction
  					collector.collect(reuseOutput);
  					// send accumulate
  					reuseOutput.replace((BaseRow) getCurrentKey(), aggResult);
  					BaseRowUtil.setAccumulate(reuseOutput);
  					collector.collect(reuseOutput);
  					// update previousState
  					previousState.update(aggResult);
  				}
  				// if the previous agg equals to the current agg, no need to send retract and accumulate
  			}
  			// the first fire for the window, only send accumulate
  			else {
  				// send accumulate
  				reuseOutput.replace((BaseRow) getCurrentKey(), aggResult);
  				BaseRowUtil.setAccumulate(reuseOutput);
  				collector.collect(reuseOutput);
  				// update previousState
  				previousState.update(aggResult);
  			}
  		} else {
  			reuseOutput.replace((BaseRow) getCurrentKey(), aggResult);
  			// no need to set header
  			collector.collect(reuseOutput);
  		}
  	}
docs: Emits the window result of the given window.