[FLASH-402] Skip unneeded columns when decode row

dbms/src/Storages/Transaction/Codec.cpp

@@ -65,6 +65,24 @@ String DecodeBytes(size_t & cursor, const String & raw_value)
     return ss.str();
 }
 
+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;
+    }
+}
+
 String DecodeCompactBytes(size_t & cursor, const String & raw_value)
 {
     size_t size = DecodeVarInt(cursor, raw_value);
@@ -73,13 +91,24 @@ String DecodeCompactBytes(size_t & cursor, const String & raw_value)
     return res;
 }
 
+void SkipCompactBytes(size_t & cursor, const String & raw_value)
+{
+    size_t size = DecodeVarInt(cursor, raw_value);
+    cursor += size;
+}
+
 Int64 DecodeVarInt(size_t & cursor, const String & raw_value)
 {
     UInt64 v = DecodeVarUInt(cursor, raw_value);
     Int64 vx = v >> 1;
     return (v & 1) ? ~vx : vx;
 }
 
+void SkipVarInt(size_t & cursor, const String & raw_value)
+{
+    SkipVarUInt(cursor, raw_value);
+}
+
 UInt64 DecodeVarUInt(size_t & cursor, const String & raw_value)
 {
     UInt64 res = 0;
@@ -99,6 +128,11 @@ UInt64 DecodeVarUInt(size_t & cursor, const String & raw_value)
     throw Exception("Wrong format. (DecodeVarUInt)", ErrorCodes::LOGICAL_ERROR);
 }
 
+void SkipVarUInt(size_t & cursor, const String & raw_value)
+{
+    std::ignore = DecodeVarUInt(cursor, raw_value);
+}
+
 inline Int8 getWords(PrecType prec, ScaleType scale)
 {
     Int8 scale_word = scale / 9 + (scale % 9 > 0);
@@ -204,6 +238,15 @@ Decimal DecodeDecimal(size_t & cursor, const String & raw_value)
     return Decimal(value, prec, frac);
 }
 
+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;
+}
+
 Field DecodeDatum(size_t & cursor, const String & raw_value)
 {
     switch (raw_value[cursor++])
@@ -233,6 +276,43 @@ Field DecodeDatum(size_t & cursor, const String & raw_value)
     }
 }
 
+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);
+    }
+}
+
 void EncodeFloat64(Float64 num, std::stringstream & ss)
 {
     UInt64 u = enforce_cast<UInt64>(num);

dbms/src/Storages/Transaction/Codec.h

@@ -41,6 +41,18 @@ Decimal DecodeDecimal(size_t & cursor, const String & raw_value);
 
 Field DecodeDatum(size_t & cursor, const String & raw_value);
 
+void SkipBytes(size_t & cursor, const String & raw_value);
+
+void SkipCompactBytes(size_t & cursor, const String & raw_value);
+
+void SkipVarInt(size_t & cursor, const String & raw_value);
+
+void SkipVarUInt(size_t & cursor, const String & raw_value);
+
+void SkipDecimal(size_t & cursor, const String & raw_value);
+
+void SkipDatum(size_t & cursor, const String & raw_value);
+
 template <typename T>
 inline std::enable_if_t<std::is_unsigned_v<T>, void> EncodeUInt(T u, std::stringstream & ss)
 {

dbms/src/Storages/Transaction/RegionBlockReader.cpp

@@ -125,16 +125,32 @@ 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;
-    for (const auto & column_info : table_info.columns)
+    std::unordered_map<ColumnID, size_t> column_id_to_info_index_map;
+    std::unordered_set<ColumnID> column_ids_to_read;
+    std::unordered_set<ColumnID> schema_all_column_ids;
+    for (size_t i = 0; i < table_info.columns.size(); i++)
     {
+        auto & column_info = table_info.columns[i];
         ColumnID col_id = column_info.id;
         String col_name = column_info.name;
+        schema_all_column_ids.insert(col_id);
+        if (std::find(column_names_to_read.begin(), column_names_to_read.end(), col_name) == column_names_to_read.end())
+        {
+            continue;
+        }
         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());
         if (table_info.pk_is_handle && column_info.hasPriKeyFlag())
             handle_col_id = col_id;
+        else
+        {
+            column_ids_to_read.emplace(col_id);
+            column_id_to_info_index_map.emplace(std::make_pair(col_id, i));
+        }
     }
+    if (column_names_to_read.size() - 3 != column_ids_to_read.size())
+        throw Exception("schema doesn't contain needed columns.", ErrorCodes::LOGICAL_ERROR);
 
     if (!table_info.pk_is_handle)
     {
@@ -168,15 +184,19 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
     const auto & date_lut = DateLUT::instance();
 
-    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;
 
     Block block;
 
     // optimize for only need handle, tso, delmark.
     if (column_names_to_read.size() > 3)
     {
+        // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
+        std::vector<ColumnID> decoded_col_ids;
+        std::vector<Field> decoded_fields;
+        decoded_col_ids.reserve(target_col_size);
+        decoded_fields.reserve(target_col_size);
+
         for (const auto & [handle, write_type, commit_ts, value_ptr] : data_list)
         {
             std::ignore = handle;
@@ -185,94 +205,82 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             if (commit_ts > start_ts)
                 continue;
 
-            // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
-
-            std::vector<Field> row;
-
+            decoded_col_ids.clear();
+            decoded_fields.clear();
             if (write_type == Region::DelFlag)
             {
-                row.reserve(table_info.columns.size() * 2);
-                for (const TiDB::ColumnInfo & column : table_info.columns)
+                for (auto col_id : column_ids_to_read)
                 {
-                    if (handle_col_id == column.id)
-                        continue;
+                    const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
 
-                    row.push_back(Field(column.id));
-                    row.push_back(GenDecodeRow(column.getCodecFlag()));
+                    decoded_col_ids.push_back(column.id);
+                    decoded_fields.emplace_back(GenDecodeRow(column.getCodecFlag()));
                 }
             }
             else
-                row = RecordKVFormat::DecodeRow(*value_ptr);
-
-            if (row.size() == 1 && row[0].isNull())
             {
-                // all field is null
-                row.clear();
+                bool schema_matches = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, decoded_col_ids, decoded_fields, schema_all_column_ids);
+                if (!schema_matches && !force_decode)
+                {
+                    return std::make_tuple(block, false);
+                }
+                if (decoded_col_ids.empty() && decoded_fields.size() == 1 && decoded_fields[0].isNull())
+                {
+                    // all field is null
+                    decoded_fields.clear();
+                }
             }
 
-            if (row.size() & 1)
+            if (decoded_col_ids.size() != decoded_fields.size())
                 throw Exception("row size is wrong.", ErrorCodes::LOGICAL_ERROR);
 
             /// Modify `row` by adding missing column values or removing useless column values.
-
-            col_id_included.clear();
-            for (size_t i = 0; i < row.size(); i += 2)
-                col_id_included.emplace(row[i].get<ColumnID>());
-
-            // Fill in missing column values.
-            for (const TiDB::ColumnInfo & column : table_info.columns)
+            if (unlikely(decoded_col_ids.size() > column_ids_to_read.size()))
             {
-                if (handle_col_id == column.id)
-                    continue;
-                if (col_id_included.count(column.id))
-                    continue;
-
-                if (!force_decode)
-                    return std::make_tuple(block, false);
-
-                row.emplace_back(Field(column.id));
-                if (column.hasNoDefaultValueFlag())
-                    // Fill `zero` value if NOT NULL specified or else NULL.
-                    row.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
-                else
-                    // Fill default value.
-                    row.push_back(column.defaultValueToField());
+                throw Exception("read unexpected columns.", ErrorCodes::LOGICAL_ERROR);
             }
 
-            // Remove values of non-existing columns, which could be data inserted (but not flushed) before DDLs that drop some columns.
-            // TODO: May need to log this.
-            for (int i = int(row.size()) - 2; i >= 0; i -= 2)
+            // redundant column values (column id not in current schema) has been dropped when decoding row
+            // this branch handles the case when the row doesn't contain all the needed column
+            if (decoded_col_ids.size() < column_ids_to_read.size())
             {
-                Field & col_id = row[i];
-                if (column_map.find(col_id.get<ColumnID>()) == column_map.end())
+                std::unordered_set<ColumnID> decoded_col_ids_set(decoded_col_ids.begin(), decoded_col_ids.end());
+
+                for (auto col_id : column_ids_to_read)
                 {
-                    if (!force_decode)
-                        return std::make_tuple(block, false);
+                    if (decoded_col_ids_set.count(col_id))
+                        continue;
 
-                    row.erase(row.begin() + i, row.begin() + i + 2);
+                    const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
+                    decoded_col_ids.push_back(column.id);
+                    if (column.hasNoDefaultValueFlag())
+                        // Fill `zero` value if NOT NULL specified or else NULL.
+                        decoded_fields.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
+                    else
+                        // Fill default value.
+                        decoded_fields.push_back(column.defaultValueToField());
                 }
             }
 
-            if (row.size() != target_col_size)
+            if (decoded_col_ids.size() != target_col_size || decoded_fields.size() != target_col_size)
                 throw Exception("decode row error.", ErrorCodes::LOGICAL_ERROR);
 
             /// Transform `row` to columnar format.
-
-            for (size_t i = 0; i < row.size(); i += 2)
+            for (size_t i = 0; i < decoded_col_ids.size(); i++)
             {
-                Field & col_id = row[i];
-                auto it = column_map.find(col_id.get<ColumnID>());
+                ColumnID col_id = decoded_col_ids[i];
+                auto it = column_map.find(col_id);
                 if (it == column_map.end())
                     throw Exception("col_id not found in column_map", ErrorCodes::LOGICAL_ERROR);
 
                 const auto & tp = it->second.second.type;
                 if (tp->isDateOrDateTime()
                     || (tp->isNullable() && dynamic_cast<const DataTypeNullable *>(tp.get())->getNestedType()->isDateOrDateTime()))
                 {
-                    Field & field = row[i + 1];
+                    Field & field = decoded_fields[i];
                     if (field.isNull())
                     {
-                        it->second.first->insert(row[i + 1]);
+                        it->second.first->insert(decoded_fields[i]);
                         continue;
                     }
                     UInt64 packed = field.get<UInt64>();
@@ -317,7 +325,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 }
                 else
                 {
-                    it->second.first->insert(row[i + 1]);
+                    it->second.first->insert(decoded_fields[i]);
 
                     // Check overflow for potential un-synced data type widen,
                     // i.e. schema is old and narrow, meanwhile data is new and wide.
@@ -335,14 +343,14 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                         {
                             // Unsigned checking by bitwise compare.
                             UInt64 inserted = nested_column.get64(inserted_index);
-                            UInt64 orig = row[i + 1].get<UInt64>();
+                            UInt64 orig = decoded_fields[i].get<UInt64>();
                             overflow = inserted != orig;
                         }
                         else
                         {
                             // Singed checking by arithmetical cast.
                             Int64 inserted = nested_column.getInt(inserted_index);
-                            Int64 orig = row[i + 1].get<Int64>();
+                            Int64 orig = decoded_fields[i].get<Int64>();
                             overflow = inserted != orig;
                         }
                         if (overflow)
@@ -352,7 +360,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                             // Otherwise return false to outer, outer should sync schema and try again.
                             if (force_decode)
                                 throw Exception(
-                                    "Detected overflow for data " + std::to_string(row[i + 1].get<UInt64>()) + " of type " + tp->getName(),
+                                    "Detected overflow for data " + std::to_string(decoded_fields[i].get<UInt64>()) + " of type " + tp->getName(),
                                     ErrorCodes::LOGICAL_ERROR);
 
                             return std::make_tuple(block, false);
@@ -364,6 +372,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
         }
     }
 
+
     for (const auto & name : column_names_to_read)
     {
         if (name == MutableSupport::delmark_column_name)