diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java index b4a9e78222a90..0a959488a4326 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java @@ -765,7 +765,7 @@ public boolean persistAllProgressIndex(final long timeoutInMs) { persistThread.setDaemon(true); LOGGER.info( - "Start to persist all pipe progress indexes during shutdown, pipe count {}, timeout {} ms.", + "Start to persist all pipe progress indexes during shutdown, pipe count: {}, timeout: {} ms.", getPipeCount(), normalizedTimeoutInMs); persistThread.start(); @@ -816,7 +816,8 @@ private boolean persistAllProgressIndexInternal() { .mapToInt(ByteBuffer::remaining) .sum(); LOGGER.info( - "Collected pipe metas for shutdown progress persist, pipe count {}, pipe meta count {}, pipe meta size {} bytes, cost {} ms.", + "Collected pipe metas for shutdown progress persist, local pipe count: {}, " + + "collected pipe meta count: {}, total serialized size: {} bytes, cost: {} ms.", pipeCount, pipeMetaCount, pipeMetaSizeInBytes, @@ -825,7 +826,8 @@ private boolean persistAllProgressIndexInternal() { if (resp.getPipeMetaList().isEmpty()) { if (pipeCount != 0) { LOGGER.info( - "Collected empty pipe metas during shutdown while pipe count is {}.", pipeCount); + "Collected empty pipe metas during shutdown while local pipe count is {}.", + pipeCount); return false; } return true; @@ -834,7 +836,8 @@ private boolean persistAllProgressIndexInternal() { try (final ConfigNodeClient configNodeClient = ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { LOGGER.info( - "Start to push heartbeat shutdown pipe meta to ConfigNode, data node id {}, pipe count {}, pipe meta count {}, pipe meta size {} bytes.", + "Start to push shutdown pipe meta heartbeat to ConfigNode, dataNodeId: {}, " + + "local pipe count: {}, collected pipe meta count: {}, total serialized size: {} bytes.", IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), pipeCount, pipeMetaCount, @@ -845,20 +848,21 @@ private boolean persistAllProgressIndexInternal() { IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), resp); final long pushCostTime = System.currentTimeMillis() - pushStartTime; if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != result.getCode()) { - LOGGER.warn("Failed to persist progress index to ConfigNode, status: {}", result); + LOGGER.warn("Failed to persist progress index to configNode, status: {}", result); LOGGER.warn( - "Failed to push heartbeat shutdown pipe meta to ConfigNode, status {}, cost {} ms.", + "Failed to push shutdown pipe meta heartbeat to ConfigNode, status: {}, cost {} ms.", result, pushCostTime); return false; } else { LOGGER.info( - "Successfully finished pushing heartbeat shutdown pipe meta to ConfigNode, pipe count {}, pipe meta count {}, pipe meta size {} bytes, cost {} ms.", + "Successfully pushed shutdown pipe meta heartbeat to ConfigNode, local pipe count: {}, " + + "collected pipe meta count: {}, total serialized size: {} bytes, cost {} ms.", pipeCount, pipeMetaCount, pipeMetaSizeInBytes, pushCostTime); - LOGGER.info("Successfully persisted all pipe's info to ConfigNode."); + LOGGER.info("Successfully persisted all pipe's info to configNode."); return true; } } 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 1d984f9e18f45..cfb268cb95ffe 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 @@ -406,15 +406,17 @@ public synchronized void start() { originalUnSequenceTsFileCount, startIndex); + final HistoricalTsFileExtractionStatistics statistics = + new HistoricalTsFileExtractionStatistics(); final Collection sequenceTsFileResources = tsFileManager.getTsFileList(true).stream() .peek(originalResourceList::add) - .filter(this::shouldExtractTsFileResource) + .filter(resource -> shouldExtractTsFileResource(resource, statistics)) .collect(Collectors.toList()); final Collection unSequenceTsFileResources = tsFileManager.getTsFileList(false).stream() .peek(originalResourceList::add) - .filter(this::shouldExtractTsFileResource) + .filter(resource -> shouldExtractTsFileResource(resource, statistics)) .collect(Collectors.toList()); filteredTsFileResources.addAll(sequenceTsFileResources); filteredTsFileResources.addAll(unSequenceTsFileResources); @@ -428,6 +430,7 @@ public synchronized void start() { .pinTsFileResource(resource, shouldTransferModFile, pipeName); return false; } catch (final IOException e) { + ++statistics.pinFailedCount; LOGGER.warn("Pipe: failed to pin TsFileResource {}", resource.getTsFilePath(), e); return true; } @@ -456,6 +459,22 @@ public synchronized void start() { filteredTsFileResources.size(), originalSequenceTsFileCount + originalUnSequenceTsFileCount, System.currentTimeMillis() - startHistoricalExtractionTime); + LOGGER.info( + "Pipe {}@{}: historical TsFile selection summary, selected by progress uncovered {}, " + + "selected by unclosed/closing {}, filtered by time/path {} (time {}, path {}), " + + "skipped covered {}, skipped deleted {}, skipped generated by pipe {}, " + + "pin failed {}", + pipeName, + dataRegionId, + statistics.selectedByProgressUncoveredCount, + statistics.selectedByUnclosedOrClosingCount, + statistics.filteredByTimeOrPathCount, + statistics.filteredByTimeCount, + statistics.filteredByPathCount, + statistics.skippedCoveredCount, + statistics.skippedDeletedCount, + statistics.skippedGeneratedByPipeCount, + statistics.pinFailedCount); } finally { tsFileManager.readUnlock(); } @@ -567,43 +586,54 @@ private int compareTsFileResourcesByQueryPriority( return resource1.getTsFilePath().compareTo(resource2.getTsFilePath()); } - private boolean shouldExtractTsFileResource(final TsFileResource resource) { + private boolean shouldExtractTsFileResource( + final TsFileResource resource, final HistoricalTsFileExtractionStatistics statistics) { if (!isHistoricalSourceEnabled) { return false; } // Some resource is marked as deleted but not removed from the list. if (resource.isDeleted()) { + ++statistics.skippedDeletedCount; return false; } // Some resource is generated by pipe. We ignore them if the pipe should not transfer pipe // requests. if (resource.isGeneratedByPipe() && !isForwardingPipeRequests) { + ++statistics.skippedGeneratedByPipeCount; return false; } - // Some resource may not be closed due to the control of PIPE_MIN_FLUSH_INTERVAL_IN_MS. We - // simply ignore them. + // If the tsFile is not already marked closing, it is not captured by the pipe realtime module. + // Thus, we can wait for the realtime sync module to handle this, to avoid blocking the pipe + // sync process. if (!resource.isClosed() && Optional.ofNullable(resource.getProcessor()) .map(TsFileProcessor::alreadyMarkedClosing) .orElse(true)) { + ++statistics.selectedByUnclosedOrClosingCount; return true; } if (!mayTsFileContainUnprocessedData(resource)) { + ++statistics.skippedCoveredCount; return false; } if (!isTsFileResourceOverlappedWithTimeRange(resource)) { + ++statistics.filteredByTimeOrPathCount; + ++statistics.filteredByTimeCount; return false; } if (!mayTsFileResourceOverlappedWithPattern(resource)) { + ++statistics.filteredByTimeOrPathCount; + ++statistics.filteredByPathCount; return false; } + ++statistics.selectedByProgressUncoveredCount; return true; } @@ -678,6 +708,19 @@ private TimeWindowStateProgressIndex getTimeWindowStateProgressIndex( : progressIndex.getProgressIndexByType(TimeWindowStateProgressIndex.class).orElse(null); } + private static class HistoricalTsFileExtractionStatistics { + + private int selectedByProgressUncoveredCount; + private int selectedByUnclosedOrClosingCount; + private int filteredByTimeOrPathCount; + private int filteredByTimeCount; + private int filteredByPathCount; + private int skippedCoveredCount; + private int skippedDeletedCount; + private int skippedGeneratedByPipeCount; + private int pinFailedCount; + } + private boolean mayTsFileResourceOverlappedWithPattern(final TsFileResource resource) { // Trimming to avoid unnecessary file device getter if (isDbNameCoveredByPattern) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java index 76966820abb87..5b2887108f630 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java @@ -151,7 +151,7 @@ public void run() { final long shutdownProgressPersistTimeoutInMs = PipeDataNodeAgent.task().getShutdownProgressPersistTimeoutInMs(); logger.info( - "Persisting pipe progress indexes before shutdown, timeout {} ms.", + "Persisting pipe progress indexes before shutdown with timeout {} ms.", shutdownProgressPersistTimeoutInMs); if (!PipeDataNodeAgent.task().persistAllProgressIndex(shutdownProgressPersistTimeoutInMs)) { logger.warn("Pipe progress indexes were not confirmed during shutdown."); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java index 7562e54e5811b..2184889f3c329 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java @@ -39,7 +39,6 @@ import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.apache.iotdb.pipe.api.event.Event; -import com.google.common.collect.ImmutableMap; import org.apache.tsfile.file.metadata.PlainDeviceID; import org.junit.Assert; import org.junit.Test; @@ -104,21 +103,41 @@ public void testMayTsFileContainUnprocessedDataUsesEqualOrAfterCoverage() throws tempDir, "superset.tsfile", hybridProgressIndex( - new IoTProgressIndex(ImmutableMap.of(1, 100L, 2, 200L)), + iotProgressIndex(1, 100L, 2, 200L), new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))), hybridProgressIndex( - new IoTProgressIndex(1, 100L), + iotProgressIndex(1, 100L), new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 9))), false); assertMayTsFileContainUnprocessedData( tempDir, "missing-dimension.tsfile", - hybridProgressIndex(new IoTProgressIndex(1, 100L)), + hybridProgressIndex(iotProgressIndex(1, 100L)), hybridProgressIndex( - new IoTProgressIndex(1, 90L), + iotProgressIndex(1, 90L), new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))), true); + + assertMayTsFileContainUnprocessedData( + tempDir, + "larger-iot.tsfile", + hybridProgressIndex( + iotProgressIndex(1, 100L, 2, 200L), + new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))), + hybridProgressIndex( + iotProgressIndex(1, 101L), + new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))), + true); + + final ProgressIndex recoverProgressIndex = + new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10)); + assertMayTsFileContainUnprocessedData( + tempDir, + "old-sequence-recover.tsfile", + hybridProgressIndex(recoverProgressIndex, iotProgressIndex(1, 100L)), + recoverProgressIndex, + false); } finally { FileUtils.deleteFileOrDirectory(tempDir); } @@ -543,6 +562,16 @@ private static ProgressIndex hybridProgressIndex( return result; } + private static IoTProgressIndex iotProgressIndex( + final int firstPeerId, final long firstSearchIndex, final long... peerIdAndSearchIndexPairs) { + final Map peerId2SearchIndex = new HashMap<>(); + peerId2SearchIndex.put(firstPeerId, firstSearchIndex); + for (int i = 0; i < peerIdAndSearchIndexPairs.length; i += 2) { + peerId2SearchIndex.put((int) peerIdAndSearchIndexPairs[i], peerIdAndSearchIndexPairs[i + 1]); + } + return new IoTProgressIndex(peerId2SearchIndex); + } + private static void sortExtractedResources( final PipeHistoricalDataRegionTsFileSource source, final List resources) throws ReflectiveOperationException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexKeeperTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexKeeperTest.java index bfd45b708b620..d7dcc30beacdc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexKeeperTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexKeeperTest.java @@ -35,6 +35,8 @@ import java.io.File; import java.io.IOException; import java.nio.file.Files; +import java.util.HashMap; +import java.util.Map; public class PipeTsFileEpochProgressIndexKeeperTest { @@ -97,8 +99,7 @@ public void testProgressIndexCheckDoesNotLeakAcrossTaskScopes() throws IOExcepti public void testProgressIndexCheckUsesEqualOrAfterCoverage() throws IOException { final ProgressIndex registeredProgressIndex = hybridProgressIndex( - new IoTProgressIndex(1, 90L), - new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))); + iotProgressIndex(1, 90L), new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))); keeper.registerProgressIndex( DATA_REGION_ID, TASK_SCOPE_A, @@ -106,7 +107,7 @@ public void testProgressIndexCheckUsesEqualOrAfterCoverage() throws IOException Assert.assertFalse( keeper.isProgressIndexAfterOrEquals( - DATA_REGION_ID, TASK_SCOPE_A, "current.tsfile", new IoTProgressIndex(1, 100L))); + DATA_REGION_ID, TASK_SCOPE_A, "current.tsfile", iotProgressIndex(1, 100L))); Assert.assertTrue( keeper.isProgressIndexAfterOrEquals( @@ -114,7 +115,7 @@ public void testProgressIndexCheckUsesEqualOrAfterCoverage() throws IOException TASK_SCOPE_A, "current.tsfile", hybridProgressIndex( - new IoTProgressIndex(1, 100L), + iotProgressIndex(1, 100L), new RecoverProgressIndex(-1, new SimpleProgressIndex(0, 10))))); } @@ -157,4 +158,10 @@ private ProgressIndex hybridProgressIndex( } return result; } + + private IoTProgressIndex iotProgressIndex(final int peerId, final long searchIndex) { + final Map peerId2SearchIndex = new HashMap<>(); + peerId2SearchIndex.put(peerId, searchIndex); + return new IoTProgressIndex(peerId2SearchIndex); + } }