Optimize write metric update for insertTablets and insertRelationalTablet

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java

@@ -169,6 +169,9 @@ public TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion
       dataRegion.insertTablets(node);
       dataRegion.insertSeparatorToWAL();
       return StatusUtils.OK;
+    } catch (WriteProcessRejectException e) {
+      LOGGER.warn("Reject in executing plan node: {}, caused by {}", node, e.getMessage());
+      return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
     } catch (BatchProcessException e) {
       LOGGER.warn("Batch failure in executing a InsertMultiTabletsNode.");
       TSStatus firstStatus = null;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/PayloadFormatManager.java

@@ -30,6 +30,7 @@
 import java.io.IOException;
 import java.net.URL;
 import java.net.URLClassLoader;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -91,7 +92,7 @@ private static void buildMqttPluginMap() throws IOException {
     }
 
     URL[] jarURLs = getPluginJarURLs(mqttDir);
-    logger.debug("MQTT Plugin jarURLs: {}", jarURLs);
+    logger.debug("MQTT Plugin jarURLs: {}", Arrays.toString(jarURLs));
 
     for (URL jarUrl : jarURLs) {
       ClassLoader classLoader = new URLClassLoader(new URL[] {jarUrl});

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

@@ -1071,7 +1071,11 @@ private long getLastFlushTime(long timePartitionID, IDeviceID deviceID) {
   }
 
   private boolean splitAndInsert(
-      InsertTabletNode insertTabletNode, int loc, int endOffset, TSStatus[] results) {
+      InsertTabletNode insertTabletNode,
+      int loc,
+      int endOffset,
+      TSStatus[] results,
+      long[] costsForMetrics) {
     boolean noFailure = true;
 
     // before is first start point
@@ -1103,7 +1107,8 @@ private boolean splitAndInsert(
                     isSequence,
                     results,
                     beforeTimePartition,
-                    noFailure)
+                    noFailure,
+                    costsForMetrics)
                 && noFailure;
         if (before < loc) {
           insertCnt += 1;
@@ -1130,7 +1135,8 @@ private boolean splitAndInsert(
                       isSequence,
                       results,
                       beforeTimePartition,
-                      noFailure)
+                      noFailure,
+                      costsForMetrics)
                   && noFailure;
           before = loc;
           if (before < loc) {
@@ -1159,7 +1165,8 @@ private boolean splitAndInsert(
                   isSequence,
                   results,
                   beforeTimePartition,
-                  noFailure)
+                  noFailure,
+                  costsForMetrics)
               && noFailure;
       insertCnt += 1;
       logger.debug(
@@ -1193,26 +1200,13 @@ public void insertTablet(InsertTabletNode insertTabletNode)
       }
       TSStatus[] results = new TSStatus[insertTabletNode.getRowCount()];
       Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
-      boolean noFailure;
-      int loc = insertTabletNode.checkTTL(results, i -> getTTL(insertTabletNode));
-      noFailure = loc == 0;
-      List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
-          insertTabletNode.splitByDevice(loc, insertTabletNode.getRowCount());
-      int start = loc;
-      for (Pair<IDeviceID, Integer> deviceEndOffsetPair : deviceEndOffsetPairs) {
-        int end = deviceEndOffsetPair.getRight();
-        noFailure = noFailure && splitAndInsert(insertTabletNode, start, end, results);
-        start = end;
-      }
+      long[] costsForMetrics = new long[4];
+      boolean noFailure = executeInsertTablet(insertTabletNode, results, costsForMetrics);
 
-      if (CommonDescriptor.getInstance().getConfig().isLastCacheEnable()
-          && !insertTabletNode.isGeneratedByRemoteConsensusLeader()) {
-        // disable updating last cache on follower
-        startTime = System.nanoTime();
-        tryToUpdateInsertTabletLastCache(insertTabletNode);
-        PERFORMANCE_OVERVIEW_METRICS.recordScheduleUpdateLastCacheCost(
-            System.nanoTime() - startTime);
-      }
+      PERFORMANCE_OVERVIEW_METRICS.recordCreateMemtableBlockCost(costsForMetrics[0]);
+      PERFORMANCE_OVERVIEW_METRICS.recordScheduleMemoryBlockCost(costsForMetrics[1]);
+      PERFORMANCE_OVERVIEW_METRICS.recordScheduleWalCost(costsForMetrics[2]);
+      PERFORMANCE_OVERVIEW_METRICS.recordScheduleMemTableCost(costsForMetrics[3]);
 
       if (!noFailure) {
         throw new BatchProcessException(results);
@@ -1222,6 +1216,31 @@ public void insertTablet(InsertTabletNode insertTabletNode)
     }
   }
 
+  private boolean executeInsertTablet(
+      InsertTabletNode insertTabletNode, TSStatus[] results, long[] costsForMetrics)
+      throws OutOfTTLException {
+    boolean noFailure;
+    int loc = insertTabletNode.checkTTL(results, i -> getTTL(insertTabletNode));
+    noFailure = loc == 0;
+    List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
+        insertTabletNode.splitByDevice(loc, insertTabletNode.getRowCount());
+    int start = loc;
+    for (Pair<IDeviceID, Integer> deviceEndOffsetPair : deviceEndOffsetPairs) {
+      int end = deviceEndOffsetPair.getRight();
+      noFailure =
+          noFailure && splitAndInsert(insertTabletNode, start, end, results, costsForMetrics);
+      start = end;
+    }
+    if (CommonDescriptor.getInstance().getConfig().isLastCacheEnable()
+        && !insertTabletNode.isGeneratedByRemoteConsensusLeader()) {
+      // disable updating last cache on follower
+      long startTime = System.nanoTime();
+      tryToUpdateInsertTabletLastCache(insertTabletNode);
+      PERFORMANCE_OVERVIEW_METRICS.recordScheduleUpdateLastCacheCost(System.nanoTime() - startTime);
+    }
+    return noFailure;
+  }
+
   private void initFlushTimeMap(long timePartitionId) {
     if (config.isEnableSeparateData()
         && !lastFlushTimeMap.checkAndCreateFlushedTimePartition(timePartitionId, true)) {
@@ -1301,7 +1320,8 @@ private boolean insertTabletToTsFileProcessor(
       boolean sequence,
       TSStatus[] results,
       long timePartitionId,
-      boolean noFailure) {
+      boolean noFailure,
+      long[] costsForMetrics) {
     // return when start >= end or all measurement failed
     if (start >= end || insertTabletNode.allMeasurementFailed()) {
       if (logger.isDebugEnabled()) {
@@ -1328,7 +1348,8 @@ private boolean insertTabletToTsFileProcessor(
     registerToTsFile(insertTabletNode, tsFileProcessor);
 
     try {
-      tsFileProcessor.insertTablet(insertTabletNode, start, end, results, noFailure);
+      tsFileProcessor.insertTablet(
+          insertTabletNode, start, end, results, noFailure, costsForMetrics);
     } catch (WriteProcessRejectException e) {
       logger.warn("insert to TsFileProcessor rejected, {}", e.getMessage());
       return false;
@@ -1424,7 +1445,7 @@ private List<InsertRowNode> insertToTsFileProcessors(
       TsFileProcessor tsFileProcessor = entry.getKey();
       InsertRowsNode subInsertRowsNode = entry.getValue();
       try {
-        tsFileProcessor.insert(subInsertRowsNode, costsForMetrics);
+        tsFileProcessor.insertRows(subInsertRowsNode, costsForMetrics);
       } catch (WriteProcessException e) {
         insertRowsNode
             .getResults()
@@ -3492,7 +3513,7 @@ public void insert(InsertRowsOfOneDeviceNode insertRowsOfOneDeviceNode)
         TsFileProcessor tsFileProcessor = entry.getKey();
         InsertRowsNode subInsertRowsNode = entry.getValue();
         try {
-          tsFileProcessor.insert(subInsertRowsNode, costsForMetrics);
+          tsFileProcessor.insertRows(subInsertRowsNode, costsForMetrics);
         } catch (WriteProcessException e) {
           insertRowsOfOneDeviceNode
               .getResults()
@@ -3604,30 +3625,56 @@ public void insert(InsertRowsNode insertRowsNode)
    * @param insertMultiTabletsNode batch of tablets belongs to multiple devices
    */
   public void insertTablets(InsertMultiTabletsNode insertMultiTabletsNode)
-      throws BatchProcessException {
-    for (int i = 0; i < insertMultiTabletsNode.getInsertTabletNodeList().size(); i++) {
-      InsertTabletNode insertTabletNode = insertMultiTabletsNode.getInsertTabletNodeList().get(i);
-      try {
-        insertTablet(insertTabletNode);
-      } catch (WriteProcessException e) {
-        insertMultiTabletsNode
-            .getResults()
-            .put(i, RpcUtils.getStatus(e.getErrorCode(), e.getMessage()));
-      } catch (BatchProcessException e) {
-        // for each error
-        TSStatus firstStatus = null;
-        for (TSStatus status : e.getFailingStatus()) {
-          if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-            firstStatus = status;
-          }
-          // return WRITE_PROCESS_REJECT directly for the consensus retry logic
-          if (status.getCode() == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) {
-            insertMultiTabletsNode.getResults().put(i, status);
-            throw new BatchProcessException("Rejected inserting multi tablets");
+      throws BatchProcessException, WriteProcessRejectException {
+
+    StorageEngine.blockInsertionIfReject();
+    long startTime = System.nanoTime();
+    writeLock("insertTablets");
+    PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() - startTime);
+    try {
+      if (deleted) {
+        logger.info(
+            "Won't insert tablets {}, because region is deleted",
+            insertMultiTabletsNode.getSearchIndex());
+        return;
+      }
+      long[] costsForMetrics = new long[4];
+      for (int i = 0; i < insertMultiTabletsNode.getInsertTabletNodeList().size(); i++) {
+        InsertTabletNode insertTabletNode = insertMultiTabletsNode.getInsertTabletNodeList().get(i);
+        TSStatus[] results = new TSStatus[insertTabletNode.getRowCount()];
+        Arrays.fill(results, RpcUtils.SUCCESS_STATUS);
+        boolean noFailure = false;
+        try {
+          noFailure = executeInsertTablet(insertTabletNode, results, costsForMetrics);
+        } catch (WriteProcessException e) {
+          insertMultiTabletsNode
+              .getResults()
+              .put(i, RpcUtils.getStatus(e.getErrorCode(), e.getMessage()));
+        }
+        if (!noFailure) {
+          // for each error
+          TSStatus firstStatus = null;
+          for (TSStatus status : results) {
+            if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+              firstStatus = status;
+            }
+            // return WRITE_PROCESS_REJECT directly for the consensus retry logic
+            if (status.getCode() == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) {
+              insertMultiTabletsNode.getResults().put(i, status);
+              throw new BatchProcessException("Rejected inserting multi tablets");
+            }
           }
+          insertMultiTabletsNode.getResults().put(i, firstStatus);
         }
-        insertMultiTabletsNode.getResults().put(i, firstStatus);
       }
+
+      PERFORMANCE_OVERVIEW_METRICS.recordCreateMemtableBlockCost(costsForMetrics[0]);
+      PERFORMANCE_OVERVIEW_METRICS.recordScheduleMemoryBlockCost(costsForMetrics[1]);
+      PERFORMANCE_OVERVIEW_METRICS.recordScheduleWalCost(costsForMetrics[2]);
+      PERFORMANCE_OVERVIEW_METRICS.recordScheduleMemTableCost(costsForMetrics[3]);
+
+    } finally {
+      writeUnlock();
     }
 
     if (!insertMultiTabletsNode.getResults().isEmpty()) {

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

@@ -351,7 +351,7 @@ public void insert(InsertRowNode insertRowNode, long[] costsForMetrics)
     costsForMetrics[3] += System.nanoTime() - startTime;
   }
 
-  public void insert(InsertRowsNode insertRowsNode, long[] costsForMetrics)
+  public void insertRows(InsertRowsNode insertRowsNode, long[] costsForMetrics)
       throws WriteProcessException {
 
     ensureMemTable(costsForMetrics);
@@ -449,14 +449,20 @@ private void createNewWorkingMemTable() {
     walNode.onMemTableCreated(workMemTable, tsFileResource.getTsFilePath());
   }
 
-  private long[] checkMemCost(
-      InsertTabletNode insertTabletNode, int start, int end, TSStatus[] results, boolean noFailure)
+  private long[] scheduleMemoryBlock(
+      InsertTabletNode insertTabletNode,
+      int start,
+      int end,
+      TSStatus[] results,
+      boolean noFailure,
+      long[] costsForMetrics)
       throws WriteProcessException {
     long[] memIncrements;
     try {
-      long startTime = System.nanoTime();
+      long memControlStartTime = System.nanoTime();
       memIncrements = checkMemCost(insertTabletNode, start, end, noFailure, results);
-      PERFORMANCE_OVERVIEW_METRICS.recordScheduleMemoryBlockCost(System.nanoTime() - startTime);
+      // recordScheduleMemoryBlockCost
+      costsForMetrics[1] += System.nanoTime() - memControlStartTime;
     } catch (WriteProcessException e) {
       for (int i = start; i < end; i++) {
         results[i] = RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_REJECT, e.getMessage());
@@ -525,18 +531,18 @@ private long[] checkAlignedMemCost(
    * @param results result array
    */
   public void insertTablet(
-      InsertTabletNode insertTabletNode, int start, int end, TSStatus[] results, boolean noFailure)
+      InsertTabletNode insertTabletNode,
+      int start,
+      int end,
+      TSStatus[] results,
+      boolean noFailure,
+      long[] costsForMetrics)
       throws WriteProcessException {
 
-    if (workMemTable == null) {
-      long startTime = System.nanoTime();
-      createNewWorkingMemTable();
-      PERFORMANCE_OVERVIEW_METRICS.recordCreateMemtableBlockCost(System.nanoTime() - startTime);
-      WritingMetrics.getInstance()
-          .recordActiveMemTableCount(dataRegionInfo.getDataRegion().getDataRegionId(), 1);
-    }
+    ensureMemTable(costsForMetrics);
 
-    long[] memIncrements = checkMemCost(insertTabletNode, start, end, results, noFailure);
+    long[] memIncrements =
+        scheduleMemoryBlock(insertTabletNode, start, end, results, noFailure, costsForMetrics);
 
     long startTime = System.nanoTime();
     WALFlushListener walFlushListener;
@@ -552,7 +558,8 @@ public void insertTablet(
       rollbackMemoryInfo(memIncrements);
       throw new WriteProcessException(e);
     } finally {
-      PERFORMANCE_OVERVIEW_METRICS.recordScheduleWalCost(System.nanoTime() - startTime);
+      // recordScheduleWalCost
+      costsForMetrics[2] += System.nanoTime() - startTime;
     }
 
     startTime = System.nanoTime();
@@ -610,7 +617,8 @@ public void insertTablet(
 
     tsFileResource.updateProgressIndex(insertTabletNode.getProgressIndex());
 
-    PERFORMANCE_OVERVIEW_METRICS.recordScheduleMemTableCost(System.nanoTime() - startTime);
+    // recordScheduleMemTableCost
+    costsForMetrics[3] += System.nanoTime() - startTime;
   }
 
   @SuppressWarnings("squid:S3776") // High Cognitive Complexity