You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/12 06:50:55 UTC
(iotdb) 20/38: tmp
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 <jt...@163.com>
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