[FLASH-402] Skip unneeded columns when decode row

dbms/src/Storages/Transaction/Codec.h

@@ -251,6 +251,96 @@ inline Field DecodeDatum(size_t & cursor, const String & raw_value)
     }
 }
 
+inline void SkipBytes(size_t & cursor, const String & raw_value)
+{
+    while (true)
+    {
+        size_t next_cursor = cursor + 9;
+        if (next_cursor > raw_value.size())
+            throw Exception("Wrong format, cursor over buffer size. (DecodeBytes)", ErrorCodes::LOGICAL_ERROR);
+        UInt8 marker = (UInt8)raw_value[cursor + 8];
+        UInt8 pad_size = ENC_MARKER - marker;
+
+        if (pad_size > 8)
+            throw Exception("Wrong format, too many padding bytes. (DecodeBytes)", ErrorCodes::LOGICAL_ERROR);
+        cursor = next_cursor;
+        if (pad_size != 0)
+            break;
+    }
+}
+
+inline void SkipCompactBytes(size_t & cursor, const String & raw_value)
+{
+    size_t size = DecodeVarInt(cursor, raw_value);
+    cursor += size;
+}
+
+inline void SkipVarUInt(size_t & cursor, const String & raw_value)
+{
+    for (int i = 0; cursor < raw_value.size(); i++)
+    {
+        UInt64 v = raw_value[cursor++];
+        if (v < 0x80)
+        {
+            if (i > 9 || (i == 9 && v > 1))
+                throw Exception("Overflow when DecodeVarUInt", ErrorCodes::LOGICAL_ERROR);
+            return;
+        }
+    }
+    throw Exception("Wrong format. (DecodeVarUInt)", ErrorCodes::LOGICAL_ERROR);
+}
+
+inline void SkipVarInt(size_t & cursor, const String & raw_value)
+{
+    SkipVarUInt(cursor, raw_value);
+}
+
+inline void SkipDecimal(size_t & cursor, const String & raw_value)
+{
+    PrecType prec = raw_value[cursor++];
+    ScaleType frac = raw_value[cursor++];
+
+    int binSize = getBytes(prec, frac);
+    cursor += binSize;
+}
+
+inline void SkipDatum(size_t & cursor, const String & raw_value)
+{
+    switch (raw_value[cursor++])
+    {
+        case TiDB::CodecFlagNil:
+            return;
+        case TiDB::CodecFlagInt:
+            cursor += sizeof(Int64);
+            return;
+        case TiDB::CodecFlagUInt:
+            cursor += sizeof(UInt64);
+            return;
+        case TiDB::CodecFlagBytes:
+            SkipBytes(cursor, raw_value);
+            return;
+        case TiDB::CodecFlagCompactBytes:
+            SkipCompactBytes(cursor, raw_value);
+            return;
+        case TiDB::CodecFlagFloat:
+            cursor += sizeof(UInt64);
+            return;
+        case TiDB::CodecFlagVarUInt:
+            SkipVarUInt(cursor, raw_value);
+            return;
+        case TiDB::CodecFlagVarInt:
+            SkipVarInt(cursor, raw_value);
+            return;
+        case TiDB::CodecFlagDuration:
+            throw Exception("Not implented yet. DecodeDatum: CodecFlagDuration", ErrorCodes::LOGICAL_ERROR);
+        case TiDB::CodecFlagDecimal:
+            SkipDecimal(cursor, raw_value);
+            return;
+        default:
+            throw Exception("Unknown Type:" + std::to_string(raw_value[cursor - 1]), ErrorCodes::LOGICAL_ERROR);
+    }
+}
+
 template <typename T>
 inline void writeIntBinary(const T & x, std::stringstream & ss)
 {

dbms/src/Storages/Transaction/RegionBlockReader.cpp

@@ -125,10 +125,18 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     ColumnID handle_col_id = InvalidColumnID;
 
     std::unordered_map<ColumnID, std::pair<MutableColumnPtr, NameAndTypePair>> column_map;
+    std::unordered_set<ColumnID> column_ids_to_read;
+    std::unordered_set<ColumnID> all_column_ids;
     for (const auto & column_info : table_info.columns)
     {
         ColumnID col_id = column_info.id;
         String col_name = column_info.name;
+        all_column_ids.emplace(col_id);
+        if (std::find(column_names_to_read.begin(), column_names_to_read.end(), col_name) == column_names_to_read.end())
+        {
+            continue;
+        }
+        column_ids_to_read.emplace(col_id);
         auto ch_col = columns.getPhysical(col_name);
         column_map[col_id] = std::make_pair(ch_col.type->createColumn(), ch_col);
         column_map[col_id].first->reserve(data_list.size());
@@ -170,7 +178,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
     std::unordered_set<ColumnID> col_id_included;
 
-    const size_t target_col_size = (!table_info.pk_is_handle ? table_info.columns.size() : table_info.columns.size() - 1) * 2;
+    const size_t target_col_size = (column_names_to_read.size() - 3) * 2;
 
     Block block;
 
@@ -188,6 +196,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
 
             std::vector<Field> row;
+            bool has_unknown_col_id = false;
 
             if (write_type == Region::DelFlag)
             {
@@ -197,12 +206,15 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                     if (handle_col_id == column.id)
                         continue;
 
+                    if (!column_ids_to_read.count(column.id))
+                        continue;
+
                     row.push_back(Field(column.id));
                     row.push_back(GenDecodeRow(column.getCodecFlag()));
                 }
             }
             else
-                row = RecordKVFormat::DecodeRow(*value_ptr);
+                std::tie(row, has_unknown_col_id) = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, all_column_ids);
 
             if (row.size() == 1 && row[0].isNull())
             {
@@ -213,6 +225,12 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             if (row.size() & 1)
                 throw Exception("row size is wrong.", ErrorCodes::LOGICAL_ERROR);
 
+            /// row contains unknown column id
+            /// if force_decode is false, the schema may be too old
+            /// if force decode is true, the column may have been dropped and just ignore it
+            if (has_unknown_col_id && !force_decode)
+                return std::make_tuple(block, false);
+
             /// Modify `row` by adding missing column values or removing useless column values.
 
             col_id_included.clear();
@@ -230,6 +248,9 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 if (!force_decode)
                     return std::make_tuple(block, false);
 
+                if (!column_ids_to_read.count(column.id))
+                    continue;
+
                 row.emplace_back(Field(column.id));
                 if (column.hasNoDefaultValueFlag())
                     // Fill `zero` value if NOT NULL specified or else NULL.
@@ -246,9 +267,6 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 Field & col_id = row[i];
                 if (column_map.find(col_id.get<ColumnID>()) == column_map.end())
                 {
-                    if (!force_decode)
-                        return std::make_tuple(block, false);
-
                     row.erase(row.begin() + i, row.begin() + i + 2);
                 }
             }

dbms/src/Storages/Transaction/TiKVRecordFormat.h

@@ -37,20 +37,36 @@ static const UInt64 SIGN_MARK = UInt64(1) << 63;
 static const size_t RAW_KEY_NO_HANDLE_SIZE = 1 + 8 + 2;
 static const size_t RAW_KEY_SIZE = RAW_KEY_NO_HANDLE_SIZE + 8;
 
-inline std::vector<Field> DecodeRow(const TiKVValue & value)
+inline std::tuple<std::vector<Field>, bool> DecodeRow(const TiKVValue & value, std::unordered_set<ColumnID> column_ids_to_read, std::unordered_set<ColumnID> all_column_ids)
 {
     std::vector<Field> vec;
     const String & raw_value = value.getStr();
     size_t cursor = 0;
+    bool has_unknown_col_id = false;
     while (cursor < raw_value.size())
     {
-        vec.push_back(DecodeDatum(cursor, raw_value));
+        Field f = DecodeDatum(cursor, raw_value);
+        if (f.isNull())
+            break;
+        if (!all_column_ids.count(f.get<ColumnID>()))
+        {
+            has_unknown_col_id = true;
+        }
+        if (!column_ids_to_read.count(f.get<ColumnID>()))
+        {
+            SkipDatum(cursor, raw_value);
+        }
+        else
+        {
+            vec.push_back(f);
+            vec.push_back(DecodeDatum(cursor, raw_value));
+        }
     }
 
     if (cursor != raw_value.size())
         throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
 
-    return vec;
+    return std::make_tuple(vec, has_unknown_col_id);
 }
 
 // Key format is here: