You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/10/24 11:33:44 UTC
(iotdb) 01/01: Revert "Load: Parallelly load files into different target data partitions (#13893)"
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch revert-load-shit
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7d6c7cf3da6a3614caf9d2265d7034ab56876846
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Thu Oct 24 19:33:27 2024 +0800
Revert "Load: Parallelly load files into different target data partitions (#13893)"
This reverts commit dbb99bc88dea50c6effffae4081abbba1ea5f76a.
---
.../db/storageengine/load/LoadTsFileManager.java | 57 +++++++---------------
1 file changed, 17 insertions(+), 40 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java
index f785d0992eb..775dad1df46 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java
@@ -61,11 +61,8 @@ import java.io.IOException;
import java.nio.file.DirectoryNotEmptyException;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
@@ -455,47 +452,27 @@ public class LoadTsFileManager {
if (isClosed) {
throw new IOException(String.format(MESSAGE_WRITER_MANAGER_HAS_BEEN_CLOSED, taskDir));
}
-
for (Map.Entry<DataPartitionInfo, ModificationFile> entry :
dataPartition2ModificationFile.entrySet()) {
entry.getValue().close();
}
-
- final List<Map.Entry<DataPartitionInfo, TsFileIOWriter>> dataPartition2WriterList =
- new ArrayList<>(dataPartition2Writer.entrySet());
- Collections.shuffle(dataPartition2WriterList);
-
- final AtomicReference<Exception> exception = new AtomicReference<>();
- dataPartition2WriterList.parallelStream()
- .forEach(
- entry -> {
- try {
- final TsFileIOWriter writer = entry.getValue();
- if (writer.isWritingChunkGroup()) {
- writer.endChunkGroup();
- }
- writer.endFile();
-
- final DataRegion dataRegion = entry.getKey().getDataRegion();
- dataRegion.loadNewTsFile(
- generateResource(writer, progressIndex), true, isGeneratedByPipe);
-
- // Metrics
- dataRegion
- .getNonSystemDatabaseName()
- .ifPresent(
- databaseName ->
- updateWritePointCountMetrics(
- dataRegion,
- databaseName,
- getTsFileWritePointCount(writer),
- false));
- } catch (final Exception e) {
- exception.set(e);
- }
- });
- if (exception.get() != null) {
- throw new LoadFileException(exception.get());
+ for (Map.Entry<DataPartitionInfo, TsFileIOWriter> entry : dataPartition2Writer.entrySet()) {
+ TsFileIOWriter writer = entry.getValue();
+ if (writer.isWritingChunkGroup()) {
+ writer.endChunkGroup();
+ }
+ writer.endFile();
+
+ DataRegion dataRegion = entry.getKey().getDataRegion();
+ dataRegion.loadNewTsFile(generateResource(writer, progressIndex), true, isGeneratedByPipe);
+
+ // Metrics
+ dataRegion
+ .getNonSystemDatabaseName()
+ .ifPresent(
+ databaseName ->
+ updateWritePointCountMetrics(
+ dataRegion, databaseName, getTsFileWritePointCount(writer), false));
}
}