Test shared mod file

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java

@@ -158,6 +158,21 @@ public void testUnsupportedValueFilter() throws SQLException {
             "701: The column 'attr1' does not exist or is not a tag column", e.getMessage());
       }
 
+      try {
+        statement.execute("DELETE FROM vehicle1  WHERE s1 = 'text'");
+        fail("should not reach here!");
+      } catch (SQLException e) {
+        assertEquals("701: The column 's1' does not exist or is not a tag column", e.getMessage());
+      }
+
+      try {
+        statement.execute("DELETE FROM vehicle1  WHERE attr1 = 'text'");
+        fail("should not reach here!");
+      } catch (SQLException e) {
+        assertEquals(
+            "701: The column 'attr1' does not exist or is not a tag column", e.getMessage());
+      }
+
       try {
         statement.execute("DELETE FROM vehicle1  WHERE s3 = 'text'");
         fail("should not reach here!");
@@ -299,8 +314,7 @@ public void testDelAfterFlush() throws SQLException {
       statement.execute("CREATE DATABASE ln3");
       statement.execute("use ln3");
       statement.execute(
-          String.format(
-              "CREATE TABLE vehicle3(deviceId STRING TAG, s0 INT32 FIELD, s1 INT64 FIELD, s2 FLOAT FIELD, s3 TEXT FIELD, s4 BOOLEAN FIELD)"));
+          "CREATE TABLE vehicle3(deviceId STRING TAG, s0 INT32 FIELD, s1 INT64 FIELD, s2 FLOAT FIELD, s3 TEXT FIELD, s4 BOOLEAN FIELD)");
 
       statement.execute(
           "INSERT INTO vehicle3(time, deviceId, s4) " + "values(1509465600000, 'd0', true)");
@@ -1117,8 +1131,8 @@ public void testConcurrentFlushAndRandomDeletionWithRestart()
     AtomicLong deletedPointCounter = new AtomicLong(0);
     ExecutorService writeDeletionThreadPool = Executors.newCachedThreadPool();
     ExecutorService restartThreadPool = Executors.newCachedThreadPool();
-    int fileNumMax = 1000;
-    int pointPerFile = 1000;
+    int fileNumMax = 100;
+    int pointPerFile = 100;
     int deviceNum = 4;
     Future<Void> writeThread =
         writeDeletionThreadPool.submit(
@@ -1145,7 +1159,7 @@ public void testConcurrentFlushAndRandomDeletionWithRestart()
                     deletionRange,
                     minIntervalToRecord,
                     testNum));
-    int restartTargetPointWritten = 100000;
+    int restartTargetPointWritten = 5000;
     Future<Void> restartThread =
         restartThreadPool.submit(
             () -> restart(writtenPointCounter, restartTargetPointWritten, writeDeletionThreadPool));

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java

@@ -121,27 +121,29 @@ public void testRestart() throws SQLException {
     }
   }
 
-  @Ignore // data deletion
+  @Ignore
   @Test
   public void testRestartDelete() throws SQLException {
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
         Statement statement = connection.createStatement()) {
+      statement.execute("create database test");
       statement.execute("use \"test\"");
-      statement.execute("insert into root.turbine.d1(time,s1) values(1,1)");
-      statement.execute("insert into root.turbine.d1(time,s1) values(2,2)");
-      statement.execute("insert into root.turbine.d1(time,s1) values(3,3)");
+      statement.execute("create table turbine (id1 string id, s1 float measurement)");
+      statement.execute("insert into turbine(id1, time,s1) values('d1', 1,1.0)");
+      statement.execute("insert into turbine(id1, time,s1) values('d1', 2,2.0)");
+      statement.execute("insert into turbine(id1, time,s1) values('d1', 3,3.0)");
     }
 
     TestUtils.restartDataNodes();
 
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
         Statement statement = connection.createStatement()) {
       statement.execute("use \"test\"");
-      statement.execute("delete from root.turbine.d1.s1 where time<=1");
+      statement.execute("delete from turbine where time<=1");
 
-      ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.turbine.d1");
+      ResultSet resultSet = statement.executeQuery("SELECT Time,s1 FROM turbine");
       assertNotNull(resultSet);
-      String[] exp = new String[] {"2,2.0", "3,3.0"};
+      String[] exp = new String[] {"1970-01-01T00:00:00.002Z,2.0", "1970-01-01T00:00:00.003Z,3.0"};
       int cnt = 0;
       try {
         while (resultSet.next()) {
@@ -151,10 +153,10 @@ public void testRestartDelete() throws SQLException {
         }
 
         statement.execute("flush");
-        statement.execute("delete from root.turbine.d1.s1 where time<=2");
+        statement.execute("delete from turbine where time<=2");
 
-        exp = new String[] {"3,3.0"};
-        resultSet = statement.executeQuery("SELECT s1 FROM root.turbine.d1");
+        exp = new String[] {"1970-01-01T00:00:00.003Z,3.0"};
+        resultSet = statement.executeQuery("SELECT Time,s1 FROM turbine");
         assertNotNull(resultSet);
         cnt = 0;
         while (resultSet.next()) {

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

@@ -1222,6 +1222,15 @@ public class IoTDBConfig {
 
   private CompressionType WALCompressionAlgorithm = CompressionType.LZ4;
 
+  /** the number of shared mod files in each level of each partition * */
+  private int levelModFileNumThreshold = 30;
+
+  /**
+   * when the size of a shared mod file reaches this value, new mod file will be alloacted to new
+   * TsFiles as long as the number of shared mod files does not exceed levelModFileNumThreshold.*
+   */
+  private long singleModFileSizeThresholdByte = 16 * 1024L;
+
   IoTDBConfig() {}
 
   public int getMaxLogEntriesNumPerBatch() {
@@ -4334,4 +4343,20 @@ public CompressionType getWALCompressionAlgorithm() {
   public void setWALCompressionAlgorithm(CompressionType WALCompressionAlgorithm) {
     this.WALCompressionAlgorithm = WALCompressionAlgorithm;
   }
+
+  public long getSingleModFileSizeThresholdByte() {
+    return singleModFileSizeThresholdByte;
+  }
+
+  public void setSingleModFileSizeThresholdByte(long singleModFileSizeThresholdByte) {
+    this.singleModFileSizeThresholdByte = singleModFileSizeThresholdByte;
+  }
+
+  public int getLevelModFileNumThreshold() {
+    return levelModFileNumThreshold;
+  }
+
+  public void setLevelModFileNumThreshold(int levelModFileNumThreshold) {
+    this.levelModFileNumThreshold = levelModFileNumThreshold;
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBDataRegionAirGapConnector.java

@@ -35,6 +35,7 @@
 import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
+import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
 import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
 import org.apache.iotdb.pipe.api.annotation.TableModel;
 import org.apache.iotdb.pipe.api.annotation.TreeModel;
@@ -45,6 +46,7 @@
 import org.apache.iotdb.pipe.api.exception.PipeException;
 import org.apache.iotdb.rpc.TSStatusCode;
 
+import org.apache.tsfile.utils.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -297,6 +299,50 @@ private void doTransferWrapper(
     }
   }
 
+  /**
+   * Combine the exclusive mod file and the shared mod file of the sender as the receiver's
+   * exclusive mod file.
+   *
+   * @return the combined mod file and its length
+   */
+  private Pair<File, Long> doTransferModFile(
+      final AirGapSocket socket, final PipeTsFileInsertionEvent pipeTsFileInsertionEvent)
+      throws IOException {
+    final String pipeName = pipeTsFileInsertionEvent.getPipeName();
+    final long creationTime = pipeTsFileInsertionEvent.getCreationTime();
+    final File tsFile = pipeTsFileInsertionEvent.getTsFile();
+    File targetModFile = ModificationFile.getExclusiveMods(tsFile);
+    long lengthSent = 0;
+    if (pipeTsFileInsertionEvent.isWithExclusiveMod()) {
+      transferFilePieces(
+          pipeName,
+          creationTime,
+          pipeTsFileInsertionEvent.getExclusiveModFile(),
+          0,
+          targetModFile,
+          0,
+          socket,
+          true);
+      lengthSent = pipeTsFileInsertionEvent.getExclusiveModFile().length();
+    }
+
+    if (pipeTsFileInsertionEvent.isWithSharedMod()) {
+      transferFilePieces(
+          pipeName,
+          creationTime,
+          pipeTsFileInsertionEvent.getSharedModFile(),
+          pipeTsFileInsertionEvent.getSharedModFileOffset(),
+          targetModFile,
+          lengthSent,
+          socket,
+          true);
+      lengthSent +=
+          pipeTsFileInsertionEvent.getSharedModFile().length()
+              - pipeTsFileInsertionEvent.getSharedModFileOffset();
+    }
+    return new Pair<>(targetModFile, lengthSent);
+  }
+
   private void doTransfer(
       final AirGapSocket socket, final PipeTsFileInsertionEvent pipeTsFileInsertionEvent)
       throws PipeException, IOException {
@@ -306,18 +352,19 @@ private void doTransfer(
     final String errorMessage = String.format("Seal file %s error. Socket %s.", tsFile, socket);
 
     // 1. Transfer file piece by piece, and mod if needed
-    if (pipeTsFileInsertionEvent.isWithMod() && supportModsIfIsDataNodeReceiver) {
-      final File modFile = pipeTsFileInsertionEvent.getModFile();
-      transferFilePieces(pipeName, creationTime, modFile, socket, true);
+    boolean modFileExists =
+        pipeTsFileInsertionEvent.isWithExclusiveMod() || pipeTsFileInsertionEvent.isWithSharedMod();
+    if (modFileExists && supportModsIfIsDataNodeReceiver) {
+      final Pair<File, Long> modFileAndLength = doTransferModFile(socket, pipeTsFileInsertionEvent);
       transferFilePieces(pipeName, creationTime, tsFile, socket, true);
       // 2. Transfer file seal signal with mod, which means the file is transferred completely
       if (!send(
           pipeName,
           creationTime,
           socket,
           PipeTransferTsFileSealWithModReq.toTPipeTransferBytes(
-              modFile.getName(),
-              modFile.length(),
+              modFileAndLength.getLeft().getName(),
+              modFileAndLength.getRight(),
               tsFile.getName(),
               tsFile.length(),
               pipeTsFileInsertionEvent.isTableModelEvent()