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 ea0b0099227 Load: Fix active load reliability issues (#18224)
ea0b0099227 is described below
commit ea0b0099227a850254040ab87436eb5bbd092d60
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 21 09:52:37 2026 +0800
Load: Fix active load reliability issues (#18224)
* Fix active load reliability issues
* Test active load file group isolation
* Address active load review comments
---
.../org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 59 ++++++++
.../plan/node/load/LoadSingleTsFileNode.java | 28 ++--
.../plan/scheduler/load/LoadTsFileScheduler.java | 15 +-
.../load/active/ActiveLoadDirScanner.java | 11 +-
.../load/active/ActiveLoadPendingQueue.java | 20 ++-
.../load/active/ActiveLoadTsFileLoader.java | 40 +++++-
.../iotdb/db/storageengine/load/util/LoadUtil.java | 120 ++++++++++++----
.../plan/planner/node/load/LoadTsFileNodeTest.java | 46 ++++++
.../scheduler/load/LoadTsFileSchedulerTest.java | 42 ++++++
.../load/active/ActiveLoadDirScannerTest.java | 157 +++++++++++++++++++++
.../load/active/ActiveLoadTsFileLoaderTest.java | 20 +++
.../db/storageengine/load/util/LoadUtilTest.java | 141 ++++++++++++++++++
12 files changed, 649 insertions(+), 50 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 1eda0f06226..8a0800771cf 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.commons.schema.column.ColumnHeaderConstant;
import org.apache.iotdb.db.it.utils.TestUtils;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -985,6 +988,62 @@ 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), false)) {
+ modificationFile.write(
+ new TreeDeletionEntry(
+ new MeasurementPath(
+ SchemaConfig.DEVICE_0,
SchemaConfig.MEASUREMENT_00.getMeasurementName()),
+ 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.getMeasurementName()
+ + ") from "
+ + SchemaConfig.DEVICE_0,
+ Collections.singletonMap(
+ "count("
+ + SchemaConfig.DEVICE_0
+ + "."
+ + SchemaConfig.MEASUREMENT_00.getMeasurementName()
+ + ")",
+ 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 dae0a63e974..ecaf0aea6ec 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
@@ -25,6 +25,7 @@ import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+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.i18n.DataNodeQueryMessages;
@@ -241,17 +242,24 @@ public class LoadSingleTsFileNode extends WritePlanNode {
}
public void clean() {
+ if (!deleteAfterLoad) {
+ return;
+ }
+ deleteFile(tsFile);
+ deleteFile(new
File(LoadUtil.getTsFileResourcePath(tsFile.getAbsolutePath())));
+ deleteFile(ModificationFile.getExclusiveMods(tsFile));
+ deleteFile(new
File(LoadUtil.getTsFileModsV1Path(tsFile.getAbsolutePath())));
+ }
+
+ private void deleteFile(final File file) {
try {
- if (deleteAfterLoad) {
- Files.deleteIfExists(tsFile.toPath());
- Files.deleteIfExists(
- new
File(LoadUtil.getTsFileResourcePath(tsFile.getAbsolutePath())).toPath());
-
Files.deleteIfExists(ModificationFile.getExclusiveMods(tsFile).toPath());
- Files.deleteIfExists(
- new
File(LoadUtil.getTsFileModsV1Path(tsFile.getAbsolutePath())).toPath());
- }
- } catch (final IOException e) {
- LOGGER.warn(DataNodeQueryMessages.DELETE_AFTER_LOADING_ERROR, tsFile, e);
+ RetryUtils.retryOnException(
+ () -> {
+ Files.deleteIfExists(file.toPath());
+ return null;
+ });
+ } catch (final Exception e) {
+ LOGGER.warn(DataNodeQueryMessages.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 d3beefc0802..ef2e948c0e6 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
@@ -800,8 +800,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
@@ -886,7 +885,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())) {
@@ -900,7 +899,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 cb3460d78e7..9d1ad235ecb 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
@@ -108,21 +108,22 @@ public class ActiveLoadDirScanner extends
ActiveLoadScheduledExecutorService {
FileUtils.streamFiles(listeningDirFile, true, (String[]) null)) {
try {
fileStream
+ .map(file -> new
File(LoadUtil.getTsFilePath(file.getAbsolutePath())))
+ .distinct()
.filter(file ->
!activeLoadTsFileLoader.isFilePendingOrLoading(file))
.filter(File::exists)
- .map(file -> LoadUtil.getTsFilePath(file.getAbsolutePath()))
- .filter(this::isTsFileCompleted)
+ .filter(file -> isTsFileCompleted(file.getAbsolutePath()))
.limit(currentAllowedPendingSize)
.forEach(
- filePath -> {
- final File tsFile = new File(filePath);
+ tsFile -> {
final Map<String, String> attributes =
ActiveLoadPathHelper.parseAttributes(tsFile,
listeningDirFile);
final File parentFile = tsFile.getParentFile();
final boolean isTableModel =
ActiveLoadPathHelper.containsDatabaseName(attributes)
- || (parentFile != null
+ || (attributes.isEmpty()
+ && parentFile != null
&& !Objects.equals(
parentFile.getAbsoluteFile(),
listeningDirFile.getAbsoluteFile()));
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 88c18b19cb6..060935f51d0 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
@@ -61,9 +61,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) {
@@ -78,6 +78,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 ff5222e5754..1f0c1786608 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
@@ -160,12 +160,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(
StorageEngineMessages.STILL_NOT_EXIT_AFTER_30S,
ThreadName.ACTIVE_LOAD_TSFILE_LOADER.getName());
@@ -175,6 +178,12 @@ public class ActiveLoadTsFileLoader {
StorageEngineMessages.STILL_NOT_EXIT_AFTER_30S,
ThreadName.ACTIVE_LOAD_TSFILE_LOADER.getName());
Thread.currentThread().interrupt();
+ } finally {
+ if (isTerminated) {
+ pendingQueue.clear();
+ } else {
+ pendingQueue.clearPending();
+ }
}
}
@@ -213,6 +222,7 @@ public class ActiveLoadTsFileLoader {
handleOtherException(loadEntry.get(), e);
} finally {
pendingQueue.removeFromLoading(loadEntry.get().getFile());
+ cleanupEmptyDirectories(loadEntry.get());
}
}
} finally {
@@ -364,6 +374,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(StorageEngineMessages.FAILED_DELETE_FOLDER_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/util/LoadUtil.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java
index 88c71fb1cef..4a67b362a5b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.storageengine.load.util;
import org.apache.iotdb.commons.disk.FolderManager;
import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException;
+import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.commons.utils.RetryUtils;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -34,19 +35,22 @@ import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.load.active.ActiveLoadPathHelper;
import org.apache.iotdb.db.storageengine.load.disk.ILoadDiskSelector;
+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.Comparator;
import java.util.LinkedHashMap;
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 LoadUtil {
@@ -130,13 +134,14 @@ public class LoadUtil {
final Map<String, String> attributes =
appendCurrentUserIfAbsent(loadAttributes);
final File targetDir =
ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes);
- loadTsFileAsyncToTargetDir(
- targetDir, new File(getTsFileResourcePath(file.getAbsolutePath())),
isDeleteAfterLoad);
- loadTsFileAsyncToTargetDir(
- targetDir, new File(getTsFileModsV1Path(file.getAbsolutePath())),
isDeleteAfterLoad);
- loadTsFileAsyncToTargetDir(
- targetDir, new File(getTsFileModsV2Path(file.getAbsolutePath())),
isDeleteAfterLoad);
- loadTsFileAsyncToTargetDir(targetDir, file, isDeleteAfterLoad);
+ transferFilesToActiveDir(
+ targetDir,
+ Arrays.asList(
+ new File(getTsFileResourcePath(file.getAbsolutePath())),
+ new File(getTsFileModsV1Path(file.getAbsolutePath())),
+ new File(getTsFileModsV2Path(file.getAbsolutePath())),
+ file),
+ isDeleteAfterLoad);
return true;
}
@@ -183,32 +188,93 @@ public class LoadUtil {
final Map<String, String> attributes =
appendCurrentUserIfAbsent(loadAttributes);
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(LoadUtil::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(
- StorageEngineMessages.FAILED_TO_CREATE_TARGET_DIR +
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(StorageEngineMessages.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 e5898108946..37cedc50b34 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,7 +23,9 @@ import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
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.iotdb.db.storageengine.load.util.LoadUtil;
import org.apache.tsfile.exception.NotImplementedException;
import org.junit.Assert;
@@ -31,6 +33,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;
@@ -89,4 +92,47 @@ 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(LoadUtil.getTsFileResourcePath(tsFile.getAbsolutePath()));
+ final File modsV2File = ModificationFile.getExclusiveMods(tsFile);
+ final File modsV1File = new
File(LoadUtil.getTsFileModsV1Path(tsFile.getAbsolutePath()));
+ Assert.assertTrue(resourceFile.createNewFile());
+ Assert.assertTrue(modsV2File.createNewFile());
+ Assert.assertTrue(modsV1File.createNewFile());
+
+ final LoadSingleTsFileNode node =
+ new LoadSingleTsFileNode(
+ new PlanNodeId(""), new TsFileResource(tsFile), false, null,
true, 0L, false);
+ node.clean();
+
+ Assert.assertTrue(tsFile.exists());
+ Assert.assertFalse(resourceFile.exists());
+ Assert.assertFalse(modsV2File.exists());
+ Assert.assertFalse(modsV1File.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 19d97490c81..1b738f6496d 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
@@ -29,6 +29,7 @@ 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.queryengine.plan.statement.crud.LoadTsFileStatement;
+import
org.apache.iotdb.db.storageengine.load.memory.LoadTsFileDataCacheMemoryBlock;
import org.junit.Assert;
import org.junit.Before;
@@ -37,6 +38,8 @@ import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import java.io.File;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
import java.lang.reflect.Method;
import static org.mockito.Mockito.mock;
@@ -118,4 +121,43 @@ public class LoadTsFileSchedulerTest {
Assert.assertEquals(2, statement.getDatabaseLevel());
Assert.assertTrue(statement.isGeneratedByPipe());
}
+
+ @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..60532f749d8
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.modification.v1.ModificationFileV1;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.load.util.LoadUtil;
+
+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 java.util.Map;
+
+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);
+ LoadUtil.updateLoadDiskSelector();
+ deleteRecursively(tempDir);
+ }
+
+ @Test
+ public void testScanDeduplicatesTsFileAndCompanionFiles() throws Exception {
+ // The recursive scanner sees all four 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() +
ModificationFileV1.FILE_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), eq(false));
+ }
+
+ @Test
+ public void testAttributeAndTransferDirectoriesDoNotImplyTableModel() throws
Exception {
+ // Async tree loads add attribute and per-handoff transfer directories
below pending. These are
+ // internal directories, not table database names inferred from a
user-created subdirectory.
+ final Map<String, String> attributes =
+ ActiveLoadPathHelper.buildAttributes(null, 2, false, false, null,
false, "test-user");
+ final File attributeDir =
ActiveLoadPathHelper.resolveTargetDir(pendingDir, attributes);
+ final File transferDir = new File(attributeDir, "transfer-id");
+ Assert.assertTrue(transferDir.mkdirs());
+ final File tsFile = createCompletedTsFile(transferDir, "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), 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 3208dc16ef5..aa81de4c53b 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
@@ -35,6 +35,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;
@@ -107,6 +108,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, 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/util/LoadUtilTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java
new file mode 100644
index 00000000000..f9537835da7
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.util;
+
+import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1;
+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 LoadUtilTest {
+
+ 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();
+ LoadUtil.transferFilesToActiveDir(targetDir, sourceFiles, true);
+
+ for (final File sourceFile : sourceFiles) {
+ Files.write(
+ sourceFile.toPath(), ("second-" +
sourceFile.getName()).getBytes(StandardCharsets.UTF_8));
+ }
+ LoadUtil.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 {
+ LoadUtil.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 modsV1File = new File(tsFile.getAbsolutePath() +
ModificationFileV1.FILE_SUFFIX);
+ final File modsV2File = new File(tsFile.getAbsolutePath() +
ModificationFile.FILE_SUFFIX);
+ final List<File> sourceFiles = Arrays.asList(resourceFile, modsV1File,
modsV2File, 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());
+ }
+}
