Introduce delay analyzer to get watermark between sequence data and unsequece data based on ICDE 2022

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -394,6 +394,24 @@ public class IoTDBConfig {
   /** The interval to check whether unsequence memtables need flushing. Unit: ms */
   private long unseqMemtableFlushCheckInterval = 30 * 1000L;
 
+  /**
+   * Whether to enable delay analyzer for tracking data arrival delays and calculating safe
+   * watermark
+   */
+  private boolean enableDelayAnalyzer = false;
+
+  /** Default window size for delay analyzer, empirical value: 10000 sample points */
+  private int delayAnalyzerWindowSize = 10000;
+
+  /** Minimum window size for delay analyzer validation */
+  private int delayAnalyzerMinWindowSize = 1000;
+
+  /** Maximum window size for delay analyzer validation */
+  private int delayAnalyzerMaxWindowSize = 100000;
+
+  /** Default confidence level for delay analyzer: 99% */
+  private double delayAnalyzerConfidenceLevel = 0.99;
+
   /** The sort algorithm used in TVList */
   private TVListSortAlgorithm tvListSortAlgorithm = TVListSortAlgorithm.TIM;
 
@@ -2183,6 +2201,46 @@ public void setUnseqMemtableFlushCheckInterval(long unseqMemtableFlushCheckInter
     this.unseqMemtableFlushCheckInterval = unseqMemtableFlushCheckInterval;
   }
 
+  public boolean isEnableDelayAnalyzer() {
+    return enableDelayAnalyzer;
+  }
+
+  public void setEnableDelayAnalyzer(boolean enableDelayAnalyzer) {
+    this.enableDelayAnalyzer = enableDelayAnalyzer;
+  }
+
+  public int getDelayAnalyzerWindowSize() {
+    return delayAnalyzerWindowSize;
+  }
+
+  public void setDelayAnalyzerWindowSize(int delayAnalyzerWindowSize) {
+    this.delayAnalyzerWindowSize = delayAnalyzerWindowSize;
+  }
+
+  public int getDelayAnalyzerMinWindowSize() {
+    return delayAnalyzerMinWindowSize;
+  }
+
+  public void setDelayAnalyzerMinWindowSize(int delayAnalyzerMinWindowSize) {
+    this.delayAnalyzerMinWindowSize = delayAnalyzerMinWindowSize;
+  }
+
+  public int getDelayAnalyzerMaxWindowSize() {
+    return delayAnalyzerMaxWindowSize;
+  }
+
+  public void setDelayAnalyzerMaxWindowSize(int delayAnalyzerMaxWindowSize) {
+    this.delayAnalyzerMaxWindowSize = delayAnalyzerMaxWindowSize;
+  }
+
+  public double getDelayAnalyzerConfidenceLevel() {
+    return delayAnalyzerConfidenceLevel;
+  }
+
+  public void setDelayAnalyzerConfidenceLevel(double delayAnalyzerConfidenceLevel) {
+    this.delayAnalyzerConfidenceLevel = delayAnalyzerConfidenceLevel;
+  }
+
   public TVListSortAlgorithm getTvListSortAlgorithm() {
     return tvListSortAlgorithm;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -1119,6 +1119,52 @@ public void loadProperties(TrimProperties properties) throws BadNodeUrlException
             properties.getProperty(
                 "include_null_value_in_write_throughput_metric",
                 String.valueOf(conf.isIncludeNullValueInWriteThroughputMetric()))));
+
+    conf.setEnableDelayAnalyzer(
+        Boolean.parseBoolean(
+            properties.getProperty(
+                "enable_delay_analyzer",
+                ConfigurationFileUtils.getConfigurationDefaultValue("enable_delay_analyzer"))));
+
+    int delayAnalyzerWindowSize =
+        Integer.parseInt(
+            properties.getProperty(
+                "delay_analyzer_window_size",
+                ConfigurationFileUtils.getConfigurationDefaultValue("delay_analyzer_window_size")));
+    if (delayAnalyzerWindowSize > 0) {
+      LOGGER.warn("[DelayAnalyzer] Set delay_analyzer_window_size to {}", delayAnalyzerWindowSize);
+      conf.setDelayAnalyzerWindowSize(delayAnalyzerWindowSize);
+    }
+
+    int delayAnalyzerMinWindowSize =
+        Integer.parseInt(
+            properties.getProperty(
+                "delay_analyzer_min_window_size",
+                ConfigurationFileUtils.getConfigurationDefaultValue(
+                    "delay_analyzer_min_window_size")));
+    if (delayAnalyzerMinWindowSize > 0) {
+      conf.setDelayAnalyzerMinWindowSize(delayAnalyzerMinWindowSize);
+    }
+
+    int delayAnalyzerMaxWindowSize =
+        Integer.parseInt(
+            properties.getProperty(
+                "delay_analyzer_max_window_size",
+                ConfigurationFileUtils.getConfigurationDefaultValue(
+                    "delay_analyzer_max_window_size")));
+    if (delayAnalyzerMaxWindowSize > 0) {
+      conf.setDelayAnalyzerMaxWindowSize(delayAnalyzerMaxWindowSize);
+    }
+
+    double delayAnalyzerConfidenceLevel =
+        Double.parseDouble(
+            properties.getProperty(
+                "delay_analyzer_confidence_level",
+                ConfigurationFileUtils.getConfigurationDefaultValue(
+                    "delay_analyzer_confidence_level")));
+    if (delayAnalyzerConfidenceLevel > 0 && delayAnalyzerConfidenceLevel <= 1) {
+      conf.setDelayAnalyzerConfidenceLevel(delayAnalyzerConfidenceLevel);
+    }
   }
 
   private void loadFixedSizeLimitForQuery(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

@@ -152,6 +152,7 @@
 import org.apache.iotdb.db.storageengine.rescon.memory.TimePartitionManager;
 import org.apache.iotdb.db.storageengine.rescon.memory.TsFileResourceManager;
 import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeSpaceQuotaManager;
+import org.apache.iotdb.db.tools.DelayAnalyzer;
 import org.apache.iotdb.db.tools.settle.TsFileAndModSettleTool;
 import org.apache.iotdb.db.utils.CommonUtils;
 import org.apache.iotdb.db.utils.DateTimeUtils;
@@ -369,6 +370,9 @@ public class DataRegion implements IDataRegionForQuery {
   private ILoadDiskSelector ordinaryLoadDiskSelector;
   private ILoadDiskSelector pipeAndIoTV2LoadDiskSelector;
 
+  /** Delay analyzer for tracking data arrival delays and calculating safe watermarks */
+  private final DelayAnalyzer delayAnalyzer;
+
   /**
    * Construct a database processor.
    *
@@ -387,6 +391,14 @@ public DataRegion(
     this.dataRegionId = new DataRegionId(Integer.parseInt(dataRegionIdString));
     this.databaseName = databaseName;
     this.fileFlushPolicy = fileFlushPolicy;
+    this.delayAnalyzer =
+        config.isEnableDelayAnalyzer()
+            ? new DelayAnalyzer(
+                config.getDelayAnalyzerWindowSize(),
+                config.getDelayAnalyzerMinWindowSize(),
+                config.getDelayAnalyzerMaxWindowSize(),
+                config.getDelayAnalyzerConfidenceLevel())
+            : null;
     acquireDirectBufferMemory();
 
     dataRegionSysDir = SystemFileFactory.INSTANCE.getFile(systemDir, dataRegionIdString);
@@ -453,6 +465,14 @@ public DataRegion(String databaseName, String dataRegionIdString) {
     partitionMaxFileVersions.put(0L, 0L);
     upgradeModFileThreadPool = null;
     this.metrics = new DataRegionMetrics(this);
+    this.delayAnalyzer =
+        config.isEnableDelayAnalyzer()
+            ? new DelayAnalyzer(
+                config.getDelayAnalyzerWindowSize(),
+                config.getDelayAnalyzerMinWindowSize(),
+                config.getDelayAnalyzerMaxWindowSize(),
+                config.getDelayAnalyzerConfidenceLevel())
+            : null;
 
     initDiskSelector();
   }
@@ -1151,6 +1171,12 @@ public void insert(InsertRowNode insertRowNode) throws WriteProcessException {
       if (deleted) {
         return;
       }
+      if (delayAnalyzer != null) {
+        long arrivalTime = System.currentTimeMillis();
+        long generationTime = insertRowNode.getTime();
+        delayAnalyzer.update(generationTime, arrivalTime);
+      }
+
       // init map
       long timePartitionId = TimePartitionUtils.getTimePartitionId(insertRowNode.getTime());
       initFlushTimeMap(timePartitionId);
@@ -1315,6 +1341,13 @@ public void insertTablet(InsertTabletNode insertTabletNode)
             "Won't insert tablet {}, because region is deleted", insertTabletNode.getSearchIndex());
         return;
       }
+      if (delayAnalyzer != null) {
+        long arrivalTime = System.currentTimeMillis();
+        long[] times = insertTabletNode.getTimes();
+        for (long generationTime : times) {
+          delayAnalyzer.update(generationTime, arrivalTime);
+        }
+      }
       TSStatus[] results = new TSStatus[insertTabletNode.getRowCount()];
       Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
       long[] infoForMetrics = new long[5];
@@ -4108,6 +4141,10 @@ public void insert(InsertRowsOfOneDeviceNode insertRowsOfOneDeviceNode)
       Map<TsFileProcessor, InsertRowsNode> tsFileProcessorMap = new HashMap<>();
       for (int i = 0; i < insertRowsOfOneDeviceNode.getInsertRowNodeList().size(); i++) {
         InsertRowNode insertRowNode = insertRowsOfOneDeviceNode.getInsertRowNodeList().get(i);
+        if (delayAnalyzer != null) {
+          long arrivalTime = System.currentTimeMillis();
+          delayAnalyzer.update(insertRowNode.getTime(), arrivalTime);
+        }
         if (!CommonUtils.isAlive(insertRowNode.getTime(), ttl)) {
           // we do not need to write these part of data, as they can not be queried
           // or the sub-plan has already been executed, we are retrying other sub-plans
@@ -4225,6 +4262,10 @@ public void insert(InsertRowsNode insertRowsNode)
       long[] timePartitionIds = new long[insertRowsNode.getInsertRowNodeList().size()];
       for (int i = 0; i < insertRowsNode.getInsertRowNodeList().size(); i++) {
         InsertRowNode insertRowNode = insertRowsNode.getInsertRowNodeList().get(i);
+        if (delayAnalyzer != null) {
+          long arrivalTime = System.currentTimeMillis();
+          delayAnalyzer.update(insertRowNode.getTime(), arrivalTime);
+        }
         long ttl = getTTL(insertRowNode);
         if (!CommonUtils.isAlive(insertRowNode.getTime(), ttl)) {
           insertRowsNode
@@ -4313,9 +4354,16 @@ public void insertTablets(InsertMultiTabletsNode insertMultiTabletsNode)
       // infoForMetrics[1]: ScheduleMemoryBlockTimeCost
       // infoForMetrics[2]: ScheduleWalTimeCost
       // infoForMetrics[3]: ScheduleMemTableTimeCost
-      // infoForMetrics[4]: InsertedPointsNumber
+      // infoForMetrics[4]: InsertedPointsNumbe
       for (int i = 0; i < insertMultiTabletsNode.getInsertTabletNodeList().size(); i++) {
         InsertTabletNode insertTabletNode = insertMultiTabletsNode.getInsertTabletNodeList().get(i);
+        long[] times = insertTabletNode.getTimes();
+        if (delayAnalyzer != null) {
+          for (long generationTime : times) {
+            long arrivalTime = System.currentTimeMillis();
+            delayAnalyzer.update(generationTime, arrivalTime);
+          }
+        }
         TSStatus[] results = new TSStatus[insertTabletNode.getRowCount()];
         Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
         boolean noFailure = false;
@@ -4619,6 +4667,16 @@ public TsFileManager getTsFileManager() {
     return tsFileManager;
   }
 
+  /**
+   * Get the delay analyzer instance for this data region
+   *
+   * @return DelayAnalyzer instance for tracking data arrival delays and calculating safe
+   *     watermarks, or null if delay analyzer is disabled
+   */
+  public DelayAnalyzer getDelayAnalyzer() {
+    return delayAnalyzer;
+  }
+
   private long getTTL(InsertNode insertNode) {
     if (insertNode.getTableName() == null) {
       return DataNodeTTLCache.getInstance().getTTLForTree(insertNode.getTargetPath().getNodes());