Storage: Adapt to ingesting snapshot with irregular region range (#10151) (release-8.5) (#10157)

close #10147

Storage: Adapt to ingesting snapshot with irregular region range
If the key format is "t${tableID}_r${handleID1}{AnySuffix}", and `AnySuffix` is not empty, `"t${tableID}_r${handleID1}{AnySuffix}" > "t${tableID}_r${handleID1}"` according to the the comparison semantics of "Key" on TiDB/TiKV.

For example,
If a table is non-clustered, that is with Int64 as the HandleID type.
And the range of a Region is `["xxx_r{handleId1}{AnySuffix}", "xxx_r{handleId2}{AnySuffix}")`, where handleId1 = 100, handleId2 = 200.

If `AnySuffix` is empty, then Region contains the left-closed-right-open range of `[handleID1, handleID2)`
If `AnySuffix` is not empty, then Region contains the range `(handleID1, handleID2]`, which is actually the left-closed-right-open range of `[handleID1+1, handleID2+1)`.

Signed-off-by: ti-chi-bot <[email protected]>
Signed-off-by: JaySon-Huang <[email protected]>

Co-authored-by: JaySon <[email protected]>
Co-authored-by: JaySon-Huang <[email protected]>

Author: ti-chi-bot

dbms/src/Storages/DeltaMerge/DeltaMergeStore_Ingest.cpp

@@ -29,6 +29,7 @@
 #include <Storages/KVStore/MultiRaft/Disagg/FastAddPeerContext.h>
 #include <Storages/KVStore/TMTContext.h>
 #include <Storages/PathPool.h>
+#include <common/logger_useful.h>
 
 #include <magic_enum.hpp>
 
@@ -131,7 +132,7 @@ void DeltaMergeStore::cleanPreIngestFiles(
         {
             // For disagg mode
             // - if the job has been finished, it means the local files is likely all uploaded to S3
-            // - if the job is intrrupted, it means the `SSTFilesToDTFilesOutputStream::cancel` is called
+            // - if the job is interrupted, it means the `SSTFilesToDTFilesOutputStream::cancel` is called
             //   and local files are also removed.
             // So we ignore the files on disk.
             removePreIngestFile(f.id, false);
@@ -269,9 +270,9 @@ Segments DeltaMergeStore::ingestDTFilesUsingSplit(
                 /*throw_if_notfound*/ true);
 
             const auto delete_range = remaining_delete_range.shrink(segment->getRowKeyRange());
+            // as remaining_delete_range is not none, we expect the shrunk range to be not none.
             RUNTIME_CHECK(
-                !delete_range
-                     .none(), // as remaining_delete_range is not none, we expect the shrinked range to be not none.
+                !delete_range.none(),
                 delete_range.toDebugString(),
                 segment->simpleInfo(),
                 remaining_delete_range.toDebugString());
@@ -616,13 +617,13 @@ UInt64 DeltaMergeStore::ingestFiles(
         }
 
         // Check whether all external files are contained by the range.
-        if (dm_context->global_context.getSettingsRef().dt_enable_ingest_check)
+        for (const auto & ext_file : external_files)
         {
-            for (const auto & ext_file : external_files)
+            if (dm_context->global_context.getSettingsRef().dt_enable_ingest_check)
             {
                 RUNTIME_CHECK_MSG(
                     range.getStart() <= ext_file.range.getStart() && range.getEnd() >= ext_file.range.getEnd(),
-                    "Detected illegal region boundary: range={} file_range={} keyspace={} table_id={}. "
+                    "Detected illegal region boundary: keyspace={} table_id={} range={} file_range={}. "
                     "TiFlash will exit to prevent data inconsistency. "
                     "If you accept data inconsistency and want to continue the service, "
                     "set profiles.default.dt_enable_ingest_check=false .",
@@ -631,6 +632,21 @@ UInt64 DeltaMergeStore::ingestFiles(
                     range.toDebugString(),
                     ext_file.range.toDebugString());
             }
+            else
+            {
+                // If the check is disabled, we just log a warning for better diagnosing.
+                if (unlikely(
+                        !(range.getStart() <= ext_file.range.getStart() && range.getEnd() >= ext_file.range.getEnd())))
+                {
+                    LOG_WARNING(
+                        log,
+                        "Detected illegal region boundary: keyspace={} table_id={} range={} file_range={}",
+                        keyspace_id,
+                        physical_table_id,
+                        range.toDebugString(),
+                        ext_file.range.toDebugString());
+                }
+            }
         }
     }
 
@@ -862,9 +878,9 @@ std::vector<SegmentPtr> DeltaMergeStore::ingestSegmentsUsingSplit(
                 /*throw_if_notfound*/ true);
 
             const auto delete_range = remaining_delete_range.shrink(segment->getRowKeyRange());
+            // as remaining_delete_range is not none, we expect the shrunk range to be not none.
             RUNTIME_CHECK(
-                !delete_range
-                     .none(), // as remaining_delete_range is not none, we expect the shrinked range to be not none.
+                !delete_range.none(),
                 delete_range.toDebugString(),
                 segment->simpleInfo(),
                 remaining_delete_range.toDebugString());

dbms/src/Storages/DeltaMerge/RowKeyRange.cpp

@@ -46,6 +46,59 @@ const RowKeyValue RowKeyValue::COMMON_HANDLE_MAX_KEY
     = RowKeyValue(true, std::make_shared<String>(1, TiDB::CodecFlag::CodecFlagMax), 0);
 const RowKeyValue RowKeyValue::EMPTY_STRING_KEY = RowKeyValue(true, std::make_shared<String>(""), 0);
 
+std::pair<RowKeyValue, std::string_view> RowKeyValue::fromHandleWithSuffix(
+    bool is_common_handle_,
+    const HandleValuePtr value_)
+{
+    if (is_common_handle_)
+    {
+        RowKeyValue rowkey_value(is_common_handle_, value_, 0);
+        return {rowkey_value, std::string_view{nullptr, 0}};
+    }
+
+    assert(!is_common_handle_);
+    // According to the tidb encoding rule, the int handle must be larger or equal to 8 bytes.
+    RUNTIME_CHECK_MSG(
+        value_->size() >= sizeof(Int64),
+        "Meet illegal int handle size less than expected, value_size={} value={}",
+        value_->size(),
+        Redact::keyToHexString(value_->data(), value_->size()));
+
+    size_t cursor = 0;
+    Int64 int_value = DB::DecodeInt64(cursor, *value_);
+    if (unlikely(value_->size() > sizeof(UInt64)))
+    {
+        // For int type handle, the standard key encoding format should be t{table_id}_r{handle_value}.
+        // But TiKV may generate region range keys which are not strictly following the standard format.
+        // More concretely, the key may be t{table_id}_r{handle_value}{any_suffix}.
+        // We need to adapt the key to the standard format.
+        // For example, the key may be t100_r1000 + '0x00', '0x01' or any other suffix, we need to adapt it to t100_r1001.
+        // This is ok, because
+        //  1) if the key is the start range, then [t100_r1000 + 0x00, xxx) has the same semantics with [t100_r1001, xxx)
+        //  2) if the key is the end range, then [xxx, t100_r1000 + 0x00) also has the same semantics with [xxx, t100_r1001)
+        //
+        // Note if the `int_value` is Int64::max_value,
+        // it is a value generated by tiflash itself to means +inf()(which is RowKeyValue::INT_HANDLE_MAX_KEY).
+        // So we can just ignore it.
+        if (int_value < std::numeric_limits<Int64>::max())
+        {
+            int_value = int_value + 1;
+            WriteBufferFromOwnString ss;
+            DB::EncodeInt64(int_value, ss);
+            HandleValuePtr fixed_value = std::make_shared<String>(ss.releaseStr());
+            return {
+                RowKeyValue(is_common_handle_, fixed_value, int_value),
+                std::string_view{value_->data() + sizeof(int_value), value_->size() - sizeof(int_value)},
+            };
+        }
+        else
+        {
+            // keep the int_value unchanged if it is RowKeyValue::INT_HANDLE_MAX_KEY
+        }
+    }
+    return {RowKeyValue(is_common_handle_, value_, int_value), std::string_view{nullptr, 0}};
+}
+
 RowKeyValue RowKeyValueRef::toRowKeyValue() const
 {
     if (data == nullptr)

dbms/src/Storages/DeltaMerge/RowKeyRange.h

@@ -22,11 +22,14 @@
 #include <Storages/DeltaMerge/DeltaMergeHelpers.h>
 #include <Storages/DeltaMerge/dtpb/column_file.pb.h>
 #include <Storages/KVStore/Decode/DecodedTiKVKeyValue.h>
-#include <Storages/KVStore/MultiRaft/RegionRangeKeys.h>
 #include <Storages/KVStore/TiKVHelpers/TiKVRecordFormat.h>
 #include <Storages/KVStore/Types.h>
 #include <TiDB/Decode/DatumCodec.h>
 
+namespace DB
+{
+class RegionRangeKeys;
+}
 namespace DB::DM
 {
 using HandleValuePtr = std::shared_ptr<String>;
@@ -86,62 +89,9 @@ struct RowKeyValue
         , int_value(int_value_)
     {}
 
-    RowKeyValue(bool is_common_handle_, HandleValuePtr value_)
-        : is_common_handle(is_common_handle_)
-        , value(value_)
-    {
-        if (is_common_handle)
-            int_value = 0;
-        else
-        {
-            size_t cursor = 0;
-            int_value = DB::DecodeInt64(cursor, *value);
-            if (unlikely(value->size() != sizeof(Int64)))
-            {
-                // For int type handle, the standard key enconding format should be t{table_id}_r{handle_value}.
-                // But TiKV may generate region range keys which are not strictly following the standard format.
-                // More concretely, the key may be t{table_id}_r{handle_value} + some other bytes.
-                // We need to adapt the key to the standard format.
-                // For example, the key may be t100_r1000 + 0x00, we need to adapt it to t100_r1001.
-                // This is ok, because
-                //  1) if the key is the start range, then [t100_r1000 + 0x00, xxx) has the same semantics with [t100_r1001, xxx)
-                //  2) if the key is the end range, then [xxx, t100_r1000 + 0x00) also has the same semantics with [xxx, t100_r1001)
-                //
-                // Note if the `int_value` is Int64::max_value,
-                // it is a value generated by tiflash itself to means +inf()(which is RowKeyValue::INT_HANDLE_MAX_KEY).
-                // So we can just ignore it.
-                if (value->size() != sizeof(UInt64) + 1 || value->back() != 0x00)
-                {
-                    LOG_WARNING(
-                        Logger::get(),
-                        "Meet rowkey {} with unexpected encoding format",
-                        Redact::keyToDebugString(value->data(), value->size()));
-                }
-                else
-                {
-                    if (int_value < std::numeric_limits<Int64>::max())
-                    {
-                        LOG_WARNING(
-                            Logger::get(),
-                            "Meet rowkey {} which has an extra zero suffix",
-                            Redact::keyToDebugString(value->data(), value->size()));
-                        int_value = int_value + 1;
-                        WriteBufferFromOwnString ss;
-                        DB::EncodeInt64(int_value, ss);
-                        value = std::make_shared<String>(ss.releaseStr());
-                    }
-                    else
-                    {
-                        // ignore RowKeyValue::INT_HANDLE_MAX_KEY
-                    }
-                }
-            }
-        }
-    }
-
     explicit RowKeyValue(const RowKeyValueRef & rowkey_value)
+        : is_common_handle(rowkey_value.is_common_handle)
     {
-        is_common_handle = rowkey_value.is_common_handle;
         if (is_common_handle)
             value = std::make_shared<String>(rowkey_value.data, rowkey_value.size);
         else
@@ -153,12 +103,32 @@ struct RowKeyValue
         int_value = rowkey_value.int_value;
     }
 
-    static RowKeyValue fromHandle(Handle value)
+#ifndef NDEBUG
+    // Generate from int handle
+    static RowKeyValue fromIntHandle(Handle value)
     {
         WriteBufferFromOwnString ss;
         DB::EncodeInt64(value, ss);
         return RowKeyValue(false, std::make_shared<String>(ss.releaseStr()), value);
     }
+#endif
+
+    static RowKeyValue fromHandle(bool is_common_handle_, const HandleValuePtr value_)
+    {
+        return fromHandleWithSuffix(is_common_handle_, value_).first;
+    }
+
+    /**
+     * Parse the handle from `value_`.
+     * If the handle is a common handle, the return RowKeyValue is the same with `value_`.
+     * If the handle is an int handle, the `int_value` will be the int value.
+     * Specially, for int handle, if there is any suffix rather than the regular format,
+     * a.k.a "t${tableID}_r{handleID}{AnySuffix}" `AnySuffix` is not empty, then
+     * the `int_value` will be the next int value, and the `AnySuffix` will be returned.
+     *
+     * Note that the `AnySuffix` rely on the lifetime of `value_`
+     */
+    static std::pair<RowKeyValue, std::string_view> fromHandleWithSuffix(bool is_common_handle_, HandleValuePtr value_);
 
     // Format as a string
     String toString() const;
@@ -171,18 +141,6 @@ struct RowKeyValue
         return RowKeyValueRef{is_common_handle, value->data(), value->size(), int_value};
     }
 
-    DecodedTiKVKeyPtr toRegionKey(TableID table_id) const
-    {
-        // FIXME: move this to TiKVRecordFormat.h
-        WriteBufferFromOwnString ss;
-        ss.write('t');
-        EncodeInt64(table_id, ss);
-        ss.write('_');
-        ss.write('r');
-        String prefix = ss.releaseStr();
-        return std::make_shared<DecodedTiKVKey>(prefix + *value);
-    }
-
     bool operator==(const RowKeyValue & v) const
     {
         return is_common_handle == v.is_common_handle && (*value) == (*v.value) && int_value == v.int_value;
@@ -249,7 +207,7 @@ struct RowKeyValue
         readBoolText(is_common_handle, buf);
         readStringBinary(value, buf);
         HandleValuePtr start_ptr = std::make_shared<String>(value);
-        return RowKeyValue(is_common_handle, start_ptr);
+        return RowKeyValue::fromHandle(is_common_handle, start_ptr);
     }
 
     bool is_common_handle = false;
@@ -577,8 +535,8 @@ struct RowKeyRange
             end_ptr = RowKeyValue::COMMON_HANDLE_MAX_KEY.value;
         }
         return RowKeyRange(
-            RowKeyValue(is_common_handle, start_ptr),
-            RowKeyValue(is_common_handle, end_ptr),
+            RowKeyValue::fromHandle(is_common_handle, start_ptr),
+            RowKeyValue::fromHandle(is_common_handle, end_ptr),
             is_common_handle,
             rowkey_column_size);
     }
@@ -598,8 +556,8 @@ struct RowKeyRange
             end_ptr = RowKeyValue::COMMON_HANDLE_MAX_KEY.value;
         }
         return RowKeyRange(
-            RowKeyValue(is_common_handle, start_ptr),
-            RowKeyValue(is_common_handle, end_ptr),
+            RowKeyValue::fromHandle(is_common_handle, start_ptr),
+            RowKeyValue::fromHandle(is_common_handle, end_ptr),
             is_common_handle,
             rowkey_column_size);
     }
@@ -758,9 +716,10 @@ struct RowKeyRange
             String end = ss.releaseStr();
             /// when handle_range.end == HandleRange::MAX, according to previous implementation, it should be +Inf
             return RowKeyRange(
-                RowKeyValue(is_common_handle, std::make_shared<String>(start)),
-                handle_range.end == HandleRange::MAX ? RowKeyValue::COMMON_HANDLE_MAX_KEY
-                                                     : RowKeyValue(is_common_handle, std::make_shared<String>(end)),
+                RowKeyValue::fromHandle(is_common_handle, std::make_shared<String>(start)),
+                handle_range.end == HandleRange::MAX
+                    ? RowKeyValue::COMMON_HANDLE_MAX_KEY
+                    : RowKeyValue::fromHandle(is_common_handle, std::make_shared<String>(end)),
                 /*is_common_handle=*/is_common_handle,
                 1);
         }
@@ -794,64 +753,17 @@ struct RowKeyRange
 
     static RowKeyRange fromRegionRange(
         const std::shared_ptr<const RegionRangeKeys> & region_range,
-        const TableID table_id,
+        TableID table_id,
         bool is_common_handle,
-        size_t rowkey_column_size)
-    {
-        return fromRegionRange(
-            region_range->rawKeys(),
-            region_range->getMappedTableID(),
-            table_id,
-            is_common_handle,
-            rowkey_column_size);
-    }
+        size_t rowkey_column_size,
+        const String & tracing_msg = "");
     static RowKeyRange fromRegionRange(
         const std::pair<DecodedTiKVKeyPtr, DecodedTiKVKeyPtr> & raw_keys,
-        const TableID table_id_in_raw_key,
-        const TableID table_id,
+        TableID table_id_in_raw_key,
+        TableID table_id,
         bool is_common_handle,
-        size_t rowkey_column_size)
-    {
-        if (likely(table_id_in_raw_key == table_id))
-        {
-            auto & start_key = *raw_keys.first;
-            auto & end_key = *raw_keys.second;
-            auto keyspace_id = start_key.getKeyspaceID();
-            const auto & table_range_min_max = getTableMinMaxData(keyspace_id, table_id, is_common_handle);
-            RowKeyValue start_value, end_value;
-            if (start_key <= *table_range_min_max.min)
-            {
-                if (is_common_handle)
-                    start_value = RowKeyValue::COMMON_HANDLE_MIN_KEY;
-                else
-                    start_value = RowKeyValue::INT_HANDLE_MIN_KEY;
-            }
-            else
-            {
-                start_value = RowKeyValue(
-                    is_common_handle,
-                    std::make_shared<std::string>(RecordKVFormat::getRawTiDBPKView(start_key)));
-            }
-            if (end_key >= *table_range_min_max.max)
-            {
-                if (is_common_handle)
-                    end_value = RowKeyValue::COMMON_HANDLE_MAX_KEY;
-                else
-                    end_value = RowKeyValue::INT_HANDLE_MAX_KEY;
-            }
-            else
-                end_value = RowKeyValue(
-                    is_common_handle,
-                    std::make_shared<std::string>(RecordKVFormat::getRawTiDBPKView(end_key)));
-            return RowKeyRange(start_value, end_value, is_common_handle, rowkey_column_size);
-        }
-        else
-        {
-            /// if table id is not the same, just return none range
-            /// maybe should throw exception since it should not happen
-            return newNone(is_common_handle, rowkey_column_size);
-        }
-    }
+        size_t rowkey_column_size,
+        const String & tracing_msg);
 
     // Format as a string
     String toString() const;