You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2024/07/12 12:32:02 UTC
(iotdb) 01/01: Fix Unseq data can't update the seq data bug
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 <ja...@gmail.com>
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