Fix empty tsfile and resource generated when insert, load, kill -9 and restart (#16215)

* Fix empty tsfile and resource generated when insert, load, kill -9 and restart

* Fix more

* Add IT

Author: HTHou

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

@@ -24,9 +24,13 @@
 import org.apache.iotdb.db.it.utils.TestUtils;
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.it.utils.TsFileGenerator;
 import org.apache.iotdb.itbase.category.ClusterIT;
 import org.apache.iotdb.itbase.category.LocalStandaloneIT;
 
+import org.apache.commons.io.FileUtils;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.schema.MeasurementSchema;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -35,10 +39,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.File;
+import java.nio.file.Files;
 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Collections;
 
 import static org.apache.iotdb.db.utils.constant.TestConstant.TIMESTAMP_STR;
 import static org.junit.Assert.assertEquals;
@@ -371,4 +378,56 @@ public void testRecoverFromFlushMemTableError() throws Exception {
       }
     }
   }
+
+  @Test
+  public void testInsertLoadAndRecover() throws Exception {
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+      statement.execute("create timeseries root.sg.d1.s1 with datatype=int32");
+      statement.execute("insert into root.sg.d1(time,s1) values(2,2)");
+      statement.execute("flush");
+    }
+    File tmpDir = new File(Files.createTempDirectory("load").toUri());
+    File tsfile = new File(tmpDir, "0-0-0-0.tsfile");
+    try {
+      try (final TsFileGenerator generator = new TsFileGenerator(tsfile)) {
+        generator.registerTimeseries(
+            "root.sg.d1", Collections.singletonList(new MeasurementSchema("s1", TSDataType.INT32)));
+        generator.generateData("root.sg.d1", 1, 2, false);
+      }
+      try (Connection connection = EnvFactory.getEnv().getConnection();
+          Statement statement = connection.createStatement()) {
+        statement.execute("insert into root.sg.d1(time,s1) values(1,1)");
+        statement.execute(String.format("load \"%s\" ", tsfile.getAbsolutePath()));
+        try (ResultSet resultSet = statement.executeQuery("select s1 from root.sg.d1")) {
+          assertNotNull(resultSet);
+          int cnt = 0;
+          while (resultSet.next()) {
+            assertEquals(String.valueOf(cnt + 1), resultSet.getString(1));
+            cnt++;
+          }
+          assertEquals(2, cnt);
+        }
+      }
+
+      // restart dn
+      TestUtils.stopForciblyAndRestartDataNodes();
+
+      try (Connection connection = EnvFactory.getEnv().getConnection();
+          Statement statement = connection.createStatement()) {
+        try (ResultSet resultSet = statement.executeQuery("select s1 from root.sg.d1")) {
+          assertNotNull(resultSet);
+          int cnt = 0;
+          while (resultSet.next()) {
+            assertEquals(String.valueOf(cnt + 1), resultSet.getString(1));
+            cnt++;
+          }
+          assertEquals(2, cnt);
+        }
+      }
+
+    } finally {
+      FileUtils.deleteDirectory(tmpDir);
+    }
+  }
 }

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

@@ -581,9 +581,10 @@ private void recover() throws DataRegionException {
           }
         }
       }
-      for (List<TsFileResource> value : partitionTmpUnseqTsFiles.values()) {
+      for (List<TsFileResource> unseqTsFiles : partitionTmpUnseqTsFiles.values()) {
+        List<TsFileResource> unsealedTsFiles = new ArrayList<>();
         // tsFiles without resource file are unsealed
-        for (TsFileResource resource : value) {
+        for (TsFileResource resource : unseqTsFiles) {
           if (resource.resourceFileExists()) {
             FileMetrics.getInstance()
                 .addTsFile(
@@ -592,6 +593,13 @@ private void recover() throws DataRegionException {
                     resource.getTsFile().length(),
                     false,
                     resource.getTsFile().getName());
+          } else {
+            WALRecoverListener recoverListener =
+                recoverUnsealedTsFile(resource, dataRegionRecoveryContext, false);
+            if (recoverListener != null) {
+              recoverListeners.add(recoverListener);
+            }
+            unsealedTsFiles.add(resource);
           }
           if (ModificationFile.getExclusiveMods(resource.getTsFile()).exists()) {
             // update mods file metrics
@@ -600,19 +608,7 @@ private void recover() throws DataRegionException {
             resource.upgradeModFile(upgradeModFileThreadPool);
           }
         }
-        while (!value.isEmpty()) {
-          TsFileResource tsFileResource = value.get(value.size() - 1);
-          if (tsFileResource.resourceFileExists()) {
-            break;
-          } else {
-            value.remove(value.size() - 1);
-            WALRecoverListener recoverListener =
-                recoverUnsealedTsFile(tsFileResource, dataRegionRecoveryContext, false);
-            if (recoverListener != null) {
-              recoverListeners.add(recoverListener);
-            }
-          }
-        }
+        unseqTsFiles.removeAll(unsealedTsFiles);
       }
       // signal wal recover manager to recover this region's files
       WALRecoverManager.getInstance().getAllDataRegionScannedLatch().countDown();
@@ -964,7 +960,12 @@ private void recoverSealedTsFiles(
         new SealedTsFileRecoverPerformer(sealedTsFile)) {
       recoverPerformer.recover();
       sealedTsFile.close();
-      tsFileResourceManager.registerSealedTsFileResource(sealedTsFile);
+      if (!TsFileValidator.getInstance().validateTsFile(sealedTsFile)) {
+        sealedTsFile.remove();
+        tsFileManager.remove(sealedTsFile, sealedTsFile.isSeq());
+      } else {
+        tsFileResourceManager.registerSealedTsFileResource(sealedTsFile);
+      }
     } catch (Throwable e) {
       logger.error("Fail to recover sealed TsFile {}, skip it.", sealedTsFile.getTsFilePath(), e);
     } finally {
@@ -1075,7 +1076,9 @@ private void syncRecoverFilesInPartition(
                     lastFlushTimeMap.getMemSize(partitionId)));
       }
       for (TsFileResource tsFileResource : resourceList) {
-        updateDeviceLastFlushTime(tsFileResource);
+        if (!tsFileResource.isDeleted()) {
+          updateDeviceLastFlushTime(tsFileResource);
+        }
       }
       TimePartitionManager.getInstance()
           .updateAfterFlushing(