private Map<String, Pair<Long, ByteBuffer>> processRow(
      final Row row, final RowCollector rowCollector, final AtomicReference<Exception> exception) {
    final Map<String, Pair<Long, ByteBuffer>> resultMap = new HashMap<>();

    final long timestamp = row.getTime();
    for (int index = 0, size = row.size(); index < size; ++index) {
      // Do not calculate null values
      if (row.isNull(index)) {
        continue;
      }

      // All the timeSeries we stored are without database here if the parameters "outputDatabase"
      // is configured, because we do not support the same timeSeries (all the same except database)
      // in that mode, without the database we can save space and prevent string replacing problems.
      final String timeSeries =
          (outputDatabaseWithPathSeparator.isEmpty()
                  ? row.getDeviceId()
                  : row.getDeviceId().replaceFirst(databaseWithPathSeparator, ""))
              + TsFileConstant.PATH_SEPARATOR
              + row.getColumnName(index);

      final AtomicReference<TimeSeriesRuntimeState> stateReference =
          pipeName2timeSeries2TimeSeriesRuntimeStateMap
              .get(pipeName)
              .computeIfAbsent(
                  timeSeries,
                  key ->
                      new AtomicReference<>(
                          new TimeSeriesRuntimeState(
                              outputName2OperatorMap,
                              intermediateResultName2OperatorSupplierMap,
                              systemParameters,
                              windowingProcessor)));

      final Pair<List<WindowOutput>, Pair<Long, ByteBuffer>> result;
      synchronized (stateReference) {
        final TimeSeriesRuntimeState state = stateReference.get();
        try {
          switch (row.getDataType(index)) {
            case BOOLEAN:
              result =
                  state.updateWindows(
                      timestamp, row.getBoolean(index), outputMinReportIntervalMilliseconds);
              break;
            case INT32:
              result =
                  state.updateWindows(
                      timestamp, row.getInt(index), outputMinReportIntervalMilliseconds);
              break;
            case DATE:
              result =
                  state.updateWindows(
                      timestamp, row.getDate(index), outputMinReportIntervalMilliseconds);
              break;
            case INT64:
            case TIMESTAMP:
              result =
                  state.updateWindows(
                      timestamp, row.getLong(index), outputMinReportIntervalMilliseconds);
              break;
            case FLOAT:
              result =
                  state.updateWindows(
                      timestamp, row.getFloat(index), outputMinReportIntervalMilliseconds);
              break;
            case DOUBLE:
              result =
                  state.updateWindows(
                      timestamp, row.getDouble(index), outputMinReportIntervalMilliseconds);
              break;
            case TEXT:
            case STRING:
              result =
                  state.updateWindows(
                      timestamp, row.getString(index), outputMinReportIntervalMilliseconds);
              break;
            case BLOB:
              result =
                  state.updateWindows(
                      timestamp, row.getBinary(index), outputMinReportIntervalMilliseconds);
              break;
            default:
              throw new UnsupportedOperationException(
                  String.format("The type %s is not supported", row.getDataType(index)));
          }
          if (Objects.nonNull(result)) {
            collectWindowOutputs(result.getLeft(), timeSeries, rowCollector);
            if (Objects.nonNull(result.getRight())) {
              resultMap.put(timeSeries, result.getRight());
            }
          }
        } catch (final IOException | UnsupportedOperationException e) {
          exception.set(e);
        }
      }
    }
    return resultMap;
  }
{@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