@Override
  public void process(
      final TabletInsertionEvent tabletInsertionEvent, final EventCollector eventCollector)
      throws Exception {
    if (!(tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent)
        && !(tabletInsertionEvent instanceof PipeRawTabletInsertionEvent)) {
      eventCollector.collect(tabletInsertionEvent);
      return;
    }

    pipeName2LastValueReceiveTimeMap
        .computeIfAbsent(pipeName, key -> new AtomicLong(System.currentTimeMillis()))
        .set(System.currentTimeMillis());

    final AtomicReference<Exception> exception = new AtomicReference<>();
    final TimeWindowStateProgressIndex[] progressIndex = {
      new TimeWindowStateProgressIndex(new ConcurrentHashMap<>())
    };

    final Iterable<TabletInsertionEvent> outputEvents =
        tabletInsertionEvent.processRowByRow(
            (row, rowCollector) ->
                progressIndex[0] =
                    (TimeWindowStateProgressIndex)
                        progressIndex[0].updateToMinimumEqualOrIsAfterProgressIndex(
                            new TimeWindowStateProgressIndex(
                                processRow(row, rowCollector, exception))));

    // Must reset progressIndex before collection
    ((EnrichedEvent) tabletInsertionEvent).bindProgressIndex(progressIndex[0]);

    outputEvents.forEach(
        event -> {
          try {
            eventCollector.collect(event);
          } catch (Exception e) {
            exception.set(e);
          }
        });

    if (Objects.nonNull(exception.get())) {
      throw exception.get();
    }
  }
{@link AggregateProcessor} is a {@link PipeProcessor} that can adopt different implementations of
{@link AbstractWindowingProcessor} as windowing strategy and use calculation methods from all the
{@link AbstractOperatorProcessor}s to calculate the given operators. Both the {@link
AbstractWindowingProcessor} and {@link AbstractOperatorProcessor} can be implemented by user and
loaded as a normal {@link PipeProcessor}
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java