diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java index 9edb53ed932b7..bf9e658f565ee 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java +++ b/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; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 311190f018107..1e63a2d4af39f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/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 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(), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java index e32789776479e..0ed563d52b58b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java +++ b/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() diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java index c43b2265b17e3..1b4b32b1283d5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java +++ b/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 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() { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java index 2146808222cc2..d5094fa668841 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java +++ b/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 {