This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 6658413f77d6e6df67c51ec438e3ff795036bf20 Author: Tian Jiang <[email protected]> AuthorDate: Mon Sep 21 18:42:25 2026 +0800 tmp --- .../operator/source/AlignedSeriesScanUtil.java | 17 ++++- .../memtable/AlignedWritableMemChunk.java | 14 ++++- .../db/utils/datastructure/AlignedTVList.java | 73 +++++++++++++++++++--- 3 files changed, 91 insertions(+), 13 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java index 6fe02e754e2..3907ca3029e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java @@ -92,7 +92,7 @@ public class AlignedSeriesScanUtil extends SeriesScanUtil { boolean queryAllSensors, List<TSDataType> givenDataTypes, List<Integer> rowKeyColumnIndices) { - super(seriesPath, scanOrder, scanOptions, context); + super(seriesPath, scanOrder, scanOptions, context, !rowKeyColumnIndices.isEmpty()); isAligned = true; this.dataTypes = givenDataTypes != null @@ -114,12 +114,16 @@ public class AlignedSeriesScanUtil extends SeriesScanUtil { @Override protected PriorityMergeReader getPriorityMergeReader() { - return new FlatDevicePriorityMergeReader(); + return currentDeviceIsFlat + ? new FlatDevicePriorityMergeReader() + : new AlignedPriorityMergeReader(); } @Override protected DescPriorityMergeReader getDescPriorityMergeReader() { - return new FlatDeviceDescPriorityMergeReader(); + return currentDeviceIsFlat + ? new FlatDeviceDescPriorityMergeReader() + : new AlignedDescPriorityMergeReader(); } @Override @@ -142,6 +146,13 @@ public class AlignedSeriesScanUtil extends SeriesScanUtil { @Override protected IPointReader getPointReader(TsBlock tsBlock) { + if (rowKeyColumnIndices.length > 0) { + return new FlatDevicePointReader( + tsBlock, + dataTypes, + Arrays.stream(rowKeyColumnIndices).boxed().collect(Collectors.toList()), + orderUtils.getAscending()); + } return tsBlock.getTsBlockAlignedRowIterator(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java index eea675f7a6a..5c1658c69e6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java @@ -43,6 +43,7 @@ import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.common.TimeRange; import org.apache.tsfile.read.filter.basic.Filter; import org.apache.tsfile.utils.Binary; @@ -56,6 +57,7 @@ import org.apache.tsfile.write.schema.MeasurementSchema; import java.io.DataInputStream; import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; @@ -569,7 +571,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { List<TimeRange> columnDeletionList = valueColumnsDeletionList.get(column); int[] deleteCursor = valueColumnDeleteCursor.get(column); if (columnDeletionList != null && !columnDeletionList.isEmpty()) { - if (!alignedTVList.isSorted()) { + if (!alignedTVList.isSorted() || !alignedTVList.isTimeOrdered()) { deleteCursor[0] = 0; } if (isPointDeleted(timestamp, columnDeletionList, deleteCursor)) { @@ -808,6 +810,10 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { BlockingQueue<Object> ioTaskQueue, long maxNumberOfPointsInChunk, int maxNumberOfPointsInPage) { + if (useColumnSortedList) { + encodeFlatDevice(ioTaskQueue, maxNumberOfPointsInChunk, maxNumberOfPointsInPage); + return; + } List<IMeasurementSchema> activeSchemaList = getActiveSchemaList(); if (activeSchemaList.isEmpty() && ignoreAllNullRows) { return; @@ -1251,6 +1257,12 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { encodeInfo.maxNumberOfPointsInChunk, (encodeInfo.targetChunkSize / getAvgPointSizeOfLargestColumn())); + if (useColumnSortedList) { + encodeFlatDevice( + ioTaskQueue, encodeInfo.maxNumberOfPointsInChunk, encodeInfo.maxNumberOfPointsInPage); + return; + } + if (TVLIST_SORT_THRESHOLD == 0) { encodeWorkingAlignedTVList( ioTaskQueue, encodeInfo.maxNumberOfPointsInChunk, encodeInfo.maxNumberOfPointsInPage); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java index 807ecea68eb..d0bc9bb67b8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java @@ -192,6 +192,26 @@ public abstract class AlignedTVList extends TVList { return 0; } + /** Whether sorting this list guarantees globally ordered timestamps, independent of TAGs. */ + public boolean isTimeOrdered() { + return getRowKeyColumnCount() == 0; + } + + private boolean isDeletedInRowOrder(long time, List<TimeRange> deletions, int[] cursor) { + return isDeletedInRowOrder(time, deletions, cursor, Ordering.ASC); + } + + private boolean isDeletedInRowOrder( + long time, List<TimeRange> deletions, int[] cursor, Ordering ordering) { + if (deletions == null || deletions.isEmpty()) { + return false; + } + // A cursor advanced for one TAG cannot be reused after time decreases at the next TAG. + return isTimeOrdered() + ? isPointDeleted(time, deletions, cursor, ordering) + : isPointDeleted(time, deletions); + } + /** Returns whether two sorted rows in this list represent the same logical row. */ public boolean isSameRowKey(int leftRowIndex, int rightRowIndex) { return compareRowKey(leftRowIndex, this, rightRowIndex) == 0; @@ -1844,7 +1864,7 @@ public abstract class AlignedTVList extends TVList { } long timestamp = getTime(sortedRowIndex); if ((nextRowIndex == rowCount || !isSameRowKey(sortedRowIndex, nextRowIndex)) - && !isPointDeleted(timestamp, timeColumnDeletion, deleteCursor)) { + && !isDeletedInRowOrder(timestamp, timeColumnDeletion, deleteCursor)) { timeBuilder.writeLong(getTime(sortedRowIndex)); validRowCount++; } else { @@ -1902,7 +1922,7 @@ public abstract class AlignedTVList extends TVList { originRowIndex = getValueIndex(sortedRowIndex); } if (isNullValue(originRowIndex, columnIndex) - || isPointDeleted( + || isDeletedInRowOrder( getTime(sortedRowIndex), Objects.isNull(deletionList) ? null : deletionList.get(columnIndex), deleteCursor)) { @@ -2510,6 +2530,35 @@ public abstract class AlignedTVList extends TVList { } } + public boolean isTimeOrderedScan() { + return isTimeOrdered(); + } + + @Override + protected boolean isCurrentTimeExceedTimeRange(long time) { + return isTimeOrderedScan() && super.isCurrentTimeExceedTimeRange(time); + } + + @Override + protected boolean isTimeSatisfied(long time) { + return super.isTimeSatisfied(time) + && (isTimeOrderedScan() || timeRange == null || timeRange.contains(time)); + } + + @Override + protected void skipToCurrentTimeRangeStartPosition() { + if (isTimeOrdered()) { + super.skipToCurrentTimeRangeStartPosition(); + } else { + probeNext = false; + } + } + + @Override + public long getUsedMemorySize() { + return 0; + } + @Override @SuppressWarnings("java:S6541") protected void prepareNext() { @@ -2528,7 +2577,7 @@ public abstract class AlignedTVList extends TVList { continue; } long time = getTime(getScanOrderIndex(index)); - if (isPointDeleted(time, timeColumnDeletion, timeDeleteCursor, scanOrder)) { + if (isDeletedInRowOrder(time, timeColumnDeletion, timeDeleteCursor, scanOrder)) { index++; continue; } @@ -2589,7 +2638,7 @@ public abstract class AlignedTVList extends TVList { for (int columnIndex = 0; columnIndex < dataTypeList.size(); columnIndex++) { if (selectedIndices[columnIndex] == -1 || (valueColumnsDeletionList != null - && isPointDeleted( + && isDeletedInRowOrder( time, valueColumnsDeletionList.get(columnIndex), valueColumnDeleteCursor.get(columnIndex), @@ -2662,7 +2711,13 @@ public abstract class AlignedTVList extends TVList { if (validColumnIndex < 0 || validColumnIndex >= dataTypes.size()) { return null; } - if (outer.isNullValue(valueIndex, validColumnIndex)) { + if (outer.isNullValue(valueIndex, validColumnIndex) + || (valueColumnsDeletionList != null + && isDeletedInRowOrder( + getTime(rowIndex), + valueColumnsDeletionList.get(columnIndex), + valueColumnDeleteCursor.get(columnIndex), + scanOrder))) { return null; } switch (dataTypeList.get(columnIndex)) { @@ -2727,7 +2782,7 @@ public abstract class AlignedTVList extends TVList { if ((allValueColDeletedMap != null && allValueColDeletedMap.isMarked(getValueIndex(getScanOrderIndex(rowIndex)))) || isTimeDeleted(getScanOrderIndex(rowIndex)) - || isPointDeleted(time, timeColumnDeletion, deleteCursor, scanOrder)) { + || isDeletedInRowOrder(time, timeColumnDeletion, deleteCursor, scanOrder)) { return true; } @@ -2766,8 +2821,8 @@ public abstract class AlignedTVList extends TVList { int[] deleteCursor = {0}; int startIndex = index; - boolean isVerbose = this.getQueryContext().isVerbose(); - long[] filteredRowsByTimeFilter = this.getQueryContext().isVerbose() ? new long[] {0} : null; + boolean isVerbose = getQueryContext() != null && getQueryContext().isVerbose(); + long[] filteredRowsByTimeFilter = isVerbose ? new long[] {0} : null; // time column for (; index < rows; index++) { @@ -2888,7 +2943,7 @@ public abstract class AlignedTVList extends TVList { originRowIndex = getValueIndex(getScanOrderIndex(sortedRowIndex)); } if (outer.isNullValue(originRowIndex, validColumnIndex) - || isPointDeleted( + || isDeletedInRowOrder( getTime(getScanOrderIndex(sortedRowIndex)), Objects.isNull(valueColumnsDeletionList) ? null
