From 6abfa1d0cc52615e9854e99bf201ce287a390d7a Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Sat, 10 Aug 2019 16:46:04 +0800
Subject: [PATCH 01/37] skip unneeded column

---
 dbms/src/Storages/Transaction/Codec.h         | 90 +++++++++++++++++++
 .../Transaction/RegionBlockReader.cpp         | 12 ++-
 .../Storages/Transaction/TiKVRecordFormat.h   | 18 +++-
 3 files changed, 116 insertions(+), 4 deletions(-)

diff --git a/dbms/src/Storages/Transaction/Codec.h b/dbms/src/Storages/Transaction/Codec.h
index ff32d38e39d..c71f0949133 100644
--- a/dbms/src/Storages/Transaction/Codec.h
+++ b/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)
 {
diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index c2b98e4edbf..ffed8a347a4 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -125,10 +125,16 @@ 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;
     for (const auto & column_info : table_info.columns)
     {
         ColumnID col_id = column_info.id;
         String col_name = column_info.name;
+        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 +176,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_ids_to_read.size() - 3) * 2;
 
     Block block;
 
@@ -202,7 +208,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 }
             }
             else
-                row = RecordKVFormat::DecodeRow(*value_ptr);
+                row = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read);
 
             if (row.size() == 1 && row[0].isNull())
             {
@@ -226,6 +232,8 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                     continue;
                 if (col_id_included.count(column.id))
                     continue;
+                if (!column_ids_to_read.count(column.id))
+                    continue;
 
                 if (!force_decode)
                     return std::make_tuple(block, false);
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index f18dc82b70a..9438b7438cc 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -37,14 +37,28 @@ 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::vector<Field> DecodeRow(const TiKVValue & value, std::unordered_set<ColumnID> column_ids_to_read)
 {
     std::vector<Field> vec;
     const String & raw_value = value.getStr();
     size_t cursor = 0;
+    size_t field_index = 0;
     while (cursor < raw_value.size())
     {
-        vec.push_back(DecodeDatum(cursor, raw_value));
+        if (field_index % 2 == 0)
+        {
+            Field f = DecodeDatum(cursor, raw_value);
+            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));
+            }
+        }
+        field_index++;
     }
 
     if (cursor != raw_value.size())

From b83a4b5ba10062e64dacf394e0395c6babc9949a Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Sun, 11 Aug 2019 23:16:14 +0800
Subject: [PATCH 02/37] small fix

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index ffed8a347a4..de59a48a313 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -176,7 +176,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
     std::unordered_set<ColumnID> col_id_included;
 
-    const size_t target_col_size = (column_ids_to_read.size() - 3) * 2;
+    const size_t target_col_size = (column_names_to_read.size() - 3) * 2;
 
     Block block;
 

From 445fc4d98d57b1ab182bcd3465bab48ff93c99c2 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Mon, 12 Aug 2019 23:35:30 +0800
Subject: [PATCH 03/37] check unknown column id when decode row

---
 .../Transaction/RegionBlockReader.cpp         | 22 ++++++++++++++-----
 .../Storages/Transaction/TiKVRecordFormat.h   |  9 ++++++--
 2 files changed, 23 insertions(+), 8 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index de59a48a313..f7bf0fc317c 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -126,10 +126,12 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
     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;
@@ -194,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;
 
             if (write_type == Region::DelFlag)
             {
@@ -203,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, column_ids_to_read);
+                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())
             {
@@ -219,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();
@@ -232,12 +244,13 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                     continue;
                 if (col_id_included.count(column.id))
                     continue;
-                if (!column_ids_to_read.count(column.id))
-                    continue;
 
                 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.
@@ -254,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);
                 }
             }
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 9438b7438cc..b2e78eced12 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -37,17 +37,22 @@ 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, std::unordered_set<ColumnID> column_ids_to_read)
+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;
     size_t field_index = 0;
+    bool has_unknown_col_id = false;
     while (cursor < raw_value.size())
     {
         if (field_index % 2 == 0)
         {
             Field f = DecodeDatum(cursor, raw_value);
+            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);
@@ -64,7 +69,7 @@ inline std::vector<Field> DecodeRow(const TiKVValue & value, std::unordered_set<
     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:

From fce6f1580f3cb5ea5b67ed0d606790ad7bf32f5c Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 13 Aug 2019 09:10:23 +0800
Subject: [PATCH 04/37] add initializer for boolean variable

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index f7bf0fc317c..b761c56e564 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -196,7 +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;
+            bool has_unknown_col_id = false;
 
             if (write_type == Region::DelFlag)
             {

From 0231b951c6178fd3f3b55ce74a70237f5b834ff7 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 13 Aug 2019 13:25:57 +0800
Subject: [PATCH 05/37] fix the case when tikv value contains just a nill flag

---
 .../Storages/Transaction/TiKVRecordFormat.h   | 31 +++++++++----------
 1 file changed, 14 insertions(+), 17 deletions(-)

diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index b2e78eced12..4622bf44cb4 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -42,28 +42,25 @@ inline std::tuple<std::vector<Field>, bool> DecodeRow(const TiKVValue & value, s
     std::vector<Field> vec;
     const String & raw_value = value.getStr();
     size_t cursor = 0;
-    size_t field_index = 0;
     bool has_unknown_col_id = false;
     while (cursor < raw_value.size())
     {
-        if (field_index % 2 == 0)
+        Field f = DecodeDatum(cursor, raw_value);
+        if (f.isNull())
+            break;
+        if (!all_column_ids.count(f.get<ColumnID>()))
         {
-            Field f = DecodeDatum(cursor, raw_value);
-            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));
-            }
+            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));
         }
-        field_index++;
     }
 
     if (cursor != raw_value.size())

From 03661425c85f0b2262558accce2da2e7ef53721a Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 13 Aug 2019 19:37:25 +0800
Subject: [PATCH 06/37] remove unnecessary copy

---
 dbms/src/Storages/Transaction/TiKVRecordFormat.h | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 4622bf44cb4..86a6975c4d9 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -37,7 +37,7 @@ 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::tuple<std::vector<Field>, bool> DecodeRow(const TiKVValue & value, std::unordered_set<ColumnID> column_ids_to_read, std::unordered_set<ColumnID> all_column_ids)
+inline std::tuple<std::vector<Field>, bool> DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read, const std::unordered_set<ColumnID> & all_column_ids)
 {
     std::vector<Field> vec;
     const String & raw_value = value.getStr();
@@ -58,7 +58,7 @@ inline std::tuple<std::vector<Field>, bool> DecodeRow(const TiKVValue & value, s
         }
         else
         {
-            vec.push_back(f);
+            vec.push_back(std::move(f));
             vec.push_back(DecodeDatum(cursor, raw_value));
         }
     }

From f5013e92199c27c7034cc6e78cf79e7119ec952a Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 13 Aug 2019 21:58:35 +0800
Subject: [PATCH 07/37] small improvement

---
 dbms/src/Storages/Transaction/TiKVRecordFormat.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 86a6975c4d9..0db7ee2be8e 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -48,7 +48,7 @@ inline std::tuple<std::vector<Field>, bool> DecodeRow(const TiKVValue & value, c
         Field f = DecodeDatum(cursor, raw_value);
         if (f.isNull())
             break;
-        if (!all_column_ids.count(f.get<ColumnID>()))
+        if (!has_unknown_col_id && !all_column_ids.count(f.get<ColumnID>()))
         {
             has_unknown_col_id = true;
         }

From b0d3e9c27d92e326cac82a91cc42700e7d9641b4 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Wed, 14 Aug 2019 12:32:16 +0800
Subject: [PATCH 08/37] fix comment

---
 dbms/src/Storages/Transaction/TiKVRecordFormat.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 0db7ee2be8e..70f1602be28 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -66,7 +66,7 @@ inline std::tuple<std::vector<Field>, bool> DecodeRow(const TiKVValue & value, c
     if (cursor != raw_value.size())
         throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
 
-    return std::make_tuple(vec, has_unknown_col_id);
+    return std::make_tuple(std::move(vec), has_unknown_col_id);
 }
 
 // Key format is here:

From 93a10a7b2259a805748fdd73814ef74018970680 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Wed, 14 Aug 2019 20:15:29 +0800
Subject: [PATCH 09/37] don't store column id in field

---
 .../Transaction/RegionBlockReader.cpp         | 80 +++++++++++--------
 .../Storages/Transaction/TiKVRecordFormat.h   | 19 +++--
 2 files changed, 59 insertions(+), 40 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index b761c56e564..6e413baec5a 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -176,9 +176,7 @@ 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 = (column_names_to_read.size() - 3) * 2;
+    const size_t target_col_size = column_names_to_read.size() - 3;
 
     Block block;
 
@@ -195,12 +193,16 @@ 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;
+            std::vector<ColumnID> col_ids;
+            std::vector<Field> fields;
+            std::set<ColumnID> row_all_col_ids;
             bool has_unknown_col_id = false;
+            bool all_column_is_null = false;
 
             if (write_type == Region::DelFlag)
             {
-                row.reserve(table_info.columns.size() * 2);
+                col_ids.reserve(target_col_size);
+                fields.reserve(target_col_size);
                 for (const TiDB::ColumnInfo & column : table_info.columns)
                 {
                     if (handle_col_id == column.id)
@@ -209,20 +211,21 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                     if (!column_ids_to_read.count(column.id))
                         continue;
 
-                    row.push_back(Field(column.id));
-                    row.push_back(GenDecodeRow(column.getCodecFlag()));
+                    col_ids.push_back(column.id);
+                    fields.push_back(GenDecodeRow(column.getCodecFlag()));
                 }
             }
             else
-                std::tie(row, has_unknown_col_id) = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, all_column_ids);
+                std::tie(col_ids, fields, row_all_col_ids, has_unknown_col_id) = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, all_column_ids);
 
-            if (row.size() == 1 && row[0].isNull())
+            if (col_ids.empty() && fields.size() == 1 && fields[0].isNull())
             {
                 // all field is null
-                row.clear();
+                fields.clear();
+                all_column_is_null = true;
             }
 
-            if (row.size() & 1)
+            if (col_ids.size() != fields.size())
                 throw Exception("row size is wrong.", ErrorCodes::LOGICAL_ERROR);
 
             /// row contains unknown column id
@@ -232,10 +235,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 return std::make_tuple(block, false);
 
             /// 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>());
+            std::unordered_set<ColumnID> col_id_included(col_ids.begin(), col_ids.end());
 
             // Fill in missing column values.
             for (const TiDB::ColumnInfo & column : table_info.columns)
@@ -245,41 +245,53 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 if (col_id_included.count(column.id))
                     continue;
 
-                if (!force_decode)
+                if (!row_all_col_ids.count(column.id) && !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())
+                col_ids.push_back(column.id);
+                if (all_column_is_null)
+                {
+                    if (column.hasNotNullFlag())
+                    {
+                        throw Exception("not null column contains null value", ErrorCodes::LOGICAL_ERROR);
+                    }
+                    else
+                    {
+                        fields.push_back(Field());
+                    }
+                }
+                else if (column.hasNoDefaultValueFlag())
                     // Fill `zero` value if NOT NULL specified or else NULL.
-                    row.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
+                    fields.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
                 else
                     // Fill default value.
-                    row.push_back(column.defaultValueToField());
+                    fields.push_back(column.defaultValueToField());
             }
 
             // 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)
+            for (int i = int(col_ids.size()) - 1; i >= 0; i--)
             {
-                Field & col_id = row[i];
-                if (column_map.find(col_id.get<ColumnID>()) == column_map.end())
+                ColumnID col_id = col_ids[i];
+                if (column_map.find(col_id) == column_map.end())
                 {
-                    row.erase(row.begin() + i, row.begin() + i + 2);
+                    col_ids.erase(col_ids.begin() + i, col_ids.begin() + i + 1);
+                    fields.erase(fields.begin() + i, fields.begin() + i + 1);
                 }
             }
 
-            if (row.size() != target_col_size)
+            if (col_ids.size() != target_col_size || 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 < col_ids.size(); i++)
             {
-                Field & col_id = row[i];
-                auto it = column_map.find(col_id.get<ColumnID>());
+                ColumnID col_id = 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);
 
@@ -287,10 +299,10 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 if (tp->isDateOrDateTime()
                     || (tp->isNullable() && dynamic_cast<const DataTypeNullable *>(tp.get())->getNestedType()->isDateOrDateTime()))
                 {
-                    Field & field = row[i + 1];
+                    Field & field = fields[i];
                     if (field.isNull())
                     {
-                        it->second.first->insert(row[i + 1]);
+                        it->second.first->insert(fields[i]);
                         continue;
                     }
                     UInt64 packed = field.get<UInt64>();
@@ -335,7 +347,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 }
                 else
                 {
-                    it->second.first->insert(row[i + 1]);
+                    it->second.first->insert(fields[i]);
 
                     // Check overflow for potential un-synced data type widen,
                     // i.e. schema is old and narrow, meanwhile data is new and wide.
@@ -353,14 +365,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 = 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 = fields[i].get<Int64>();
                             overflow = inserted != orig;
                         }
                         if (overflow)
@@ -370,7 +382,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(fields[i].get<UInt64>()) + " of type " + tp->getName(),
                                     ErrorCodes::LOGICAL_ERROR);
 
                             return std::make_tuple(block, false);
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 70f1602be28..3bee49d67cb 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -37,9 +37,11 @@ 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::tuple<std::vector<Field>, bool> DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read, const std::unordered_set<ColumnID> & all_column_ids)
+inline std::tuple<std::vector<ColumnID>, std::vector<Field>, std::set<ColumnID>, bool> DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read, const std::unordered_set<ColumnID> & all_column_ids)
 {
-    std::vector<Field> vec;
+    std::vector<ColumnID> col_ids;
+    std::vector<Field> fields;
+    std::set<ColumnID> row_all_col_ids;
     const String & raw_value = value.getStr();
     size_t cursor = 0;
     bool has_unknown_col_id = false;
@@ -47,26 +49,31 @@ inline std::tuple<std::vector<Field>, bool> DecodeRow(const TiKVValue & value, c
     {
         Field f = DecodeDatum(cursor, raw_value);
         if (f.isNull())
+        {
+            fields.push_back(std::move(f));
             break;
+        }
         if (!has_unknown_col_id && !all_column_ids.count(f.get<ColumnID>()))
         {
             has_unknown_col_id = true;
         }
-        if (!column_ids_to_read.count(f.get<ColumnID>()))
+        ColumnID col_id = f.get<ColumnID>();
+        row_all_col_ids.insert(col_id);
+        if (!column_ids_to_read.count(col_id))
         {
             SkipDatum(cursor, raw_value);
         }
         else
         {
-            vec.push_back(std::move(f));
-            vec.push_back(DecodeDatum(cursor, raw_value));
+            col_ids.push_back(col_id);
+            fields.push_back(DecodeDatum(cursor, raw_value));
         }
     }
 
     if (cursor != raw_value.size())
         throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
 
-    return std::make_tuple(std::move(vec), has_unknown_col_id);
+    return std::make_tuple(std::move(col_ids), std::move(fields), std::move(row_all_col_ids), has_unknown_col_id);
 }
 
 // Key format is here:

From 8a5bb7b0f462d7f6bbf874029690ebf6903644ab Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Thu, 15 Aug 2019 13:13:24 +0800
Subject: [PATCH 10/37] small improvement

---
 .../Transaction/RegionBlockReader.cpp         | 39 ++++++-------------
 .../Storages/Transaction/TiKVRecordFormat.h   | 13 ++-----
 2 files changed, 15 insertions(+), 37 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 6e413baec5a..52efc72a208 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -126,12 +126,12 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
     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;
+    std::unordered_set<ColumnID> schema_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);
+        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;
@@ -195,9 +195,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
             std::vector<ColumnID> col_ids;
             std::vector<Field> fields;
-            std::set<ColumnID> row_all_col_ids;
-            bool has_unknown_col_id = false;
-            bool all_column_is_null = false;
+            std::unordered_set<ColumnID> row_all_column_ids;
 
             if (write_type == Region::DelFlag)
             {
@@ -216,24 +214,23 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 }
             }
             else
-                std::tie(col_ids, fields, row_all_col_ids, has_unknown_col_id) = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, all_column_ids);
+            {
+                std::tie(col_ids, fields, row_all_column_ids) = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read);
+                if ((schema_all_column_ids != row_all_column_ids) && !force_decode)
+                {
+                    return std::make_tuple(block, false);
+                }
+            }
 
             if (col_ids.empty() && fields.size() == 1 && fields[0].isNull())
             {
                 // all field is null
                 fields.clear();
-                all_column_is_null = true;
             }
 
             if (col_ids.size() != fields.size())
                 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.
             std::unordered_set<ColumnID> col_id_included(col_ids.begin(), col_ids.end());
 
@@ -245,25 +242,11 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 if (col_id_included.count(column.id))
                     continue;
 
-                if (!row_all_col_ids.count(column.id) && !force_decode)
-                    return std::make_tuple(block, false);
-
                 if (!column_ids_to_read.count(column.id))
                     continue;
 
                 col_ids.push_back(column.id);
-                if (all_column_is_null)
-                {
-                    if (column.hasNotNullFlag())
-                    {
-                        throw Exception("not null column contains null value", ErrorCodes::LOGICAL_ERROR);
-                    }
-                    else
-                    {
-                        fields.push_back(Field());
-                    }
-                }
-                else if (column.hasNoDefaultValueFlag())
+                if (column.hasNoDefaultValueFlag())
                     // Fill `zero` value if NOT NULL specified or else NULL.
                     fields.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
                 else
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 3bee49d67cb..ed213dcf647 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -37,14 +37,13 @@ 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::tuple<std::vector<ColumnID>, std::vector<Field>, std::set<ColumnID>, bool> DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read, const std::unordered_set<ColumnID> & all_column_ids)
+inline std::tuple<std::vector<ColumnID>, std::vector<Field>, std::unordered_set<ColumnID>> DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read)
 {
     std::vector<ColumnID> col_ids;
     std::vector<Field> fields;
-    std::set<ColumnID> row_all_col_ids;
+    std::unordered_set<ColumnID> row_all_column_ids;
     const String & raw_value = value.getStr();
     size_t cursor = 0;
-    bool has_unknown_col_id = false;
     while (cursor < raw_value.size())
     {
         Field f = DecodeDatum(cursor, raw_value);
@@ -53,12 +52,8 @@ inline std::tuple<std::vector<ColumnID>, std::vector<Field>, std::set<ColumnID>,
             fields.push_back(std::move(f));
             break;
         }
-        if (!has_unknown_col_id && !all_column_ids.count(f.get<ColumnID>()))
-        {
-            has_unknown_col_id = true;
-        }
         ColumnID col_id = f.get<ColumnID>();
-        row_all_col_ids.insert(col_id);
+        row_all_column_ids.insert(col_id);
         if (!column_ids_to_read.count(col_id))
         {
             SkipDatum(cursor, raw_value);
@@ -73,7 +68,7 @@ inline std::tuple<std::vector<ColumnID>, std::vector<Field>, std::set<ColumnID>,
     if (cursor != raw_value.size())
         throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
 
-    return std::make_tuple(std::move(col_ids), std::move(fields), std::move(row_all_col_ids), has_unknown_col_id);
+    return std::make_tuple(std::move(col_ids), std::move(fields), std::move(row_all_column_ids));
 }
 
 // Key format is here:

From dcc42655f390666d0724f94704b443761ae20e61 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Thu, 15 Aug 2019 13:58:58 +0800
Subject: [PATCH 11/37] remove unnecessary blank line

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 52efc72a208..7713f93fd1d 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -220,12 +220,11 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 {
                     return std::make_tuple(block, false);
                 }
-            }
-
-            if (col_ids.empty() && fields.size() == 1 && fields[0].isNull())
-            {
-                // all field is null
-                fields.clear();
+                if (col_ids.empty() && fields.size() == 1 && fields[0].isNull())
+                {
+                    // all field is null
+                    fields.clear();
+                }
             }
 
             if (col_ids.size() != fields.size())
@@ -270,7 +269,6 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 throw Exception("decode row error.", ErrorCodes::LOGICAL_ERROR);
 
             /// Transform `row` to columnar format.
-
             for (size_t i = 0; i < col_ids.size(); i++)
             {
                 ColumnID col_id = col_ids[i];

From a92d8782ff63522839d42ce4d899a905dd54ba45 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Thu, 15 Aug 2019 15:17:06 +0800
Subject: [PATCH 12/37] add log for performance debug

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 13 +++++++++++++
 dbms/src/Storages/Transaction/RegionTable.cpp       |  6 +++---
 2 files changed, 16 insertions(+), 3 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 7713f93fd1d..f71eb9204da 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -183,8 +183,12 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     // optimize for only need handle, tso, delmark.
     if (column_names_to_read.size() > 3)
     {
+        unsigned long sum_decode = 0;
+        unsigned long sum_transform = 0;
+
         for (const auto & [handle, write_type, commit_ts, value_ptr] : data_list)
         {
+            auto start_time = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
             std::ignore = handle;
 
             // Ignore data after the start_ts.
@@ -268,6 +272,8 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             if (col_ids.size() != target_col_size || fields.size() != target_col_size)
                 throw Exception("decode row error.", ErrorCodes::LOGICAL_ERROR);
 
+            auto mid_time = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
+
             /// Transform `row` to columnar format.
             for (size_t i = 0; i < col_ids.size(); i++)
             {
@@ -372,8 +378,15 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                     // TODO: Consider other kind of type change? I.e. arbitrary type change.
                 }
             }
+            auto end_time = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
+
+            sum_decode += (mid_time - start_time);
+            sum_transform += (end_time - mid_time);
         }
+        std::cout << "decode time: " << sum_decode << "ms" << std::endl;
+        std::cout << "transform time: " << sum_transform << "ms" << std::endl;
     }
+    
 
     for (const auto & name : column_names_to_read)
     {
diff --git a/dbms/src/Storages/Transaction/RegionTable.cpp b/dbms/src/Storages/Transaction/RegionTable.cpp
index dd499677124..13093ca114a 100644
--- a/dbms/src/Storages/Transaction/RegionTable.cpp
+++ b/dbms/src/Storages/Transaction/RegionTable.cpp
@@ -417,7 +417,7 @@ void RegionTable::tryFlushRegion(RegionID region_id)
     if (!status)
         return;
 
-    flushRegion(table_id, region_id, cache_bytes, false);
+    //flushRegion(table_id, region_id, cache_bytes, false);
 
     func_update_region([&](InternalRegion & region) -> bool {
         region.pause_flush = false;
@@ -443,8 +443,8 @@ bool RegionTable::tryFlushRegions()
         });
     }
 
-    for (auto & [id, cache_bytes] : to_flush)
-        flushRegion(id.first, id.second, cache_bytes);
+//    for (auto & [id, cache_bytes] : to_flush)
+//        flushRegion(id.first, id.second, cache_bytes);
 
     { // Now reset status information.
         Timepoint now = Clock::now();

From 1881a5ccb3255e5ded62a157c9e0e2664f0ff7a8 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Thu, 15 Aug 2019 15:54:18 +0800
Subject: [PATCH 13/37] adjust time point

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index f71eb9204da..dc76e972eeb 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -231,6 +231,8 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 }
             }
 
+            auto mid_time = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
+
             if (col_ids.size() != fields.size())
                 throw Exception("row size is wrong.", ErrorCodes::LOGICAL_ERROR);
 
@@ -272,7 +274,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             if (col_ids.size() != target_col_size || fields.size() != target_col_size)
                 throw Exception("decode row error.", ErrorCodes::LOGICAL_ERROR);
 
-            auto mid_time = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
+
 
             /// Transform `row` to columnar format.
             for (size_t i = 0; i < col_ids.size(); i++)
@@ -386,7 +388,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
         std::cout << "decode time: " << sum_decode << "ms" << std::endl;
         std::cout << "transform time: " << sum_transform << "ms" << std::endl;
     }
-    
+
 
     for (const auto & name : column_names_to_read)
     {

From eae234bb9714a392b42b5f1e436247fdfbf18dc3 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Thu, 15 Aug 2019 16:23:15 +0800
Subject: [PATCH 14/37] small fix and adjust log

---
 .../src/Storages/Transaction/RegionBlockReader.cpp | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index dc76e972eeb..2643c4aa33b 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -184,7 +184,10 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     if (column_names_to_read.size() > 3)
     {
         unsigned long sum_decode = 0;
+        unsigned long sum_mid = 0;
         unsigned long sum_transform = 0;
+        std::unordered_set<ColumnID> col_id_included;
+        std::unordered_set<ColumnID> row_all_column_ids;
 
         for (const auto & [handle, write_type, commit_ts, value_ptr] : data_list)
         {
@@ -199,7 +202,6 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
             std::vector<ColumnID> col_ids;
             std::vector<Field> fields;
-            std::unordered_set<ColumnID> row_all_column_ids;
 
             if (write_type == Region::DelFlag)
             {
@@ -237,7 +239,9 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 throw Exception("row size is wrong.", ErrorCodes::LOGICAL_ERROR);
 
             /// Modify `row` by adding missing column values or removing useless column values.
-            std::unordered_set<ColumnID> col_id_included(col_ids.begin(), col_ids.end());
+            col_id_included.clear();
+            for (size_t i = 0; i < col_ids.size(); i++)
+                col_id_included.emplace(col_ids[i]);
 
             // Fill in missing column values.
             for (const TiDB::ColumnInfo & column : table_info.columns)
@@ -274,7 +278,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             if (col_ids.size() != target_col_size || fields.size() != target_col_size)
                 throw Exception("decode row error.", ErrorCodes::LOGICAL_ERROR);
 
-
+            auto mid_time2 = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
 
             /// Transform `row` to columnar format.
             for (size_t i = 0; i < col_ids.size(); i++)
@@ -383,9 +387,11 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             auto end_time = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
 
             sum_decode += (mid_time - start_time);
-            sum_transform += (end_time - mid_time);
+            sum_mid += (mid_time2 - mid_time);
+            sum_transform += (end_time - mid_time2);
         }
         std::cout << "decode time: " << sum_decode << "ms" << std::endl;
+        std::cout << "mid time: " << sum_mid << "ms" << std::endl;
         std::cout << "transform time: " << sum_transform << "ms" << std::endl;
     }
 

From 43472ba022982fe023844d21423c6f3b0ebb1a3b Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Thu, 15 Aug 2019 20:38:15 +0800
Subject: [PATCH 15/37] small improvement

---
 .../Transaction/RegionBlockReader.cpp         | 28 -------------------
 1 file changed, 28 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 2643c4aa33b..95183e9bafd 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -183,15 +183,11 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     // optimize for only need handle, tso, delmark.
     if (column_names_to_read.size() > 3)
     {
-        unsigned long sum_decode = 0;
-        unsigned long sum_mid = 0;
-        unsigned long sum_transform = 0;
         std::unordered_set<ColumnID> col_id_included;
         std::unordered_set<ColumnID> row_all_column_ids;
 
         for (const auto & [handle, write_type, commit_ts, value_ptr] : data_list)
         {
-            auto start_time = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
             std::ignore = handle;
 
             // Ignore data after the start_ts.
@@ -233,8 +229,6 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 }
             }
 
-            auto mid_time = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
-
             if (col_ids.size() != fields.size())
                 throw Exception("row size is wrong.", ErrorCodes::LOGICAL_ERROR);
 
@@ -263,23 +257,9 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                     fields.push_back(column.defaultValueToField());
             }
 
-            // 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(col_ids.size()) - 1; i >= 0; i--)
-            {
-                ColumnID col_id = col_ids[i];
-                if (column_map.find(col_id) == column_map.end())
-                {
-                    col_ids.erase(col_ids.begin() + i, col_ids.begin() + i + 1);
-                    fields.erase(fields.begin() + i, fields.begin() + i + 1);
-                }
-            }
-
             if (col_ids.size() != target_col_size || fields.size() != target_col_size)
                 throw Exception("decode row error.", ErrorCodes::LOGICAL_ERROR);
 
-            auto mid_time2 = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
-
             /// Transform `row` to columnar format.
             for (size_t i = 0; i < col_ids.size(); i++)
             {
@@ -384,15 +364,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                     // TODO: Consider other kind of type change? I.e. arbitrary type change.
                 }
             }
-            auto end_time = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now().time_since_epoch()).count();
-
-            sum_decode += (mid_time - start_time);
-            sum_mid += (mid_time2 - mid_time);
-            sum_transform += (end_time - mid_time2);
         }
-        std::cout << "decode time: " << sum_decode << "ms" << std::endl;
-        std::cout << "mid time: " << sum_mid << "ms" << std::endl;
-        std::cout << "transform time: " << sum_transform << "ms" << std::endl;
     }
 
 

From 1ddb5d2ba7cad0c8108310bac6a980b521778c28 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Fri, 16 Aug 2019 09:38:23 +0800
Subject: [PATCH 16/37] small improvement

---
 .../Transaction/RegionBlockReader.cpp         | 30 +++++++++----------
 1 file changed, 15 insertions(+), 15 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 95183e9bafd..9ddf350231a 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -125,10 +125,12 @@ 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_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 (const auto & column_info : table_info.columns)
+    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);
@@ -136,13 +138,19 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
         {
             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());
         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)
     {
@@ -203,13 +211,9 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             {
                 col_ids.reserve(target_col_size);
                 fields.reserve(target_col_size);
-                for (const TiDB::ColumnInfo & column : table_info.columns)
+                for (auto col_id : column_ids_to_read)
                 {
-                    if (handle_col_id == column.id)
-                        continue;
-
-                    if (!column_ids_to_read.count(column.id))
-                        continue;
+                    const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
 
                     col_ids.push_back(column.id);
                     fields.push_back(GenDecodeRow(column.getCodecFlag()));
@@ -238,16 +242,12 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 col_id_included.emplace(col_ids[i]);
 
             // Fill in missing column values.
-            for (const TiDB::ColumnInfo & column : table_info.columns)
+            for (auto col_id : column_ids_to_read)
             {
-                if (handle_col_id == column.id)
-                    continue;
-                if (col_id_included.count(column.id))
-                    continue;
-
-                if (!column_ids_to_read.count(column.id))
+                if (col_id_included.count(col_id))
                     continue;
 
+                const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
                 col_ids.push_back(column.id);
                 if (column.hasNoDefaultValueFlag())
                     // Fill `zero` value if NOT NULL specified or else NULL.

From 19522723133fdf8a9e24b6fe813253999a400f6b Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Fri, 16 Aug 2019 11:40:21 +0800
Subject: [PATCH 17/37] remove unnecessary construction

---
 .../Storages/Transaction/RegionBlockReader.cpp   | 16 ++++++++++------
 dbms/src/Storages/Transaction/TiKVRecordFormat.h |  8 ++------
 2 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 9ddf350231a..0d6d6130cf2 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -194,6 +194,12 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
         std::unordered_set<ColumnID> col_id_included;
         std::unordered_set<ColumnID> row_all_column_ids;
 
+        // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
+        std::vector<ColumnID> col_ids;
+        std::vector<Field> fields;
+        col_ids.reserve(target_col_size);
+        fields.reserve(target_col_size);
+
         for (const auto & [handle, write_type, commit_ts, value_ptr] : data_list)
         {
             std::ignore = handle;
@@ -202,11 +208,8 @@ 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<ColumnID> col_ids;
-            std::vector<Field> fields;
-
+            col_ids.clear();
+            fields.clear();
             if (write_type == Region::DelFlag)
             {
                 col_ids.reserve(target_col_size);
@@ -221,7 +224,8 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             }
             else
             {
-                std::tie(col_ids, fields, row_all_column_ids) = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read);
+                row_all_column_ids.clear();
+                RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, row_all_column_ids);
                 if ((schema_all_column_ids != row_all_column_ids) && !force_decode)
                 {
                     return std::make_tuple(block, false);
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index ed213dcf647..0135ff58dad 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -37,11 +37,9 @@ 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::tuple<std::vector<ColumnID>, std::vector<Field>, std::unordered_set<ColumnID>> DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read)
+inline void DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read,
+        std::vector<ColumnID> & col_ids, std::vector<Field> & fields,  std::unordered_set<ColumnID> & row_all_column_ids)
 {
-    std::vector<ColumnID> col_ids;
-    std::vector<Field> fields;
-    std::unordered_set<ColumnID> row_all_column_ids;
     const String & raw_value = value.getStr();
     size_t cursor = 0;
     while (cursor < raw_value.size())
@@ -67,8 +65,6 @@ inline std::tuple<std::vector<ColumnID>, std::vector<Field>, std::unordered_set<
 
     if (cursor != raw_value.size())
         throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
-
-    return std::make_tuple(std::move(col_ids), std::move(fields), std::move(row_all_column_ids));
 }
 
 // Key format is here:

From d051cbb22a848bf9259ad5e160e713885127929a Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Fri, 16 Aug 2019 12:33:01 +0800
Subject: [PATCH 18/37] small improvement

---
 .../Transaction/RegionBlockReader.cpp         | 37 ++++++++++---------
 1 file changed, 20 insertions(+), 17 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 0d6d6130cf2..b087245dda4 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -241,24 +241,27 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 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 < col_ids.size(); i++)
-                col_id_included.emplace(col_ids[i]);
-
-            // Fill in missing column values.
-            for (auto col_id : column_ids_to_read)
+            if (col_ids.size() != column_ids_to_read.size())
             {
-                if (col_id_included.count(col_id))
-                    continue;
-
-                const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
-                col_ids.push_back(column.id);
-                if (column.hasNoDefaultValueFlag())
-                    // Fill `zero` value if NOT NULL specified or else NULL.
-                    fields.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
-                else
-                    // Fill default value.
-                    fields.push_back(column.defaultValueToField());
+                col_id_included.clear();
+                for (size_t i = 0; i < col_ids.size(); i++)
+                    col_id_included.emplace(col_ids[i]);
+
+                // Fill in missing column values.
+                for (auto col_id : column_ids_to_read)
+                {
+                    if (col_id_included.count(col_id))
+                        continue;
+
+                    const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
+                    col_ids.push_back(column.id);
+                    if (column.hasNoDefaultValueFlag())
+                        // Fill `zero` value if NOT NULL specified or else NULL.
+                        fields.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
+                    else
+                        // Fill default value.
+                        fields.push_back(column.defaultValueToField());
+                }
             }
 
             if (col_ids.size() != target_col_size || fields.size() != target_col_size)

From 9191506f1e44b38ccf35531a36c3ca0d92732b47 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Fri, 16 Aug 2019 14:15:19 +0800
Subject: [PATCH 19/37] uncomment flushregion

---
 dbms/src/Storages/Transaction/RegionTable.cpp | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionTable.cpp b/dbms/src/Storages/Transaction/RegionTable.cpp
index 13093ca114a..dd499677124 100644
--- a/dbms/src/Storages/Transaction/RegionTable.cpp
+++ b/dbms/src/Storages/Transaction/RegionTable.cpp
@@ -417,7 +417,7 @@ void RegionTable::tryFlushRegion(RegionID region_id)
     if (!status)
         return;
 
-    //flushRegion(table_id, region_id, cache_bytes, false);
+    flushRegion(table_id, region_id, cache_bytes, false);
 
     func_update_region([&](InternalRegion & region) -> bool {
         region.pause_flush = false;
@@ -443,8 +443,8 @@ bool RegionTable::tryFlushRegions()
         });
     }
 
-//    for (auto & [id, cache_bytes] : to_flush)
-//        flushRegion(id.first, id.second, cache_bytes);
+    for (auto & [id, cache_bytes] : to_flush)
+        flushRegion(id.first, id.second, cache_bytes);
 
     { // Now reset status information.
         Timepoint now = Clock::now();

From a64e2c5f0f2c7b67a3bf64e72d887ceff2085302 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Mon, 19 Aug 2019 21:13:07 +0800
Subject: [PATCH 20/37] skip uint by calling DecodeVarUInt

---
 dbms/src/Storages/Transaction/Codec.cpp | 12 +-----------
 1 file changed, 1 insertion(+), 11 deletions(-)

diff --git a/dbms/src/Storages/Transaction/Codec.cpp b/dbms/src/Storages/Transaction/Codec.cpp
index e8e8b21b0dd..12251ec9a87 100644
--- a/dbms/src/Storages/Transaction/Codec.cpp
+++ b/dbms/src/Storages/Transaction/Codec.cpp
@@ -133,17 +133,7 @@ UInt64 DecodeVarUInt(size_t & cursor, const String & raw_value)
 
 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);
+    std::ignore = DecodeVarUInt(cursor, raw_value);
 }
 
 inline Int8 getWords(PrecType prec, ScaleType scale)

From 171b2f9527d6e5843dd550aa07c425ee36c12b4e Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 11:27:37 +0800
Subject: [PATCH 21/37] avoid insert column id

---
 .../Storages/Transaction/RegionBlockReader.cpp  |  7 +++----
 .../src/Storages/Transaction/TiKVRecordFormat.h | 17 ++++++++++++++---
 2 files changed, 17 insertions(+), 7 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index b087245dda4..232e84b6dd3 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -192,13 +192,13 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     if (column_names_to_read.size() > 3)
     {
         std::unordered_set<ColumnID> col_id_included;
-        std::unordered_set<ColumnID> row_all_column_ids;
 
         // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
         std::vector<ColumnID> col_ids;
         std::vector<Field> fields;
         col_ids.reserve(target_col_size);
         fields.reserve(target_col_size);
+        bool schema_not_match = false;
 
         for (const auto & [handle, write_type, commit_ts, value_ptr] : data_list)
         {
@@ -224,9 +224,8 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             }
             else
             {
-                row_all_column_ids.clear();
-                RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, row_all_column_ids);
-                if ((schema_all_column_ids != row_all_column_ids) && !force_decode)
+                schema_not_match = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
+                if (schema_not_match && !force_decode)
                 {
                     return std::make_tuple(block, false);
                 }
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 0135ff58dad..7af90e2bc10 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -37,11 +37,13 @@ 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 void DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read,
-        std::vector<ColumnID> & col_ids, std::vector<Field> & fields,  std::unordered_set<ColumnID> & row_all_column_ids)
+inline bool DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read,
+        std::vector<ColumnID> & col_ids, std::vector<Field> & fields, std::unordered_set<ColumnID> schema_all_column_ids)
 {
     const String & raw_value = value.getStr();
     size_t cursor = 0;
+    bool schema_not_match = false;
+    size_t column_cnt = 0;
     while (cursor < raw_value.size())
     {
         Field f = DecodeDatum(cursor, raw_value);
@@ -51,7 +53,11 @@ inline void DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID
             break;
         }
         ColumnID col_id = f.get<ColumnID>();
-        row_all_column_ids.insert(col_id);
+        column_cnt++;
+        if (!schema_all_column_ids.count(col_id))
+        {
+            schema_not_match = true;
+        }
         if (!column_ids_to_read.count(col_id))
         {
             SkipDatum(cursor, raw_value);
@@ -62,9 +68,14 @@ inline void DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID
             fields.push_back(DecodeDatum(cursor, raw_value));
         }
     }
+    if (column_cnt != schema_all_column_ids.size())
+    {
+        schema_not_match = true;
+    }
 
     if (cursor != raw_value.size())
         throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
+    return schema_not_match;
 }
 
 // Key format is here:

From a8a25ca255e59b40d70aefbee85d121fc335a35d Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 12:08:48 +0800
Subject: [PATCH 22/37] small fix

---
 dbms/src/Storages/Transaction/TiKVRecordFormat.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 42db157ab98..ef3c056b14b 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -36,7 +36,7 @@ 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 bool DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read,
-        std::vector<ColumnID> & col_ids, std::vector<Field> & fields, std::unordered_set<ColumnID> schema_all_column_ids)
+        std::vector<ColumnID> & col_ids, std::vector<Field> & fields, std::unordered_set<ColumnID> & schema_all_column_ids)
 {
     const String & raw_value = value.getStr();
     size_t cursor = 0;

From 72beacce5ebfa30e6b958f42472e547d2bc70fc2 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 12:27:48 +0800
Subject: [PATCH 23/37] add exception message

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index cacbae01856..8c31be93ec1 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -240,7 +240,11 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 throw Exception("row size is wrong.", ErrorCodes::LOGICAL_ERROR);
 
             /// Modify `row` by adding missing column values or removing useless column values.
-            if (col_ids.size() != column_ids_to_read.size())
+            if (unlikely(col_ids.size() > column_ids_to_read.size()))
+            {
+                throw Exception("read unexpected columns.", ErrorCodes::LOGICAL_ERROR);
+            }
+            if (col_ids.size() < column_ids_to_read.size())
             {
                 col_id_included.clear();
                 for (size_t i = 0; i < col_ids.size(); i++)

From 237b7f28177e5902fa3516fbc57cd3735decae29 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 12:28:53 +0800
Subject: [PATCH 24/37] comment flushRegion

---
 dbms/src/Storages/Transaction/RegionTable.cpp | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionTable.cpp b/dbms/src/Storages/Transaction/RegionTable.cpp
index a11cbf25bea..9903596d34d 100644
--- a/dbms/src/Storages/Transaction/RegionTable.cpp
+++ b/dbms/src/Storages/Transaction/RegionTable.cpp
@@ -417,7 +417,7 @@ void RegionTable::tryFlushRegion(RegionID region_id)
     if (!status)
         return;
 
-    flushRegion(table_id, region_id, cache_bytes, false);
+    //flushRegion(table_id, region_id, cache_bytes, false);
 
     func_update_region([&](InternalRegion & region) -> bool {
         region.pause_flush = false;
@@ -443,8 +443,8 @@ bool RegionTable::tryFlushRegions()
         });
     }
 
-    for (auto & [id, cache_bytes] : to_flush)
-        flushRegion(id.first, id.second, cache_bytes);
+//    for (auto & [id, cache_bytes] : to_flush)
+//        flushRegion(id.first, id.second, cache_bytes);
 
     { // Now reset status information.
         Timepoint now = Clock::now();

From ad152f8e3cd048e4f432697a09eee47a9e821203 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 13:43:40 +0800
Subject: [PATCH 25/37] uncomment flushRegion and other minor fix

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 3 +--
 dbms/src/Storages/Transaction/RegionTable.cpp       | 6 +++---
 2 files changed, 4 insertions(+), 5 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 8c31be93ec1..fe562e41153 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -198,7 +198,6 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
         std::vector<Field> fields;
         col_ids.reserve(target_col_size);
         fields.reserve(target_col_size);
-        bool schema_not_match = false;
 
         for (const auto & [handle, write_type, commit_ts, value_ptr] : data_list)
         {
@@ -224,7 +223,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             }
             else
             {
-                schema_not_match = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
+                bool schema_not_match = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
                 if (schema_not_match && !force_decode)
                 {
                     return std::make_tuple(block, false);
diff --git a/dbms/src/Storages/Transaction/RegionTable.cpp b/dbms/src/Storages/Transaction/RegionTable.cpp
index 9903596d34d..a11cbf25bea 100644
--- a/dbms/src/Storages/Transaction/RegionTable.cpp
+++ b/dbms/src/Storages/Transaction/RegionTable.cpp
@@ -417,7 +417,7 @@ void RegionTable::tryFlushRegion(RegionID region_id)
     if (!status)
         return;
 
-    //flushRegion(table_id, region_id, cache_bytes, false);
+    flushRegion(table_id, region_id, cache_bytes, false);
 
     func_update_region([&](InternalRegion & region) -> bool {
         region.pause_flush = false;
@@ -443,8 +443,8 @@ bool RegionTable::tryFlushRegions()
         });
     }
 
-//    for (auto & [id, cache_bytes] : to_flush)
-//        flushRegion(id.first, id.second, cache_bytes);
+    for (auto & [id, cache_bytes] : to_flush)
+        flushRegion(id.first, id.second, cache_bytes);
 
     { // Now reset status information.
         Timepoint now = Clock::now();

From c2afabba2b0d028285aa0db722474680dd44fcbf Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 13:48:38 +0800
Subject: [PATCH 26/37] modify push_back to emplace_back

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index fe562e41153..2ebcfef3539 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -218,7 +218,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                     const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
 
                     col_ids.push_back(column.id);
-                    fields.push_back(GenDecodeRow(column.getCodecFlag()));
+                    fields.emplace_back(GenDecodeRow(column.getCodecFlag()));
                 }
             }
             else

From f50a6f4081c51a5e0e7e6a964cb7843ff94adb80 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 13:54:33 +0800
Subject: [PATCH 27/37] add const

---
 dbms/src/Storages/Transaction/TiKVRecordFormat.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index ef3c056b14b..3d45b4681a6 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -36,7 +36,7 @@ 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 bool DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read,
-        std::vector<ColumnID> & col_ids, std::vector<Field> & fields, std::unordered_set<ColumnID> & schema_all_column_ids)
+        std::vector<ColumnID> & col_ids, std::vector<Field> & fields, const std::unordered_set<ColumnID> & schema_all_column_ids)
 {
     const String & raw_value = value.getStr();
     size_t cursor = 0;

From 9daf9743785e9c2b43dd885060cb123749ec1b98 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 13:57:13 +0800
Subject: [PATCH 28/37] fix comment

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 4 ++--
 dbms/src/Storages/Transaction/TiKVRecordFormat.h    | 8 ++++----
 2 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 2ebcfef3539..3c3f1fafa2d 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -223,8 +223,8 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             }
             else
             {
-                bool schema_not_match = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
-                if (schema_not_match && !force_decode)
+                bool schema_is_match = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
+                if (!schema_is_match && !force_decode)
                 {
                     return std::make_tuple(block, false);
                 }
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 3d45b4681a6..319984fdb62 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -40,7 +40,7 @@ inline bool DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID
 {
     const String & raw_value = value.getStr();
     size_t cursor = 0;
-    bool schema_not_match = false;
+    bool schema_is_match = true;
     size_t column_cnt = 0;
     while (cursor < raw_value.size())
     {
@@ -54,7 +54,7 @@ inline bool DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID
         column_cnt++;
         if (!schema_all_column_ids.count(col_id))
         {
-            schema_not_match = true;
+            schema_is_match = false;
         }
         if (!column_ids_to_read.count(col_id))
         {
@@ -68,12 +68,12 @@ inline bool DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID
     }
     if (column_cnt != schema_all_column_ids.size())
     {
-        schema_not_match = true;
+        schema_is_match = false;
     }
 
     if (cursor != raw_value.size())
         throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
-    return schema_not_match;
+    return schema_is_match;
 }
 
 // Key format is here:

From b6cb23fa36f5d1d11866215ed86c28c30af1313c Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 14:50:39 +0800
Subject: [PATCH 29/37] fix comment

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp |  4 ++--
 dbms/src/Storages/Transaction/TiKVRecordFormat.h    | 12 ++++++------
 2 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 3c3f1fafa2d..ff7cf46eb51 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -223,8 +223,8 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             }
             else
             {
-                bool schema_is_match = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
-                if (!schema_is_match && !force_decode)
+                bool schema_matches = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
+                if (!schema_matches && !force_decode)
                 {
                     return std::make_tuple(block, false);
                 }
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 319984fdb62..83ddc71368a 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -40,21 +40,21 @@ inline bool DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID
 {
     const String & raw_value = value.getStr();
     size_t cursor = 0;
-    bool schema_is_match = true;
+    bool schema_matches = true;
     size_t column_cnt = 0;
     while (cursor < raw_value.size())
     {
         Field f = DecodeDatum(cursor, raw_value);
         if (f.isNull())
         {
-            fields.push_back(std::move(f));
+            fields.emplace_back(std::move(f));
             break;
         }
         ColumnID col_id = f.get<ColumnID>();
         column_cnt++;
         if (!schema_all_column_ids.count(col_id))
         {
-            schema_is_match = false;
+            schema_matches = false;
         }
         if (!column_ids_to_read.count(col_id))
         {
@@ -63,17 +63,17 @@ inline bool DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID
         else
         {
             col_ids.push_back(col_id);
-            fields.push_back(DecodeDatum(cursor, raw_value));
+            fields.emplace_back(DecodeDatum(cursor, raw_value));
         }
     }
     if (column_cnt != schema_all_column_ids.size())
     {
-        schema_is_match = false;
+        schema_matches = false;
     }
 
     if (cursor != raw_value.size())
         throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
-    return schema_is_match;
+    return schema_matches;
 }
 
 // Key format is here:

From 8b89ea1053a8fb164487c74f5bbb925c6a164d22 Mon Sep 17 00:00:00 2001
From: Tong Zhigao <tongzhigao@pingcap.com>
Date: Tue, 20 Aug 2019 15:06:06 +0800
Subject: [PATCH 30/37] optimize by using dense_hash_map&dense_hash_set

---
 .../Transaction/RegionBlockReader.cpp         | 67 +++++++++++++++++--
 .../Storages/Transaction/TiKVRecordFormat.h   | 42 ------------
 2 files changed, 60 insertions(+), 49 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 3c3f1fafa2d..be6266aa36d 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -9,6 +9,8 @@
 #include <Storages/Transaction/Region.h>
 #include <Storages/Transaction/RegionBlockReader.h>
 #include <Storages/Transaction/TiDB.h>
+#include <sparsehash/dense_hash_map>
+#include <sparsehash/dense_hash_set>
 
 namespace DB
 {
@@ -112,6 +114,48 @@ void setPKVersionDel(ColumnUInt8 & delmark_col,
     }
 }
 
+inline bool DecodeRow(const TiKVValue & value, const google::dense_hash_set<ColumnID> & column_ids_to_read, std::vector<ColumnID> & col_ids,
+    std::vector<Field> & fields, const google::dense_hash_set<ColumnID> & schema_all_column_ids)
+{
+    const String & raw_value = value.getStr();
+    size_t cursor = 0;
+    bool schema_is_match = true;
+    size_t column_cnt = 0;
+    while (cursor < raw_value.size())
+    {
+        Field f = DecodeDatum(cursor, raw_value);
+        if (f.isNull())
+        {
+            fields.push_back(std::move(f));
+            break;
+        }
+        ColumnID col_id = f.get<ColumnID>();
+        column_cnt++;
+        if (!schema_all_column_ids.count(col_id))
+        {
+            schema_is_match = false;
+        }
+        if (!column_ids_to_read.count(col_id))
+        {
+            SkipDatum(cursor, raw_value);
+        }
+        else
+        {
+            col_ids.push_back(col_id);
+            fields.push_back(DecodeDatum(cursor, raw_value));
+        }
+    }
+    if (column_cnt != schema_all_column_ids.size())
+    {
+        schema_is_match = false;
+    }
+
+    if (cursor != raw_value.size())
+        throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
+    return schema_is_match;
+}
+
+
 std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     const ColumnsDescription & columns,
     const Names & column_names_to_read,
@@ -124,10 +168,19 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
     ColumnID handle_col_id = InvalidColumnID;
 
+    constexpr ColumnID EmptyColumnID = InvalidColumnID - 1;
+
     std::unordered_map<ColumnID, std::pair<MutableColumnPtr, NameAndTypePair>> column_map;
-    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;
+
+    google::dense_hash_map<ColumnID, size_t> column_id_to_info_index_map;
+    column_id_to_info_index_map.set_empty_key(EmptyColumnID);
+
+    google::dense_hash_set<ColumnID> column_ids_to_read;
+    column_ids_to_read.set_empty_key(EmptyColumnID);
+
+    google::dense_hash_set<ColumnID> schema_all_column_ids;
+    schema_all_column_ids.set_empty_key(EmptyColumnID);
+
     for (size_t i = 0; i < table_info.columns.size(); i++)
     {
         auto & column_info = table_info.columns[i];
@@ -145,8 +198,8 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             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));
+            column_ids_to_read.insert(col_id);
+            column_id_to_info_index_map.insert(std::make_pair(col_id, i));
         }
     }
     if (column_names_to_read.size() - 3 != column_ids_to_read.size())
@@ -191,7 +244,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     // optimize for only need handle, tso, delmark.
     if (column_names_to_read.size() > 3)
     {
-        std::unordered_set<ColumnID> col_id_included;
+        google::dense_hash_set<ColumnID> col_id_included;
 
         // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
         std::vector<ColumnID> col_ids;
@@ -247,7 +300,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             {
                 col_id_included.clear();
                 for (size_t i = 0; i < col_ids.size(); i++)
-                    col_id_included.emplace(col_ids[i]);
+                    col_id_included.insert(col_ids[i]);
 
                 // Fill in missing column values.
                 for (auto col_id : column_ids_to_read)
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 319984fdb62..8ffb2a0babc 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -14,7 +14,6 @@
 #include <Storages/Transaction/TiKVVarInt.h>
 #include <Storages/Transaction/Types.h>
 
-
 namespace DB
 {
 
@@ -35,47 +34,6 @@ static const size_t SHORT_VALUE_MAX_LEN = 64;
 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 bool DecodeRow(const TiKVValue & value, const std::unordered_set<ColumnID> & column_ids_to_read,
-        std::vector<ColumnID> & col_ids, std::vector<Field> & fields, const std::unordered_set<ColumnID> & schema_all_column_ids)
-{
-    const String & raw_value = value.getStr();
-    size_t cursor = 0;
-    bool schema_is_match = true;
-    size_t column_cnt = 0;
-    while (cursor < raw_value.size())
-    {
-        Field f = DecodeDatum(cursor, raw_value);
-        if (f.isNull())
-        {
-            fields.push_back(std::move(f));
-            break;
-        }
-        ColumnID col_id = f.get<ColumnID>();
-        column_cnt++;
-        if (!schema_all_column_ids.count(col_id))
-        {
-            schema_is_match = false;
-        }
-        if (!column_ids_to_read.count(col_id))
-        {
-            SkipDatum(cursor, raw_value);
-        }
-        else
-        {
-            col_ids.push_back(col_id);
-            fields.push_back(DecodeDatum(cursor, raw_value));
-        }
-    }
-    if (column_cnt != schema_all_column_ids.size())
-    {
-        schema_is_match = false;
-    }
-
-    if (cursor != raw_value.size())
-        throw Exception("DecodeRow cursor is not end", ErrorCodes::LOGICAL_ERROR);
-    return schema_is_match;
-}
-
 // Key format is here:
 // https://docs.google.com/document/d/1J9Dsp8l5Sbvzjth77hK8yx3SzpEJ4SXaR_wIvswRhro/edit
 // https://github.com/tikv/tikv/blob/289ce2ddac505d7883ec616c078e184c00844d17/src/util/codec/bytes.rs#L33-L63

From 8970e153c0cfa472be7f8d2b8f5084ea767f4aa2 Mon Sep 17 00:00:00 2001
From: Tong Zhigao <tongzhigao@pingcap.com>
Date: Tue, 20 Aug 2019 15:06:23 +0800
Subject: [PATCH 31/37] fix

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index be6266aa36d..77d3a8b706f 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -276,7 +276,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             }
             else
             {
-                bool schema_is_match = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
+                bool schema_is_match = DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
                 if (!schema_is_match && !force_decode)
                 {
                     return std::make_tuple(block, false);

From 1422d0a95fd5a8b5b88b0f6e8a77eab9b994d0f1 Mon Sep 17 00:00:00 2001
From: Tong Zhigao <tongzhigao@pingcap.com>
Date: Tue, 20 Aug 2019 15:07:37 +0800
Subject: [PATCH 32/37] fix

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 1 +
 1 file changed, 1 insertion(+)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 77d3a8b706f..f6bf0e98293 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -245,6 +245,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     if (column_names_to_read.size() > 3)
     {
         google::dense_hash_set<ColumnID> col_id_included;
+        col_id_included.set_empty_key(EmptyColumnID);
 
         // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
         std::vector<ColumnID> col_ids;

From 8b2e25657de60c8489c4ac806b793e61e18cf474 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 15:19:42 +0800
Subject: [PATCH 33/37] fix comment

---
 .../Transaction/RegionBlockReader.cpp         | 67 +++++++++----------
 1 file changed, 33 insertions(+), 34 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index ff7cf46eb51..bde588c4a9e 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -191,13 +191,11 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     // optimize for only need handle, tso, delmark.
     if (column_names_to_read.size() > 3)
     {
-        std::unordered_set<ColumnID> col_id_included;
-
         // TODO: optimize columns' insertion, use better implementation rather than Field, it's terrible.
-        std::vector<ColumnID> col_ids;
-        std::vector<Field> fields;
-        col_ids.reserve(target_col_size);
-        fields.reserve(target_col_size);
+        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)
         {
@@ -207,72 +205,73 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             if (commit_ts > start_ts)
                 continue;
 
-            col_ids.clear();
-            fields.clear();
+            decoded_col_ids.clear();
+            decoded_fields.clear();
             if (write_type == Region::DelFlag)
             {
-                col_ids.reserve(target_col_size);
-                fields.reserve(target_col_size);
+                decoded_col_ids.reserve(target_col_size);
+                decoded_fields.reserve(target_col_size);
                 for (auto col_id : column_ids_to_read)
                 {
                     const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
 
-                    col_ids.push_back(column.id);
-                    fields.emplace_back(GenDecodeRow(column.getCodecFlag()));
+                    decoded_col_ids.push_back(column.id);
+                    decoded_fields.emplace_back(GenDecodeRow(column.getCodecFlag()));
                 }
             }
             else
             {
-                bool schema_matches = RecordKVFormat::DecodeRow(*value_ptr, column_ids_to_read, col_ids, fields, schema_all_column_ids);
+                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 (col_ids.empty() && fields.size() == 1 && fields[0].isNull())
+                if (decoded_col_ids.empty() && decoded_fields.size() == 1 && decoded_fields[0].isNull())
                 {
                     // all field is null
-                    fields.clear();
+                    decoded_fields.clear();
                 }
             }
 
-            if (col_ids.size() != fields.size())
+            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.
-            if (unlikely(col_ids.size() > column_ids_to_read.size()))
+            if (unlikely(decoded_col_ids.size() > column_ids_to_read.size()))
             {
                 throw Exception("read unexpected columns.", ErrorCodes::LOGICAL_ERROR);
             }
-            if (col_ids.size() < column_ids_to_read.size())
+
+            // 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())
             {
-                col_id_included.clear();
-                for (size_t i = 0; i < col_ids.size(); i++)
-                    col_id_included.emplace(col_ids[i]);
+                std::unordered_set<ColumnID> decoded_col_ids_set(decoded_col_ids.begin(), decoded_col_ids.end());
 
                 // Fill in missing column values.
                 for (auto col_id : column_ids_to_read)
                 {
-                    if (col_id_included.count(col_id))
+                    if (decoded_col_ids_set.count(col_id))
                         continue;
 
                     const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];
-                    col_ids.push_back(column.id);
+                    decoded_col_ids.push_back(column.id);
                     if (column.hasNoDefaultValueFlag())
                         // Fill `zero` value if NOT NULL specified or else NULL.
-                        fields.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
+                        decoded_fields.push_back(column.hasNotNullFlag() ? GenDecodeRow(column.getCodecFlag()) : Field());
                     else
                         // Fill default value.
-                        fields.push_back(column.defaultValueToField());
+                        decoded_fields.push_back(column.defaultValueToField());
                 }
             }
 
-            if (col_ids.size() != target_col_size || fields.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 < col_ids.size(); i++)
+            for (size_t i = 0; i < decoded_col_ids.size(); i++)
             {
-                ColumnID col_id = col_ids[i];
+                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);
@@ -281,10 +280,10 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 if (tp->isDateOrDateTime()
                     || (tp->isNullable() && dynamic_cast<const DataTypeNullable *>(tp.get())->getNestedType()->isDateOrDateTime()))
                 {
-                    Field & field = fields[i];
+                    Field & field = decoded_fields[i];
                     if (field.isNull())
                     {
-                        it->second.first->insert(fields[i]);
+                        it->second.first->insert(decoded_fields[i]);
                         continue;
                     }
                     UInt64 packed = field.get<UInt64>();
@@ -329,7 +328,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 }
                 else
                 {
-                    it->second.first->insert(fields[i]);
+                    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.
@@ -347,14 +346,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 = fields[i].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 = fields[i].get<Int64>();
+                            Int64 orig = decoded_fields[i].get<Int64>();
                             overflow = inserted != orig;
                         }
                         if (overflow)
@@ -364,7 +363,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(fields[i].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);

From b574f5b364e62d808391510b31b9c19298c757ed Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 15:34:22 +0800
Subject: [PATCH 34/37] small fix

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 2 --
 1 file changed, 2 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index bde588c4a9e..03f9da4f32d 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -209,8 +209,6 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             decoded_fields.clear();
             if (write_type == Region::DelFlag)
             {
-                decoded_col_ids.reserve(target_col_size);
-                decoded_fields.reserve(target_col_size);
                 for (auto col_id : column_ids_to_read)
                 {
                     const auto & column = table_info.columns[column_id_to_info_index_map[col_id]];

From 5119717c3cbcedf747d8e0de715d853a3d9e6fd6 Mon Sep 17 00:00:00 2001
From: linkmyth <lidezhu@pingcap.com>
Date: Tue, 20 Aug 2019 15:38:23 +0800
Subject: [PATCH 35/37] remove useless comment

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 1 -
 1 file changed, 1 deletion(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 03f9da4f32d..332c69cb26d 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -246,7 +246,6 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             {
                 std::unordered_set<ColumnID> decoded_col_ids_set(decoded_col_ids.begin(), decoded_col_ids.end());
 
-                // Fill in missing column values.
                 for (auto col_id : column_ids_to_read)
                 {
                     if (decoded_col_ids_set.count(col_id))

From ad9bb5194cf06c6dcaf2b65aac67f85431489633 Mon Sep 17 00:00:00 2001
From: Tong Zhigao <tongzhigao@pingcap.com>
Date: Tue, 20 Aug 2019 15:43:31 +0800
Subject: [PATCH 36/37] format

---
 dbms/src/Storages/Transaction/RegionBlockReader.cpp | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index 4f1bfe52249..e1adb853fd9 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -275,8 +275,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             }
             else
             {
-                bool schema_matches
-                    = DecodeRow(*value_ptr, column_ids_to_read, decoded_col_ids, decoded_fields, schema_all_column_ids);
+                bool schema_matches = 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);

From f886ff3a00b16d36860866f3a76b0aa9b22181ab Mon Sep 17 00:00:00 2001
From: Tong Zhigao <tongzhigao@pingcap.com>
Date: Tue, 20 Aug 2019 18:47:16 +0800
Subject: [PATCH 37/37] [FLASH-402] optimize column_map (#2)

---
 dbms/src/Storages/Transaction/Codec.cpp       | 41 ++++++++-----------
 .../Transaction/RegionBlockReader.cpp         | 39 ++++++++++--------
 .../txn_mock/data_only_in_region.test         | 10 +++++
 3 files changed, 48 insertions(+), 42 deletions(-)

diff --git a/dbms/src/Storages/Transaction/Codec.cpp b/dbms/src/Storages/Transaction/Codec.cpp
index 9923df7dbbf..7dadf74c2dc 100644
--- a/dbms/src/Storages/Transaction/Codec.cpp
+++ b/dbms/src/Storages/Transaction/Codec.cpp
@@ -44,9 +44,9 @@ Float64 DecodeFloat64(size_t & cursor, const String & raw_value)
     return enforce_cast<Float64>(num);
 }
 
-String DecodeBytes(size_t & cursor, const String & raw_value)
+template <typename StringStream>
+void DecodeBytes(size_t & cursor, const String & raw_value, StringStream & ss)
 {
-    std::stringstream ss;
     while (true)
     {
         size_t next_cursor = cursor + 9;
@@ -62,25 +62,24 @@ String DecodeBytes(size_t & cursor, const String & raw_value)
         if (pad_size != 0)
             break;
     }
+}
+
+String DecodeBytes(size_t & cursor, const String & raw_value)
+{
+    std::stringstream ss;
+    DecodeBytes(cursor, raw_value, ss);
     return ss.str();
 }
 
-void SkipBytes(size_t & cursor, const String & raw_value)
+struct NullStringStream
 {
-    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;
+    void write(const char *, size_t) {}
+};
 
-        if (pad_size > 8)
-            throw Exception("Wrong format, too many padding bytes. (DecodeBytes)", ErrorCodes::LOGICAL_ERROR);
-        cursor = next_cursor;
-        if (pad_size != 0)
-            break;
-    }
+void SkipBytes(size_t & cursor, const String & raw_value)
+{
+    NullStringStream ss;
+    DecodeBytes(cursor, raw_value, ss);
 }
 
 String DecodeCompactBytes(size_t & cursor, const String & raw_value)
@@ -104,10 +103,7 @@ Int64 DecodeVarInt(size_t & cursor, const String & raw_value)
     return (v & 1) ? ~vx : vx;
 }
 
-void SkipVarInt(size_t & cursor, const String & raw_value)
-{
-    SkipVarUInt(cursor, raw_value);
-}
+void SkipVarInt(size_t & cursor, const String & raw_value) { SkipVarUInt(cursor, raw_value); }
 
 UInt64 DecodeVarUInt(size_t & cursor, const String & raw_value)
 {
@@ -128,10 +124,7 @@ 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);
-}
+void SkipVarUInt(size_t & cursor, const String & raw_value) { std::ignore = DecodeVarUInt(cursor, raw_value); }
 
 inline Int8 getWords(PrecType prec, ScaleType scale)
 {
diff --git a/dbms/src/Storages/Transaction/RegionBlockReader.cpp b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
index a7280e54987..7d1cf0b1303 100644
--- a/dbms/src/Storages/Transaction/RegionBlockReader.cpp
+++ b/dbms/src/Storages/Transaction/RegionBlockReader.cpp
@@ -170,7 +170,9 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
 
     constexpr ColumnID EmptyColumnID = InvalidColumnID - 1;
 
-    std::unordered_map<ColumnID, std::pair<MutableColumnPtr, NameAndTypePair>> column_map;
+    using ColTypePair = std::pair<MutableColumnPtr, NameAndTypePair>;
+    google::dense_hash_map<ColumnID, std::shared_ptr<ColTypePair>> column_map;
+    column_map.set_empty_key(EmptyColumnID);
 
     google::dense_hash_map<ColumnID, size_t> column_id_to_info_index_map;
     column_id_to_info_index_map.set_empty_key(EmptyColumnID);
@@ -192,8 +194,8 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
             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());
+        column_map.insert(std::make_pair(col_id, std::make_shared<ColTypePair>(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
@@ -208,11 +210,11 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
     if (!table_info.pk_is_handle)
     {
         auto ch_col = columns.getPhysical(MutableSupport::tidb_pk_column_name);
-        column_map[handle_col_id] = std::make_pair(ch_col.type->createColumn(), ch_col);
-        column_map[handle_col_id].first->reserve(data_list.size());
+        column_map.insert(std::make_pair(handle_col_id, std::make_shared<ColTypePair>(ch_col.type->createColumn(), ch_col)));
+        column_map[handle_col_id]->first->reserve(data_list.size());
     }
 
-    const TMTPKType pk_type = getTMTPKType(*column_map[handle_col_id].second.type);
+    const TMTPKType pk_type = getTMTPKType(*column_map[handle_col_id]->second.type);
 
     if (pk_type == TMTPKType::UINT64)
         ReorderRegionDataReadList(data_list);
@@ -232,7 +234,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 break;
         }
 
-        func(*delmark_col, *version_col, column_map[handle_col_id].first, data_list, start_ts);
+        func(*delmark_col, *version_col, column_map[handle_col_id]->first, data_list, start_ts);
     }
 
     const auto & date_lut = DateLUT::instance();
@@ -330,14 +332,16 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                 if (it == column_map.end())
                     throw Exception("col_id not found in column_map", ErrorCodes::LOGICAL_ERROR);
 
-                const auto & tp = it->second.second.type;
+                auto & name_type = it->second->second;
+                auto & mut_col = it->second->first;
+                const auto & tp = name_type.type;
                 if (tp->isDateOrDateTime()
                     || (tp->isNullable() && dynamic_cast<const DataTypeNullable *>(tp.get())->getNestedType()->isDateOrDateTime()))
                 {
                     Field & field = decoded_fields[i];
                     if (field.isNull())
                     {
-                        it->second.first->insert(decoded_fields[i]);
+                        mut_col->insert(decoded_fields[i]);
                         continue;
                     }
                     UInt64 packed = field.get<UInt64>();
@@ -371,27 +375,26 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
                             }
                             datetime = date_lut.makeDateTime(year, month, day, hour, minute, second);
                         }
-                        it->second.first->insert(static_cast<Int64>(datetime));
+                        mut_col->insert(static_cast<Int64>(datetime));
                     }
                     else
                     {
                         auto date = date_lut.makeDayNum(year, month, day);
                         Field date_field(static_cast<Int64>(date));
-                        it->second.first->insert(date_field);
+                        mut_col->insert(date_field);
                     }
                 }
                 else
                 {
-                    it->second.first->insert(decoded_fields[i]);
+                    mut_col->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.
                     // So far only integers is possible of overflow.
-                    auto nested_tp = tp->isNullable() ? dynamic_cast<const DataTypeNullable *>(tp.get())->getNestedType() : tp;
-                    auto & orig_column = *it->second.first;
-                    auto & nested_column = it->second.first->isColumnNullable()
-                        ? dynamic_cast<ColumnNullable &>(*it->second.first).getNestedColumn()
-                        : *it->second.first;
+                    auto & nested_tp = tp->isNullable() ? dynamic_cast<const DataTypeNullable *>(tp.get())->getNestedType() : tp;
+                    auto & orig_column = *mut_col;
+                    auto & nested_column
+                        = mut_col->isColumnNullable() ? dynamic_cast<ColumnNullable &>(*mut_col).getNestedColumn() : *mut_col;
                     auto inserted_index = orig_column.size() - 1;
                     if (!orig_column.isNullAt(inserted_index) && nested_tp->isInteger())
                     {
@@ -443,7 +446,7 @@ std::tuple<Block, bool> readRegionBlock(const TiDB::TableInfo & table_info,
         else
         {
             Int64 col_id = table_info.getColumnID(name);
-            block.insert({std::move(column_map[col_id].first), column_map[col_id].second.type, name});
+            block.insert({std::move(column_map[col_id]->first), column_map[col_id]->second.type, name});
         }
     }
 
diff --git a/tests/mutable-test/txn_mock/data_only_in_region.test b/tests/mutable-test/txn_mock/data_only_in_region.test
index 20c7f10a250..095b75a2ebf 100644
--- a/tests/mutable-test/txn_mock/data_only_in_region.test
+++ b/tests/mutable-test/txn_mock/data_only_in_region.test
@@ -68,5 +68,15 @@
 │    1205 │
 └─────────┘
 
+=> select count(col_1) from default.test
+┌─count(col_1)─┐
+│         1205 │
+└──────────────┘
+
+=> select count(col_2) from default.test
+┌─count(col_2)─┐
+│         1205 │
+└──────────────┘
+
 => DBGInvoke __drop_tidb_table(default, test)
 => drop table if exists default.test