[IOTDB-1131] Improve wal by using differential logging

server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java

@@ -436,6 +436,7 @@ private Map<Long, List<TsFileResource>> splitResourcesByPartition(
 
   /** recover from file */
   private void recover() throws StorageGroupProcessorException {
+    long recoverStart = System.currentTimeMillis();
     logger.info(
         String.format(
             "start recovering virtual storage group %s[%s]",
@@ -464,12 +465,16 @@ private void recover() throws StorageGroupProcessorException {
           splitResourcesByPartition(tmpSeqTsFiles);
       Map<Long, List<TsFileResource>> partitionTmpUnseqTsFiles =
           splitResourcesByPartition(tmpUnseqTsFiles);
+      long innerStart = System.currentTimeMillis();
       for (List<TsFileResource> value : partitionTmpSeqTsFiles.values()) {
         recoverTsFiles(value, true);
       }
+      logger.info("Recovery of seq files costs {}ms", System.currentTimeMillis() - innerStart);
+      innerStart = System.currentTimeMillis();
       for (List<TsFileResource> value : partitionTmpUnseqTsFiles.values()) {
         recoverTsFiles(value, false);
       }
+      logger.info("Recovery of unseq files costs {}ms", System.currentTimeMillis() - innerStart);
 
       String taskName =
           logicalStorageGroupName + "-" + virtualStorageGroupId + "-" + System.currentTimeMillis();
@@ -487,6 +492,7 @@ private void recover() throws StorageGroupProcessorException {
               taskName,
               IoTDBDescriptor.getInstance().getConfig().isForceFullMerge(),
               logicalStorageGroupName);
+      innerStart = System.currentTimeMillis();
       logger.info(
           "{} - {} a RecoverMergeTask {} starts...",
           logicalStorageGroupName,
@@ -497,6 +503,12 @@ private void recover() throws StorageGroupProcessorException {
       if (!IoTDBDescriptor.getInstance().getConfig().isContinueMergeAfterReboot()) {
         mergingMods.delete();
       }
+      logger.info(
+          "{} - {} a RecoverMergeTask {} ends after {}ms",
+          logicalStorageGroupName,
+          virtualStorageGroupId,
+          taskName,
+          System.currentTimeMillis() - innerStart);
       recoverCompaction();
       for (TsFileResource resource : tsFileManagement.getTsFileList(true)) {
         long partitionNum = resource.getTimePartition();
@@ -543,11 +555,11 @@ private void recover() throws StorageGroupProcessorException {
             timePartitionId, IoTDBDescriptor.getInstance().getConfig().isForceFullMerge());
       }
     }
-
     logger.info(
-        String.format(
-            "the virtual storage group %s[%s] is recovered successfully",
-            logicalStorageGroupName, virtualStorageGroupId));
+        "the virtual storage group {}[{}] is recovered successfully after {}ms",
+        logicalStorageGroupName,
+        virtualStorageGroupId,
+        System.currentTimeMillis() - recoverStart);
   }
 
   private void recoverCompaction() {
@@ -753,6 +765,7 @@ private void recoverTsFiles(List<TsFileResource> tsFiles, boolean isSeq) {
 
       RestorableTsFileIOWriter writer;
       try {
+        long start = System.currentTimeMillis();
         // this tsfile is not zero level, no need to perform redo wal
         if (LevelCompactionTsFileManagement.getMergeLevel(tsFileResource.getTsFile()) > 0) {
           writer =
@@ -768,6 +781,10 @@ private void recoverTsFiles(List<TsFileResource> tsFiles, boolean isSeq) {
           writer =
               recoverPerformer.recover(true, this::getWalDirectByteBuffer, this::releaseWalBuffer);
         }
+        logger.debug(
+            "Recovery of {} costs {}ms",
+            tsFileResource.getTsFile(),
+            System.currentTimeMillis() - start);
       } catch (StorageGroupProcessorException e) {
         logger.warn(
             "Skip TsFile: {} because of error in recover: ", tsFileResource.getTsFilePath(), e);

server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java

@@ -68,6 +68,7 @@
 import org.apache.iotdb.db.qp.physical.sys.StartTriggerPlan;
 import org.apache.iotdb.db.qp.physical.sys.StopTriggerPlan;
 import org.apache.iotdb.db.qp.physical.sys.StorageGroupMNodePlan;
+import org.apache.iotdb.db.utils.datastructure.RandomAccessArrayDeque;
 import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
 
 import java.io.DataOutputStream;
@@ -111,10 +112,6 @@ protected PhysicalPlan(boolean isQuery, Operator.OperatorType operatorType) {
     this.operatorType = operatorType;
   }
 
-  public String printQueryPlan() {
-    return "abstract plan";
-  }
-
   public abstract List<PartialPath> getPaths();
 
   public void setPaths(List<PartialPath> paths) {}
@@ -181,6 +178,26 @@ public void deserialize(ByteBuffer buffer) throws IllegalPathException, IOExcept
     throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
   }
 
+  /**
+   * Serialize the plan into the given buffer. This is provided for WAL, so fields that can be
+   * recovered will not be serialized.
+   *
+   * @param buffer
+   */
+  public void serialize(ByteBuffer buffer, PhysicalPlan base) {
+    throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
+  }
+
+  /**
+   * Deserialize the plan from the given buffer. This is provided for WAL, and must be used with
+   * serializeToWAL.
+   *
+   * @param buffer
+   */
+  public void deserialize(ByteBuffer buffer, PhysicalPlan base) {
+    throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
+  }
+
   protected void putString(ByteBuffer buffer, String value) {
     if (value == null) {
       buffer.putInt(NULL_VALUE_LEN);
@@ -237,18 +254,24 @@ public void setLoginUserName(String loginUserName) {
 
   public static class Factory {
 
+    public static final String UNRECOGNIZED_LOG_TYPE = "unrecognized log type ";
+
     private Factory() {
       // hidden initializer
     }
 
     public static PhysicalPlan create(ByteBuffer buffer) throws IOException, IllegalPathException {
       int typeNum = buffer.get();
       if (typeNum >= PhysicalPlanType.values().length) {
-        throw new IOException("unrecognized log type " + typeNum);
+        throw new IOException(UNRECOGNIZED_LOG_TYPE + typeNum);
       }
       PhysicalPlanType type = PhysicalPlanType.values()[typeNum];
+      return create(type, buffer);
+    }
+
+    public static PhysicalPlan create(PhysicalPlanType type, ByteBuffer buffer)
+        throws IllegalPathException, IOException {
       PhysicalPlan plan;
-      // TODO-Cluster: support more plans
       switch (type) {
         case INSERT:
           plan = new InsertRowPlan();
@@ -419,11 +442,52 @@ public static PhysicalPlan create(ByteBuffer buffer) throws IOException, Illegal
           plan = new SetSystemModePlan();
           break;
         default:
-          throw new IOException("unrecognized log type " + type);
+          throw new IOException(UNRECOGNIZED_LOG_TYPE + type);
       }
       plan.deserialize(buffer);
       return plan;
     }
+
+    public static PhysicalPlan create(
+        ByteBuffer buffer, RandomAccessArrayDeque<PhysicalPlan> planWindow)
+        throws IOException, IllegalPathException {
+      short baseIndex = buffer.getShort();
+      int typeNum = buffer.get();
+      if (typeNum >= PhysicalPlanType.values().length) {
+        throw new IOException(UNRECOGNIZED_LOG_TYPE + typeNum);
+      }
+      PhysicalPlanType type = PhysicalPlanType.values()[typeNum];
+      PhysicalPlan plan;
+      switch (type) {
+        case INSERT:
+          plan = new InsertRowPlan();
+          if (baseIndex < 0) {
+            plan.deserialize(buffer);
+          } else {
+            InsertRowPlan baseInsertRowPlan = (InsertRowPlan) getPlan(planWindow, baseIndex);
+            plan.deserialize(buffer, baseInsertRowPlan);
+          }
+          break;
+        case BATCHINSERT:
+          plan = new InsertTabletPlan();
+          if (baseIndex < 0) {
+            plan.deserialize(buffer);
+          } else {
+            InsertTabletPlan baseInsertTabletPlan =
+                (InsertTabletPlan) getPlan(planWindow, baseIndex);
+            plan.deserialize(buffer, baseInsertTabletPlan);
+          }
+          break;
+        default:
+          plan = create(type, buffer);
+      }
+      return plan;
+    }
+
+    private static PhysicalPlan getPlan(
+        RandomAccessArrayDeque<PhysicalPlan> planWindow, int index) {
+      return planWindow.get(index);
+    }
   }
 
   /** If you want to add new PhysicalPlanType, you must add it in the last. */
@@ -495,6 +559,62 @@ public void setIndex(long index) {
     this.index = index;
   }
 
+  protected void putDiffTime(long time, long base, ByteBuffer buffer) {
+    long timeDiff = time - base;
+    TimeDiffType diffType = TimeDiffType.fromDiff(timeDiff);
+    buffer.put((byte) diffType.ordinal());
+    switch (diffType) {
+      case INT:
+        buffer.putInt((int) timeDiff);
+        break;
+      case BYTE:
+        buffer.put((byte) timeDiff);
+        break;
+      case SHORT:
+        buffer.putShort((short) timeDiff);
+        break;
+      case LONG:
+      default:
+        // NOTICE HERE
+        buffer.putLong(time);
+    }
+  }
+
+  protected long getDiffTime(ByteBuffer buffer, long base) {
+    TimeDiffType diffType = TimeDiffType.values()[buffer.get()];
+    switch (diffType) {
+      case INT:
+        return buffer.getInt() + base;
+      case BYTE:
+        return buffer.get() + base;
+      case SHORT:
+        return buffer.getShort() + base;
+      case LONG:
+      default:
+        // NOTICE HERE
+        return buffer.getLong();
+    }
+  }
+
+  public enum TimeDiffType {
+    BYTE,
+    SHORT,
+    INT,
+    LONG;
+
+    public static TimeDiffType fromDiff(long timeDiff) {
+      if (Byte.MIN_VALUE <= timeDiff && timeDiff <= Byte.MAX_VALUE) {
+        return BYTE;
+      } else if (Short.MIN_VALUE <= timeDiff && timeDiff <= Short.MAX_VALUE) {
+        return SHORT;
+      } else if (Integer.MIN_VALUE <= timeDiff && timeDiff <= Integer.MAX_VALUE) {
+        return INT;
+      } else {
+        return LONG;
+      }
+    }
+  }
+
   /**
    * Check the integrity of the plan in case that the plan is generated by a careless user through
    * Session API.