Fix snapshots of deleted TsFile resources during DataNode removal

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java

@@ -635,7 +635,7 @@ private StorageEngineMessages() {}
   public static final String CANNOT_SNAPSHOT_UNCLOSED_TSFILE =
       "Cannot create snapshot because TsFile {} is not closed";
   public static final String HARD_LINK_TARGET_DIR_NOT_EXIST = "Hard link target dir {} doesn't exist";
-  public static final String HARD_LINK_SOURCE_FILE_NOT_EXIST = "Hard link source file {} doesn't exist, this file will be ignored.";
+  public static final String HARD_LINK_SOURCE_FILE_NOT_EXIST = "Hard link source file {} doesn't exist.";
   public static final String COPY_TARGET_DIR_NOT_EXIST = "Copy target dir {} doesn't exist";
   public static final String COPY_SOURCE_FILE_NOT_EXIST = "Copy source file {} doesn't exist";
   public static final String CANNOT_CREATE_DIRECTORY = "Cannot create directory ";

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java

@@ -616,7 +616,7 @@ private StorageEngineMessages() {}
   public static final String CANNOT_SNAPSHOT_UNCLOSED_TSFILE =
       "无法创建快照，因为 TsFile {} 尚未关闭";
   public static final String HARD_LINK_TARGET_DIR_NOT_EXIST = "硬链接目标目录 {} 不存在";
-  public static final String HARD_LINK_SOURCE_FILE_NOT_EXIST = "硬链接源文件 {} 不存在，该文件将被忽略。";
+  public static final String HARD_LINK_SOURCE_FILE_NOT_EXIST = "硬链接源文件 {} 不存在。";
   public static final String COPY_TARGET_DIR_NOT_EXIST = "复制目标目录 {} 不存在";
   public static final String COPY_SOURCE_FILE_NOT_EXIST = "复制源文件 {} 不存在";
   public static final String CANNOT_CREATE_DIRECTORY = "无法创建目录: ";

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

@@ -35,6 +35,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.util.Arrays;
@@ -226,6 +227,16 @@ private void readUnlockTheFile() {
 
   private boolean createSnapshot(List<TsFileResource> resources, String snapshotId) {
     try {
+      // Closing empty files can delete resources after readLockTheFile() captured them.
+      // Remove them from subsequent snapshot processing and balance their captured read locks.
+      resources.removeIf(
+          resource -> {
+            if (resource.isDeleted()) {
+              resource.readUnlock();
+              return true;
+            }
+            return false;
+          });
       for (TsFileResource resource : resources) {
         if (!resource.isClosed()) {
           LOGGER.error(StorageEngineMessages.CANNOT_SNAPSHOT_UNCLOSED_TSFILE, resource);
@@ -279,7 +290,7 @@ private void createHardLink(File target, File source) throws IOException {
       LOGGER.error(StorageEngineMessages.HARD_LINK_TARGET_DIR_NOT_EXIST, target.getParentFile());
     }
     if (!checkHardLinkSourceFile(source, 10)) {
-      return;
+      throw new FileNotFoundException(source.getAbsolutePath());
     }
     Files.deleteIfExists(target.toPath());
     SnapshotFileUtils.createLinkOrCopy(target.toPath(), source.toPath());

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

@@ -70,6 +70,7 @@
 
 import java.io.BufferedOutputStream;
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -332,6 +333,9 @@ private ITimeIndex getTimeIndexForSerialization() throws IOException {
   /** deserialize from disk */
   public void deserialize() throws IOException {
     try (InputStream inputStream = fsFactory.getBufferedInputStream(file + RESOURCE_SUFFIX)) {
+      if (inputStream == null) {
+        throw new FileNotFoundException(file + RESOURCE_SUFFIX);
+      }
       // The first byte is VERSION_NUMBER, second byte is timeIndexType.
       ReadWriteIOUtils.readByte(inputStream);
       timeIndex =

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java

@@ -33,6 +33,7 @@
 import org.apache.iotdb.commons.path.NonAlignedFullPath;
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.commons.utils.TimePartitionUtils;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -83,6 +84,8 @@
 import org.apache.iotdb.db.storageengine.dataregion.modification.v1.Deletion;
 import org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1;
 import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+import org.apache.iotdb.db.storageengine.dataregion.snapshot.SnapshotLoader;
+import org.apache.iotdb.db.storageengine.dataregion.snapshot.SnapshotTaker;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
@@ -614,6 +617,42 @@ private void setMemoryAdmissionTestThreshold(String name, Object value) throws E
     field.set(SystemInfo.getInstance(), value);
   }
 
+  @Test
+  public void testSnapshotAfterDeletingUnflushedData() throws Exception {
+    dataRegion.setAllowCompaction(false);
+    TSRecord record = new TSRecord(deviceId, 1);
+    record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, "10"));
+    dataRegion.insert(buildInsertRowNodeByTSRecord(record));
+    TsFileResource resource = dataRegion.getTsFileManager().getTsFileList(true).get(0);
+    Assert.assertFalse(resource.isClosed());
+    MeasurementPath path = new MeasurementPath(deviceId, measurementId);
+    dataRegion.deleteByDevice(
+        path,
+        new DeleteDataNode(
+            new PlanNodeId("delete"),
+            Collections.singletonList(path),
+            Long.MIN_VALUE,
+            Long.MAX_VALUE));
+
+    File snapshotDir = new File(TestConstant.OUTPUT_DATA_DIR, "snapshot-deleted-unflushed");
+    try {
+      Assert.assertTrue(
+          new SnapshotTaker(dataRegion).takeFullSnapshot(snapshotDir.getAbsolutePath(), true));
+      Assert.assertTrue(resource.isDeleted());
+      Assert.assertFalse(resource.tsFileExists());
+      Assert.assertTrue(resource.tryWriteLock());
+      resource.writeUnlock();
+      DataRegion loaded =
+          new SnapshotLoader(snapshotDir.getAbsolutePath(), storageGroup, "0")
+              .loadSnapshotForStateMachine();
+      Assert.assertNotNull(loaded);
+      Assert.assertTrue(loaded.getTsFileManager().getTsFileList(true).isEmpty());
+      Assert.assertTrue(loaded.getTsFileManager().getTsFileList(false).isEmpty());
+    } finally {
+      FileUtils.recursivelyDeleteFolder(snapshotDir.getAbsolutePath());
+    }
+  }
+
   @Test
   public void testUnseqUnsealedDelete()
       throws WriteProcessException, IOException, MetadataException {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/IoTDBSnapshotTest.java

@@ -148,6 +148,36 @@ public void testCreateSnapshot()
     }
   }
 
+  @Test
+  public void testRejectSnapshotWithMissingResource() throws Exception {
+    String[][] originDataDirs = IoTDBDescriptor.getInstance().getConfig().getTierDataDirs();
+    IoTDBDescriptor.getInstance().getConfig().setTierDataDirs(testDataDirs);
+    TierManager.getInstance().resetFolders();
+    File snapshotDir = new File("target/snapshot-missing-resource");
+    try {
+      List<TsFileResource> resources = writeTsFiles(testDataDirs[0]);
+      TsFileResource resource = resources.get(0);
+      Files.delete(new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).toPath());
+      DataRegion region = new DataRegion(testSgName, "0");
+      region.getTsFileManager().add(resource, true);
+      Assert.assertFalse(
+          new SnapshotTaker(region).takeFullSnapshot(snapshotDir.getAbsolutePath(), true));
+      SnapshotLogAnalyzer analyzer =
+          new SnapshotLogAnalyzer(new File(snapshotDir, SnapshotLogger.SNAPSHOT_LOG_NAME));
+      try {
+        Assert.assertFalse(analyzer.isSnapshotComplete());
+      } finally {
+        analyzer.close();
+      }
+      Assert.assertTrue(resource.tryWriteLock());
+      resource.writeUnlock();
+    } finally {
+      FileUtils.recursivelyDeleteFolder(snapshotDir.getAbsolutePath());
+      IoTDBDescriptor.getInstance().getConfig().setTierDataDirs(originDataDirs);
+      TierManager.getInstance().resetFolders();
+    }
+  }
+
   @Test
   public void testRejectSnapshotWithUnclosedTsFile()
       throws IOException, WriteProcessException, DirectoryNotLegalException {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java

@@ -99,6 +99,11 @@ public void tearDown() throws IOException {
     }
   }
 
+  @Test(expected = IOException.class)
+  public void testDeserializeMissingResource() throws IOException {
+    tsFileResource.deserialize();
+  }
+
   @Test
   public void testSerializeAndDeserialize() throws IOException {
     tsFileResource.serialize();