This is an automated email from the ASF dual-hosted git repository. Wei-hao-Li pushed a commit to branch mppEx in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ee86f1c393c82f7b186e125aa61376d4258188ce Author: Weihao Li <[email protected]> AuthorDate: Mon Sep 14 15:46:59 2026 +0800 make offset int Signed-off-by: Weihao Li <[email protected]> --- .../queryengine/execution/exchange/MPPDataExchangeManager.java | 10 +++++----- .../db/queryengine/execution/exchange/sink/SinkChannel.java | 8 ++++---- .../db/queryengine/execution/exchange/source/SourceHandle.java | 4 ++-- iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift | 4 ++-- 4 files changed, 13 insertions(+), 13 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java index f8327673a80..ea3dce68281 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java @@ -182,17 +182,17 @@ public class MPPDataExchangeManager implements IMPPDataExchangeManager { IoTDBDescriptor.getInstance() .getConfig() .getMppDataExchangeMaxPayloadSizeInBytes(); - long offset = req.getOffset(); + int offset = req.getOffset(); for (int i = req.getStartSequenceId(); i < req.getEndSequenceId(); i++) { try { ByteBuffer serializedTsBlock = sinkChannel.getSerializedTsBlock(i); - long blockOffset = i == req.getStartSequenceId() ? offset : 0L; - long remainingBlockSize = serializedTsBlock.remaining() - blockOffset; + int blockOffset = i == req.getStartSequenceId() ? offset : 0; + int remainingBlockSize = serializedTsBlock.remaining() - blockOffset; if (remainingBlockSize <= remainingPayloadSize) { resp.addToTsBlocks( sinkChannel.getSerializedTsBlockFragment( - i, blockOffset, Math.toIntExact(remainingBlockSize))); - remainingPayloadSize -= Math.toIntExact(remainingBlockSize); + i, blockOffset, remainingBlockSize)); + remainingPayloadSize -= remainingBlockSize; if (remainingPayloadSize == 0) { break; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java index 533387afdf0..86c66b6677e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java @@ -432,7 +432,7 @@ public class SinkChannel implements ISinkChannel { } public synchronized ByteBuffer getSerializedTsBlockFragment( - int sequenceId, long offset, int maxBytes) throws IOException { + int sequenceId, int offset, int maxBytes) throws IOException { ByteBuffer serializedTsBlock = getSerializedTsBlock(sequenceId); if (offset < 0 || offset > serializedTsBlock.remaining() || maxBytes <= 0) { throw new IllegalArgumentException( @@ -441,10 +441,10 @@ public class SinkChannel implements ISinkChannel { "serialized TsBlock", "fragment range")); } - int length = (int) Math.min(maxBytes, serializedTsBlock.remaining() - offset); + int length = Math.min(maxBytes, serializedTsBlock.remaining() - offset); ByteBuffer fragment = serializedTsBlock.duplicate(); - fragment.position(Math.toIntExact(offset)); - fragment.limit(Math.toIntExact(offset + length)); + fragment.position(offset); + fragment.limit(offset + length); return fragment.slice(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java index 1deba48e8ad..db4f1dfd408 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java @@ -794,7 +794,7 @@ public class SourceHandle implements ISourceHandle { private final int endSequenceId; private final List<ByteBuffer> tsBlocks; private int nextSequenceId; - private long offset; + private int offset; private ByteArrayOutputStream partialTsBlock; private DataBlockFetchProgress(int startSequenceId, int endSequenceId) { @@ -855,7 +855,7 @@ public class SourceHandle implements ISourceHandle { partialTsBlock.writeBytes(bytes); } - private void updateOffset(long nextOffset) throws TException { + private void updateOffset(int nextOffset) throws TException { if (nextOffset <= offset || nextOffset != partialTsBlock.size()) { throw new TException( DataNodeQueryMessages.EXCEPTION_UNEXPECTED_DATA_BLOCK_RESPONSE_SIZE_A7DD7E33); diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift index 77dacf07a6b..cf9aca63b17 100644 --- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift +++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift @@ -91,13 +91,13 @@ struct TGetDataBlockRequest { // Index of upstream SinkChannel 4: required i32 index // Optional byte range for fetching one serialized TsBlock in fragments. - 5: optional i64 offset + 5: optional i32 offset } struct TGetDataBlockResponse { 1: required list<binary> tsBlocks // The start offset of the next fragment. It is set only when the last element in tsBlocks is a fragment. - 2: optional i64 offset + 2: optional i32 offset } struct TAcknowledgeDataBlockEvent {
