This is an automated email from the ASF dual-hosted git repository. jackietien pushed a commit to branch fixQueryBug in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 219b13370a04df9a1bcb8ef69c615eb0dcf58131 Author: JackieTien97 <[email protected]> AuthorDate: Fri Jul 12 20:31:44 2024 +0800 Fix Unseq data can't update the seq data bug --- .../execution/operator/source/SeriesScanUtil.java | 2 +- .../read/reader/common/PriorityMergeReader.java | 23 +++++++++++++++------- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java index 2c687dcc1a5..f1a20ef86be 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java @@ -1199,7 +1199,7 @@ public class SeriesScanUtil implements Accountable { VersionPageReader( QueryContext context, long version, long offset, IPageReader data, boolean isSeq) { this.context = context; - this.version = new PriorityMergeReader.MergeReaderPriority(version, offset); + this.version = new PriorityMergeReader.MergeReaderPriority(version, offset, isSeq); this.data = data; this.isSeq = isSeq; this.isAligned = data instanceof AlignedPageReader || data instanceof MemAlignedPageReader; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReader.java index 7fc89dce439..122421ac2dd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReader.java @@ -62,7 +62,7 @@ public class PriorityMergeReader implements IPointReader { public void addReader(IPointReader reader, long priority) throws IOException { if (reader.hasNextTimeValuePair()) { heap.add( - new Element(reader, reader.nextTimeValuePair(), new MergeReaderPriority(priority, 0))); + new Element(reader, reader.nextTimeValuePair(), new MergeReaderPriority(priority, 0, false))); } else { reader.close(); } @@ -189,20 +189,29 @@ public class PriorityMergeReader implements IPointReader { } public static class MergeReaderPriority implements Comparable<MergeReaderPriority> { - long version; - long offset; + final long version; + final long offset; - public MergeReaderPriority(long version, long offset) { + final boolean isSeq; + + public MergeReaderPriority(long version, long offset, boolean isSeq) { this.version = version; this.offset = offset; + this.isSeq = isSeq; } @Override public int compareTo(MergeReaderPriority o) { - if (version < o.version) { - return -1; + if (isSeq != o.isSeq) { + // one is seq and another is unseq, unseq always win + return isSeq ? -1 : 1; + } else { + // both seq or both unseq, using version + offset to compare + if (version < o.version) { + return -1; + } + return ((version > o.version) ? 1 : (Long.compare(offset, o.offset))); } - return ((version > o.version) ? 1 : (Long.compare(offset, o.offset))); } @Override
