From f6779c117b83adc3b6fccfdba984c010219cfc2b Mon Sep 17 00:00:00 2001
From: Wenxuan <breezewish@outlook.com>
Date: Sat, 4 Feb 2023 10:11:56 +0800
Subject: [PATCH] Fix incorrect string value when doing int_handle.toNext
 (#6724)

close pingcap/tiflash#6699

Signed-off-by: ywqzzy <592838129@qq.com>
---
 dbms/src/Storages/DeltaMerge/RowKeyRange.h    | 13 ++++----
 .../DeltaMerge/tests/gtest_key_range.cpp      | 31 +++++++++++++++++++
 2 files changed, 38 insertions(+), 6 deletions(-)

diff --git a/dbms/src/Storages/DeltaMerge/RowKeyRange.h b/dbms/src/Storages/DeltaMerge/RowKeyRange.h
index acbcda2ec73..96491471875 100644
--- a/dbms/src/Storages/DeltaMerge/RowKeyRange.h
+++ b/dbms/src/Storages/DeltaMerge/RowKeyRange.h
@@ -130,7 +130,7 @@ struct RowKeyValue
         return std::make_shared<DecodedTiKVKey>(prefix + *value);
     }
 
-    bool operator==(const RowKeyValue & v)
+    bool operator==(const RowKeyValue & v) const
     {
         return is_common_handle == v.is_common_handle && (*value) == (*v.value) && int_value == v.int_value;
     }
@@ -173,14 +173,15 @@ struct RowKeyValue
      */
     RowKeyValue toNext() const
     {
+        // We want to always ensure that the IntHandle.stringValue == IntHandle.intValue.
+        if (!is_common_handle)
+            return toPrefixNext();
+
         HandleValuePtr next_value = std::make_shared<String>(value->begin(), value->end());
         next_value->push_back(0x0);
 
-        Int64 next_int_value = int_value;
-        if (!is_common_handle && next_int_value != std::numeric_limits<Int64>::max())
-            next_int_value++;
-
-        return RowKeyValue(is_common_handle, next_value, next_int_value);
+        // For common handle, int_value will not be used in compare. Let's keep it unchanged.
+        return RowKeyValue(/* is_common_handle */ true, next_value, int_value);
     }
 
     void serialize(WriteBuffer & buf) const
diff --git a/dbms/src/Storages/DeltaMerge/tests/gtest_key_range.cpp b/dbms/src/Storages/DeltaMerge/tests/gtest_key_range.cpp
index 40a0cf782e6..6ce6fe00064 100644
--- a/dbms/src/Storages/DeltaMerge/tests/gtest_key_range.cpp
+++ b/dbms/src/Storages/DeltaMerge/tests/gtest_key_range.cpp
@@ -148,6 +148,37 @@ TEST(RowKey, ToNextKeyCommonHandle)
     EXPECT_EQ(0, compare(my_next.toRowKeyValueRef(), next.toRowKeyValueRef()));
 }
 
+TEST(RowKey, NextIntHandleCompare)
+{
+    auto int_max = RowKeyValue::INT_HANDLE_MAX_KEY;
+    auto int_max_i64 = RowKeyValue::fromHandle(Handle(std::numeric_limits<HandleID>::max()));
+
+    EXPECT_EQ(1, compare(int_max.toRowKeyValueRef(), int_max_i64.toRowKeyValueRef()));
+
+    auto int_max_i64_pnext = int_max_i64.toPrefixNext();
+    EXPECT_EQ(int_max, int_max_i64_pnext);
+    EXPECT_EQ(0, compare(int_max.toRowKeyValueRef(), int_max_i64_pnext.toRowKeyValueRef()));
+    EXPECT_EQ(0, compare(int_max_i64_pnext.toRowKeyValueRef(), int_max.toRowKeyValueRef()));
+
+    auto int_max_i64_next = int_max_i64.toNext();
+    EXPECT_EQ(int_max, int_max_i64_next);
+    EXPECT_EQ(0, compare(int_max.toRowKeyValueRef(), int_max_i64_next.toRowKeyValueRef()));
+    EXPECT_EQ(0, compare(int_max_i64_next.toRowKeyValueRef(), int_max.toRowKeyValueRef()));
+}
+
+TEST(RowKey, NextIntHandleMinMax)
+{
+    auto v0 = RowKeyValue::fromHandle(Handle(1178400));
+    auto v0_next = v0.toNext();
+    auto v1 = RowKeyValue::fromHandle(Handle(1178401));
+
+    EXPECT_EQ(v0, min(v0, v1));
+    EXPECT_EQ(v0, min(v0, v0_next));
+
+    EXPECT_EQ(v1, max(v0, v1));
+    EXPECT_EQ(v1, max(v0, v0_next));
+}
+
 } // namespace tests
 } // namespace DM
 } // namespace DB