This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 350c6c7bcd2 Optimize pipe TsFile cleanup on drop (#18163)
350c6c7bcd2 is described below
commit 350c6c7bcd25b883eca53897b294b57637850119
Author: Zhenyu Luo <[email protected]>
AuthorDate: Fri Jul 10 14:15:25 2026 +0800
Optimize pipe TsFile cleanup on drop (#18163)
* Optimize pipe TsFile cleanup on drop
* Use ThreadName for pipe TsFile cleanup executor
* Make pipe TsFile cleanup thread lazy
* Use IoTDB thread pool factory for pipe cleanup
* Clarify pipe TsFile cleanup deletion guard
* Reuse periodical cleaner for Pipe TsFile cleanup
* Clarify stale Pipe dir collection
* Revert "Clarify stale Pipe dir collection"
This reverts commit 25625fb1cf2a76fb54360800b990d27449b12feb.
* Avoid duplicate stale Pipe dir collection
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 22 ++++++
...aNodeHardlinkOrCopiedFileDirStartupCleaner.java | 85 +++++++++++++++-------
.../pipe/resource/tsfile/PipeTsFileResource.java | 8 +-
.../resource/tsfile/PipeTsFileResourceManager.java | 39 +++++++++-
6 files changed, 131 insertions(+), 27 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 e7c6c85ca5c..67932c73967 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
@@ -1410,6 +1410,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 =
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 54b68d18e0d..1e2954d80ff 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
@@ -1326,6 +1326,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 =
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 df08311151a..dce7ee5f6d9 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.PipeDataNodeSinglePipeMetrics;
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 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
@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().cleanPipeTsFileDir(pipeTsFileResourcePipeName);
return true;
}
@@ -319,6 +327,15 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
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 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
}
if (!super.dropPipe(pipeName)) {
+ if (Objects.nonNull(pipeTsFileResourcePipeName)) {
+ PipeDataNodeResourceManager.tsfile()
+ .unmarkPipeTsFileDirUnderDeletion(pipeTsFileResourcePipeName);
+ }
return false;
}
@@ -339,6 +360,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
final String taskId = pipeName + "_" + creationTime;
PipeTsFileToTabletsMetrics.getInstance().deregister(taskId);
PipeDataNodeSinglePipeMetrics.getInstance().deregister(taskId);
+
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 bb1f760f449..db60fc4645b 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.SimpleFileVisitor;
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
@@ -71,7 +75,8 @@ public class
PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner {
moveAsideAndCollect(pipeHardLinkDir, pipeHardlinkBaseDirName,
stalePipeDirs);
}
}
- registerPeriodicalCleanupJob(periodicalJobRegistrar, stalePipeDirs);
+ PERIODICAL_STALE_PIPE_DIR_CLEANER.addStalePipeDirs(stalePipeDirs);
+ registerPeriodicalCleanupJob(periodicalJobRegistrar);
}
private static void collectInterruptedStalePipeDirs(
@@ -82,7 +87,7 @@ public class
PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner {
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 +142,47 @@ public class
PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner {
}
private static void registerPeriodicalCleanupJob(
- final PeriodicalJobRegistrar periodicalJobRegistrar, final List<File>
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 +265,36 @@ public class
PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner {
private static class PeriodicalStalePipeDirCleaner {
- private final List<File> stalePipeDirs;
- private int currentDirIndex;
- private boolean finished;
+ private final ConcurrentLinkedQueue<File> stalePipeDirs = new
ConcurrentLinkedQueue<>();
+ private File currentStalePipeDir;
- private PeriodicalStalePipeDirCleaner(final List<File> 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<File> 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 +302,11 @@ public class
PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner {
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/PipeTsFileResource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java
index 51421cdcec8..a8a7aa7e195 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 class PipeTsFileResource implements AutoCloseable {
}
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 4fc1c3063df..5bcdaec14a1 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
@@ -24,6 +24,7 @@ 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;
@@ -39,6 +40,7 @@ import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
public class PipeTsFileResourceManager {
@@ -54,6 +56,9 @@ public class PipeTsFileResourceManager {
// PipeName -> TsFilePath -> PipeTsFileResource
private final Map<String, Map<String, PipeTsFileResource>>
hardlinkOrCopiedFileToPipeTsFileResourceMap = new ConcurrentHashMap<>();
+ private final Map<String, String> pipeNameToPipeTsFileDirPathMap = new
ConcurrentHashMap<>();
+ private final Set<String> pipeTsFileResourcePipeNameSetUnderDeletion =
+ ConcurrentHashMap.newKeySet();
private final PipeTsFileResourceSegmentLock segmentLock = new
PipeTsFileResourceSegmentLock();
public static String getPipeTsFileResourcePipeName(
@@ -126,6 +131,8 @@ public class PipeTsFileResourceManager {
// 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 +232,8 @@ public class PipeTsFileResourceManager {
try {
final String filePath = hardlinkOrCopiedFile.getPath();
final PipeTsFileResource resource =
getResourceMap(pipeName).get(filePath);
- if (resource != null && resource.decreaseReferenceCount()) {
+ if (resource != null
+ &&
resource.decreaseReferenceCount(shouldDeleteFileWhenNoReference(pipeName))) {
getResourceMap(pipeName).remove(filePath);
}
} finally {
@@ -246,6 +254,35 @@ public class PipeTsFileResourceManager {
decreaseFileReference(new File(getCommonFilePath(file)), null);
}
+ private boolean shouldDeleteFileWhenNoReference(
+ final @Nullable String pipeTsFileResourcePipeName) {
+ return Objects.isNull(pipeTsFileResourcePipeName)
+ ||
!pipeTsFileResourcePipeNameSetUnderDeletion.contains(pipeTsFileResourcePipeName);
+ }
+
+ public void markPipeTsFileDirUnderDeletion(final @Nonnull String
pipeTsFileResourcePipeName) {
+ pipeTsFileResourcePipeNameSetUnderDeletion.add(pipeTsFileResourcePipeName);
+ }
+
+ public void unmarkPipeTsFileDirUnderDeletion(final @Nonnull String
pipeTsFileResourcePipeName) {
+
pipeTsFileResourcePipeNameSetUnderDeletion.remove(pipeTsFileResourcePipeName);
+ }
+
+ public void cleanPipeTsFileDir(final @Nonnull String
pipeTsFileResourcePipeName) {
+ final String pipeTsFileDirPath =
+ pipeNameToPipeTsFileDirPathMap.remove(pipeTsFileResourcePipeName);
+
hardlinkOrCopiedFileToPipeTsFileResourceMap.remove(pipeTsFileResourcePipeName);
+
+ if (Objects.isNull(pipeTsFileDirPath)) {
+
pipeTsFileResourcePipeNameSetUnderDeletion.remove(pipeTsFileResourcePipeName);
+ return;
+ }
+
+
PipeDataNodeHardlinkOrCopiedFileDirStartupCleaner.submitStalePipeDirForPeriodicalCleanup(
+ new File(pipeTsFileDirPath));
+
pipeTsFileResourcePipeNameSetUnderDeletion.remove(pipeTsFileResourcePipeName);
+ }
+
// 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
