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 2025/05/26 12:49:43 UTC
(iotdb) branch fix-15426 created (now c989b855408)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch fix-15426
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at c989b855408 Pipe/Load: Assign distinct progress indexes for loading tsfiles in time partitions to reduce pipe reprocessing after restart
This branch includes the following new commits:
new c989b855408 Pipe/Load: Assign distinct progress indexes for loading tsfiles in time partitions to reduce pipe reprocessing after restart
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: Pipe/Load: Assign distinct progress indexes for loading tsfiles in time partitions to reduce pipe reprocessing after restart
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch fix-15426
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c989b8554085326276f4516304fb3f0ed4945eb3
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon May 26 20:49:33 2025 +0800
Pipe/Load: Assign distinct progress indexes for loading tsfiles in time partitions to reduce pipe reprocessing after restart
---
...oricalDataRegionTsFileAndDeletionExtractor.java | 13 ++------
.../impl/DataNodeInternalRPCServiceImpl.java | 21 +++++++-----
.../scheduler/load/LoadTsFileDispatcherImpl.java | 26 +++++++++------
.../plan/scheduler/load/LoadTsFileScheduler.java | 39 +++++++++++++++++-----
.../iotdb/db/storageengine/StorageEngine.java | 5 +--
.../db/storageengine/load/LoadTsFileManager.java | 22 +++++++++---
.../src/main/thrift/datanode.thrift | 2 +-
7 files changed, 83 insertions(+), 45 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
index 181c2e440f3..e223686c1cd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
@@ -615,17 +615,10 @@ public class PipeHistoricalDataRegionTsFileAndDeletionExtractor
}
if (startIndex instanceof StateProgressIndex) {
- // Some different tsFiles may share the same max progressIndex, thus tsFiles with an
- // "equals" max progressIndex must be transmitted to avoid data loss
- final ProgressIndex innerProgressIndex =
- ((StateProgressIndex) startIndex).getInnerProgressIndex();
- return !innerProgressIndex.isAfter(resource.getMaxProgressIndexAfterClose())
- && !innerProgressIndex.equals(resource.getMaxProgressIndexAfterClose());
+ startIndex = ((StateProgressIndex) startIndex).getInnerProgressIndex();
}
-
- // Some different tsFiles may share the same max progressIndex, thus tsFiles with an
- // "equals" max progressIndex must be transmitted to avoid data loss
- return !startIndex.isAfter(resource.getMaxProgressIndexAfterClose());
+ return !startIndex.isAfter(resource.getMaxProgressIndexAfterClose())
+ && !startIndex.equals(resource.getMaxProgressIndexAfterClose());
}
private boolean mayTsFileResourceOverlappedWithPattern(final TsFileResource resource) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index d427703cce5..68c9cb8aa15 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.common.rpc.thrift.TSettleReq;
import org.apache.iotdb.common.rpc.thrift.TShowConfigurationResp;
import org.apache.iotdb.common.rpc.thrift.TTestConnectionResp;
import org.apache.iotdb.common.rpc.thrift.TTestConnectionResult;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.client.request.AsyncRequestContext;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.conf.CommonConfig;
@@ -517,14 +518,18 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@Override
public TLoadResp sendLoadCommand(TLoadCommandReq req) {
- final ProgressIndex progressIndex;
- if (req.isSetProgressIndex()) {
- progressIndex = ProgressIndexType.deserializeFrom(ByteBuffer.wrap(req.getProgressIndex()));
+ final Map<TTimePartitionSlot, ProgressIndex> timePartitionProgressIndexMap = new HashMap<>();
+ if (req.isSetTimePartition2ProgressIndex()) {
+ for (Map.Entry<TTimePartitionSlot, ByteBuffer> entry :
+ req.getTimePartition2ProgressIndex().entrySet()) {
+ timePartitionProgressIndexMap.put(
+ entry.getKey(), ProgressIndexType.deserializeFrom(entry.getValue()));
+ }
} else {
- // fallback to use local generated progress index for compatibility
- progressIndex = PipeDataNodeAgent.runtime().getNextProgressIndexForTsFileLoad();
- LOGGER.info(
- "Use local generated load progress index {} for uuid {}.", progressIndex, req.uuid);
+ final TSStatus status = new TSStatus();
+ status.setCode(TSStatusCode.LOAD_FILE_ERROR.getStatusCode());
+ status.setMessage("Load command requires time partition to progress index map");
+ return createTLoadResp(status);
}
return createTLoadResp(
@@ -533,7 +538,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
LoadTsFileScheduler.LoadCommand.values()[req.commandType],
req.uuid,
req.isSetIsGeneratedByPipe() && req.isGeneratedByPipe,
- progressIndex));
+ timePartitionProgressIndexMap));
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
index 5227f230007..819cd3abbfe 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.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.TSStatus;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
@@ -56,8 +57,10 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
@@ -256,17 +259,18 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher {
private void dispatchLocally(TLoadCommandReq loadCommandReq)
throws FragmentInstanceDispatchException {
- final ProgressIndex progressIndex;
- if (loadCommandReq.isSetProgressIndex()) {
- progressIndex =
- ProgressIndexType.deserializeFrom(ByteBuffer.wrap(loadCommandReq.getProgressIndex()));
+ final Map<TTimePartitionSlot, ProgressIndex> timePartitionProgressIndexMap = new HashMap<>();
+ if (loadCommandReq.isSetTimePartition2ProgressIndex()) {
+ for (Map.Entry<TTimePartitionSlot, ByteBuffer> entry :
+ loadCommandReq.getTimePartition2ProgressIndex().entrySet()) {
+ timePartitionProgressIndexMap.put(
+ entry.getKey(), ProgressIndexType.deserializeFrom(entry.getValue()));
+ }
} else {
- // fallback to use local generated progress index for compatibility
- progressIndex = PipeDataNodeAgent.runtime().getNextProgressIndexForTsFileLoad();
- LOGGER.info(
- "Use local generated load progress index {} for uuid {}.",
- progressIndex,
- loadCommandReq.uuid);
+ final TSStatus status = new TSStatus();
+ status.setCode(TSStatusCode.LOAD_FILE_ERROR.getStatusCode());
+ status.setMessage("Load command requires time partition to progress index map");
+ throw new FragmentInstanceDispatchException(status);
}
final TSStatus resultStatus =
@@ -275,7 +279,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher {
LoadTsFileScheduler.LoadCommand.values()[loadCommandReq.commandType],
loadCommandReq.uuid,
loadCommandReq.isSetIsGeneratedByPipe() && loadCommandReq.isGeneratedByPipe,
- progressIndex);
+ timePartitionProgressIndexMap);
if (!RpcUtils.SUCCESS_STATUS.equals(resultStatus)) {
throw new FragmentInstanceDispatchException(resultStatus);
}
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 ead789b1bcd..6b5002e1891 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
@@ -29,6 +29,7 @@ import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
@@ -140,6 +141,7 @@ public class LoadTsFileScheduler implements IScheduler {
private final PlanFragmentId fragmentId;
private final Set<TRegionReplicaSet> allReplicaSets;
private final boolean isGeneratedByPipe;
+ private final Map<TTimePartitionSlot, ProgressIndex> timePartitionSlotToProgressIndex;
private final LoadTsFileDataCacheMemoryBlock block;
public LoadTsFileScheduler(
@@ -158,6 +160,7 @@ public class LoadTsFileScheduler implements IScheduler {
this.partitionFetcher = new DataPartitionBatchFetcher(partitionFetcher);
this.allReplicaSets = new HashSet<>();
this.isGeneratedByPipe = isGeneratedByPipe;
+ this.timePartitionSlotToProgressIndex = new HashMap<>();
this.block = LoadTsFileMemoryManager.getInstance().allocateDataCacheMemoryBlock();
for (FragmentInstance fragmentInstance : distributedQueryPlan.getInstances()) {
@@ -408,7 +411,26 @@ public class LoadTsFileScheduler implements IScheduler {
try {
loadCommandReq.setIsGeneratedByPipe(isGeneratedByPipe);
- loadCommandReq.setProgressIndex(assignProgressIndex(tsFileResource));
+ loadCommandReq.setTimePartition2ProgressIndex(
+ timePartitionSlotToProgressIndex.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry -> {
+ try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ final DataOutputStream dataOutputStream =
+ new DataOutputStream(byteArrayOutputStream)) {
+ entry.getValue().serialize(dataOutputStream);
+ return ByteBuffer.wrap(
+ byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
+ } catch (final IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Serialize Progress Index error, isFirstPhaseSuccess: %s, uuid: %s, tsFile: %s",
+ isFirstPhaseSuccess, uuid, tsFile.getAbsolutePath()),
+ e);
+ }
+ })));
Future<FragInstanceDispatchResult> dispatchResultFuture =
dispatcher.dispatchCommand(loadCommandReq, allReplicaSets);
@@ -431,14 +453,6 @@ public class LoadTsFileScheduler implements IScheduler {
stateMachine.transitionToFailed(status);
return false;
}
- } catch (IOException e) {
- LOGGER.warn(
- "Serialize Progress Index error, isFirstPhaseSuccess: {}, uuid: {}, tsFile: {}",
- isFirstPhaseSuccess,
- uuid,
- tsFile.getAbsolutePath());
- stateMachine.transitionToFailed(e);
- return false;
} catch (InterruptedException | ExecutionException e) {
if (e instanceof InterruptedException) {
Thread.currentThread().interrupt();
@@ -639,6 +653,12 @@ public class LoadTsFileScheduler implements IScheduler {
return null;
}
+ private void computeTimePartitionSlotToProgressIndexIfAbsent(
+ final TTimePartitionSlot timePartitionSlot) {
+ timePartitionSlotToProgressIndex.putIfAbsent(
+ timePartitionSlot, PipeDataNodeAgent.runtime().getNextProgressIndexForTsFileLoad());
+ }
+
public enum LoadCommand {
EXECUTE,
ROLLBACK
@@ -686,6 +706,7 @@ public class LoadTsFileScheduler implements IScheduler {
nonDirectionalChunkData.add(chunkData);
dataSize += chunkData.getDataSize();
block.addMemoryUsage(chunkData.getDataSize());
+ scheduler.computeTimePartitionSlotToProgressIndexIfAbsent(chunkData.getTimePartitionSlot());
if (!isMemoryEnough()) {
routeChunkData();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index cf24235f9ed..2c2281dcf66 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSetConfigurationReq;
import org.apache.iotdb.common.rpc.thrift.TSetTTLReq;
+import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.concurrent.ExceptionalCountDownLatch;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
@@ -980,13 +981,13 @@ public class StorageEngine implements IService {
LoadTsFileScheduler.LoadCommand loadCommand,
String uuid,
boolean isGeneratedByPipe,
- ProgressIndex progressIndex) {
+ Map<TTimePartitionSlot, ProgressIndex> timePartitionProgressIndexMap) {
TSStatus status = new TSStatus();
try {
switch (loadCommand) {
case EXECUTE:
- if (loadTsFileManager.loadAll(uuid, isGeneratedByPipe, progressIndex)) {
+ if (loadTsFileManager.loadAll(uuid, isGeneratedByPipe, timePartitionProgressIndexMap)) {
status = RpcUtils.SUCCESS_STATUS;
} else {
status.setCode(TSStatusCode.LOAD_FILE_ERROR.getStatusCode());
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 74bf838a8d3..e928fcc4d03 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
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Metric;
@@ -273,7 +274,10 @@ public class LoadTsFileManager {
return FOLDER_MANAGER.get().getNextFolder();
}
- public boolean loadAll(String uuid, boolean isGeneratedByPipe, ProgressIndex progressIndex)
+ public boolean loadAll(
+ String uuid,
+ boolean isGeneratedByPipe,
+ Map<TTimePartitionSlot, ProgressIndex> timePartitionProgressIndexMap)
throws IOException, LoadFileException {
if (!uuid2WriterManager.containsKey(uuid)) {
return false;
@@ -282,7 +286,7 @@ public class LoadTsFileManager {
final Optional<CleanupTask> cleanupTask = Optional.of(uuid2CleanupTask.get(uuid));
cleanupTask.ifPresent(CleanupTask::markLoadTaskRunning);
try {
- uuid2WriterManager.get(uuid).loadAll(isGeneratedByPipe, progressIndex);
+ uuid2WriterManager.get(uuid).loadAll(isGeneratedByPipe, timePartitionProgressIndexMap);
} finally {
cleanupTask.ifPresent(CleanupTask::markLoadTaskNotRunning);
}
@@ -497,7 +501,9 @@ public class LoadTsFileManager {
}
}
- private void loadAll(boolean isGeneratedByPipe, ProgressIndex progressIndex)
+ private void loadAll(
+ boolean isGeneratedByPipe,
+ Map<TTimePartitionSlot, ProgressIndex> timePartitionProgressIndexMap)
throws IOException, LoadFileException {
if (isClosed) {
throw new IOException(String.format(MESSAGE_WRITER_MANAGER_HAS_BEEN_CLOSED, taskDir));
@@ -516,7 +522,11 @@ public class LoadTsFileManager {
final DataRegion dataRegion = entry.getKey().getDataRegion();
final TsFileResource tsFileResource = dataPartition2Resource.get(entry.getKey());
- endTsFileResource(writer, tsFileResource, progressIndex);
+ endTsFileResource(
+ writer,
+ tsFileResource,
+ timePartitionProgressIndexMap.getOrDefault(
+ entry.getKey().getTimePartitionSlot(), MinimumProgressIndex.INSTANCE));
dataRegion.loadNewTsFile(tsFileResource, true, isGeneratedByPipe);
// Metrics
@@ -681,6 +691,10 @@ public class LoadTsFileManager {
return dataRegion;
}
+ public TTimePartitionSlot getTimePartitionSlot() {
+ return timePartitionSlot;
+ }
+
@Override
public String toString() {
return String.join(
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index 59ed2f10acb..70456a952f3 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -396,7 +396,7 @@ struct TLoadCommandReq {
1: required i32 commandType
2: required string uuid
3: optional bool isGeneratedByPipe
- 4: optional binary progressIndex
+ 4: optional map<common.TTimePartitionSlot, binary> timePartition2ProgressIndex
}
struct TAttributeUpdateReq {