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 72fd1eddb51b24ddee62b4d9ebce7ffcc919bc49 Author: Tian Jiang <[email protected]> AuthorDate: Mon Sep 21 18:33:52 2026 +0800 tmp --- .../operator/source/AlignedSeriesScanUtil.java | 8 ++++++-- .../utils/writer/AbstractCompactionWriter.java | 2 +- .../utils/writer/AbstractCrossCompactionWriter.java | 2 +- .../reader/common/AlignedPriorityMergeReader.java | 20 +++++++++++++++++++- .../datastructure/ColumnSortedAlignedTVList.java | 9 ++------- 5 files changed, 29 insertions(+), 12 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 4dd97e629e3..6fe02e754e2 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 @@ -26,6 +26,9 @@ import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.AlignedDescPriorityMergeReader; import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.AlignedPriorityMergeReader; import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.DescPriorityMergeReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.FlatDeviceDescPriorityMergeReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.FlatDevicePointReader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.FlatDevicePriorityMergeReader; import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.PriorityMergeReader; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -37,6 +40,7 @@ import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.write.schema.IMeasurementSchema; import java.io.IOException; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.stream.Collectors; @@ -110,12 +114,12 @@ public class AlignedSeriesScanUtil extends SeriesScanUtil { @Override protected PriorityMergeReader getPriorityMergeReader() { - return new AlignedPriorityMergeReader(); + return new FlatDevicePriorityMergeReader(); } @Override protected DescPriorityMergeReader getDescPriorityMergeReader() { - return new AlignedDescPriorityMergeReader(); + return new FlatDeviceDescPriorityMergeReader(); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java index d25f0264378..c5e873975c6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java @@ -172,7 +172,7 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { while (rows.hasNextTimeValuePair()) { TimeValuePair row = rows.nextTimeValuePair(); if (lastRowKey != null - && AlignedPriorityMergeReader.compareAlignedRowKey( + && AlignedPriorityMergeReader.compareFlatDeviceRowKey( lastRowKey, row, types, rowKeyColumnIndices) >= 0) { throw new CompactionLastTimeCheckFailedException( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java index 880d688080a..89bdaf29870 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCrossCompactionWriter.java @@ -53,7 +53,7 @@ public abstract class AbstractCrossCompactionWriter extends AbstractCompactionWr protected List<CompactionTsFileWriter> targetFileWriters = new ArrayList<>(); // source tsfiles - private List<TsFileResource> seqTsFileResources; + protected List<TsFileResource> seqTsFileResources; // Each sub task has its corresponding seq file index. // The index of the array corresponds to subTaskId. diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReader.java index 771da6c7698..63355b3160a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/AlignedPriorityMergeReader.java @@ -63,6 +63,24 @@ public class AlignedPriorityMergeReader extends PriorityMergeReader { if (result != 0) { return result; } + return compareTagValues(left, right, dataTypes, rowKeyColumnIndices); + } + + /** Flat storage and scans order logical devices before timestamps within each device. */ + public static int compareFlatDeviceRowKey( + TimeValuePair left, + TimeValuePair right, + List<TSDataType> dataTypes, + List<Integer> rowKeyColumnIndices) { + int result = compareTagValues(left, right, dataTypes, rowKeyColumnIndices); + return result != 0 ? result : Long.compare(left.getTimestamp(), right.getTimestamp()); + } + + private static int compareTagValues( + TimeValuePair left, + TimeValuePair right, + List<TSDataType> dataTypes, + List<Integer> rowKeyColumnIndices) { TsPrimitiveType[] leftValues = left.getValue().getVector(); TsPrimitiveType[] rightValues = right.getValue().getVector(); for (int columnIndex : rowKeyColumnIndices) { @@ -77,7 +95,7 @@ public class AlignedPriorityMergeReader extends PriorityMergeReader { if (rightValue == null || rightValue.getValue() == null) { return 1; } - result = compareValue(leftValue, rightValue, dataTypes.get(columnIndex)); + int result = compareValue(leftValue, rightValue, dataTypes.get(columnIndex)); if (result != 0) { return result; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java index 8bead9f3a3f..2d0e7da1c27 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java @@ -378,11 +378,6 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { @Override public int compareRowKey(int leftRowIndex, AlignedTVList rightList, int rightRowIndex) { - int result = Long.compare(getTime(leftRowIndex), rightList.getTime(rightRowIndex)); - if (result != 0) { - return result; - } - List<Integer> rightSortColumnIndices = rightList instanceof ColumnSortedAlignedTVList rightColumnSortedList ? rightColumnSortedList.sortColumnIndices @@ -395,14 +390,14 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { keyIndex < sortColumnIndices.size() ? sortColumnIndices.get(keyIndex) : -1; int rightColumnIndex = keyIndex < rightSortColumnIndices.size() ? rightSortColumnIndices.get(keyIndex) : -1; - result = + int result = compareValues( leftValueIndex, rightList, rightValueIndex, leftColumnIndex, rightColumnIndex); if (result != 0) { return result; } } - return 0; + return Long.compare(getTime(leftRowIndex), rightList.getTime(rightRowIndex)); } private int compareValues(
