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 <[email protected]> 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 {
