@Override
  protected void processRow(
      Row row,
      RowCollector rowCollector,
      String deviceSuffix,
      AtomicReference<Exception> exception) {
    final PipeRemarkableRow remarkableRow = new PipeRemarkableRow((PipeRow) row);

    boolean hasNonNullMeasurements = false;
    for (int i = 0, size = row.size(); i < size; i++) {
      if (row.isNull(i)) {
        continue;
      }

      final String timeSeriesSuffix =
          deviceSuffix + TsFileConstant.PATH_SEPARATOR + row.getColumnName(i);
      final ChangingValueFilter filter =
          pathLastObjectCache.getPartialPathLastObject(timeSeriesSuffix);

      if (filter != null) {
        if (filter.filter(row.getTime(), row.getObject(i))) {
          hasNonNullMeasurements = true;
        } else {
          remarkableRow.markNull(i);
        }
      } else {
        hasNonNullMeasurements = true;
        pathLastObjectCache.setPartialPathLastObject(
            timeSeriesSuffix, new ChangingValueFilter<>(this, row.getTime(), row.getObject(i)));
      }
    }

    if (hasNonNullMeasurements) {
      try {
        rowCollector.collectRow(remarkableRow);
      } catch (IOException e) {
        exception.set(e);
      }
    }
  }
The maximum time distance between two stored data points if current point time to the last
stored point time distance >= compressionMaxTimeInterval, current point will be stored
regardless of compression deviation
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/changing/ChangingValueSamplingProcessor.java