Fix Unseq data can't update the seq data bug (#12918)

(cherry picked from commit e8385b9)

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBDuplicateTimeIT.java

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.it;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBDuplicateTimeIT {
+
+  @Before
+  public void setUp() throws Exception {
+    EnvFactory.getEnv().getConfig().getCommonConfig().setAvgSeriesPointNumberThreshold(2);
+    // Adjust memstable threshold size to make it flush automatically
+    EnvFactory.getEnv().initClusterEnvironment();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  @Test
+  public void testDuplicateTime() throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+
+      statement.execute("create timeseries root.db.d1.s1 with datatype=INT32,encoding=PLAIN");
+      // version-1 tsfile
+      statement.execute("insert into root.db.d1(time,s1) values (2,2)");
+      statement.execute("insert into root.db.d1(time,s1) values (3,3)");
+
+      // version-2 unseq work memtable
+      statement.execute("insert into root.db.d1(time,s1) values (2,20)");
+
+      // version-3 tsfile
+      statement.execute("insert into root.db.d1(time,s1) values (5,5)");
+      statement.execute("insert into root.db.d1(time,s1) values (6,6)");
+
+      // version-2 unseq work memtable -> unseq tsfile
+      statement.execute("insert into root.db.d1(time,s1) values (5,50)");
+
+      try (ResultSet set = statement.executeQuery("SELECT s1 FROM root.db.d1 where time = 5")) {
+        int cnt = 0;
+        while (set.next()) {
+          assertEquals(5L, set.getLong(1));
+          assertEquals(50, set.getInt(2));
+
+          cnt++;
+        }
+        assertEquals(1, cnt);
+      }
+    }
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java

@@ -1181,7 +1181,7 @@ protected static class VersionPageReader {
     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;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/common/PriorityMergeReader.java

@@ -54,7 +54,8 @@ public PriorityMergeReader() {
   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();
     }
@@ -162,20 +163,29 @@ public void close() throws IOException {
   }
 
   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