support bypassing shared lock(#10667) (#10770)

close #10704

This PR support parsing the TiKV's shared lock and bypass them in conflict check.

Signed-off-by: you06 <you1474600@gmail.com>

Author: you06

contrib/kvproto

@@ -477,6 +477,7 @@ static_assert(RAFT_REGION_BIG_WRITE_THRES * 4 < RAFT_REGION_BIG_WRITE_MAX, "Inva
       F(type_apply_snapshot_write, {"type", "apply_snapshot_write"}),                                                               \
       F(type_large_txn_lock_put, {"type", "large_txn_lock_put"}),                                                                   \
       F(type_large_txn_lock_del, {"type", "large_txn_lock_del"}),                                                                   \
+      F(type_shared_lock_put, {"type", "shared_lock_put"}),                                                                         \
       F(type_ingest_sst, {"type", "ingest_sst"}))                                                                                   \
     M(tiflash_raft_apply_write_command_duration_seconds,                                                                            \
       "Bucketed histogram of applying write command Raft logs",                                                                     \

dbms/src/Common/TiFlashMetrics.h

@@ -72,6 +72,10 @@ RegionDataRes RegionCFDataBase<RegionLockCFDataTrait>::insert(TiKVKey && key, Ti
             {
                 GET_METRIC(tiflash_raft_process_keys, type_large_txn_lock_put).Increment(1);
             }
+            if (decoded->lock_type == kvrpcpb::Op::SharedLock)
+            {
+                GET_METRIC(tiflash_raft_process_keys, type_shared_lock_put).Increment(1);
+            }
         }
     }
     // according to the process of pessimistic lock, just overwrite.

dbms/src/Storages/KVStore/MultiRaft/RegionCFDataBase.cpp

@@ -225,7 +225,7 @@ DecodedLockCFValuePtr RegionData::getLockInfo(const RegionLockReadQuery & query)
         const auto & lock_info = *lock_info_ptr;
 
         if (lock_info.lock_version > query.read_tso || lock_info.lock_type == kvrpcpb::Op::Lock
-            || lock_info.lock_type == kvrpcpb::Op::PessimisticLock)
+            || lock_info.lock_type == kvrpcpb::Op::PessimisticLock || lock_info.lock_type == kvrpcpb::Op::SharedLock)
             continue;
         if (lock_info.min_commit_ts > query.read_tso)
             continue;

dbms/src/Storages/KVStore/MultiRaft/RegionData.cpp

@@ -44,6 +44,10 @@ inline void decodeLockCfValue(DecodedLockCFValue & res)
     case LockType::Pessimistic:
         lock_type = kvrpcpb::Op::PessimisticLock;
         break;
+    case LockType::Shared:
+        // Skip parsing shared locks as they do not block any read requests.
+        res.lock_type = kvrpcpb::Op::SharedLock;
+        return;
     }
     res.lock_type = lock_type;
     res.primary_lock = readVarString<std::string_view>(data, len);

dbms/src/Storages/KVStore/TiKVHelpers/DecodedLockCFValue.cpp

@@ -370,6 +370,7 @@ enum LockType : UInt8
     Delete = 'D',
     Lock = 'L',
     Pessimistic = 'S',
+    Shared = 'H',
 };
 
 struct InnerDecodedWriteCFValue

dbms/src/Storages/KVStore/TiKVHelpers/TiKVRecordFormat.h

@@ -675,6 +675,18 @@ TEST_F(RegionKVStoreOldTest, RegionReadWrite)
         }
         region->clearAllData();
     }
+    {
+        region->insert(
+            "lock",
+            RecordKVFormat::genKey(table_id, 3),
+            RecordKVFormat::encodeLockCfValue(RecordKVFormat::LockType::Shared, "PK", 3, 20, nullptr, 5));
+        {
+            auto iter = region->createCommittedScanner(true, true);
+            auto lock = iter.getLockInfo({100, nullptr});
+            ASSERT_EQ(lock, nullptr);
+        }
+        region->clearAllData();
+    }
     {
         // Test duplicate and tryCompactionFilter
         region->insert(