This is an automated email from the ASF dual-hosted git repository. jackietien pushed a commit to branch rc/1.3.5 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c0d70886f923ae05dc05d607715351a4ec8528c2 Author: Caideyipi <[email protected]> AuthorDate: Tue Jul 22 09:34:27 2025 +0800 [To dev/1.3] Pipe: Fixed hard-link problem (#15984) (#15986) * Update PipeTsFileResourceManager.java * Update PipeDataRegionAssigner.java * Update PipeTsFileResourceManager.java * Update PipeTsFileResourceManager.java --- .../realtime/assigner/PipeDataRegionAssigner.java | 9 ---- .../resource/tsfile/PipeTsFileResourceManager.java | 60 +++++++++++----------- 2 files changed, 31 insertions(+), 38 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java index 8e415772438..0b83876b186 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java @@ -19,9 +19,6 @@ package org.apache.iotdb.db.pipe.extractor.dataregion.realtime.assigner; -import org.apache.iotdb.commons.consensus.index.ProgressIndex; -import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; -import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.commons.pipe.metric.PipeEventCounter; @@ -39,14 +36,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.Closeable; -import java.util.concurrent.atomic.AtomicReference; public class PipeDataRegionAssigner implements Closeable { private static final Logger LOGGER = LoggerFactory.getLogger(PipeDataRegionAssigner.class); - private static final PipeConfig PIPE_CONFIG = PipeConfig.getInstance(); - /** * The {@link PipeDataRegionMatcher} is used to match the event with the extractor based on the * pattern. @@ -58,9 +52,6 @@ public class PipeDataRegionAssigner implements Closeable { private final String dataRegionId; - private final AtomicReference<ProgressIndex> maxProgressIndexForRealtimeEvent = - new AtomicReference<>(MinimumProgressIndex.INSTANCE); - private final PipeEventCounter eventCounter = new PipeDataRegionEventCounter(); public String getDataRegionId() { 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 bf8b6b86b43..86a5d40d375 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 @@ -56,6 +56,11 @@ public class PipeTsFileResourceManager { hardlinkOrCopiedFileToPipeTsFileResourceMap = new ConcurrentHashMap<>(); private final PipeTsFileResourceSegmentLock segmentLock = new PipeTsFileResourceSegmentLock(); + public File increaseFileReference( + final File file, final boolean isTsFile, final @Nullable String pipeName) throws IOException { + return increaseFileReference(file, isTsFile, pipeName, null); + } + /** * Given a file, 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. @@ -72,16 +77,24 @@ public class PipeTsFileResourceManager { * @param file tsfile, resource file or mod file. can be original file or hardlink/copy of * original file * @param isTsFile {@code true} to create hardlink, {@code false} to copy file + * @param pipeName Nonnull if the pipe is from historical or assigner -> extractors, null if is + * dataRegion -> assigner + * @param sourceFile for inner use, historical extractor will use this to create hardlink from + * pipe tsFile -> common tsFile * @return the hardlink or copied file * @throws IOException when create hardlink or copy file failed */ - public File increaseFileReference( - final File file, final boolean isTsFile, final @Nullable String pipeName) throws IOException { + private File increaseFileReference( + final File file, + final boolean isTsFile, + final @Nullable String pipeName, + final @Nullable File sourceFile) + throws IOException { // If the file is already a hardlink or copied file, // just increase reference count and return it segmentLock.lock(file); try { - if (increaseReferenceIfExists(file, pipeName)) { + if (increaseReferenceIfExists(file, pipeName, isTsFile)) { return file; } } finally { @@ -90,19 +103,22 @@ public class PipeTsFileResourceManager { // If the file is not a hardlink or copied file, check if there is a related hardlink or // copied file in pipe dir. if so, increase reference count and return it - final File hardlinkOrCopiedFile = getHardlinkOrCopiedFileInPipeDir(file, pipeName); + final File hardlinkOrCopiedFile = + Objects.isNull(sourceFile) ? getHardlinkOrCopiedFileInPipeDir(file, pipeName) : file; segmentLock.lock(hardlinkOrCopiedFile); try { - if (increaseReferenceIfExists(hardlinkOrCopiedFile, pipeName)) { + if (increaseReferenceIfExists(hardlinkOrCopiedFile, pipeName, isTsFile)) { return getResourceMap(pipeName).get(hardlinkOrCopiedFile.getPath()).getFile(); } // If the file is a tsfile, create a hardlink in pipe dir and will return it. // otherwise, copy the file (.mod or .resource) to pipe dir and will return it. + final File source = Objects.isNull(sourceFile) ? file : sourceFile; + final File resultFile = isTsFile - ? FileUtils.createHardLink(file, hardlinkOrCopiedFile) - : FileUtils.copyFile(file, hardlinkOrCopiedFile); + ? FileUtils.createHardLink(source, hardlinkOrCopiedFile) + : FileUtils.copyFile(source, hardlinkOrCopiedFile); // If the file is not a hardlink or copied file, and there is no related hardlink or copied // file in pipe dir, create a hardlink or copy it to pipe dir, maintain a reference count for @@ -116,7 +132,7 @@ public class PipeTsFileResourceManager { resultFile.getPath(), new PipeTsFilePublicResource(resultFile)); } - increasePublicReference(resultFile, pipeName); + increasePublicReference(resultFile, pipeName, isTsFile); return resultFile; } finally { @@ -124,34 +140,26 @@ public class PipeTsFileResourceManager { } } - private boolean increaseReferenceIfExists(final File file, final @Nullable String pipeName) { + private boolean increaseReferenceIfExists( + final File file, final @Nullable String pipeName, final boolean isTsFile) throws IOException { final String path = file.getPath(); final PipeTsFileResource resource = getResourceMap(pipeName).get(path); if (resource != null) { resource.increaseReferenceCount(); - increasePublicReference(file, pipeName); + increasePublicReference(file, pipeName, isTsFile); return true; } return false; } - private void increasePublicReference(final File file, final String pipeName) { + private void increasePublicReference( + final File file, final String pipeName, final boolean isTsFile) throws IOException { if (Objects.isNull(pipeName)) { return; } // Increase the assigner's file to avoid hard-link or memory cache cleaning // Note that it does not exist for historical files - final String path = getCommonFilePath(file); - hardlinkOrCopiedFileToTsFilePublicResourceMap.compute( - path, - (k, v) -> { - if (Objects.isNull(v)) { - return new PipeTsFilePublicResource(new File(path)); - } else { - v.increaseReferenceCount(); - return v; - } - }); + increaseFileReference(new File(getCommonFilePath(file)), isTsFile, null, file); } public static File getHardlinkOrCopiedFileInPipeDir( @@ -228,13 +236,7 @@ public class PipeTsFileResourceManager { } // Increase the assigner's file to avoid hard-link or memory cache cleaning // Note that it does not exist for historical files - final String commonFilePath = getCommonFilePath(file); - if (hardlinkOrCopiedFileToTsFilePublicResourceMap.containsKey(commonFilePath) - && hardlinkOrCopiedFileToTsFilePublicResourceMap - .get(commonFilePath) - .decreaseReferenceCount()) { - hardlinkOrCopiedFileToPipeTsFileResourceMap.remove(commonFilePath); - } + decreaseFileReference(new File(getCommonFilePath(file)), null); } // Warning: Shall not be called by the assigner
