From c03d06ca2cd282de49a4bf6da89a7bdce3a1d906 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 12:02:47 +0800 Subject: [PATCH 1/9] Optimize pipe TsFile cleanup on drop --- .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../agent/task/PipeDataNodeTaskAgent.java | 22 ++++++++++ .../resource/tsfile/PipeTsFileResource.java | 8 +++- .../tsfile/PipeTsFileResourceManager.java | 44 ++++++++++++++++++- 5 files changed, 76 insertions(+), 2 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index bcdab68abfd17..e02ef8f64ddc8 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2487,5 +2487,7 @@ private DataNodePipeMessages() {} "Reported %s pipe metas."; public static final String MESSAGE_TRANSFER_FILE_ARG_ERROR_RESULT_STATUS_ARG_E565D9FD = "Transfer file %s error, result status %s."; + public static final String CLEANED_UP_PIPE_TSFILE_DIRS_FOR_PIPE = + "Cleaned up Pipe TsFile dirs for pipe {}: {}"; } diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 3a51486d8474f..3620a364610a3 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2325,5 +2325,7 @@ private DataNodePipeMessages() {} "已上报 %s 个 pipe meta。"; public static final String MESSAGE_TRANSFER_FILE_ARG_ERROR_RESULT_STATUS_ARG_E565D9FD = "传输文件 %s 出错，结果状态为 %s。"; + public static final String CLEANED_UP_PIPE_TSFILE_DIRS_FOR_PIPE = + "已清理 pipe {} 的 Pipe TsFile 目录：{}"; } 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 df08311151a94..d33eb86f28946 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 @@ -59,6 +59,7 @@ import org.apache.iotdb.db.pipe.metric.overview.PipeTsFileToTabletsMetrics; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager; +import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter; import org.apache.iotdb.db.pipe.source.dataregion.realtime.listener.PipeInsertionDataNodeListener; import org.apache.iotdb.db.pipe.source.schemaregion.SchemaRegionListeningFilter; @@ -304,13 +305,20 @@ protected void freezeRate(final String pipeName, final long creationTime) { @Override protected boolean dropPipe(final String pipeName, final long creationTime) { + final String pipeTsFileResourcePipeName = + PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); + PipeDataNodeResourceManager.tsfile().markPipeTsFileDirUnderDeletion(pipeTsFileResourcePipeName); + if (!super.dropPipe(pipeName, creationTime)) { + PipeDataNodeResourceManager.tsfile() + .unmarkPipeTsFileDirUnderDeletion(pipeTsFileResourcePipeName); return false; } final String taskId = pipeName + "_" + creationTime; PipeTsFileToTabletsMetrics.getInstance().deregister(taskId); PipeDataNodeSinglePipeMetrics.getInstance().deregister(taskId); + PipeDataNodeResourceManager.tsfile().cleanPipeTsFileDirAsync(pipeTsFileResourcePipeName); return true; } @@ -319,6 +327,15 @@ protected boolean dropPipe(final String pipeName, final long creationTime) { protected boolean dropPipe(final String pipeName) { // Get the pipe meta first because it is removed after super#dropPipe(pipeName) final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName); + final String pipeTsFileResourcePipeName = + Objects.isNull(pipeMeta) + ? null + : PipeTsFileResourceManager.getPipeTsFileResourcePipeName( + pipeName, pipeMeta.getStaticMeta().getCreationTime()); + if (Objects.nonNull(pipeTsFileResourcePipeName)) { + PipeDataNodeResourceManager.tsfile() + .markPipeTsFileDirUnderDeletion(pipeTsFileResourcePipeName); + } // Record whether there are pipe tasks before dropping the pipe final boolean hasPipeTasks; @@ -331,6 +348,10 @@ protected boolean dropPipe(final String pipeName) { } if (!super.dropPipe(pipeName)) { + if (Objects.nonNull(pipeTsFileResourcePipeName)) { + PipeDataNodeResourceManager.tsfile() + .unmarkPipeTsFileDirUnderDeletion(pipeTsFileResourcePipeName); + } return false; } @@ -339,6 +360,7 @@ protected boolean dropPipe(final String pipeName) { final String taskId = pipeName + "_" + creationTime; PipeTsFileToTabletsMetrics.getInstance().deregister(taskId); PipeDataNodeSinglePipeMetrics.getInstance().deregister(taskId); + PipeDataNodeResourceManager.tsfile().cleanPipeTsFileDirAsync(pipeTsFileResourcePipeName); // When the pipe contains no pipe tasks, there is no corresponding prefetching queue for the // subscribed pipe, so the subscription needs to be manually marked as completed. if (!hasPipeTasks && PipeStaticMeta.isSubscriptionPipe(pipeName)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java index 51421cdcec8da..a8a7aa7e19525 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java @@ -70,9 +70,15 @@ public void increaseReferenceCount() { } public boolean decreaseReferenceCount() { + return decreaseReferenceCount(true); + } + + public boolean decreaseReferenceCount(final boolean deleteFileWhenNoReference) { final int finalReferenceCount = referenceCount.addAndGet(-1); if (finalReferenceCount == 0) { - close(); + if (deleteFileWhenNoReference) { + close(); + } return true; } if (finalReferenceCount < 0) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java index 4fc1c3063df7c..e75a559149891 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.pipe.resource.tsfile; +import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.utils.FileUtils; @@ -39,7 +40,9 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; public class PipeTsFileResourceManager { @@ -54,6 +57,10 @@ public class PipeTsFileResourceManager { // PipeName -> TsFilePath -> PipeTsFileResource private final Map> hardlinkOrCopiedFileToPipeTsFileResourceMap = new ConcurrentHashMap<>(); + private final Map pipeNameToPipeTsFileDirPathMap = new ConcurrentHashMap<>(); + private final Set pipeNameSetUnderDeletion = ConcurrentHashMap.newKeySet(); + private final ExecutorService pipeTsFileDirCleanupExecutor = + IoTDBThreadPoolFactory.newSingleThreadExecutor("Pipe-TsFile-Dir-Cleanup"); private final PipeTsFileResourceSegmentLock segmentLock = new PipeTsFileResourceSegmentLock(); public static String getPipeTsFileResourcePipeName( @@ -126,6 +133,8 @@ private File increaseFileReference( // file in pipe dir, create a hardlink or copy it to pipe dir, maintain a reference count for // the hardlink or copied file, and return the hardlink or copied file. if (Objects.nonNull(pipeName)) { + pipeNameToPipeTsFileDirPathMap.putIfAbsent( + pipeName, hardlinkOrCopiedFile.getParentFile().getPath()); hardlinkOrCopiedFileToPipeTsFileResourceMap .computeIfAbsent(pipeName, k -> new ConcurrentHashMap<>()) .put(resultFile.getPath(), new PipeTsFileResource(resultFile)); @@ -225,7 +234,9 @@ public void decreaseFileReference( try { final String filePath = hardlinkOrCopiedFile.getPath(); final PipeTsFileResource resource = getResourceMap(pipeName).get(filePath); - if (resource != null && resource.decreaseReferenceCount()) { + if (resource != null + && resource.decreaseReferenceCount( + Objects.isNull(pipeName) || !pipeNameSetUnderDeletion.contains(pipeName))) { getResourceMap(pipeName).remove(filePath); } } finally { @@ -246,6 +257,37 @@ private void decreasePublicReferenceIfExists(final File file, final @Nullable St decreaseFileReference(new File(getCommonFilePath(file)), null); } + public void markPipeTsFileDirUnderDeletion(final @Nonnull String pipeName) { + pipeNameSetUnderDeletion.add(pipeName); + } + + public void unmarkPipeTsFileDirUnderDeletion(final @Nonnull String pipeName) { + pipeNameSetUnderDeletion.remove(pipeName); + } + + public void cleanPipeTsFileDirAsync(final @Nonnull String pipeName) { + final String pipeTsFileDirPath = pipeNameToPipeTsFileDirPathMap.remove(pipeName); + hardlinkOrCopiedFileToPipeTsFileResourceMap.remove(pipeName); + + if (Objects.isNull(pipeTsFileDirPath)) { + pipeNameSetUnderDeletion.remove(pipeName); + return; + } + + pipeTsFileDirCleanupExecutor.execute( + () -> { + try { + FileUtils.deleteFileOrDirectory(new File(pipeTsFileDirPath), true); + LOGGER.info( + DataNodePipeMessages.CLEANED_UP_PIPE_TSFILE_DIRS_FOR_PIPE, + pipeName, + pipeTsFileDirPath); + } finally { + pipeNameSetUnderDeletion.remove(pipeName); + } + }); + } + // Warning: Shall not be called by the assigner private String getCommonFilePath(final @Nonnull File file) { // If the parent or grandparent is null then this is testing scenario From 5587d7909a5b8d85af1d44f5bfdf73e7a51b7339 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 14:40:17 +0800 Subject: [PATCH 2/9] Use ThreadName for pipe TsFile cleanup executor --- .../db/pipe/resource/tsfile/PipeTsFileResourceManager.java | 3 ++- .../java/org/apache/iotdb/commons/concurrent/ThreadName.java | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java index e75a559149891..d76ddceac1627 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.pipe.resource.tsfile; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; +import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.utils.FileUtils; @@ -60,7 +61,7 @@ public class PipeTsFileResourceManager { private final Map pipeNameToPipeTsFileDirPathMap = new ConcurrentHashMap<>(); private final Set pipeNameSetUnderDeletion = ConcurrentHashMap.newKeySet(); private final ExecutorService pipeTsFileDirCleanupExecutor = - IoTDBThreadPoolFactory.newSingleThreadExecutor("Pipe-TsFile-Dir-Cleanup"); + IoTDBThreadPoolFactory.newSingleThreadExecutor(ThreadName.PIPE_TSFILE_DIR_CLEANUP.getName()); private final PipeTsFileResourceSegmentLock segmentLock = new PipeTsFileResourceSegmentLock(); public static String getPipeTsFileResourcePipeName( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java index e9e01efd2d48b..c0e95a48a051e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java @@ -145,6 +145,7 @@ public enum ThreadName { PIPE_CONFIGNODE_EXECUTOR_POOL("Pipe-ConfigNode-Executor-Pool"), PIPE_SUBTASK_CALLBACK_EXECUTOR_POOL("Pipe-SubTask-Callback-Executor-Pool"), PIPE_TSFILE_ASYNC_SEND_POOL("Pipe-TsFile-Async-Send-Pool"), + PIPE_TSFILE_DIR_CLEANUP("Pipe-TsFile-Dir-Cleanup"), PIPE_RUNTIME_META_SYNCER("Pipe-Runtime-Meta-Syncer"), PIPE_RUNTIME_HEARTBEAT("Pipe-Runtime-Heartbeat"), PIPE_RUNTIME_PROCEDURE_SUBMITTER("Pipe-Runtime-Procedure-Submitter"), @@ -313,6 +314,7 @@ public enum ThreadName { IOT_CONSENSUS_V2_EXECUTOR_POOL, PIPE_CONFIGNODE_EXECUTOR_POOL, PIPE_SUBTASK_CALLBACK_EXECUTOR_POOL, + PIPE_TSFILE_DIR_CLEANUP, PIPE_RUNTIME_META_SYNCER, PIPE_RUNTIME_HEARTBEAT, PIPE_RUNTIME_PROCEDURE_SUBMITTER, From f9d7f44403ab56c72aa2cf717b5a15521897905f Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 14:52:55 +0800 Subject: [PATCH 3/9] Make pipe TsFile cleanup thread lazy --- .../tsfile/PipeTsFileResourceManager.java | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java index d76ddceac1627..8c28325945bb4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java @@ -19,8 +19,9 @@ package org.apache.iotdb.db.pipe.resource.tsfile; -import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; +import org.apache.iotdb.commons.concurrent.IoTThreadFactory; import org.apache.iotdb.commons.concurrent.ThreadName; +import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.utils.FileUtils; @@ -44,6 +45,8 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; public class PipeTsFileResourceManager { @@ -60,10 +63,20 @@ public class PipeTsFileResourceManager { hardlinkOrCopiedFileToPipeTsFileResourceMap = new ConcurrentHashMap<>(); private final Map pipeNameToPipeTsFileDirPathMap = new ConcurrentHashMap<>(); private final Set pipeNameSetUnderDeletion = ConcurrentHashMap.newKeySet(); - private final ExecutorService pipeTsFileDirCleanupExecutor = - IoTDBThreadPoolFactory.newSingleThreadExecutor(ThreadName.PIPE_TSFILE_DIR_CLEANUP.getName()); + private final ExecutorService pipeTsFileDirCleanupExecutor = createPipeTsFileDirCleanupExecutor(); private final PipeTsFileResourceSegmentLock segmentLock = new PipeTsFileResourceSegmentLock(); + private static ExecutorService createPipeTsFileDirCleanupExecutor() { + return new WrappedThreadPoolExecutor( + 0, + 1, + 60L, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + new IoTThreadFactory(ThreadName.PIPE_TSFILE_DIR_CLEANUP.getName()), + ThreadName.PIPE_TSFILE_DIR_CLEANUP.getName()); + } + public static String getPipeTsFileResourcePipeName( final @Nullable String pipeName, final long creationTime) { return Objects.isNull(pipeName) ? null : pipeName + "_" + creationTime; From 11b53a2982cda5d34c1ffecf898df2ccd4588353 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 15:01:30 +0800 Subject: [PATCH 4/9] Use IoTDB thread pool factory for pipe cleanup --- .../resource/tsfile/PipeTsFileResourceManager.java | 14 +++----------- 1 file changed, 3 insertions(+), 11 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java index 8c28325945bb4..e72679b4cef17 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java @@ -19,9 +19,8 @@ package org.apache.iotdb.db.pipe.resource.tsfile; -import org.apache.iotdb.commons.concurrent.IoTThreadFactory; +import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; -import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.utils.FileUtils; @@ -45,7 +44,6 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; -import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; public class PipeTsFileResourceManager { @@ -67,14 +65,8 @@ public class PipeTsFileResourceManager { private final PipeTsFileResourceSegmentLock segmentLock = new PipeTsFileResourceSegmentLock(); private static ExecutorService createPipeTsFileDirCleanupExecutor() { - return new WrappedThreadPoolExecutor( - 0, - 1, - 60L, - TimeUnit.SECONDS, - new LinkedBlockingQueue<>(), - new IoTThreadFactory(ThreadName.PIPE_TSFILE_DIR_CLEANUP.getName()), - ThreadName.PIPE_TSFILE_DIR_CLEANUP.getName()); + return IoTDBThreadPoolFactory.newFixedThreadPoolWithIdleThreadTimeout( + 1, 60L, TimeUnit.SECONDS, ThreadName.PIPE_TSFILE_DIR_CLEANUP.getName()); } public static String getPipeTsFileResourcePipeName( From a4bea5da63bfda2d349c6d7e190fbfa1a26afac1 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 15:06:59 +0800 Subject: [PATCH 5/9] Clarify pipe TsFile cleanup deletion guard --- .../tsfile/PipeTsFileResourceManager.java | 33 +++++++++++-------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java index e72679b4cef17..1386a243a4ba0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java @@ -60,7 +60,8 @@ public class PipeTsFileResourceManager { private final Map> hardlinkOrCopiedFileToPipeTsFileResourceMap = new ConcurrentHashMap<>(); private final Map pipeNameToPipeTsFileDirPathMap = new ConcurrentHashMap<>(); - private final Set pipeNameSetUnderDeletion = ConcurrentHashMap.newKeySet(); + private final Set pipeTsFileResourcePipeNameSetUnderDeletion = + ConcurrentHashMap.newKeySet(); private final ExecutorService pipeTsFileDirCleanupExecutor = createPipeTsFileDirCleanupExecutor(); private final PipeTsFileResourceSegmentLock segmentLock = new PipeTsFileResourceSegmentLock(); @@ -241,8 +242,7 @@ public void decreaseFileReference( final String filePath = hardlinkOrCopiedFile.getPath(); final PipeTsFileResource resource = getResourceMap(pipeName).get(filePath); if (resource != null - && resource.decreaseReferenceCount( - Objects.isNull(pipeName) || !pipeNameSetUnderDeletion.contains(pipeName))) { + && resource.decreaseReferenceCount(shouldDeleteFileWhenNoReference(pipeName))) { getResourceMap(pipeName).remove(filePath); } } finally { @@ -263,20 +263,27 @@ private void decreasePublicReferenceIfExists(final File file, final @Nullable St decreaseFileReference(new File(getCommonFilePath(file)), null); } - public void markPipeTsFileDirUnderDeletion(final @Nonnull String pipeName) { - pipeNameSetUnderDeletion.add(pipeName); + private boolean shouldDeleteFileWhenNoReference( + final @Nullable String pipeTsFileResourcePipeName) { + return Objects.isNull(pipeTsFileResourcePipeName) + || !pipeTsFileResourcePipeNameSetUnderDeletion.contains(pipeTsFileResourcePipeName); } - public void unmarkPipeTsFileDirUnderDeletion(final @Nonnull String pipeName) { - pipeNameSetUnderDeletion.remove(pipeName); + public void markPipeTsFileDirUnderDeletion(final @Nonnull String pipeTsFileResourcePipeName) { + pipeTsFileResourcePipeNameSetUnderDeletion.add(pipeTsFileResourcePipeName); } - public void cleanPipeTsFileDirAsync(final @Nonnull String pipeName) { - final String pipeTsFileDirPath = pipeNameToPipeTsFileDirPathMap.remove(pipeName); - hardlinkOrCopiedFileToPipeTsFileResourceMap.remove(pipeName); + public void unmarkPipeTsFileDirUnderDeletion(final @Nonnull String pipeTsFileResourcePipeName) { + pipeTsFileResourcePipeNameSetUnderDeletion.remove(pipeTsFileResourcePipeName); + } + + public void cleanPipeTsFileDirAsync(final @Nonnull String pipeTsFileResourcePipeName) { + final String pipeTsFileDirPath = + pipeNameToPipeTsFileDirPathMap.remove(pipeTsFileResourcePipeName); + hardlinkOrCopiedFileToPipeTsFileResourceMap.remove(pipeTsFileResourcePipeName); if (Objects.isNull(pipeTsFileDirPath)) { - pipeNameSetUnderDeletion.remove(pipeName); + pipeTsFileResourcePipeNameSetUnderDeletion.remove(pipeTsFileResourcePipeName); return; } @@ -286,10 +293,10 @@ public void cleanPipeTsFileDirAsync(final @Nonnull String pipeName) { FileUtils.deleteFileOrDirectory(new File(pipeTsFileDirPath), true); LOGGER.info( DataNodePipeMessages.CLEANED_UP_PIPE_TSFILE_DIRS_FOR_PIPE, - pipeName, + pipeTsFileResourcePipeName, pipeTsFileDirPath); } finally { - pipeNameSetUnderDeletion.remove(pipeName); + pipeTsFileResourcePipeNameSetUnderDeletion.remove(pipeTsFileResourcePipeName); } }); } From da97b47de273236f50143054dbdae698d324566f Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 18:24:27 +0800 Subject: [PATCH 6/9] Reuse periodical cleaner for Pipe TsFile cleanup --- .../iotdb/db/i18n/DataNodePipeMessages.java | 4 +- .../iotdb/db/i18n/DataNodePipeMessages.java | 4 +- .../agent/task/PipeDataNodeTaskAgent.java | 4 +- ...HardlinkOrCopiedFileDirStartupCleaner.java | 93 ++++++++++++++----- .../tsfile/PipeTsFileResourceManager.java | 28 +----- .../iotdb/commons/concurrent/ThreadName.java | 2 - 6 files changed, 79 insertions(+), 56 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index e02ef8f64ddc8..e75f22665038e 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -1343,6 +1343,8 @@ public final class DataNodePipeMessages { "Finished deleting all stale pipe hardlink dirs by periodical job."; public static final String PIPE_HARDLINK_DIR_MOVE_FAILED_DELETING_SYNC = "Failed to move pipe hardlink dir {} for periodical deletion, deleting it synchronously."; + public static final String PIPE_HARDLINK_DIR_MOVE_FAILED_SKIPPING_PERIODICAL_DELETE = + "Failed to move pipe hardlink dir {} for periodical deletion, skip registering the original dir to avoid deleting files of a recreated pipe."; public static final String PIPE_SNAPSHOT_DIR_FOUND_DELETING_IT = "Pipe snapshot dir found, deleting it: {},"; public static final String SHRINK_CALLBACK_IS_NOT_SUPPORTED_IN_PIPEFIXEDMEMORYBLOCK = @@ -2487,7 +2489,5 @@ private DataNodePipeMessages() {} "Reported %s pipe metas."; public static final String MESSAGE_TRANSFER_FILE_ARG_ERROR_RESULT_STATUS_ARG_E565D9FD = "Transfer file %s error, result status %s."; - public static final String CLEANED_UP_PIPE_TSFILE_DIRS_FOR_PIPE = - "Cleaned up Pipe TsFile dirs for pipe {}: {}"; } diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 3620a364610a3..ad1c1da8cf0c0 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -1266,6 +1266,8 @@ public final class DataNodePipeMessages { "已通过周期性任务删除全部过期的 Pipe hardlink 目录。"; public static final String PIPE_HARDLINK_DIR_MOVE_FAILED_DELETING_SYNC = "为周期性删除移动 Pipe hardlink 目录 {} 失败，改为同步删除。"; + public static final String PIPE_HARDLINK_DIR_MOVE_FAILED_SKIPPING_PERIODICAL_DELETE = + "为周期性删除移动 Pipe hardlink 目录 {} 失败，跳过注册原目录以避免删除重建 Pipe 的文件。"; public static final String PIPE_SNAPSHOT_DIR_FOUND_DELETING_IT = "发现 Pipe snapshot 目录，正在删除：{},"; public static final String SHRINK_CALLBACK_IS_NOT_SUPPORTED_IN_PIPEFIXEDMEMORYBLOCK = @@ -2325,7 +2327,5 @@ private DataNodePipeMessages() {} "已上报 %s 个 pipe meta。"; public static final String MESSAGE_TRANSFER_FILE_ARG_ERROR_RESULT_STATUS_ARG_E565D9FD = "传输文件 %s 出错，结果状态为 %s。"; - public static final String CLEANED_UP_PIPE_TSFILE_DIRS_FOR_PIPE = - "已清理 pipe {} 的 Pipe TsFile 目录：{}"; } 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 d33eb86f28946..dce7ee5f6d925 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 @@ -318,7 +318,7 @@ protected boolean dropPipe(final String pipeName, final long creationTime) { final String taskId = pipeName + "_" + creationTime; PipeTsFileToTabletsMetrics.getInstance().deregister(taskId); PipeDataNodeSinglePipeMetrics.getInstance().deregister(taskId); - PipeDataNodeResourceManager.tsfile().cleanPipeTsFileDirAsync(pipeTsFileResourcePipeName); + PipeDataNodeResourceManager.tsfile().cleanPipeTsFileDir(pipeTsFileResourcePipeName); return true; } @@ -360,7 +360,7 @@ protected boolean dropPipe(final String pipeName) { final String taskId = pipeName + "_" + creationTime; PipeTsFileToTabletsMetrics.getInstance().deregister(taskId); PipeDataNodeSinglePipeMetrics.getInstance().deregister(taskId); - PipeDataNodeResourceManager.tsfile().cleanPipeTsFileDirAsync(pipeTsFileResourcePipeName); + PipeDataNodeResourceManager.tsfile().cleanPipeTsFileDir(pipeTsFileResourcePipeName); // When the pipe contains no pipe tasks, there is no corresponding prefetching queue for the // subscribed pipe, so the subscription needs to be manually marked as completed. if (!hasPipeTasks && PipeStaticMeta.isSubscriptionPipe(pipeName)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java index bb1f760f44962..4b3cc536aaaf0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java @@ -37,6 +37,7 @@ import java.nio.file.attribute.BasicFileAttributes; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -49,6 +50,9 @@ public class PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner { "PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner#cleanTsFileDir()"; private static final long DELETE_MAX_PATH_COUNT_PER_ROUND = 100_000L; private static final long DELETE_MAX_TIME_PER_ROUND_MS = 1_000L; + private static final PeriodicalStalePipeDirCleaner PERIODICAL_STALE_PIPE_DIR_CLEANER = + new PeriodicalStalePipeDirCleaner(); + private static final AtomicBoolean PERIODICAL_CLEANUP_JOB_REGISTERED = new AtomicBoolean(false); /** * Delete the data directory and all of its subdirectories that contain the @@ -65,13 +69,22 @@ private static void cleanTsFileDir(final PeriodicalJobRegistrar periodicalJobReg for (final String dataDir : IoTDBDescriptor.getInstance().getConfig().getDataDirs()) { final File localDataDir = new File(dataDir); collectInterruptedStalePipeDirs(localDataDir, pipeHardlinkBaseDirName, stalePipeDirs); + collectInterruptedStalePipeDirs( + new File( + localDataDir, + pipeHardlinkBaseDirName + + File.separator + + PipeConfig.getInstance().getPipeHardlinkTsFileDirName()), + "", + stalePipeDirs); final File pipeHardLinkDir = new File(localDataDir, pipeHardlinkBaseDirName); if (pipeHardLinkDir.isDirectory()) { moveAsideAndCollect(pipeHardLinkDir, pipeHardlinkBaseDirName, stalePipeDirs); } } - registerPeriodicalCleanupJob(periodicalJobRegistrar, stalePipeDirs); + PERIODICAL_STALE_PIPE_DIR_CLEANER.addStalePipeDirs(stalePipeDirs); + registerPeriodicalCleanupJob(periodicalJobRegistrar); } private static void collectInterruptedStalePipeDirs( @@ -82,7 +95,7 @@ private static void collectInterruptedStalePipeDirs( localDataDir.listFiles( file -> file.isDirectory() - && file.getName().startsWith(pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX)); + && file.getName().contains(pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX)); if (stalePipeDirFiles == null) { return; } @@ -137,17 +150,47 @@ private static File moveAside(final File pipeHardLinkDir, final String pipeHardl } private static void registerPeriodicalCleanupJob( - final PeriodicalJobRegistrar periodicalJobRegistrar, final List stalePipeDirs) { - if (stalePipeDirs.isEmpty()) { + final PeriodicalJobRegistrar periodicalJobRegistrar) { + if (!PERIODICAL_CLEANUP_JOB_REGISTERED.compareAndSet(false, true)) { return; } periodicalJobRegistrar.register( PERIODICAL_CLEANUP_JOB_ID, - new PeriodicalStalePipeDirCleaner(stalePipeDirs)::cleanOneRound, + PERIODICAL_STALE_PIPE_DIR_CLEANER::cleanOneRound, PipeConfig.getInstance().getPipeSubtaskExecutorCronHeartbeatEventIntervalSeconds()); } + public static void submitStalePipeDirForPeriodicalCleanup(final File stalePipeDir) { + if (stalePipeDir == null) { + return; + } + + final File stalePipeDirToDelete; + if (stalePipeDir.isDirectory()) { + try { + stalePipeDirToDelete = moveAside(stalePipeDir, stalePipeDir.getName()); + LOGGER.info( + DataNodePipeMessages.PIPE_HARDLINK_DIR_FOUND_MOVED_TO_PERIODICAL_DELETE, + stalePipeDir, + stalePipeDirToDelete); + } catch (final IOException e) { + LOGGER.warn( + DataNodePipeMessages.PIPE_HARDLINK_DIR_MOVE_FAILED_SKIPPING_PERIODICAL_DELETE, + stalePipeDir, + e); + return; + } + } else { + stalePipeDirToDelete = stalePipeDir; + } + + LOGGER.info( + DataNodePipeMessages.PIPE_STALE_HARDLINK_DIR_FOUND_REGISTERING_PERIODICAL_DELETE, + stalePipeDirToDelete); + PERIODICAL_STALE_PIPE_DIR_CLEANER.addStalePipeDir(stalePipeDirToDelete); + } + private static CleanupRoundResult deleteQuietlyWithThrottle(final File stalePipeDir) { if (!stalePipeDir.exists()) { return CleanupRoundResult.finished(); @@ -230,33 +273,36 @@ public interface PeriodicalJobRegistrar { private static class PeriodicalStalePipeDirCleaner { - private final List stalePipeDirs; - private int currentDirIndex; - private boolean finished; + private final ConcurrentLinkedQueue stalePipeDirs = new ConcurrentLinkedQueue<>(); + private File currentStalePipeDir; - private PeriodicalStalePipeDirCleaner(final List stalePipeDirs) { - this.stalePipeDirs = stalePipeDirs; - currentDirIndex = 0; - finished = false; + private void addStalePipeDir(final File stalePipeDir) { + stalePipeDirs.offer(stalePipeDir); } - private void cleanOneRound() { - if (finished) { - return; - } + private void addStalePipeDirs(final List stalePipeDirs) { + stalePipeDirs.forEach(this::addStalePipeDir); + } + private void cleanOneRound() { long deletedPathCount = 0; - while (currentDirIndex < stalePipeDirs.size()) { - final File stalePipeDir = stalePipeDirs.get(currentDirIndex); - final CleanupRoundResult result = deleteQuietlyWithThrottle(stalePipeDir); + while (true) { + if (currentStalePipeDir == null) { + currentStalePipeDir = stalePipeDirs.poll(); + } + if (currentStalePipeDir == null) { + return; + } + + final CleanupRoundResult result = deleteQuietlyWithThrottle(currentStalePipeDir); deletedPathCount += result.deletedPathCount; if (result.finished) { LOGGER.info( DataNodePipeMessages.PIPE_HARDLINK_DIR_PERIODICAL_DELETE_FINISHED, - stalePipeDir, + currentStalePipeDir, result.success); - ++currentDirIndex; + currentStalePipeDir = null; continue; } @@ -264,14 +310,11 @@ private void cleanOneRound() { LOGGER.info( DataNodePipeMessages.PIPE_HARDLINK_DIR_PERIODICAL_DELETE_PROGRESS, deletedPathCount, - stalePipeDir, + currentStalePipeDir, result.success); } return; } - - finished = true; - LOGGER.info(DataNodePipeMessages.PIPE_HARDLINK_DIR_PERIODICAL_DELETE_ALL_FINISHED); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java index 1386a243a4ba0..5bcdaec14a11f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java @@ -19,13 +19,12 @@ package org.apache.iotdb.db.pipe.resource.tsfile; -import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; -import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.i18n.DataNodePipeMessages; +import org.apache.iotdb.db.pipe.resource.PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.tsfile.enums.TSDataType; @@ -43,8 +42,6 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeUnit; public class PipeTsFileResourceManager { @@ -62,14 +59,8 @@ public class PipeTsFileResourceManager { private final Map pipeNameToPipeTsFileDirPathMap = new ConcurrentHashMap<>(); private final Set pipeTsFileResourcePipeNameSetUnderDeletion = ConcurrentHashMap.newKeySet(); - private final ExecutorService pipeTsFileDirCleanupExecutor = createPipeTsFileDirCleanupExecutor(); private final PipeTsFileResourceSegmentLock segmentLock = new PipeTsFileResourceSegmentLock(); - private static ExecutorService createPipeTsFileDirCleanupExecutor() { - return IoTDBThreadPoolFactory.newFixedThreadPoolWithIdleThreadTimeout( - 1, 60L, TimeUnit.SECONDS, ThreadName.PIPE_TSFILE_DIR_CLEANUP.getName()); - } - public static String getPipeTsFileResourcePipeName( final @Nullable String pipeName, final long creationTime) { return Objects.isNull(pipeName) ? null : pipeName + "_" + creationTime; @@ -277,7 +268,7 @@ public void unmarkPipeTsFileDirUnderDeletion(final @Nonnull String pipeTsFileRes pipeTsFileResourcePipeNameSetUnderDeletion.remove(pipeTsFileResourcePipeName); } - public void cleanPipeTsFileDirAsync(final @Nonnull String pipeTsFileResourcePipeName) { + public void cleanPipeTsFileDir(final @Nonnull String pipeTsFileResourcePipeName) { final String pipeTsFileDirPath = pipeNameToPipeTsFileDirPathMap.remove(pipeTsFileResourcePipeName); hardlinkOrCopiedFileToPipeTsFileResourceMap.remove(pipeTsFileResourcePipeName); @@ -287,18 +278,9 @@ public void cleanPipeTsFileDirAsync(final @Nonnull String pipeTsFileResourcePipe return; } - pipeTsFileDirCleanupExecutor.execute( - () -> { - try { - FileUtils.deleteFileOrDirectory(new File(pipeTsFileDirPath), true); - LOGGER.info( - DataNodePipeMessages.CLEANED_UP_PIPE_TSFILE_DIRS_FOR_PIPE, - pipeTsFileResourcePipeName, - pipeTsFileDirPath); - } finally { - pipeTsFileResourcePipeNameSetUnderDeletion.remove(pipeTsFileResourcePipeName); - } - }); + PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.submitStalePipeDirForPeriodicalCleanup( + new File(pipeTsFileDirPath)); + pipeTsFileResourcePipeNameSetUnderDeletion.remove(pipeTsFileResourcePipeName); } // Warning: Shall not be called by the assigner diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java index c0e95a48a051e..e9e01efd2d48b 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java @@ -145,7 +145,6 @@ public enum ThreadName { PIPE_CONFIGNODE_EXECUTOR_POOL("Pipe-ConfigNode-Executor-Pool"), PIPE_SUBTASK_CALLBACK_EXECUTOR_POOL("Pipe-SubTask-Callback-Executor-Pool"), PIPE_TSFILE_ASYNC_SEND_POOL("Pipe-TsFile-Async-Send-Pool"), - PIPE_TSFILE_DIR_CLEANUP("Pipe-TsFile-Dir-Cleanup"), PIPE_RUNTIME_META_SYNCER("Pipe-Runtime-Meta-Syncer"), PIPE_RUNTIME_HEARTBEAT("Pipe-Runtime-Heartbeat"), PIPE_RUNTIME_PROCEDURE_SUBMITTER("Pipe-Runtime-Procedure-Submitter"), @@ -314,7 +313,6 @@ public enum ThreadName { IOT_CONSENSUS_V2_EXECUTOR_POOL, PIPE_CONFIGNODE_EXECUTOR_POOL, PIPE_SUBTASK_CALLBACK_EXECUTOR_POOL, - PIPE_TSFILE_DIR_CLEANUP, PIPE_RUNTIME_META_SYNCER, PIPE_RUNTIME_HEARTBEAT, PIPE_RUNTIME_PROCEDURE_SUBMITTER, From 25625fb1cf2a76fb54360800b990d27449b12feb Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 18:36:54 +0800 Subject: [PATCH 7/9] Clarify stale Pipe dir collection --- ...HardlinkOrCopiedFileDirStartupCleaner.java | 50 +++++++++++-------- 1 file changed, 30 insertions(+), 20 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java index 4b3cc536aaaf0..d2bc51a4cddd4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java @@ -68,15 +68,8 @@ private static void cleanTsFileDir(final PeriodicalJobRegistrar periodicalJobReg final List stalePipeDirs = new ArrayList<>(); for (final String dataDir : IoTDBDescriptor.getInstance().getConfig().getDataDirs()) { final File localDataDir = new File(dataDir); - collectInterruptedStalePipeDirs(localDataDir, pipeHardlinkBaseDirName, stalePipeDirs); - collectInterruptedStalePipeDirs( - new File( - localDataDir, - pipeHardlinkBaseDirName - + File.separator - + PipeConfig.getInstance().getPipeHardlinkTsFileDirName()), - "", - stalePipeDirs); + collectInterruptedPipeBaseStaleDirs(localDataDir, pipeHardlinkBaseDirName, stalePipeDirs); + collectInterruptedPipeTsFileStaleDirs(localDataDir, pipeHardlinkBaseDirName, stalePipeDirs); final File pipeHardLinkDir = new File(localDataDir, pipeHardlinkBaseDirName); if (pipeHardLinkDir.isDirectory()) { @@ -87,15 +80,33 @@ private static void cleanTsFileDir(final PeriodicalJobRegistrar periodicalJobReg registerPeriodicalCleanupJob(periodicalJobRegistrar); } - private static void collectInterruptedStalePipeDirs( + private static void collectInterruptedPipeBaseStaleDirs( + final File localDataDir, + final String pipeHardlinkBaseDirName, + final List stalePipeDirs) { + collectInterruptedStalePipeDirs( + localDataDir, pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX, stalePipeDirs); + } + + private static void collectInterruptedPipeTsFileStaleDirs( final File localDataDir, final String pipeHardlinkBaseDirName, final List stalePipeDirs) { + collectInterruptedStalePipeDirs( + new File( + localDataDir, + pipeHardlinkBaseDirName + + File.separator + + PipeConfig.getInstance().getPipeHardlinkTsFileDirName()), + STALE_PIPE_DIR_SUFFIX, + stalePipeDirs); + } + + private static void collectInterruptedStalePipeDirs( + final File parentDir, final String staleDirNameToken, final List stalePipeDirs) { final File[] stalePipeDirFiles = - localDataDir.listFiles( - file -> - file.isDirectory() - && file.getName().contains(pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX)); + parentDir.listFiles( + file -> file.isDirectory() && file.getName().contains(staleDirNameToken)); if (stalePipeDirFiles == null) { return; } @@ -129,21 +140,20 @@ private static void moveAsideAndCollect( } } - private static File moveAside(final File pipeHardLinkDir, final String pipeHardlinkBaseDirName) + private static File moveAside(final File pipeDir, final String staleDirNamePrefix) throws IOException { - final File parentDir = pipeHardLinkDir.getParentFile(); + final File parentDir = pipeDir.getParentFile(); if (parentDir == null) { throw new IOException( - DataNodePipeMessages.EXCEPTION_FAILED_TO_GET_PARENT_DIR_OF_8CE21C1D + pipeHardLinkDir); + DataNodePipeMessages.EXCEPTION_FAILED_TO_GET_PARENT_DIR_OF_8CE21C1D + pipeDir); } final long timestamp = System.currentTimeMillis(); for (int i = 0; ; ++i) { final File stalePipeDir = - new File( - parentDir, pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX + timestamp + "-" + i); + new File(parentDir, staleDirNamePrefix + STALE_PIPE_DIR_SUFFIX + timestamp + "-" + i); if (!stalePipeDir.exists()) { - Files.move(pipeHardLinkDir.toPath(), stalePipeDir.toPath()); + Files.move(pipeDir.toPath(), stalePipeDir.toPath()); return stalePipeDir; } } From ae8111fdda9558b72d054072775ae44b6a0019c4 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 9 Jul 2026 18:44:18 +0800 Subject: [PATCH 8/9] Revert "Clarify stale Pipe dir collection" This reverts commit 25625fb1cf2a76fb54360800b990d27449b12feb. --- ...HardlinkOrCopiedFileDirStartupCleaner.java | 50 ++++++++----------- 1 file changed, 20 insertions(+), 30 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java index d2bc51a4cddd4..4b3cc536aaaf0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java @@ -68,8 +68,15 @@ private static void cleanTsFileDir(final PeriodicalJobRegistrar periodicalJobReg final List stalePipeDirs = new ArrayList<>(); for (final String dataDir : IoTDBDescriptor.getInstance().getConfig().getDataDirs()) { final File localDataDir = new File(dataDir); - collectInterruptedPipeBaseStaleDirs(localDataDir, pipeHardlinkBaseDirName, stalePipeDirs); - collectInterruptedPipeTsFileStaleDirs(localDataDir, pipeHardlinkBaseDirName, stalePipeDirs); + collectInterruptedStalePipeDirs(localDataDir, pipeHardlinkBaseDirName, stalePipeDirs); + collectInterruptedStalePipeDirs( + new File( + localDataDir, + pipeHardlinkBaseDirName + + File.separator + + PipeConfig.getInstance().getPipeHardlinkTsFileDirName()), + "", + stalePipeDirs); final File pipeHardLinkDir = new File(localDataDir, pipeHardlinkBaseDirName); if (pipeHardLinkDir.isDirectory()) { @@ -80,33 +87,15 @@ private static void cleanTsFileDir(final PeriodicalJobRegistrar periodicalJobReg registerPeriodicalCleanupJob(periodicalJobRegistrar); } - private static void collectInterruptedPipeBaseStaleDirs( - final File localDataDir, - final String pipeHardlinkBaseDirName, - final List stalePipeDirs) { - collectInterruptedStalePipeDirs( - localDataDir, pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX, stalePipeDirs); - } - - private static void collectInterruptedPipeTsFileStaleDirs( + private static void collectInterruptedStalePipeDirs( final File localDataDir, final String pipeHardlinkBaseDirName, final List stalePipeDirs) { - collectInterruptedStalePipeDirs( - new File( - localDataDir, - pipeHardlinkBaseDirName - + File.separator - + PipeConfig.getInstance().getPipeHardlinkTsFileDirName()), - STALE_PIPE_DIR_SUFFIX, - stalePipeDirs); - } - - private static void collectInterruptedStalePipeDirs( - final File parentDir, final String staleDirNameToken, final List stalePipeDirs) { final File[] stalePipeDirFiles = - parentDir.listFiles( - file -> file.isDirectory() && file.getName().contains(staleDirNameToken)); + localDataDir.listFiles( + file -> + file.isDirectory() + && file.getName().contains(pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX)); if (stalePipeDirFiles == null) { return; } @@ -140,20 +129,21 @@ private static void moveAsideAndCollect( } } - private static File moveAside(final File pipeDir, final String staleDirNamePrefix) + private static File moveAside(final File pipeHardLinkDir, final String pipeHardlinkBaseDirName) throws IOException { - final File parentDir = pipeDir.getParentFile(); + final File parentDir = pipeHardLinkDir.getParentFile(); if (parentDir == null) { throw new IOException( - DataNodePipeMessages.EXCEPTION_FAILED_TO_GET_PARENT_DIR_OF_8CE21C1D + pipeDir); + DataNodePipeMessages.EXCEPTION_FAILED_TO_GET_PARENT_DIR_OF_8CE21C1D + pipeHardLinkDir); } final long timestamp = System.currentTimeMillis(); for (int i = 0; ; ++i) { final File stalePipeDir = - new File(parentDir, staleDirNamePrefix + STALE_PIPE_DIR_SUFFIX + timestamp + "-" + i); + new File( + parentDir, pipeHardlinkBaseDirName + STALE_PIPE_DIR_SUFFIX + timestamp + "-" + i); if (!stalePipeDir.exists()) { - Files.move(pipeDir.toPath(), stalePipeDir.toPath()); + Files.move(pipeHardLinkDir.toPath(), stalePipeDir.toPath()); return stalePipeDir; } } From a5fb3533f44c956bc21a3f067a3ef300e96759e9 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Fri, 10 Jul 2026 09:55:59 +0800 Subject: [PATCH 9/9] Avoid duplicate stale Pipe dir collection --- ...PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java | 8 -------- 1 file changed, 8 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java index 4b3cc536aaaf0..db60fc4645b63 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.java @@ -69,14 +69,6 @@ private static void cleanTsFileDir(final PeriodicalJobRegistrar periodicalJobReg for (final String dataDir : IoTDBDescriptor.getInstance().getConfig().getDataDirs()) { final File localDataDir = new File(dataDir); collectInterruptedStalePipeDirs(localDataDir, pipeHardlinkBaseDirName, stalePipeDirs); - collectInterruptedStalePipeDirs( - new File( - localDataDir, - pipeHardlinkBaseDirName - + File.separator - + PipeConfig.getInstance().getPipeHardlinkTsFileDirName()), - "", - stalePipeDirs); final File pipeHardLinkDir = new File(localDataDir, pipeHardlinkBaseDirName); if (pipeHardLinkDir.isDirectory()) {