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 5f73d3f2e5019928e996a96c4b1fe7a79d272e55 Author: Weihao Li <[email protected]> AuthorDate: Tue Sep 15 18:10:17 2026 +0800 merge map Signed-off-by: Weihao Li <[email protected]> --- .../execution/exchange/sink/SinkChannel.java | 72 ++++++++++------------ 1 file changed, 33 insertions(+), 39 deletions(-) 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 c98be7d55bd..d24a1552a5b 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 @@ -44,7 +44,6 @@ import org.apache.tsfile.common.conf.TSFileDescriptor; import org.apache.tsfile.external.commons.lang3.Validate; import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.read.common.block.column.TsBlockSerde; -import org.apache.tsfile.utils.Pair; import org.apache.tsfile.utils.RamUsageEstimator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -67,6 +66,19 @@ import static org.apache.iotdb.db.queryengine.metric.DataExchangeCountMetricSet. public class SinkChannel implements ISinkChannel { + private static class TsBlockInfo { + private TsBlock tsBlock; + private ByteBuffer serializedTsBlock; + private final long tsBlockSize; + + // private Class, no need to do access control + private TsBlockInfo(TsBlock tsBlock, ByteBuffer serializedTsBlock, long tsBlockSize) { + this.tsBlock = tsBlock; + this.serializedTsBlock = serializedTsBlock; + this.tsBlockSize = tsBlockSize; + } + } + private static final Logger LOGGER = LoggerFactory.getLogger(SinkChannel.class); public static final int MAX_ATTEMPT_TIMES = 3; @@ -93,12 +105,7 @@ public class SinkChannel implements ISinkChannel { // Use LinkedHashMap to meet 2 needs, // 1. Predictable iteration order so that removing buffered TsBlocks can be efficient. // 2. Fast lookup. - private final LinkedHashMap<Integer, Pair<TsBlock, Long>> sequenceIdToTsBlock = - new LinkedHashMap<>(); - - /** Serialized blocks are cached so fragmented requests do not serialize the same block again. */ - private final LinkedHashMap<Integer, ByteBuffer> sequenceIdToSerializedTsBlock = - new LinkedHashMap<>(); + private final LinkedHashMap<Integer, TsBlockInfo> sequenceIdToTsBlock = new LinkedHashMap<>(); // size for current TsBlock to reserve and free private long currentTsBlockSize; @@ -278,7 +285,7 @@ public class SinkChannel implements ISinkChannel { blocked = reserveResult.getFuture(); bufferRetainedSizeInBytes += reserveResult.getReservedBytes(); - sequenceIdToTsBlock.put(nextSequenceId, new Pair<>(tsBlock, currentTsBlockSize)); + sequenceIdToTsBlock.put(nextSequenceId, new TsBlockInfo(tsBlock, null, currentTsBlockSize)); nextSequenceId += 1; currentTsBlockSize = reserveResult.getReservedBytes(); @@ -309,7 +316,6 @@ public class SinkChannel implements ISinkChannel { return false; } sequenceIdToTsBlock.clear(); - sequenceIdToSerializedTsBlock.clear(); if (blocked != null) { bufferRetainedSizeInBytes -= localMemoryManager.getQueryPool().tryCancel(blocked); } @@ -340,7 +346,6 @@ public class SinkChannel implements ISinkChannel { return false; } sequenceIdToTsBlock.clear(); - sequenceIdToSerializedTsBlock.clear(); if (blocked != null) { bufferRetainedSizeInBytes -= localMemoryManager.getQueryPool().tryCancel(blocked); } @@ -413,12 +418,20 @@ public class SinkChannel implements ISinkChannel { throw new GetTsBlockFromClosedOrAbortedChannelException( DataNodeQueryMessages.SINKCHANNEL_IS_ABORTED_OR_CLOSED); } - ByteBuffer serializedTsBlock = sequenceIdToSerializedTsBlock.get(sequenceId); + TsBlockInfo tsBlockInfo = sequenceIdToTsBlock.get(sequenceId); + if (tsBlockInfo == null) { + LOGGER.warn( + DataNodeQueryMessages.THE_TSBLOCK_DOESNT_EXIST_SEQUENCE_ID_REMAINING, + sequenceId, + sequenceIdToTsBlock.entrySet()); + throw new IllegalStateException( + DataNodeQueryMessages.THE_DATA_BLOCK_DOESN_T_EXIST_SEQUENCE_ID + sequenceId); + } + ByteBuffer serializedTsBlock = tsBlockInfo.serializedTsBlock; if (serializedTsBlock != null) { return serializedTsBlock.duplicate(); } - Pair<TsBlock, Long> pair = sequenceIdToTsBlock.get(sequenceId); - if (pair == null || pair.left == null) { + if (tsBlockInfo.tsBlock == null) { LOGGER.warn( DataNodeQueryMessages.THE_TSBLOCK_DOESNT_EXIST_SEQUENCE_ID_REMAINING, sequenceId, @@ -426,39 +439,21 @@ public class SinkChannel implements ISinkChannel { throw new IllegalStateException( DataNodeQueryMessages.THE_DATA_BLOCK_DOESN_T_EXIST_SEQUENCE_ID + sequenceId); } - serializedTsBlock = serde.serialize(pair.left); - sequenceIdToSerializedTsBlock.put(sequenceId, serializedTsBlock); - pair.left = null; + serializedTsBlock = serde.serialize(tsBlockInfo.tsBlock); + tsBlockInfo.serializedTsBlock = serializedTsBlock; + tsBlockInfo.tsBlock = null; return serializedTsBlock.duplicate(); } - public synchronized ByteBuffer getSerializedTsBlockFragment( - int sequenceId, int offset, int maxBytes) throws IOException { - ByteBuffer serializedTsBlock = getSerializedTsBlock(sequenceId); - if (offset < 0 || offset > serializedTsBlock.remaining() || maxBytes <= 0) { - throw new IllegalArgumentException( - String.format( - DataNodeQueryMessages.EXCEPTION_INVALID_ARG_ARG_2946DBE5, - "serialized TsBlock", - "fragment range")); - } - int length = Math.min(maxBytes, serializedTsBlock.remaining() - offset); - ByteBuffer fragment = serializedTsBlock.duplicate(); - fragment.position(offset); - fragment.limit(offset + length); - return fragment.slice(); - } - public void acknowledgeTsBlock(int startSequenceId, int endSequenceId) { long freedBytes = 0L; synchronized (this) { if (aborted || closed) { return; } - Iterator<Entry<Integer, Pair<TsBlock, Long>>> iterator = - sequenceIdToTsBlock.entrySet().iterator(); + Iterator<Entry<Integer, TsBlockInfo>> iterator = sequenceIdToTsBlock.entrySet().iterator(); while (iterator.hasNext()) { - Entry<Integer, Pair<TsBlock, Long>> entry = iterator.next(); + Entry<Integer, TsBlockInfo> entry = iterator.next(); if (entry.getKey() < startSequenceId) { continue; } @@ -466,10 +461,9 @@ public class SinkChannel implements ISinkChannel { break; } - freedBytes += entry.getValue().right; - bufferRetainedSizeInBytes -= entry.getValue().right; + freedBytes += entry.getValue().tsBlockSize; + bufferRetainedSizeInBytes -= entry.getValue().tsBlockSize; iterator.remove(); - sequenceIdToSerializedTsBlock.remove(entry.getKey()); if (LOGGER.isDebugEnabled()) { LOGGER.debug(DataNodeQueryMessages.ACK_TSBLOCK, entry.getKey()); }
