This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 4ea4636a9b3 Load: Fix active load reliability issues (#18224) (#18270)
4ea4636a9b3 is described below
commit 4ea4636a9b33c5357982aa5e5b066b8a2f382d76
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 22 15:30:32 2026 +0800
Load: Fix active load reliability issues (#18224) (#18270)
* Fix active load reliability issues
* Test active load file group isolation
* Address active load review comments
---
.../org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 61 +++++++++
.../plan/node/load/LoadSingleTsFileNode.java | 26 ++--
.../plan/scheduler/load/LoadTsFileScheduler.java | 15 ++-
.../load/active/ActiveLoadDirScanner.java | 32 ++---
.../load/active/ActiveLoadPendingQueue.java | 20 ++-
.../load/active/ActiveLoadTsFileLoader.java | 40 +++++-
.../storageengine/load/active/ActiveLoadUtil.java | 116 +++++++++++++----
.../plan/planner/node/load/LoadTsFileNodeTest.java | 41 ++++++
.../scheduler/load/LoadTsFileSchedulerTest.java | 45 +++++++
.../load/active/ActiveLoadDirScannerTest.java | 145 +++++++++++++++++++++
.../load/active/ActiveLoadTsFileLoaderTest.java | 20 +++
.../load/active/ActiveLoadUtilTest.java | 139 ++++++++++++++++++++
12 files changed, 640 insertions(+), 60 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
index 03743d1bf4a..c2fbbb0e30b 100644
---
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
@@ -20,8 +20,11 @@
package org.apache.iotdb.db.it;
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
+import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.db.it.utils.TestUtils;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
+import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -1070,6 +1073,64 @@ public class IoTDBLoadTsFileIT {
}
}
+ @Test
+ public void testAsyncLoadKeepsSameNamedTsFilesWithModsIsolated() throws
Exception {
+ registerSchema();
+
+ long expectedPointCount = 0;
+ // Before each file group had its own transfer directory, these same-named
TsFiles and mods
+ // were renamed independently in one shared directory and could be paired
with the wrong file.
+ for (int i = 0; i < 2; i++) {
+ final File sourceDir = new File(tmpDir, "source-" + i);
+ Assert.assertTrue(sourceDir.mkdirs());
+ try (final TsFileGenerator generator =
+ new TsFileGenerator(new File(sourceDir, "1-0-0-0.tsfile"))) {
+ generator.resetRandom(i);
+ generator.registerTimeseries(
+ SchemaConfig.DEVICE_0,
Collections.singletonList(SchemaConfig.MEASUREMENT_00));
+ generator.generateData(SchemaConfig.DEVICE_0, 20, 1, false,
TimeUnit.SECONDS.toMillis(i));
+ // Each group contributes 20 points and its own mods deletes exactly
one. Losing either
+ // TsFile-to-mods pairing therefore leaves 39 points instead of the
expected 38.
+ expectedPointCount += generator.getTotalNumber() - 1;
+ }
+ try (final ModificationFile modificationFile =
+ new ModificationFile(
+ new File(sourceDir, "1-0-0-0.tsfile" +
ModificationFile.FILE_SUFFIX)
+ .getAbsolutePath())) {
+ modificationFile.write(
+ new Deletion(
+ new MeasurementPath(
+ SchemaConfig.DEVICE_0 + "." +
SchemaConfig.MEASUREMENT_00.getMeasurementId()),
+ Long.MAX_VALUE,
+ TimeUnit.SECONDS.toMillis(i) + 1,
+ TimeUnit.SECONDS.toMillis(i) + 1));
+ }
+ }
+
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "load \"%s\" with
('async'='true','database-level'='2','on-success'='delete')",
+ tmpDir.getAbsolutePath()));
+ }
+
+ TestUtils.assertDataEventuallyOnEnv(
+ EnvFactory.getEnv(),
+ "select count("
+ + SchemaConfig.MEASUREMENT_00.getMeasurementId()
+ + ") from "
+ + SchemaConfig.DEVICE_0,
+ Collections.singletonMap(
+ "count("
+ + SchemaConfig.DEVICE_0
+ + "."
+ + SchemaConfig.MEASUREMENT_00.getMeasurementId()
+ + ")",
+ Long.toString(expectedPointCount)),
+ 30);
+ }
+
@Test
public void testLoadTsFileWithWrongTimestampPrecision() throws Exception {
try (final TsFileGenerator generator =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
index a168228deb3..b73068d1395 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
+import org.apache.iotdb.commons.utils.RetryUtils;
import org.apache.iotdb.commons.utils.TimePartitionUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis;
@@ -215,16 +216,23 @@ public class LoadSingleTsFileNode extends WritePlanNode {
}
public void clean() {
+ if (!deleteAfterLoad) {
+ return;
+ }
+ deleteFile(tsFile);
+ deleteFile(new File(tsFile.getAbsolutePath() +
TsFileResource.RESOURCE_SUFFIX));
+ deleteFile(new File(tsFile.getAbsolutePath() +
ModificationFile.FILE_SUFFIX));
+ }
+
+ private void deleteFile(final File file) {
try {
- if (deleteAfterLoad) {
- Files.deleteIfExists(tsFile.toPath());
- Files.deleteIfExists(
- new File(tsFile.getAbsolutePath() +
TsFileResource.RESOURCE_SUFFIX).toPath());
- Files.deleteIfExists(
- new File(tsFile.getAbsolutePath() +
ModificationFile.FILE_SUFFIX).toPath());
- }
- } catch (final IOException e) {
- LOGGER.warn("Delete After Loading {} error.", tsFile, e);
+ RetryUtils.retryOnException(
+ () -> {
+ Files.deleteIfExists(file.toPath());
+ return null;
+ });
+ } catch (final Exception e) {
+ LOGGER.warn("Delete After Loading {} error.", file, e);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
index f0ef7282cc9..58159003a74 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
@@ -702,8 +702,7 @@ public class LoadTsFileScheduler implements IScheduler {
singleTsFileNode
.getTsFileResource()
.getTsFile()))); // can not just remove, because of
deletion
- dataSize -= pieceNode.getDataSize();
- block.reduceMemoryUsage(pieceNode.getDataSize());
+ releaseMemoryUsage(pieceNode.getDataSize());
if (!isDispatchSuccess) {
// Currently there is no retry, so return directly
@@ -788,7 +787,7 @@ public class LoadTsFileScheduler implements IScheduler {
boolean isAllSuccess = true;
for (Map.Entry<TConsensusGroupId, Pair<TRegionReplicaSet,
LoadTsFilePieceNode>> entry :
regionId2ReplicaSetAndNode.entrySet()) {
- block.reduceMemoryUsage(entry.getValue().getRight().getDataSize());
+ releaseMemoryUsage(entry.getValue().getRight().getDataSize());
if (isAllSuccess
&& !scheduler.dispatchOnePieceNode(
entry.getValue().getRight(), entry.getValue().getLeft())) {
@@ -802,7 +801,17 @@ public class LoadTsFileScheduler implements IScheduler {
return isAllSuccess;
}
+ private void releaseMemoryUsage(final long memorySize) {
+ dataSize -= memorySize;
+ block.reduceMemoryUsage(memorySize);
+ }
+
private void clear() {
+ if (dataSize > 0) {
+ block.reduceMemoryUsage(dataSize);
+ dataSize = 0;
+ }
+ nonDirectionalChunkData.clear();
regionId2ReplicaSetAndNode.clear();
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java
index 4a476373fa7..a5c419c4d85 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java
@@ -39,7 +39,6 @@ import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays;
-import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
@@ -109,23 +108,14 @@ public class ActiveLoadDirScanner extends
ActiveLoadScheduledExecutorService {
FileUtils.streamFiles(listeningDirFile, true, (String[]) null)) {
try {
fileStream
+ .map(file -> new File(getTsFilePath(file.getAbsolutePath())))
+ .distinct()
.filter(file ->
!activeLoadTsFileLoader.isFilePendingOrLoading(file))
.filter(File::exists)
- .map(
- file ->
- (file.getName().endsWith(RESOURCE) ||
file.getName().endsWith(MODS))
- ? getTsFilePath(file.getAbsolutePath())
- : file.getAbsolutePath())
- .filter(this::isTsFileCompleted)
+ .filter(file -> isTsFileCompleted(file.getAbsolutePath()))
.limit(currentAllowedPendingSize)
.forEach(
- filePath -> {
- final File tsFile = new File(filePath);
- final Map<String, String> attributes =
- ActiveLoadPathHelper.parseAttributes(tsFile,
listeningDirFile);
-
- final File parentFile = tsFile.getParentFile();
-
+ tsFile -> {
activeLoadTsFileLoader.tryTriggerTsFileLoad(
tsFile.getAbsolutePath(),
listeningDirFile.getAbsolutePath(),
@@ -242,11 +232,15 @@ public class ActiveLoadDirScanner extends
ActiveLoadScheduledExecutorService {
}
private static String getTsFilePath(final String
filePathWithResourceOrModsTail) {
- return filePathWithResourceOrModsTail.endsWith(RESOURCE)
- ? filePathWithResourceOrModsTail.substring(
- 0, filePathWithResourceOrModsTail.length() - RESOURCE.length())
- : filePathWithResourceOrModsTail.substring(
- 0, filePathWithResourceOrModsTail.length() - MODS.length());
+ if (filePathWithResourceOrModsTail.endsWith(RESOURCE)) {
+ return filePathWithResourceOrModsTail.substring(
+ 0, filePathWithResourceOrModsTail.length() - RESOURCE.length());
+ }
+ if (filePathWithResourceOrModsTail.endsWith(MODS)) {
+ return filePathWithResourceOrModsTail.substring(
+ 0, filePathWithResourceOrModsTail.length() - MODS.length());
+ }
+ return filePathWithResourceOrModsTail;
}
// Metrics
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java
index 3ca83956462..f50af8d8d60 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java
@@ -57,9 +57,9 @@ public class ActiveLoadPendingQueue {
}
public synchronized void removeFromLoading(final String file) {
- loadingFileSet.remove(file);
-
-
ActiveLoadingFilesNumberMetricsSet.getInstance().increaseLoadingFileCounter(-1);
+ if (loadingFileSet.remove(file)) {
+
ActiveLoadingFilesNumberMetricsSet.getInstance().increaseLoadingFileCounter(-1);
+ }
}
public synchronized boolean isFilePendingOrLoading(final String file) {
@@ -74,6 +74,20 @@ public class ActiveLoadPendingQueue {
return pendingFileQueue.isEmpty() && loadingFileSet.isEmpty();
}
+ public synchronized void clear() {
+ final int loadingFileCount = loadingFileSet.size();
+ clearPending();
+ loadingFileSet.clear();
+
ActiveLoadingFilesNumberMetricsSet.getInstance().increaseLoadingFileCounter(-loadingFileCount);
+ }
+
+ public synchronized void clearPending() {
+ final int pendingFileCount = pendingFileSet.size();
+ pendingFileSet.clear();
+ pendingFileQueue.clear();
+
ActiveLoadingFilesNumberMetricsSet.getInstance().increaseQueuingFileCounter(-pendingFileCount);
+ }
+
public static class ActiveLoadEntry {
private final String file;
private final String pendingDir;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java
index 9dac9f11759..fddd8223903 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java
@@ -157,12 +157,15 @@ public class ActiveLoadTsFileLoader {
public void stop() {
final WrappedThreadPoolExecutor executor =
activeLoadExecutor.getAndSet(null);
if (executor == null) {
+ pendingQueue.clearPending();
return;
}
- executor.shutdownNow();
+ boolean isTerminated = false;
try {
- if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
+ executor.shutdownNow();
+ isTerminated = executor.awaitTermination(30, TimeUnit.SECONDS);
+ if (!isTerminated) {
LOGGER.warn(
"{} still doesn't exit after 30s",
ThreadName.ACTIVE_LOAD_TSFILE_LOADER.getName());
}
@@ -170,6 +173,12 @@ public class ActiveLoadTsFileLoader {
LOGGER.warn(
"{} still doesn't exit after 30s",
ThreadName.ACTIVE_LOAD_TSFILE_LOADER.getName());
Thread.currentThread().interrupt();
+ } finally {
+ if (isTerminated) {
+ pendingQueue.clear();
+ } else {
+ pendingQueue.clearPending();
+ }
}
}
@@ -207,6 +216,7 @@ public class ActiveLoadTsFileLoader {
handleOtherException(loadEntry.get(), e);
} finally {
pendingQueue.removeFromLoading(loadEntry.get().getFile());
+ cleanupEmptyDirectories(loadEntry.get());
}
}
} finally {
@@ -336,6 +346,32 @@ public class ActiveLoadTsFileLoader {
}
}
+ private void cleanupEmptyDirectories(final
ActiveLoadPendingQueue.ActiveLoadEntry entry) {
+ final File pendingDir =
+ entry.getPendingDir() == null
+ ? ActiveLoadPathHelper.findPendingDirectory(new
File(entry.getFile()))
+ : new File(entry.getPendingDir());
+ if (pendingDir == null) {
+ return;
+ }
+
+ final Path pendingPath = pendingDir.toPath().toAbsolutePath().normalize();
+ Path currentPath = new
File(entry.getFile()).toPath().toAbsolutePath().normalize().getParent();
+ while (currentPath != null
+ && currentPath.startsWith(pendingPath)
+ && !currentPath.equals(pendingPath)) {
+ try {
+ Files.delete(currentPath);
+ } catch (final IOException e) {
+ if (Files.exists(currentPath)) {
+ LOGGER.debug("Failed to delete folder {} when cleaning up",
currentPath, e);
+ }
+ return;
+ }
+ currentPath = currentPath.getParent();
+ }
+ }
+
public boolean isFilePendingOrLoading(final File file) {
return pendingQueue.isFilePendingOrLoading(file.getAbsolutePath());
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtil.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtil.java
index e3dbe43507d..a27bc6882df 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtil.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtil.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.load.active;
+import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.commons.utils.RetryUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.DiskSpaceInsufficientException;
@@ -26,19 +27,22 @@ import
org.apache.iotdb.db.storageengine.load.disk.ILoadDiskSelector;
import org.apache.iotdb.db.storageengine.rescon.disk.FolderManager;
import
org.apache.iotdb.db.storageengine.rescon.disk.strategy.DirectoryStrategyType;
+import org.apache.tsfile.common.constant.TsFileConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-
-import static org.apache.iotdb.commons.utils.FileUtils.copyFileWithMD5Check;
-import static org.apache.iotdb.commons.utils.FileUtils.moveFileWithMD5Check;
+import java.util.UUID;
public class ActiveLoadUtil {
@@ -92,11 +96,13 @@ public class ActiveLoadUtil {
Objects.nonNull(loadAttributes) ? loadAttributes :
Collections.emptyMap();
final File targetDir =
ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes);
- loadTsFileAsyncToTargetDir(
- targetDir, new File(file.getAbsolutePath() + ".resource"),
isDeleteAfterLoad);
- loadTsFileAsyncToTargetDir(
- targetDir, new File(file.getAbsolutePath() + ".mods"),
isDeleteAfterLoad);
- loadTsFileAsyncToTargetDir(targetDir, file, isDeleteAfterLoad);
+ transferFilesToActiveDir(
+ targetDir,
+ Arrays.asList(
+ new File(file.getAbsolutePath() + ".resource"),
+ new File(file.getAbsolutePath() + ".mods"),
+ file),
+ isDeleteAfterLoad);
return true;
}
@@ -127,31 +133,93 @@ public class ActiveLoadUtil {
Objects.nonNull(loadAttributes) ? loadAttributes :
Collections.emptyMap();
final File targetDir =
ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes);
+ final List<File> sourceFiles = new ArrayList<>(files.size());
for (final String file : files) {
- loadTsFileAsyncToTargetDir(targetDir, new File(file), isDeleteAfterLoad);
+ sourceFiles.add(new File(file));
}
+ sourceFiles.sort(Comparator.comparing(ActiveLoadUtil::isTsFile));
+ transferFilesToActiveDir(targetDir, sourceFiles, isDeleteAfterLoad);
return true;
}
- private static void loadTsFileAsyncToTargetDir(
- final File targetDir, final File file, final boolean isDeleteAfterLoad)
throws IOException {
- if (!file.exists()) {
+ static void transferFilesToActiveDir(
+ final File targetDir, final List<File> sourceFiles, final boolean
isDeleteAfterLoad)
+ throws IOException {
+ final List<File> existingSourceFiles = new ArrayList<>(sourceFiles.size());
+ for (final File sourceFile : sourceFiles) {
+ if (sourceFile.exists()) {
+ existingSourceFiles.add(sourceFile);
+ }
+ }
+ if (existingSourceFiles.isEmpty()) {
return;
}
- if (!targetDir.exists() && !targetDir.mkdirs()) {
- if (!targetDir.exists()) {
- throw new IOException("Failed to create target directory " +
targetDir.getAbsolutePath());
+
+ final File transferDir = new File(targetDir, UUID.randomUUID().toString());
+ try {
+ Files.createDirectories(transferDir.toPath());
+ for (final File sourceFile : existingSourceFiles) {
+ final File targetFile = new File(transferDir, sourceFile.getName());
+ RetryUtils.retryOnException(
+ () -> {
+ transferFile(sourceFile, targetFile, isDeleteAfterLoad);
+ return null;
+ });
}
+ } catch (final IOException | RuntimeException e) {
+ if (transferDir.exists()) {
+ FileUtils.deleteFileOrDirectoryWithRetry(transferDir);
+ }
+ throw e;
+ }
+
+ if (isDeleteAfterLoad) {
+ deleteSourceFiles(existingSourceFiles);
}
- RetryUtils.retryOnException(
- () -> {
- if (isDeleteAfterLoad) {
- moveFileWithMD5Check(file, targetDir);
- } else {
- copyFileWithMD5Check(file, targetDir);
- }
- return null;
- });
+ }
+
+ private static void transferFile(
+ final File sourceFile, final File targetFile, final boolean useHardLink)
throws IOException {
+ Exception linkException = null;
+ if (useHardLink) {
+ try {
+ Files.createLink(targetFile.toPath(), sourceFile.toPath());
+ return;
+ } catch (final IOException | UnsupportedOperationException |
SecurityException e) {
+ linkException = e;
+ }
+ }
+
+ try {
+ Files.copy(
+ sourceFile.toPath(),
+ targetFile.toPath(),
+ StandardCopyOption.REPLACE_EXISTING,
+ StandardCopyOption.COPY_ATTRIBUTES);
+ } catch (final IOException e) {
+ if (linkException != null) {
+ e.addSuppressed(linkException);
+ }
+ throw e;
+ }
+ }
+
+ private static void deleteSourceFiles(final List<File> sourceFiles) {
+ for (final File sourceFile : sourceFiles) {
+ try {
+ RetryUtils.retryOnException(
+ () -> {
+ Files.deleteIfExists(sourceFile.toPath());
+ return null;
+ });
+ } catch (final Exception e) {
+ LOGGER.warn("Failed to delete file or dir {}", sourceFile, e);
+ }
+ }
+ }
+
+ private static boolean isTsFile(final File file) {
+ return file.getName().endsWith(TsFileConstant.TSFILE_SUFFIX);
}
public static ILoadDiskSelector updateLoadDiskSelector() {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java
index 849d45c61c8..9dc760e9a65 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadSingleTsFileNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.tsfile.exception.NotImplementedException;
@@ -31,6 +32,7 @@ import org.junit.Test;
import java.io.File;
import java.nio.ByteBuffer;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
@@ -86,4 +88,43 @@ public class LoadTsFileNodeTest {
LoadTsFilePieceNode node1 = (LoadTsFilePieceNode)
LoadTsFilePieceNode.deserialize(buffer);
Assert.assertEquals(node.getTsFile(), node1.getTsFile());
}
+
+ @Test
+ public void testCleanContinuesAfterOneFileCannotBeDeleted() throws Exception
{
+ final File tempDir = Files.createTempDirectory("load-node-clean").toFile();
+ try {
+ // A non-empty directory at the TsFile path makes that deletion fail
deterministically. The
+ // companion cleanup must still continue instead of sharing the same
try-catch block.
+ final File tsFile = new File(tempDir, "1-0-0-0.tsfile");
+ Assert.assertTrue(tsFile.mkdirs());
+ Assert.assertTrue(new File(tsFile, "non-empty").createNewFile());
+ final File resourceFile = new File(tsFile.getAbsolutePath() +
TsFileResource.RESOURCE_SUFFIX);
+ final File modsFile = new File(tsFile.getAbsolutePath() +
ModificationFile.FILE_SUFFIX);
+ Assert.assertTrue(resourceFile.createNewFile());
+ Assert.assertTrue(modsFile.createNewFile());
+
+ final LoadSingleTsFileNode node =
+ new LoadSingleTsFileNode(new PlanNodeId(""), new
TsFileResource(tsFile), true, 0L);
+ node.clean();
+
+ Assert.assertTrue(tsFile.exists());
+ Assert.assertFalse(resourceFile.exists());
+ Assert.assertFalse(modsFile.exists());
+ } finally {
+ deleteRecursively(tempDir);
+ }
+ }
+
+ private static void deleteRecursively(final File file) {
+ if (file == null || !file.exists()) {
+ return;
+ }
+ final File[] children = file.listFiles();
+ if (children != null) {
+ for (final File child : children) {
+ deleteRecursively(child);
+ }
+ }
+ Assert.assertTrue(file.delete());
+ }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java
index 161a95b4ec9..d1151aef7da 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java
@@ -27,6 +27,8 @@ import
org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.planner.plan.DistributedQueryPlan;
import org.apache.iotdb.db.queryengine.plan.planner.plan.PlanFragment;
import org.apache.iotdb.db.queryengine.plan.planner.plan.SubPlan;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadSingleTsFileNode;
+import
org.apache.iotdb.db.storageengine.load.memory.LoadTsFileDataCacheMemoryBlock;
import org.junit.Assert;
import org.junit.Before;
@@ -34,6 +36,10 @@ import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
@@ -67,4 +73,43 @@ public class LoadTsFileSchedulerTest {
Assert.assertNull(t.getTotalCpuTime());
Assert.assertNull(t.getFragmentInfo());
}
+
+ @Test
+ public void testTsFileDataManagerClearReleasesCachedMemory() throws
Exception {
+ final Constructor<LoadTsFileDataCacheMemoryBlock> memoryBlockConstructor =
+
LoadTsFileDataCacheMemoryBlock.class.getDeclaredConstructor(long.class);
+ memoryBlockConstructor.setAccessible(true);
+ final LoadTsFileDataCacheMemoryBlock memoryBlock =
+ memoryBlockConstructor.newInstance(1024 * 1024L);
+
+ final Class<?> dataManagerClass =
+ Class.forName(LoadTsFileScheduler.class.getName() +
"$TsFileDataManager");
+ final Constructor<?> dataManagerConstructor =
+ dataManagerClass.getDeclaredConstructor(
+ LoadTsFileScheduler.class,
+ LoadSingleTsFileNode.class,
+ LoadTsFileDataCacheMemoryBlock.class);
+ dataManagerConstructor.setAccessible(true);
+ final Object dataManager =
+ dataManagerConstructor.newInstance(
+ mock(LoadTsFileScheduler.class), mock(LoadSingleTsFileNode.class),
memoryBlock);
+
+ // Simulate data buffered before split or routing aborts. clear() is the
last chance to return
+ // this accounting to the shared LOAD memory block.
+ final long cachedMemorySize = 128L;
+ memoryBlock.addMemoryUsage(cachedMemorySize);
+ final Field dataSizeField = dataManagerClass.getDeclaredField("dataSize");
+ dataSizeField.setAccessible(true);
+ dataSizeField.setLong(dataManager, cachedMemorySize);
+
+ final Method clearMethod = dataManagerClass.getDeclaredMethod("clear");
+ clearMethod.setAccessible(true);
+ clearMethod.invoke(dataManager);
+
+ final Method getMemoryUsageMethod =
+
LoadTsFileDataCacheMemoryBlock.class.getDeclaredMethod("getMemoryUsageInBytes");
+ getMemoryUsageMethod.setAccessible(true);
+ Assert.assertEquals(0L, getMemoryUsageMethod.invoke(memoryBlock));
+ Assert.assertEquals(0L, dataSizeField.getLong(dataManager));
+ }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java
new file mode 100644
index 00000000000..5703ee0ec55
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.storageengine.load.active;
+
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+
+import org.apache.tsfile.write.writer.TsFileIOWriter;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.nio.file.Files;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class ActiveLoadDirScannerTest {
+
+ private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private String[] originalListeningDirs;
+ private String originalPipeDir;
+ private boolean originalListeningEnabled;
+ private File tempDir;
+ private File pendingDir;
+ private File pipeDir;
+
+ @Before
+ public void setUp() throws Exception {
+ originalListeningDirs = config.getLoadActiveListeningDirs();
+ originalPipeDir = config.getLoadActiveListeningPipeDir();
+ originalListeningEnabled = config.getLoadActiveListeningEnable();
+
+ tempDir = Files.createTempDirectory("active-load-scanner").toFile();
+ pendingDir = new File(tempDir, "pending");
+ pipeDir = new File(tempDir, "pipe");
+ Assert.assertTrue(pendingDir.mkdirs());
+ Assert.assertTrue(pipeDir.mkdirs());
+ config.setLoadActiveListeningDirs(new String[]
{pendingDir.getAbsolutePath()});
+ config.setLoadActiveListeningPipeDir(pipeDir.getAbsolutePath());
+ config.setLoadActiveListeningEnable(true);
+ }
+
+ @After
+ public void tearDown() {
+ config.setLoadActiveListeningDirs(originalListeningDirs);
+ config.setLoadActiveListeningPipeDir(originalPipeDir);
+ config.setLoadActiveListeningEnable(originalListeningEnabled);
+ ActiveLoadUtil.updateLoadDiskSelector();
+ deleteRecursively(tempDir);
+ }
+
+ @Test
+ public void testScanDeduplicatesTsFileAndCompanionFiles() throws Exception {
+ // The recursive scanner sees all three paths, but every companion maps
back to the same TsFile.
+ // Enqueuing each path separately used to consume queue capacity and
schedule duplicate loads.
+ final File tsFile = createCompletedTsFile(pendingDir, "1-0-0-0.tsfile");
+ Assert.assertTrue(
+ new File(tsFile.getAbsolutePath() +
TsFileResource.RESOURCE_SUFFIX).createNewFile());
+ Assert.assertTrue(
+ new File(tsFile.getAbsolutePath() +
ModificationFile.FILE_SUFFIX).createNewFile());
+
+ final ActiveLoadTsFileLoader loader = mock(ActiveLoadTsFileLoader.class);
+ when(loader.getCurrentAllowedPendingSize()).thenReturn(10);
+ final ActiveLoadDirScanner scanner = new ActiveLoadDirScanner(loader);
+ try {
+ final Method scanMethod =
ActiveLoadDirScanner.class.getDeclaredMethod("scan");
+ scanMethod.setAccessible(true);
+ scanMethod.invoke(scanner);
+ } finally {
+ scanner.stop();
+ }
+
+ verify(loader, times(1))
+ .tryTriggerTsFileLoad(
+ eq(tsFile.getAbsolutePath()), eq(pendingDir.getAbsolutePath()),
eq(false));
+ }
+
+ @Test
+ public void testScanLoadsTsFileWithoutCompanionFiles() throws Exception {
+ final File tsFile = createCompletedTsFile(pendingDir, "2-0-0-0.tsfile");
+
+ final ActiveLoadTsFileLoader loader = mock(ActiveLoadTsFileLoader.class);
+ when(loader.getCurrentAllowedPendingSize()).thenReturn(10);
+ final ActiveLoadDirScanner scanner = new ActiveLoadDirScanner(loader);
+ try {
+ final Method scanMethod =
ActiveLoadDirScanner.class.getDeclaredMethod("scan");
+ scanMethod.setAccessible(true);
+ scanMethod.invoke(scanner);
+ } finally {
+ scanner.stop();
+ }
+
+ verify(loader, times(1))
+ .tryTriggerTsFileLoad(
+ eq(tsFile.getAbsolutePath()), eq(pendingDir.getAbsolutePath()),
eq(false));
+ }
+
+ private static File createCompletedTsFile(final File dir, final String
fileName)
+ throws Exception {
+ final File tsFile = new File(dir, fileName);
+ try (final TsFileIOWriter writer = new TsFileIOWriter(tsFile)) {
+ writer.endFile();
+ }
+ return tsFile;
+ }
+
+ private static void deleteRecursively(final File file) {
+ if (file == null || !file.exists()) {
+ return;
+ }
+ final File[] children = file.listFiles();
+ if (children != null) {
+ for (final File child : children) {
+ deleteRecursively(child);
+ }
+ }
+ Assert.assertTrue(file.delete());
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java
index f21748c3d3e..9a805971191 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java
@@ -34,6 +34,7 @@ import org.junit.Before;
import org.junit.Test;
import java.io.File;
+import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.nio.file.Files;
@@ -103,6 +104,25 @@ public class ActiveLoadTsFileLoaderTest {
Assert.assertTrue(new File(failDir, tsFile.getName() +
ModificationFile.FILE_SUFFIX).exists());
}
+ @Test
+ public void testStopClearsPendingFilesForRestart() throws Exception {
+ final ActiveLoadTsFileLoader loader = new ActiveLoadTsFileLoader();
+ final Field pendingQueueField =
ActiveLoadTsFileLoader.class.getDeclaredField("pendingQueue");
+ pendingQueueField.setAccessible(true);
+ final ActiveLoadPendingQueue pendingQueue =
+ (ActiveLoadPendingQueue) pendingQueueField.get(loader);
+ final String tsFilePath = new File(tempDir,
"pending.tsfile").getAbsolutePath();
+ Assert.assertTrue(pendingQueue.enqueue(tsFilePath,
tempDir.getAbsolutePath(), false));
+ Assert.assertTrue(loader.isFilePendingOrLoading(new File(tsFilePath)));
+
+ // A loader restart reuses this queue. Stale pending membership otherwise
makes the scanner
+ // believe the on-disk TsFile is already scheduled and it will never
enqueue it again.
+ loader.stop();
+
+ Assert.assertFalse(loader.isFilePendingOrLoading(new File(tsFilePath)));
+ Assert.assertTrue(pendingQueue.isEmpty());
+ }
+
private File createTsFileWithCompanionFiles(final String fileName) throws
Exception {
final File tsFile = new File(tempDir, fileName);
Assert.assertTrue(tsFile.createNewFile());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtilTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtilTest.java
new file mode 100644
index 00000000000..c0aaa8c9349
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadUtilTest.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.storageengine.load.active;
+
+import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class ActiveLoadUtilTest {
+
+ private File tempDir;
+ private File sourceDir;
+ private File targetDir;
+
+ @Before
+ public void setUp() throws Exception {
+ tempDir = Files.createTempDirectory("load-util").toFile();
+ sourceDir = new File(tempDir, "source");
+ targetDir = new File(tempDir, "target");
+ Assert.assertTrue(sourceDir.mkdirs());
+ Assert.assertTrue(targetDir.mkdirs());
+ }
+
+ @After
+ public void tearDown() {
+ deleteRecursively(tempDir);
+ }
+
+ @Test
+ public void
testTransferFilesKeepsSameNamedGroupsIsolatedAndDeletesSourcesAfterHandoff()
+ throws Exception {
+ final List<File> sourceFiles = createTsFileAndCompanions();
+ ActiveLoadUtil.transferFilesToActiveDir(targetDir, sourceFiles, true);
+
+ for (final File sourceFile : sourceFiles) {
+ Files.write(
+ sourceFile.toPath(), ("second-" +
sourceFile.getName()).getBytes(StandardCharsets.UTF_8));
+ }
+ ActiveLoadUtil.transferFilesToActiveDir(targetDir, sourceFiles, true);
+
+ final File[] transferDirs = targetDir.listFiles(File::isDirectory);
+ Assert.assertNotNull(transferDirs);
+ Assert.assertEquals(2, transferDirs.length);
+ for (final File sourceFile : sourceFiles) {
+ Assert.assertFalse(sourceFile.exists());
+ }
+
+ final Set<String> transferredPrefixes = new HashSet<>();
+ for (final File transferDir : transferDirs) {
+ final File tsFile = new File(transferDir, "1-0-0-0.tsfile");
+ final String tsFileContent =
+ new String(Files.readAllBytes(tsFile.toPath()),
StandardCharsets.UTF_8);
+ final String prefix = tsFileContent.startsWith("second-") ? "second-" :
"";
+ transferredPrefixes.add(prefix);
+ for (final File sourceFile : sourceFiles) {
+ final File transferredFile = new File(transferDir,
sourceFile.getName());
+ Assert.assertTrue(transferredFile.exists());
+ Assert.assertArrayEquals(
+ (prefix + sourceFile.getName()).getBytes(StandardCharsets.UTF_8),
+ Files.readAllBytes(transferredFile.toPath()));
+ }
+ }
+ Assert.assertEquals(new HashSet<>(Arrays.asList("", "second-")),
transferredPrefixes);
+ }
+
+ @Test
+ public void testTransferFailureDoesNotDeleteSources() throws Exception {
+ final List<File> sourceFiles = createTsFileAndCompanions();
+ // A regular file cannot contain the temporary transfer directory, forcing
handoff to fail
+ // before ownership of any source file can be released.
+ final File invalidTargetDir = new File(tempDir, "target-file");
+ Assert.assertTrue(invalidTargetDir.createNewFile());
+
+ try {
+ ActiveLoadUtil.transferFilesToActiveDir(invalidTargetDir, sourceFiles,
true);
+ Assert.fail("Expected IOException");
+ } catch (final IOException ignored) {
+ // expected
+ }
+
+ for (final File sourceFile : sourceFiles) {
+ Assert.assertTrue(sourceFile.exists());
+ }
+ }
+
+ private List<File> createTsFileAndCompanions() throws Exception {
+ final File tsFile = new File(sourceDir, "1-0-0-0.tsfile");
+ final File resourceFile = new File(tsFile.getAbsolutePath() +
TsFileResource.RESOURCE_SUFFIX);
+ final File modsFile = new File(tsFile.getAbsolutePath() +
ModificationFile.FILE_SUFFIX);
+ final List<File> sourceFiles = Arrays.asList(resourceFile, modsFile,
tsFile);
+ for (final File sourceFile : sourceFiles) {
+ Files.write(sourceFile.toPath(),
sourceFile.getName().getBytes(StandardCharsets.UTF_8));
+ }
+ return sourceFiles;
+ }
+
+ private static void deleteRecursively(final File file) {
+ if (file == null || !file.exists()) {
+ return;
+ }
+ final File[] children = file.listFiles();
+ if (children != null) {
+ for (final File child : children) {
+ deleteRecursively(child);
+ }
+ }
+ Assert.assertTrue(file.delete());
+ }
+}
