Skip to content

Fix oversized Load TsFile piece dispatch #18627

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 2 commits into from
Sep 18, 2026
Merged

Fix oversized Load TsFile piece dispatch #18627

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dc324a7
Fix oversized Load TsFile piece dispatch
Caideyipi Sep 11, 2026
fdf4ac6
Address Load TsFile slicing review feedback
Caideyipi Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@
/** Compile-time i18n constants for DataNode misc subsystems (English). */
public final class DataNodeMiscMessages {

public static final String MESSAGE_MISSING_LOAD_TSFILE_SLICE_METADATA_ARG_DE4333DA =
"Missing Load TsFile slice metadata: %s";

public static final String INVALID_PIPE_NAME =
"Invalid pipeName";
public static final String READ_OBJECT_CONTENT_FROM_REMOTE_FILE =
Expand Down
5 changes: 5 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,11 @@

public final class DataNodeQueryMessages {

public static final String EXCEPTION_INVALID_THRIFT_MAXIMUM_FRAME_SIZE_ARG_FROM_ARG_A639588B =
"Invalid Thrift maximum frame size %d from %s";
public static final String MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_BECAUSE_OF_EXCEPTION_ARG_2D8A483D =
"failed to dispatch load command %s to node %s because of exception: %s";

// --- Common ---

public static final String NO_MATCHED_DATABASE_PLEASE_CHECK_THE_PATH =
Expand Down
17 changes: 17 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,23 @@

public final class StorageEngineMessages {

public static final String MESSAGE_LOAD_TSFILE_SLICE_BODY_IS_NULL_OR_EMPTY_2A65366C =
"Load TsFile slice body is null or empty";
public static final String MESSAGE_INVALID_LOAD_TSFILE_SLICE_METADATA_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_379BF1B8 =
"Invalid Load TsFile slice metadata: sliceCount=%d, originBodySize=%d";
public static final String MESSAGE_LOAD_TSFILE_SLICE_METADATA_CHANGED_SLICECOUNT_ARG_EXPECTED_ARG_ORIGINBODYSIZE_ARG_EXPECTED_ARG_B16B3122 =
"Load TsFile slice metadata changed: sliceCount=%d (expected %d), originBodySize=%d (expected %d)";
public static final String MESSAGE_UNEXPECTED_LOAD_TSFILE_SLICE_INDEX_ARG_EXPECTED_ARG_SLICECOUNT_ARG_76260F62 =
"Unexpected Load TsFile slice index %d: expected %d, sliceCount=%d";
public static final String MESSAGE_LOAD_TSFILE_SLICE_EXCEEDS_ORIGINBODYSIZE_ASSEMBLEDSIZE_ARG_SLICESIZE_ARG_ORIGINBODYSIZE_ARG_0198E0D0 =
"Load TsFile slice exceeds originBodySize: assembledSize=%d, sliceSize=%d, originBodySize=%d";
public static final String MESSAGE_LOAD_TSFILE_BODY_COMPLETED_BEFORE_THE_LAST_SLICE_RECEIVED_ARG_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_0425EA2C =
"Load TsFile body completed before the last slice: received=%d, sliceCount=%d, originBodySize=%d";
public static final String MESSAGE_LOAD_TSFILE_BODY_SIZE_MISMATCH_ASSEMBLEDSIZE_ARG_ORIGINBODYSIZE_ARG_5733FC4B =
"Load TsFile body size mismatch: assembledSize=%d, originBodySize=%d";
public static final String MESSAGE_MISSING_LOAD_TSFILE_ASSEMBLER_FOR_UUID_ARG_DATAREGION_ARG_SLICEINDEX_ARG_FF6EA463 =
"Missing Load TsFile assembler for uuid %s, DataRegion %s, sliceIndex=%d";

public static final String IS_NOT_SUPPORTED =
" is not supported";

Expand Down
3 changes: 3 additions & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@
/** 编译时国际化常量 - DataNode 杂项子系统（中文）。 */
public final class DataNodeMiscMessages {

public static final String MESSAGE_MISSING_LOAD_TSFILE_SLICE_METADATA_ARG_DE4333DA =
"缺少 Load TsFile 分片元数据：%s";

public static final String INVALID_PIPE_NAME =
"无效的 pipeName";
public static final String READ_OBJECT_CONTENT_FROM_REMOTE_FILE =
Expand Down
5 changes: 5 additions & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,11 @@

public final class DataNodeQueryMessages {

public static final String EXCEPTION_INVALID_THRIFT_MAXIMUM_FRAME_SIZE_ARG_FROM_ARG_A639588B =
"Thrift 最大帧大小 %d 无效，来源：%s";
public static final String MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_BECAUSE_OF_EXCEPTION_ARG_2D8A483D =
"向节点派发 Load 命令失败，命令：%s，节点：%s，异常：%s";

// --- Common ---

public static final String NO_MATCHED_DATABASE_PLEASE_CHECK_THE_PATH =
Expand Down
17 changes: 17 additions & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,23 @@

public final class StorageEngineMessages {

public static final String MESSAGE_LOAD_TSFILE_SLICE_BODY_IS_NULL_OR_EMPTY_2A65366C =
"Load TsFile 分片内容为 null 或为空";
public static final String MESSAGE_INVALID_LOAD_TSFILE_SLICE_METADATA_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_379BF1B8 =
"Load TsFile 分片元数据无效：sliceCount=%d，originBodySize=%d";
public static final String MESSAGE_LOAD_TSFILE_SLICE_METADATA_CHANGED_SLICECOUNT_ARG_EXPECTED_ARG_ORIGINBODYSIZE_ARG_EXPECTED_ARG_B16B3122 =
"Load TsFile 分片元数据发生变化：sliceCount=%d（预期 %d），originBodySize=%d（预期 %d）";
public static final String MESSAGE_UNEXPECTED_LOAD_TSFILE_SLICE_INDEX_ARG_EXPECTED_ARG_SLICECOUNT_ARG_76260F62 =
"Load TsFile 分片索引 %d 不符合预期：预期 %d，sliceCount=%d";
public static final String MESSAGE_LOAD_TSFILE_SLICE_EXCEEDS_ORIGINBODYSIZE_ASSEMBLEDSIZE_ARG_SLICESIZE_ARG_ORIGINBODYSIZE_ARG_0198E0D0 =
"Load TsFile 分片超出 originBodySize：assembledSize=%d，sliceSize=%d，originBodySize=%d";
public static final String MESSAGE_LOAD_TSFILE_BODY_COMPLETED_BEFORE_THE_LAST_SLICE_RECEIVED_ARG_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_0425EA2C =
"Load TsFile 内容在最后一个分片之前已达到原始大小：received=%d，sliceCount=%d，originBodySize=%d";
public static final String MESSAGE_LOAD_TSFILE_BODY_SIZE_MISMATCH_ASSEMBLEDSIZE_ARG_ORIGINBODYSIZE_ARG_5733FC4B =
"Load TsFile 内容大小不匹配：assembledSize=%d，originBodySize=%d";
public static final String MESSAGE_MISSING_LOAD_TSFILE_ASSEMBLER_FOR_UUID_ARG_DATAREGION_ARG_SLICEINDEX_ARG_FF6EA463 =
"缺少 Load TsFile 分片组装器：uuid %s，DataRegion %s，sliceIndex=%d";

public static final String IS_NOT_SUPPORTED =
" 不受支持";

Expand Down
41 changes: 40 additions & 1 deletion ...rc/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -627,12 +627,51 @@ public TSchemaFetchResponse fetchSchema(final TSchemaFetchRequest req) {
throw new UnsupportedOperationException();
}

@Override
public int getThriftMaxFrameSize() {
return IoTDBDescriptor.getInstance().getConfig().getThriftMaxFrameSize();
}

@Override
public TLoadResp sendTsFilePieceNode(final TTsFilePieceReq req) {
LOGGER.info(DataNodeMiscMessages.RECEIVE_LOAD_NODE, req.uuid);
if (!req.isSetSliceIndex() || req.sliceIndex == 0) {
LOGGER.info(DataNodeMiscMessages.RECEIVE_LOAD_NODE, req.uuid);
}

final ConsensusGroupId groupId =
ConsensusGroupId.Factory.createFromTConsensusGroupId(req.consensusGroupId);
final boolean isSliced =
req.isSetSliceIndex() || req.isSetSliceCount() || req.isSetOriginBodySize();
if (isSliced) {
if (!req.isSetSliceIndex() || !req.isSetSliceCount() || !req.isSetOriginBodySize()) {
final List<String> missingFields = new ArrayList<>(3);
if (!req.isSetSliceIndex()) {
missingFields.add("sliceIndex");
}
if (!req.isSetSliceCount()) {
missingFields.add("sliceCount");
}
if (!req.isSetOriginBodySize()) {
missingFields.add("originBodySize");
}
return createTLoadResp(
RpcUtils.getStatus(
TSStatusCode.DESERIALIZE_PIECE_OF_TSFILE_ERROR,
String.format(
DataNodeMiscMessages.MESSAGE_MISSING_LOAD_TSFILE_SLICE_METADATA_ARG_DE4333DA,
String.join(", ", missingFields))));
}
Comment thread
jt2594838 marked this conversation as resolved.
return createTLoadResp(
StorageEngine.getInstance()
.writeLoadTsFileNodeSlice(
(DataRegionId) groupId,
req.body,
req.uuid,
req.sliceIndex,
req.sliceCount,
req.originBodySize));
}

final LoadTsFilePieceNode pieceNode = (LoadTsFilePieceNode) PlanNodeType.deserialize(req.body);
if (pieceNode == null) {
return createTLoadResp(
Expand Down
169 changes: 139 additions & 30 deletions ...in/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

package org.apache.iotdb.db.queryengine.plan.scheduler.load;

import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
Expand Down Expand Up @@ -57,18 +58,21 @@
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;

import org.apache.thrift.TApplicationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
Expand All @@ -82,6 +86,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher, AutoCl

private static final int MAX_CONNECTION_TIMEOUT_MS = 24 * 60 * 60 * 1000; // 1 day
private static final int FIRST_ADJUSTMENT_TIMEOUT_MS = 6 * 60 * 60 * 1000; // 6 hours
private static final int LOAD_TSFILE_PIECE_RPC_FRAME_RESERVED_BYTES = 1024;
private static final AtomicInteger CONNECTION_TIMEOUT_MS =
new AtomicInteger(IoTDBDescriptor.getInstance().getConfig().getConnectionTimeoutInMS());

Expand All @@ -92,6 +97,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher, AutoCl
internalServiceClientManager;
private ExecutorService executor;
private final boolean isGeneratedByPipe;
private final Map<TEndPoint, Integer> endPoint2ThriftMaxFrameSize = new ConcurrentHashMap<>();

public LoadTsFileDispatcherImpl(
IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> internalServiceClientManager,
Expand Down Expand Up @@ -143,26 +149,115 @@ public Future<FragInstanceDispatchResult> dispatch(

private void dispatchOneInstance(FragmentInstance instance)
throws FragmentInstanceDispatchException {
TTsFilePieceReq loadTsFileReq = null;
ByteBuffer body = null;

for (TDataNodeLocation dataNodeLocation :
instance.getRegionReplicaSet().getDataNodeLocations()) {
TEndPoint endPoint = dataNodeLocation.getInternalEndPoint();
if (isDispatchedToLocal(endPoint)) {
dispatchLocally(instance);
} else {
if (loadTsFileReq == null) {
loadTsFileReq =
new TTsFilePieceReq(
instance.getFragment().getPlanNodeTree().serializeToByteBuffer(),
uuid,
instance.getRegionReplicaSet().getRegionId());
if (body == null) {
body = instance.getFragment().getPlanNodeTree().serializeToByteBuffer();
}
dispatchRemote(loadTsFileReq, endPoint);
dispatchRemote(body, instance.getRegionReplicaSet().getRegionId(), endPoint);
}
}
}

private int getLoadTsFilePieceBodySizeLimit(final TEndPoint endPoint) throws Exception {
final int localMaxFrameSize = IoTDBDescriptor.getInstance().getConfig().getThriftMaxFrameSize();
Integer remoteMaxFrameSize = endPoint2ThriftMaxFrameSize.get(endPoint);
if (remoteMaxFrameSize == null) {
// An oversized frame is rejected before the RPC handler runs and closes the connection, so
// the receiver's frame limit cannot be recovered from the sender's transport exception.
try (SyncDataNodeInternalServiceClient client =
internalServiceClientManager.borrowClient(endPoint)) {
remoteMaxFrameSize = client.getThriftMaxFrameSize();
} catch (Exception e) {
if (!isUnknownMethod(e)) {
throw e;
}
// Older receivers still accept unsliced pieces. The failed RPC invalidates its client,
// so dispatchRemote borrows another client before sending the piece.
remoteMaxFrameSize = localMaxFrameSize;
}
if (remoteMaxFrameSize <= 0) {
throw new IllegalArgumentException(
String.format(
DataNodeQueryMessages
.EXCEPTION_INVALID_THRIFT_MAXIMUM_FRAME_SIZE_ARG_FROM_ARG_A639588B,
remoteMaxFrameSize,
endPoint));
}
endPoint2ThriftMaxFrameSize.put(endPoint, remoteMaxFrameSize);
}
return Math.max(
1,
Math.min(localMaxFrameSize, remoteMaxFrameSize)
- LOAD_TSFILE_PIECE_RPC_FRAME_RESERVED_BYTES);
}

private static boolean isUnknownMethod(Throwable e) {
do {
if (e instanceof TApplicationException
&& ((TApplicationException) e).getType() == TApplicationException.UNKNOWN_METHOD) {
return true;
}
} while ((e = e.getCause()) != null);
return false;
}

static List<TTsFilePieceReq> splitTsFilePieceReq(
final ByteBuffer body,
final String uuid,
final TConsensusGroupId consensusGroupId,
final int bodySizeLimit) {
if (bodySizeLimit <= 0) {
throw new IllegalArgumentException();
}

final int originBodySize = body.remaining();
final int sliceCount = getSliceCount(originBodySize, bodySizeLimit);
final List<TTsFilePieceReq> requests = new ArrayList<>(sliceCount);
if (sliceCount == 1) {
requests.add(createTsFilePieceReq(body.duplicate(), uuid, consensusGroupId));
return requests;
}

final int originPosition = body.position();
for (int sliceIndex = 0; sliceIndex < sliceCount; sliceIndex++) {
final int startOffset = sliceIndex * bodySizeLimit;
final int endOffset = startOffset + Math.min(bodySizeLimit, originBodySize - startOffset);
final ByteBuffer slicedBody = body.duplicate();
slicedBody.position(originPosition + startOffset);
slicedBody.limit(originPosition + endOffset);
requests.add(
createTsFilePieceReq(slicedBody.slice(), uuid, consensusGroupId)
.setSliceIndex(sliceIndex)
.setSliceCount(sliceCount)
.setOriginBodySize(originBodySize));
}
return requests;
}

static int getSliceCount(final int bodySize, final int bodySizeLimit) {
if (bodySize < 0 || bodySizeLimit <= 0) {
throw new IllegalArgumentException();
}
return bodySize == 0 ? 1 : (bodySize - 1) / bodySizeLimit + 1;
}

private static TTsFilePieceReq createTsFilePieceReq(
final ByteBuffer body, final String uuid, final TConsensusGroupId consensusGroupId) {
final TTsFilePieceReq request =
new TTsFilePieceReq().setUuid(uuid).setConsensusGroupId(consensusGroupId);
// The generated setter copies the whole buffer, while these immutable slices remain valid until
// all replicas have been dispatched.
request.body = body;
return request;
}

public void dispatchLocally(FragmentInstance instance) throws FragmentInstanceDispatchException {
if (isGeneratedByPipe) {
LOGGER.debug(DataNodeQueryMessages.RECEIVE_LOAD_NODE_FROM_UUID, uuid);
Expand Down Expand Up @@ -222,28 +317,36 @@ public void dispatchLocally(FragmentInstance instance) throws FragmentInstanceDi
}
}

private void dispatchRemote(TTsFilePieceReq loadTsFileReq, TEndPoint endPoint)
private void dispatchRemote(
ByteBuffer body, TConsensusGroupId consensusGroupId, TEndPoint endPoint)
throws FragmentInstanceDispatchException {
boolean transferAttemptRecorded = false;
try (SyncDataNodeInternalServiceClient client =
internalServiceClientManager.borrowClient(endPoint)) {
client.setTimeout(CONNECTION_TIMEOUT_MS.get());

final TLoadResp loadResp = client.sendTsFilePieceNode(loadTsFileReq);
if (!loadResp.isAccepted()) {
recordTransferAttempt(
endPoint,
false,
loadResp.isSetStatus()
? String.valueOf(loadResp.getStatus().getCode())
: UserDataTransferErrorCode.REMOTE_REJECTED.name(),
null);
try {
final List<TTsFilePieceReq> loadTsFileReqs =
splitTsFilePieceReq(
body, uuid, consensusGroupId, getLoadTsFilePieceBodySizeLimit(endPoint));
try (SyncDataNodeInternalServiceClient client =
internalServiceClientManager.borrowClient(endPoint)) {
client.setTimeout(CONNECTION_TIMEOUT_MS.get());

for (final TTsFilePieceReq loadTsFileReq : loadTsFileReqs) {
final TLoadResp loadResp = client.sendTsFilePieceNode(loadTsFileReq);
if (!loadResp.isAccepted()) {
recordTransferAttempt(
endPoint,
false,
loadResp.isSetStatus()
? String.valueOf(loadResp.getStatus().getCode())
: UserDataTransferErrorCode.REMOTE_REJECTED.name(),
null);
transferAttemptRecorded = true;
LOGGER.warn(loadResp.message);
throw new FragmentInstanceDispatchException(loadResp.status);
}
}
recordTransferAttempt(endPoint, true, null, null);
transferAttemptRecorded = true;
LOGGER.warn(loadResp.message);
throw new FragmentInstanceDispatchException(loadResp.status);
}
recordTransferAttempt(endPoint, true, null, null);
transferAttemptRecorded = true;
} catch (Exception e) {
if (!transferAttemptRecorded) {
recordTransferAttempt(endPoint, false, null, e);
Expand All @@ -252,8 +355,11 @@ private void dispatchRemote(TTsFilePieceReq loadTsFileReq, TEndPoint endPoint)

final String exceptionMessage =
String.format(
"failed to dispatch load command %s to node %s because of exception: %s",
loadTsFileReq, endPoint, e);
DataNodeQueryMessages
.MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_BECAUSE_OF_EXCEPTION_ARG_2D8A483D,
uuid,
endPoint,
e);
LOGGER.warn(exceptionMessage, e);
throw new FragmentInstanceDispatchException(
new TSStatus()
Expand Down Expand Up @@ -361,8 +467,11 @@ private void dispatchRemote(TLoadCommandReq loadCommandReq, TEndPoint endPoint)

final String exceptionMessage =
String.format(
"failed to dispatch load command %s to node %s because of exception: %s",
loadCommandReq, endPoint, e);
DataNodeQueryMessages
.MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_BECAUSE_OF_EXCEPTION_ARG_2D8A483D,
loadCommandReq,
endPoint,
e);
LOGGER.warn(exceptionMessage, e);
throw new FragmentInstanceDispatchException(
new TSStatus()
Expand Down
Loading
Loading