[To dev/1.3] Pipe: Avoided timeIndexes' memory occupation for deleted tsFiles && Removed useless stale logic && Improved the memory calculation of tsFileResource && Avoided the potential failure for permission check when the source file is deleted (#16399) (#16415)

* Pipe: Avoided timeIndexes' memory occupation for deleted tsFiles && Removed useless stale logic && Improved the memory calculation of tsFileResource && Avoided the potential failure for permission check when the source file is deleted (#16399)

* fix

* partial

* reduce unnecessary

* Fix

* may fix

* Fix

* bishop

* fix

* nonnull

* fix

* enrich

* fix

* fix

* comp

* fix

* fix

* Refactor

* fix

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java

@@ -129,12 +129,7 @@ private void parseAndCollectEvent(final PipeTsFileInsertionEvent sourceEvent) th
       return;
     }
 
-    if (skipParsing) {
-      collectEvent(sourceEvent);
-      return;
-    }
-
-    if (!forceTabletFormat && canSkipParsing4TsFileEvent(sourceEvent)) {
+    if (skipParsing || !forceTabletFormat && canSkipParsing4TsFileEvent(sourceEvent)) {
       collectEvent(sourceEvent);
       return;
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java

@@ -86,6 +86,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent
   protected long flushPointCount = TsFileProcessor.FLUSH_POINT_COUNT_NOT_SET;
 
   protected volatile ProgressIndex overridingProgressIndex;
+  private Set<String> tableNames;
 
   public PipeTsFileInsertionEvent(final TsFileResource resource, final boolean isLoaded) {
     // The modFile must be copied before the event is assigned to the listening pipes
@@ -122,6 +123,7 @@ public PipeTsFileInsertionEvent(
     this.isGeneratedByPipe = resource.isGeneratedByPipe();
     this.isGeneratedByPipeConsensus = resource.isGeneratedByPipeConsensus();
     this.isGeneratedByHistoricalExtractor = isGeneratedByHistoricalExtractor;
+    this.tableNames = tableNames;
 
     this.dataContainer = new AtomicReference<>(null);
 
@@ -394,7 +396,7 @@ public boolean mayEventPathsOverlappedWithPattern() {
               deviceID ->
                   pipePattern.mayOverlapWithDevice(((PlainDeviceID) deviceID).toStringID()));
     } catch (final Exception e) {
-      LOGGER.warn(
+      LOGGER.info(
           "Pipe {}: failed to get devices from TsFile {}, extract it anyway",
           pipeName,
           resource.getTsFilePath(),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java

@@ -111,7 +111,7 @@ public class PipeHistoricalDataRegionTsFileSource implements PipeHistoricalDataR
   private PipePattern pipePattern;
   private boolean isDbNameCoveredByPattern = false;
 
-  private boolean isHistoricalExtractorEnabled = false;
+  private boolean isHistoricalSourceEnabled = false;
   private long historicalDataExtractionStartTime = Long.MIN_VALUE; // Event time
   private long historicalDataExtractionEndTime = Long.MAX_VALUE; // Event time
 
@@ -174,7 +174,7 @@ public void validate(final PipeParameterValidator validator) {
         EXTRACTOR_START_TIME_KEY,
         SOURCE_END_TIME_KEY,
         EXTRACTOR_END_TIME_KEY)) {
-      isHistoricalExtractorEnabled = true;
+      isHistoricalSourceEnabled = true;
 
       try {
         historicalDataExtractionStartTime =
@@ -214,7 +214,7 @@ public void validate(final PipeParameterValidator validator) {
     // enabled, the pipe will lose some historical data.
     // 2. User may set the EXTRACTOR_HISTORY_START_TIME and EXTRACTOR_HISTORY_END_TIME without
     // enabling the historical data extraction, which may affect the realtime data extraction.
-    isHistoricalExtractorEnabled =
+    isHistoricalSourceEnabled =
         parameters.getBooleanOrDefault(
                 SystemConstant.RESTART_KEY, SystemConstant.RESTART_DEFAULT_VALUE)
             || parameters.getBooleanOrDefault(
@@ -382,7 +382,7 @@ public synchronized void start() {
                 .peek(originalResourceList::add)
                 .filter(
                     resource ->
-                        isHistoricalExtractorEnabled
+                        isHistoricalSourceEnabled
                             &&
                             // Some resource is marked as deleted but not removed from the list.
                             !resource.isDeleted()
@@ -408,7 +408,7 @@ && mayTsFileResourceOverlappedWithPattern(resource)))
                 .peek(originalResourceList::add)
                 .filter(
                     resource ->
-                        isHistoricalExtractorEnabled
+                        isHistoricalSourceEnabled
                             &&
                             // Some resource is marked as deleted but not removed from the list.
                             !resource.isDeleted()

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

@@ -107,7 +107,9 @@ public class TsFileResource {
   protected TsFileResource next;
 
   /** time index */
-  private ITimeIndex timeIndex;
+  private volatile ITimeIndex timeIndex;
+
+  private final AtomicReference<Boolean> isEmpty = new AtomicReference<>();
 
   @SuppressWarnings("squid:S3077")
   private volatile ModificationFile modFile;
@@ -121,11 +123,11 @@ public class TsFileResource {
   /** used for check whether this file has internal unsorted data in compaction selection */
   private TsFileRepairStatus tsFileRepairStatus = TsFileRepairStatus.NORMAL;
 
-  private TsFileLock tsFileLock = new TsFileLock();
+  private final TsFileLock tsFileLock = new TsFileLock();
 
   private boolean isSeq;
 
-  private FSFactory fsFactory = FSFactoryProducer.getFSFactory();
+  private final FSFactory fsFactory = FSFactoryProducer.getFSFactory();
 
   private DataRegion.SettleTsFileCallBack settleTsFileCallBack;
 
@@ -647,6 +649,10 @@ public void removeModFile() throws IOException {
    */
   public boolean remove() {
     forceMarkDeleted();
+    // To release the memory occupied by pipe if held by it
+    // Note that pipe can safely handle the case that the time index does not exist
+    isEmpty();
+    degradeTimeIndex();
     try {
       fsFactory.deleteIfExists(file);
       fsFactory.deleteIfExists(
@@ -935,13 +941,18 @@ public void setModFile(ModificationFile modFile) {
    * @return resource map size
    */
   public long calculateRamSize() {
+    final ProgressIndex progressIndex = maxProgressIndex.get();
     if (timeIndex.getTimeIndexType() == ITimeIndex.FILE_TIME_INDEX_TYPE) {
-      return INSTANCE_SIZE + timeIndex.calculateRamSize();
+      return INSTANCE_SIZE
+          + timeIndex.calculateRamSize()
+          + (Objects.nonNull(progressIndex) ? progressIndex.ramBytesUsed() : 0);
     }
     if (deviceTimeIndexRamSize == 0) {
       deviceTimeIndexRamSize = timeIndex.calculateRamSize();
     }
-    return INSTANCE_SIZE + deviceTimeIndexRamSize;
+    return INSTANCE_SIZE
+        + deviceTimeIndexRamSize
+        + (Objects.nonNull(progressIndex) ? progressIndex.ramBytesUsed() : 0);
   }
 
   // used for compaction
@@ -1219,7 +1230,9 @@ public ProgressIndex getMaxProgressIndex() {
   }
 
   public boolean isEmpty() {
-    return getFileStartTime() == Long.MAX_VALUE && getFileEndTime() == Long.MIN_VALUE;
+    isEmpty.compareAndSet(
+        null, getFileStartTime() == Long.MAX_VALUE && getFileEndTime() == Long.MIN_VALUE);
+    return isEmpty.get();
   }
 
   public String getDatabaseName() {

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

@@ -44,6 +44,7 @@
 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.wal.WALManager;
+import org.apache.iotdb.db.storageengine.rescon.memory.TsFileResourceManager;
 import org.apache.iotdb.db.tools.validate.TsFileValidationTool;
 import org.apache.iotdb.db.utils.EnvironmentUtils;
 import org.apache.iotdb.db.utils.constant.TestConstant;
@@ -504,6 +505,7 @@ public void tearDown() throws IOException, StorageEngineException {
     }
     registeredTimePartitionDirs.clear();
     tsFileManager.clear();
+    TsFileResourceManager.getInstance().clear();
   }
 
   private void removeFiles() throws IOException {