diff --git a/dbms/src/Debug/dbgFuncCoprocessor.cpp b/dbms/src/Debug/dbgFuncCoprocessor.cpp
index 83828238803..594a0c4ee1c 100644
--- a/dbms/src/Debug/dbgFuncCoprocessor.cpp
+++ b/dbms/src/Debug/dbgFuncCoprocessor.cpp
@@ -49,8 +49,7 @@ BlockInputStreamPtr dbgFuncDAG(Context & context, const ASTs & args)
         region_id = safeGet<RegionID>(typeid_cast<const ASTLiteral &>(*args[1]).value);
     Timestamp start_ts = context.getTMTContext().getPDClient()->getTS();
 
-    auto [table_id, schema, dag_request] = compileQuery(
-        context, query,
+    auto [table_id, schema, dag_request] = compileQuery(context, query,
         [&](const String & database_name, const String & table_name) {
             auto storage = context.getTable(database_name, table_name);
             auto mmt = std::dynamic_pointer_cast<StorageMergeTree>(storage);
@@ -92,8 +91,7 @@ BlockInputStreamPtr dbgFuncMockDAG(Context & context, const ASTs & args)
     if (start_ts == 0)
         start_ts = context.getTMTContext().getPDClient()->getTS();
 
-    auto [table_id, schema, dag_request] = compileQuery(
-        context, query,
+    auto [table_id, schema, dag_request] = compileQuery(context, query,
         [&](const String & database_name, const String & table_name) {
             return MockTiDB::instance().getTableByName(database_name, table_name)->table_info;
         },
@@ -210,9 +208,12 @@ void compileExpr(const DAGSchema & input, ASTPtr ast, tipb::Expr * expr, std::un
                 expr->set_tp(tipb::Float64);
                 encodeDAGFloat64(lit->value.get<Float64>(), ss);
                 break;
-            case Field::Types::Which::Decimal:
+            case Field::Types::Which::Decimal32:
+            case Field::Types::Which::Decimal64:
+            case Field::Types::Which::Decimal128:
+            case Field::Types::Which::Decimal256:
                 expr->set_tp(tipb::MysqlDecimal);
-                encodeDAGDecimal(lit->value.get<Decimal>(), ss);
+                encodeDAGDecimal(lit->value, ss);
                 break;
             case Field::Types::Which::String:
                 expr->set_tp(tipb::String);
diff --git a/dbms/src/Debug/dbgFuncRegion.cpp b/dbms/src/Debug/dbgFuncRegion.cpp
index 1eefde4739a..257d7694d0a 100644
--- a/dbms/src/Debug/dbgFuncRegion.cpp
+++ b/dbms/src/Debug/dbgFuncRegion.cpp
@@ -6,7 +6,6 @@
 #include <Interpreters/executeQuery.h>
 #include <Parsers/ASTIdentifier.h>
 #include <Parsers/ASTLiteral.h>
-#include <Storages/MutableSupport.h>
 #include <Storages/StorageMergeTree.h>
 #include <Storages/Transaction/KVStore.h>
 #include <Storages/Transaction/Region.h>
@@ -140,7 +139,9 @@ void dbgFuncRegionSnapshotWithData(Context & context, const ASTs & args, DBGInvo
             }
 
             TiKVKey key = RecordKVFormat::genKey(table_id, handle_id);
-            TiKVValue value = RecordKVFormat::EncodeRow(table->table_info, fields);
+            std::stringstream ss;
+            RegionBench::encodeRow(table->table_info, fields, ss);
+            TiKVValue value(ss.str());
             UInt64 commit_ts = tso;
             UInt64 prewrite_ts = tso;
             TiKVValue commit_value;
diff --git a/dbms/src/Debug/dbgTools.cpp b/dbms/src/Debug/dbgTools.cpp
index 5e40faabd9a..305ac3f8260 100644
--- a/dbms/src/Debug/dbgTools.cpp
+++ b/dbms/src/Debug/dbgTools.cpp
@@ -8,6 +8,7 @@
 #include <Raft/RaftContext.h>
 #include <Storages/Transaction/Codec.h>
 #include <Storages/Transaction/KVStore.h>
+#include <Storages/Transaction/MyTimeParser.h>
 #include <Storages/Transaction/Region.h>
 #include <Storages/Transaction/TMTContext.h>
 #include <Storages/Transaction/TiKVRange.h>
@@ -23,6 +24,9 @@ extern const int LOGICAL_ERROR;
 namespace RegionBench
 {
 
+using TiDB::ColumnInfo;
+using TiDB::TableInfo;
+
 RegionPtr createRegion(TableID table_id, RegionID region_id, const HandleID & start, const HandleID & end)
 {
     enginepb::SnapshotRequest request;
@@ -121,6 +125,128 @@ void addRequestsToRaftCmd(enginepb::CommandRequest * cmd, RegionID region_id, co
     }
 }
 
+template <typename T>
+T convertNumber(const Field & field)
+{
+    switch (field.getType())
+    {
+        case Field::Types::Int64:
+            return static_cast<T>(field.get<Int64>());
+        case Field::Types::UInt64:
+            return static_cast<T>(field.get<UInt64>());
+        case Field::Types::Float64:
+            return static_cast<T>(field.get<Float64>());
+        case Field::Types::Decimal32:
+            return static_cast<T>(field.get<DecimalField<Decimal32>>());
+        case Field::Types::Decimal64:
+            return static_cast<T>(field.get<DecimalField<Decimal64>>());
+        case Field::Types::Decimal128:
+            return static_cast<T>(field.get<DecimalField<Decimal128>>());
+        case Field::Types::Decimal256:
+            return static_cast<T>(field.get<DecimalField<Decimal256>>());
+        default:
+            throw Exception(String("Unable to convert field type ") + field.getTypeName() + " to number", ErrorCodes::LOGICAL_ERROR);
+    }
+}
+
+Field convertDecimal(UInt32 scale, const Field & field)
+{
+    switch (field.getType())
+    {
+        case Field::Types::Int64:
+            return DecimalField(ToDecimal<Int64, Decimal64>(field.get<Int64>(), scale), scale);
+        case Field::Types::UInt64:
+            return DecimalField(ToDecimal<Int64, Decimal64>(field.get<UInt64>(), scale), scale);
+        case Field::Types::Float64:
+            return DecimalField(ToDecimal<Float64, Decimal64>(field.get<Float64>(), scale), scale);
+        case Field::Types::Decimal32:
+        case Field::Types::Decimal64:
+        case Field::Types::Decimal128:
+        case Field::Types::Decimal256:
+            return field;
+        default:
+            throw Exception(String("Unable to convert field type ") + field.getTypeName() + " to number", ErrorCodes::LOGICAL_ERROR);
+    }
+}
+
+Field convertEnum(const ColumnInfo & column_info, const Field & field)
+{
+    switch (field.getType())
+    {
+        case Field::Types::Int64:
+        case Field::Types::UInt64:
+            return convertNumber<UInt64>(field);
+        case Field::Types::String:
+            return static_cast<UInt64>(column_info.getEnumIndex(field.get<String>()));
+        default:
+            throw Exception(String("Unable to convert field type ") + field.getTypeName() + " to Enum", ErrorCodes::LOGICAL_ERROR);
+    }
+}
+
+Field convertField(const ColumnInfo & column_info, const Field & field)
+{
+    if (field.isNull())
+        return field;
+
+    switch (column_info.tp)
+    {
+        case TiDB::TypeTiny:
+        case TiDB::TypeShort:
+        case TiDB::TypeLong:
+        case TiDB::TypeLongLong:
+        case TiDB::TypeInt24:
+        case TiDB::TypeBit:
+            if (column_info.hasUnsignedFlag())
+                return convertNumber<UInt64>(field);
+            else
+                return convertNumber<Int64>(field);
+        case TiDB::TypeFloat:
+        case TiDB::TypeDouble:
+            return convertNumber<Float64>(field);
+        case TiDB::TypeDate:
+        case TiDB::TypeDatetime:
+        case TiDB::TypeTimestamp:
+            return DB::parseMyDatetime(field.get<String>());
+        case TiDB::TypeVarchar:
+        case TiDB::TypeTinyBlob:
+        case TiDB::TypeMediumBlob:
+        case TiDB::TypeLongBlob:
+        case TiDB::TypeBlob:
+        case TiDB::TypeVarString:
+        case TiDB::TypeString:
+            return field;
+        case TiDB::TypeEnum:
+            return convertEnum(column_info, field);
+        case TiDB::TypeNull:
+            return Field();
+        case TiDB::TypeDecimal:
+        case TiDB::TypeNewDecimal:
+            return convertDecimal(column_info.decimal, field);
+        case TiDB::TypeTime:
+            throw Exception(String("Unable to convert field type ") + field.getTypeName() + " to Time", ErrorCodes::LOGICAL_ERROR);
+        case TiDB::TypeYear:
+            throw Exception(String("Unable to convert field type ") + field.getTypeName() + " to Year", ErrorCodes::LOGICAL_ERROR);
+        case TiDB::TypeSet:
+            throw Exception(String("Unable to convert field type ") + field.getTypeName() + " to Set", ErrorCodes::LOGICAL_ERROR);
+        default:
+            return Field();
+    }
+}
+
+void encodeRow(const TiDB::TableInfo & table_info, const std::vector<Field> & fields, std::stringstream & ss)
+{
+    if (table_info.columns.size() != fields.size())
+        throw Exception("Encoding row has different sizes between columns and values", ErrorCodes::LOGICAL_ERROR);
+    for (size_t i = 0; i < fields.size(); i++)
+    {
+        const TiDB::ColumnInfo & column_info = table_info.columns[i];
+        EncodeDatum(Field(column_info.id), TiDB::CodecFlagInt, ss);
+        Field field = convertField(column_info, fields[i]);
+        TiDB::DatumBumpy datum = TiDB::DatumBumpy(field, column_info.tp);
+        EncodeDatum(datum.field(), column_info.getCodecFlag(), ss);
+    }
+}
+
 void insert(const TiDB::TableInfo & table_info, RegionID region_id, HandleID handle_id, ASTs::const_iterator begin,
     ASTs::const_iterator end, Context & context, const std::optional<std::tuple<Timestamp, UInt8>> & tso_del)
 {
@@ -142,7 +268,9 @@ void insert(const TiDB::TableInfo & table_info, RegionID region_id, HandleID han
     TableID table_id = RecordKVFormat::getTableId(region->getRange().first);
 
     TiKVKey key = RecordKVFormat::genKey(table_id, handle_id);
-    TiKVValue value = RecordKVFormat::EncodeRow(table_info, fields);
+    std::stringstream ss;
+    encodeRow(table_info, fields, ss);
+    TiKVValue value(ss.str());
 
     UInt64 prewrite_ts = pd_client->getTS();
     UInt64 commit_ts = pd_client->getTS();
diff --git a/dbms/src/Debug/dbgTools.h b/dbms/src/Debug/dbgTools.h
index 5ed25a4eb02..70372a2c765 100644
--- a/dbms/src/Debug/dbgTools.h
+++ b/dbms/src/Debug/dbgTools.h
@@ -23,6 +23,8 @@ RegionPtr createRegion(TableID table_id, RegionID region_id, const HandleID & st
 
 Regions createRegions(TableID table_id, size_t region_num, size_t key_num_each_region, HandleID handle_begin, RegionID new_region_id_begin);
 
+void encodeRow(const TiDB::TableInfo & table_info, const std::vector<Field> & fields, std::stringstream & ss);
+
 void insert(const TiDB::TableInfo & table_info, RegionID region_id, HandleID handle_id, ASTs::const_iterator begin,
     ASTs::const_iterator end, Context & context, const std::optional<std::tuple<Timestamp, UInt8>> & tso_del = {});
 
diff --git a/dbms/src/Flash/Coprocessor/DAGBlockOutputStream.cpp b/dbms/src/Flash/Coprocessor/DAGBlockOutputStream.cpp
index 0ef25b08700..b475ac3cb24 100644
--- a/dbms/src/Flash/Coprocessor/DAGBlockOutputStream.cpp
+++ b/dbms/src/Flash/Coprocessor/DAGBlockOutputStream.cpp
@@ -2,6 +2,7 @@
 
 #include <DataTypes/DataTypeNullable.h>
 #include <Storages/Transaction/Codec.h>
+#include <Storages/Transaction/Datum.h>
 #include <Storages/Transaction/TypeMapping.h>
 
 namespace DB
@@ -13,6 +14,9 @@ extern const int UNSUPPORTED_PARAMETER;
 extern const int LOGICAL_ERROR;
 } // namespace ErrorCodes
 
+using TiDB::DatumBumpy;
+using TiDB::TP;
+
 DAGBlockOutputStream::DAGBlockOutputStream(tipb::SelectResponse & dag_response_, Int64 records_per_chunk_, tipb::EncodeType encodeType_,
     std::vector<tipb::FieldType> && result_field_types_, Block header_)
     : dag_response(dag_response_),
@@ -71,8 +75,9 @@ void DAGBlockOutputStream::write(const Block & block)
         }
         for (size_t j = 0; j < block.columns(); j++)
         {
-            auto field = (*block.getByPosition(j).column.get())[i];
-            EncodeDatum(field, getCodecFlagByFieldType(result_field_types[j]), current_ss);
+            const auto & field = (*block.getByPosition(j).column.get())[i];
+            DatumBumpy datum(field, static_cast<TP>(result_field_types[j].tp()));
+            EncodeDatum(datum.field(), getCodecFlagByFieldType(result_field_types[j]), current_ss);
         }
         // Encode current row
         records_per_chunk++;
diff --git a/dbms/src/Flash/Coprocessor/DAGCodec.cpp b/dbms/src/Flash/Coprocessor/DAGCodec.cpp
index 9d809cc1258..2316cdcad99 100644
--- a/dbms/src/Flash/Coprocessor/DAGCodec.cpp
+++ b/dbms/src/Flash/Coprocessor/DAGCodec.cpp
@@ -26,7 +26,7 @@ void encodeDAGString(const String & s, std::stringstream & ss) { ss << s; }
 
 void encodeDAGBytes(const String & bytes, std::stringstream & ss) { ss << bytes; }
 
-void encodeDAGDecimal(const Decimal & d, std::stringstream & ss) { EncodeDecimal(d, ss); }
+void encodeDAGDecimal(const Field & field, std::stringstream & ss) { EncodeDecimal(field, ss); }
 
 Int64 decodeDAGInt64(const String & s)
 {
@@ -56,7 +56,7 @@ String decodeDAGString(const String & s) { return s; }
 
 String decodeDAGBytes(const String & s) { return s; }
 
-Decimal decodeDAGDecimal(const String & s)
+Field decodeDAGDecimal(const String & s)
 {
     size_t cursor = 0;
     return DecodeDecimal(cursor, s);
diff --git a/dbms/src/Flash/Coprocessor/DAGCodec.h b/dbms/src/Flash/Coprocessor/DAGCodec.h
index faecf74df1f..44fb9e5bc3f 100644
--- a/dbms/src/Flash/Coprocessor/DAGCodec.h
+++ b/dbms/src/Flash/Coprocessor/DAGCodec.h
@@ -12,7 +12,7 @@ void encodeDAGFloat32(Float32, std::stringstream &);
 void encodeDAGFloat64(Float64, std::stringstream &);
 void encodeDAGString(const String &, std::stringstream &);
 void encodeDAGBytes(const String &, std::stringstream &);
-void encodeDAGDecimal(const Decimal &, std::stringstream &);
+void encodeDAGDecimal(const Field &, std::stringstream &);
 
 Int64 decodeDAGInt64(const String &);
 UInt64 decodeDAGUInt64(const String &);
@@ -20,6 +20,6 @@ Float32 decodeDAGFloat32(const String &);
 Float64 decodeDAGFloat64(const String &);
 String decodeDAGString(const String &);
 String decodeDAGBytes(const String &);
-Decimal decodeDAGDecimal(const String &);
+Field decodeDAGDecimal(const String &);
 
 } // namespace DB
\ No newline at end of file
diff --git a/dbms/src/Flash/Coprocessor/DAGUtils.cpp b/dbms/src/Flash/Coprocessor/DAGUtils.cpp
index be1bbb92aa9..2a334230573 100644
--- a/dbms/src/Flash/Coprocessor/DAGUtils.cpp
+++ b/dbms/src/Flash/Coprocessor/DAGUtils.cpp
@@ -71,7 +71,19 @@ String exprToString(const tipb::Expr & expr, const NamesAndTypesList & input_col
         case tipb::ExprType::Bytes:
             return decodeDAGBytes(expr.val());
         case tipb::ExprType::MysqlDecimal:
-            return decodeDAGDecimal(expr.val()).toString();
+        {
+            auto field = decodeDAGDecimal(expr.val());
+            if (field.getType() == Field::Types::Decimal32)
+                return field.get<DecimalField<Decimal32>>().toString();
+            else if (field.getType() == Field::Types::Decimal64)
+                return field.get<DecimalField<Decimal32>>().toString();
+            else if (field.getType() == Field::Types::Decimal128)
+                return field.get<DecimalField<Decimal32>>().toString();
+            else if (field.getType() == Field::Types::Decimal256)
+                return field.get<DecimalField<Decimal32>>().toString();
+            else
+                throw Exception("Not decimal literal" + expr.DebugString(), ErrorCodes::COP_BAD_DAG_REQUEST);
+        }
         case tipb::ExprType::ColumnRef:
             column_id = decodeDAGInt64(expr.val());
             if (column_id < 0 || column_id >= (ColumnID)input_col.size())
diff --git a/dbms/src/Storages/Transaction/Codec.cpp b/dbms/src/Storages/Transaction/Codec.cpp
index 7465a04b6b5..d57c5023d77 100644
--- a/dbms/src/Storages/Transaction/Codec.cpp
+++ b/dbms/src/Storages/Transaction/Codec.cpp
@@ -410,7 +410,7 @@ inline void writeWord(String & buf, Int32 word, int size)
 }
 
 template <typename T>
-void EncodeDecimal(const T & dec, PrecType prec, ScaleType frac, std::stringstream & ss)
+void EncodeDecimalImpl(const T & dec, PrecType prec, ScaleType frac, std::stringstream & ss)
 {
     static_assert(IsDecimal<T>);
 
@@ -476,19 +476,31 @@ void EncodeDecimal(const T & dec, PrecType prec, ScaleType frac, std::stringstre
     ss.write(buf.c_str(), buf.size());
 }
 
-template <typename T>
-inline T getFieldValue(const Field & field)
+void EncodeDecimal(const Field & field, std::stringstream & ss)
 {
-    switch (field.getType())
+    if (field.getType() == Field::Types::Decimal32)
     {
-        case Field::Types::UInt64:
-            return static_cast<T>(field.get<UInt64>());
-        case Field::Types::Int64:
-            return static_cast<T>(field.get<Int64>());
-        case Field::Types::Float64:
-            return static_cast<T>(field.get<Float64>());
-        default:
-            throw Exception("Unsupport (getFieldValue): " + std::string(field.getTypeName()), ErrorCodes::LOGICAL_ERROR);
+        auto decimal_field = field.get<DecimalField<Decimal32>>();
+        return EncodeDecimalImpl(decimal_field.getValue(), decimal_field.getPrec(), decimal_field.getScale(), ss);
+    }
+    else if (field.getType() == Field::Types::Decimal64)
+    {
+        auto decimal_field = field.get<DecimalField<Decimal64>>();
+        return EncodeDecimalImpl(decimal_field.getValue(), decimal_field.getPrec(), decimal_field.getScale(), ss);
+    }
+    else if (field.getType() == Field::Types::Decimal128)
+    {
+        auto decimal_field = field.get<DecimalField<Decimal128>>();
+        return EncodeDecimalImpl(decimal_field.getValue(), decimal_field.getPrec(), decimal_field.getScale(), ss);
+    }
+    else if (field.getType() == Field::Types::Decimal256)
+    {
+        auto decimal_field = field.get<DecimalField<Decimal256>>();
+        return EncodeDecimalImpl(decimal_field.getValue(), decimal_field.getPrec(), decimal_field.getScale(), ss);
+    }
+    else
+    {
+        throw Exception("Not a decimal when decoding decimal", ErrorCodes::LOGICAL_ERROR);
     }
 }
 
@@ -503,46 +515,27 @@ void EncodeDatum(const Field & field, TiDB::CodecFlag flag, std::stringstream &
     switch (flag)
     {
         case TiDB::CodecFlagDecimal:
-            if (field.getType() == Field::Types::Decimal32)
-            {
-                auto decimal_field = field.get<DecimalField<Decimal32>>();
-                return EncodeDecimal(decimal_field.getValue(), decimal_field.getPrec(), decimal_field.getScale(), ss);
-            }
-            else if (field.getType() == Field::Types::Decimal64)
-            {
-                auto decimal_field = field.get<DecimalField<Decimal64>>();
-                return EncodeDecimal(decimal_field.getValue(), decimal_field.getPrec(), decimal_field.getScale(), ss);
-            }
-            else if (field.getType() == Field::Types::Decimal128)
-            {
-                auto decimal_field = field.get<DecimalField<Decimal128>>();
-                return EncodeDecimal(decimal_field.getValue(), decimal_field.getPrec(), decimal_field.getScale(), ss);
-            }
-            else
-            {
-                auto decimal_field = field.get<DecimalField<Decimal256>>();
-                return EncodeDecimal(decimal_field.getValue(), decimal_field.getPrec(), decimal_field.getScale(), ss);
-            }
+            return EncodeDecimal(field, ss);
         case TiDB::CodecFlagCompactBytes:
-            return EncodeCompactBytes(field.get<String>(), ss);
+            return EncodeCompactBytes(field.safeGet<String>(), ss);
         case TiDB::CodecFlagFloat:
-            return EncodeFloat64(getFieldValue<Float64>(field), ss);
+            return EncodeFloat64(field.safeGet<Float64>(), ss);
         case TiDB::CodecFlagUInt:
-            return EncodeUInt<UInt64>(getFieldValue<UInt64>(field), ss);
+            return EncodeUInt<UInt64>(field.safeGet<UInt64>(), ss);
         case TiDB::CodecFlagInt:
-            return EncodeInt64(getFieldValue<Int64>(field), ss);
+            return EncodeInt64(field.safeGet<Int64>(), ss);
         case TiDB::CodecFlagVarInt:
-            return EncodeVarInt(getFieldValue<Int64>(field), ss);
+            return EncodeVarInt(field.safeGet<Int64>(), ss);
         case TiDB::CodecFlagVarUInt:
-            return EncodeVarUInt(getFieldValue<UInt64>(field), ss);
+            return EncodeVarUInt(field.safeGet<UInt64>(), ss);
         default:
             throw Exception("Not implemented codec flag: " + std::to_string(flag), ErrorCodes::LOGICAL_ERROR);
     }
 }
 
-template void EncodeDecimal<Decimal32>(const Decimal32 &, PrecType, ScaleType, std::stringstream & ss);
-template void EncodeDecimal<Decimal64>(const Decimal64 &, PrecType, ScaleType, std::stringstream & ss);
-template void EncodeDecimal<Decimal128>(const Decimal128 &, PrecType, ScaleType, std::stringstream & ss);
-template void EncodeDecimal<Decimal256>(const Decimal256 &, PrecType, ScaleType, std::stringstream & ss);
+template void EncodeDecimalImpl<Decimal32>(const Decimal32 &, PrecType, ScaleType, std::stringstream & ss);
+template void EncodeDecimalImpl<Decimal64>(const Decimal64 &, PrecType, ScaleType, std::stringstream & ss);
+template void EncodeDecimalImpl<Decimal128>(const Decimal128 &, PrecType, ScaleType, std::stringstream & ss);
+template void EncodeDecimalImpl<Decimal256>(const Decimal256 &, PrecType, ScaleType, std::stringstream & ss);
 
 } // namespace DB
diff --git a/dbms/src/Storages/Transaction/Codec.h b/dbms/src/Storages/Transaction/Codec.h
index 53265ca4d20..38998fd04a0 100644
--- a/dbms/src/Storages/Transaction/Codec.h
+++ b/dbms/src/Storages/Transaction/Codec.h
@@ -62,8 +62,7 @@ void EncodeVarUInt(UInt64 num, std::stringstream & ss);
 
 void EncodeVarInt(Int64 num, std::stringstream & ss);
 
-template <typename T>
-void EncodeDecimal(const T & dec, PrecType prec, ScaleType frac, std::stringstream & ss);
+void EncodeDecimal(const Field & field, std::stringstream & ss);
 
 void EncodeDatum(const Field & field, TiDB::CodecFlag flag, std::stringstream & ss);
 
diff --git a/dbms/src/Storages/Transaction/Datum.cpp b/dbms/src/Storages/Transaction/Datum.cpp
index 8dc4b357c67..72301b9fa9b 100644
--- a/dbms/src/Storages/Transaction/Datum.cpp
+++ b/dbms/src/Storages/Transaction/Datum.cpp
@@ -135,6 +135,17 @@ struct DatumOp<tp, typename std::enable_if<tp == TypeTiny || tp == TypeShort ||
     }
 };
 
+/// Specialized for Enum, using unflatten/flatten to transform UInt to Int back and forth.
+template <TP tp>
+struct DatumOp<tp, typename std::enable_if<tp == TypeEnum>::type>
+{
+    static void unflatten(const Field & orig, std::optional<Field> & copy) { copy = static_cast<Int64>(orig.get<UInt64>()); }
+
+    static void flatten(const Field & orig, std::optional<Field> & copy) { copy = static_cast<UInt64>(orig.get<Int64>()); }
+
+    static bool overflow(const Field &, const ColumnInfo &) { return false; }
+};
+
 DatumFlat::DatumFlat(const DB::Field & field, TP tp) : DatumBase(field, tp)
 {
     switch (tp)
diff --git a/dbms/src/Storages/Transaction/MyTimeParser.cpp b/dbms/src/Storages/Transaction/MyTimeParser.cpp
new file mode 100644
index 00000000000..e183e587ceb
--- /dev/null
+++ b/dbms/src/Storages/Transaction/MyTimeParser.cpp
@@ -0,0 +1,172 @@
+#include <Storages/Transaction/MyTimeParser.h>
+
+#include <common/DateLUT.h>
+#include <cctype>
+#include <initializer_list>
+#include <vector>
+
+#include <Poco/StringTokenizer.h>
+#include <Storages/Transaction/TiDB.h>
+
+namespace DB
+{
+
+int adjustYear(int year)
+{
+    if (year >= 0 && year <= 69)
+        return 2000 + year;
+    if (year >= 70 && year <= 99)
+        return 1900 + year;
+    return year;
+}
+
+void scanTimeArgs(const std::vector<String> & seps, std::initializer_list<int *> && list)
+{
+    int i = 0;
+    for (auto * ptr : list)
+    {
+        *ptr = std::stoi(seps[i]);
+        i++;
+    }
+}
+
+int getFracIndex(const String & format)
+{
+    int idx = -1;
+    for (int i = int(format.size()) - 1; i >= 0; i--)
+    {
+        if (std::ispunct(format[i]))
+        {
+            if (format[i] == '.')
+            {
+                idx = i;
+            }
+            break;
+        }
+    }
+    return idx;
+}
+
+std::vector<String> parseDateFormat(String format)
+{
+    format = Poco::trimInPlace(format);
+
+    std::vector<String> seps;
+    size_t start = 0;
+    for (size_t i = 0; i < format.size(); i++)
+    {
+        if (i == 0 || i + 1 == format.size())
+        {
+            if (!std::isdigit(format[i]))
+                return {};
+            continue;
+        }
+
+        if (!std::isdigit(format[i]))
+        {
+            if (!std::isdigit(format[i - 1]))
+                return {};
+            seps.push_back(format.substr(start, i - start));
+            start = i + 1;
+        }
+    }
+    seps.push_back(format.substr(start));
+    return seps;
+}
+
+std::vector<String> splitDatetime(String format)
+{
+    int idx = getFracIndex(format);
+    if (idx > 0)
+    {
+        format = format.substr(0, idx);
+    }
+    return parseDateFormat(format);
+}
+
+Field parseMyDatetime(const String & str)
+{
+    Int32 year = 0, month = 0, day = 0, hour = 0, minute = 0, second = 0;
+
+    const auto & seps = splitDatetime(str);
+
+    switch (seps.size())
+    {
+        // No delimiter
+        case 1:
+        {
+            size_t l = seps[0].size();
+            switch (l)
+            {
+                case 14:
+                    // YYYYMMDDHHMMSS
+                    {
+                        std::sscanf(seps[0].c_str(), "%4d%2d%2d%2d%2d%2d", &year, &month, &day, &hour, &minute, &second);
+                        break;
+                    }
+                case 12:
+                {
+                    std::sscanf(seps[0].c_str(), "%2d%2d%2d%2d%2d%2d", &year, &month, &day, &hour, &minute, &second);
+                    year = adjustYear(year);
+                    break;
+                }
+                case 11:
+                {
+                    std::sscanf(seps[0].c_str(), "%2d%2d%2d%2d%2d%1d", &year, &month, &day, &hour, &minute, &second);
+                    year = adjustYear(year);
+                    break;
+                }
+                case 10:
+                {
+                    std::sscanf(seps[0].c_str(), "%2d%2d%2d%2d%2d", &year, &month, &day, &hour, &minute);
+                    year = adjustYear(year);
+                    break;
+                }
+                case 9:
+                {
+                    std::sscanf(seps[0].c_str(), "%2d%2d%2d%2d%1d", &year, &month, &day, &hour, &minute);
+                    year = adjustYear(year);
+                    break;
+                }
+                case 8:
+                {
+                    std::sscanf(seps[0].c_str(), "%4d%2d%2d", &year, &month, &day);
+                    break;
+                }
+                case 6:
+                case 5:
+                {
+                    std::sscanf(seps[0].c_str(), "%2d%2d%2d", &year, &month, &day);
+                    year = adjustYear(year);
+                    break;
+                }
+                default:
+                {
+                    throw Exception("Wrong datetime format");
+                }
+                    // TODO Process frac!
+            }
+            break;
+        }
+        case 3:
+        {
+            scanTimeArgs(seps, {&year, &month, &day});
+            break;
+        }
+        case 6:
+        {
+            scanTimeArgs(seps, {&year, &month, &day, &hour, &minute, &second});
+            break;
+        }
+        default:
+        {
+            throw Exception("Wrong datetime format");
+        }
+    }
+
+    UInt64 ymd = ((year * 13 + month) << 5) | day;
+    UInt64 hms = (hour << 12) | (minute << 6) | second;
+    return Field((ymd << 17 | hms) << 24);
+}
+
+} // namespace DB
diff --git a/dbms/src/Storages/Transaction/MyTimeParser.h b/dbms/src/Storages/Transaction/MyTimeParser.h
index 009bba29390..d58a4db258c 100644
--- a/dbms/src/Storages/Transaction/MyTimeParser.h
+++ b/dbms/src/Storages/Transaction/MyTimeParser.h
@@ -2,173 +2,10 @@
 
 #include <Core/Field.h>
 #include <Core/Types.h>
-#include <common/DateLUT.h>
-#include <cctype>
-#include <initializer_list>
-#include <vector>
-
-#include <Storages/Transaction/TiDB.h>
-#include <Poco/StringTokenizer.h>
 
 namespace DB
 {
 
-int adjustYear(int year)
-{
-    if (year >= 0 && year <= 69)
-        return 2000 + year;
-    if (year >= 70 && year <= 99)
-        return 1900 + year;
-    return year;
-}
-
-void scanTimeArgs(const std::vector<String> & seps, std::initializer_list<int *> && list)
-{
-    int i = 0;
-    for (auto * ptr : list)
-    {
-        *ptr = std::stoi(seps[i]);
-        i++;
-    }
-}
-
-int getFracIndex(const String & format)
-{
-    int idx = -1;
-    for (int i = int(format.size()) - 1; i >= 0; i--)
-    {
-        if (std::ispunct(format[i]))
-        {
-            if (format[i] == '.')
-            {
-                idx = i;
-            }
-            break;
-        }
-    }
-    return idx;
-}
-
-std::vector<String> parseDateFormat(String format)
-{
-    format = Poco::trimInPlace(format);
-
-    std::vector<String> seps;
-    size_t start = 0;
-    for (size_t i = 0; i < format.size(); i++)
-    {
-        if (i == 0 || i + 1 == format.size())
-        {
-            if (!std::isdigit(format[i]))
-                return {};
-            continue;
-        }
-
-        if (!std::isdigit(format[i]))
-        {
-            if (!std::isdigit(format[i - 1]))
-                return {};
-            seps.push_back(format.substr(start, i - start));
-            start = i + 1;
-        }
-    }
-    seps.push_back(format.substr(start));
-    return seps;
-}
-
-std::vector<String> splitDatetime(String format)
-{
-    int idx = getFracIndex(format);
-    if (idx > 0)
-    {
-        format = format.substr(0, idx);
-    }
-    return parseDateFormat(format);
-}
-
-Field parseMyDatetime(const String & str)
-{
-    Int32 year = 0, month = 0, day = 0, hour = 0, minute = 0, second = 0;
-
-    const auto & seps = splitDatetime(str);
-
-    switch (seps.size())
-    {
-        // No delimiter
-        case 1:
-        {
-            size_t l = seps[0].size();
-            switch (l)
-            {
-                case 14:
-                    // YYYYMMDDHHMMSS
-                    {
-                        std::sscanf(seps[0].c_str(), "%4d%2d%2d%2d%2d%2d", &year, &month, &day, &hour, &minute, &second);
-                        break;
-                    }
-                case 12:
-                {
-                    std::sscanf(seps[0].c_str(), "%2d%2d%2d%2d%2d%2d", &year, &month, &day, &hour, &minute, &second);
-                    year = adjustYear(year);
-                    break;
-                }
-                case 11:
-                {
-                    std::sscanf(seps[0].c_str(), "%2d%2d%2d%2d%2d%1d", &year, &month, &day, &hour, &minute, &second);
-                    year = adjustYear(year);
-                    break;
-                }
-                case 10:
-                {
-                    std::sscanf(seps[0].c_str(), "%2d%2d%2d%2d%2d", &year, &month, &day, &hour, &minute);
-                    year = adjustYear(year);
-                    break;
-                }
-                case 9:
-                {
-                    std::sscanf(seps[0].c_str(), "%2d%2d%2d%2d%1d", &year, &month, &day, &hour, &minute);
-                    year = adjustYear(year);
-                    break;
-                }
-                case 8:
-                {
-                    std::sscanf(seps[0].c_str(), "%4d%2d%2d", &year, &month, &day);
-                    break;
-                }
-                case 6:
-                case 5:
-                {
-                    std::sscanf(seps[0].c_str(), "%2d%2d%2d", &year, &month, &day);
-                    year = adjustYear(year);
-                    break;
-                }
-                default:
-                {
-                    throw Exception("Wrong datetime format");
-                }
-                    // TODO Process frac!
-            }
-            break;
-        }
-        case 3:
-        {
-            scanTimeArgs(seps, {&year, &month, &day});
-            break;
-        }
-        case 6:
-        {
-            scanTimeArgs(seps, {&year, &month, &day, &hour, &minute, &second});
-            break;
-        }
-        default:
-        {
-            throw Exception("Wrong datetime format");
-        }
-    }
-
-    UInt64 ymd = ((year * 13 + month) << 5) | day;
-    UInt64 hms = (hour << 12) | (minute << 6) | second;
-    return Field((ymd << 17 | hms) << 24);
-}
+Field parseMyDatetime(const String & str);
 
 } // namespace DB
diff --git a/dbms/src/Storages/Transaction/TiDB.cpp b/dbms/src/Storages/Transaction/TiDB.cpp
index f1bd1d52d3e..bbdca6b2877 100644
--- a/dbms/src/Storages/Transaction/TiDB.cpp
+++ b/dbms/src/Storages/Transaction/TiDB.cpp
@@ -25,6 +25,7 @@ Field ColumnInfo::defaultValueToField() const
     }
     switch (tp)
     {
+        // TODO: Consider unsigned?
         // Integer Type.
         case TypeTiny:
         case TypeShort:
@@ -157,8 +158,7 @@ catch (const Poco::Exception & e)
         std::string(__PRETTY_FUNCTION__) + ": Serialize TiDB schema JSON failed (ColumnInfo): " + e.displayText(), DB::Exception(e));
 }
 
-void ColumnInfo::deserialize(Poco::JSON::Object::Ptr json)
-try
+void ColumnInfo::deserialize(Poco::JSON::Object::Ptr json) try
 {
     id = json->getValue<Int64>("id");
     name = json->getObject("name")->getValue<String>("L");
@@ -192,8 +192,7 @@ catch (const Poco::Exception & e)
 
 PartitionDefinition::PartitionDefinition(Poco::JSON::Object::Ptr json) { deserialize(json); }
 
-Poco::JSON::Object::Ptr PartitionDefinition::getJSONObject() const
-try
+Poco::JSON::Object::Ptr PartitionDefinition::getJSONObject() const try
 {
     Poco::JSON::Object::Ptr json = new Poco::JSON::Object();
     json->set("id", id);
@@ -214,8 +213,7 @@ catch (const Poco::Exception & e)
         std::string(__PRETTY_FUNCTION__) + ": Serialize TiDB schema JSON failed (PartitionDef): " + e.displayText(), DB::Exception(e));
 }
 
-void PartitionDefinition::deserialize(Poco::JSON::Object::Ptr json)
-try
+void PartitionDefinition::deserialize(Poco::JSON::Object::Ptr json) try
 {
     id = json->getValue<Int64>("id");
     name = json->getObject("name")->getValue<String>("L");
@@ -230,8 +228,7 @@ catch (const Poco::Exception & e)
 
 PartitionInfo::PartitionInfo(Poco::JSON::Object::Ptr json) { deserialize(json); }
 
-Poco::JSON::Object::Ptr PartitionInfo::getJSONObject() const
-try
+Poco::JSON::Object::Ptr PartitionInfo::getJSONObject() const try
 {
     Poco::JSON::Object::Ptr json = new Poco::JSON::Object();
 
@@ -260,8 +257,7 @@ catch (const Poco::Exception & e)
         std::string(__PRETTY_FUNCTION__) + ": Serialize TiDB schema JSON failed (PartitionInfo): " + e.displayText(), DB::Exception(e));
 }
 
-void PartitionInfo::deserialize(Poco::JSON::Object::Ptr json)
-try
+void PartitionInfo::deserialize(Poco::JSON::Object::Ptr json) try
 {
     type = static_cast<PartitionType>(json->getValue<Int32>("type"));
     expr = json->getValue<String>("expr");
@@ -285,8 +281,7 @@ catch (const Poco::Exception & e)
 
 TableInfo::TableInfo(const String & table_info_json) { deserialize(table_info_json); }
 
-String TableInfo::serialize(bool escaped) const
-try
+String TableInfo::serialize(bool escaped) const try
 {
     std::stringstream buf;
 
@@ -344,8 +339,7 @@ catch (const Poco::Exception & e)
         std::string(__PRETTY_FUNCTION__) + ": Serialize TiDB schema JSON failed (TableInfo): " + e.displayText(), DB::Exception(e));
 }
 
-void DBInfo::deserialize(const String & json_str)
-try
+void DBInfo::deserialize(const String & json_str) try
 {
     Poco::JSON::Parser parser;
     Poco::Dynamic::Var result = parser.parse(json_str);
@@ -363,8 +357,7 @@ catch (const Poco::Exception & e)
         DB::Exception(e));
 }
 
-void TableInfo::deserialize(const String & json_str)
-try
+void TableInfo::deserialize(const String & json_str) try
 {
     if (json_str.empty())
     {
diff --git a/dbms/src/Storages/Transaction/TiDB.h b/dbms/src/Storages/Transaction/TiDB.h
index 128f5ddd13a..efa0e8af5ed 100644
--- a/dbms/src/Storages/Transaction/TiDB.h
+++ b/dbms/src/Storages/Transaction/TiDB.h
@@ -179,12 +179,13 @@ struct ColumnInfo
     COLUMN_FLAGS(M)
 #undef M
 
+    DB::Field convertField(const DB::Field &) const;
     DB::Field defaultValueToField() const;
+    Int64 getEnumIndex(const String &) const;
     CodecFlag getCodecFlag() const;
 
 private:
     DB::Field getDecimalDefaultValue(const String & str) const;
-    Int64 getEnumIndex(const String &) const;
 };
 
 enum PartitionType
diff --git a/dbms/src/Storages/Transaction/TiKVRecordFormat.h b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
index 12b779be25e..5d27978f2f2 100644
--- a/dbms/src/Storages/Transaction/TiKVRecordFormat.h
+++ b/dbms/src/Storages/Transaction/TiKVRecordFormat.h
@@ -57,21 +57,6 @@ inline UInt64 decodeUInt64Desc(const UInt64 x) { return ~decodeUInt64(x); }
 
 inline Int64 decodeInt64(const UInt64 x) { return static_cast<Int64>(decodeUInt64(x) ^ SIGN_MASK); }
 
-inline TiKVValue EncodeRow(const TiDB::TableInfo & table_info, const std::vector<Field> & fields)
-{
-    if (table_info.columns.size() != fields.size())
-        throw Exception("Encoding row has different sizes between columns and values", ErrorCodes::LOGICAL_ERROR);
-    std::stringstream ss;
-    for (size_t i = 0; i < fields.size(); i++)
-    {
-        const TiDB::ColumnInfo & column_info = table_info.columns[i];
-        EncodeDatum(Field(column_info.id), TiDB::CodecFlagInt, ss);
-        TiDB::DatumBumpy datum = TiDB::DatumBumpy(fields[i], column_info.tp);
-        EncodeDatum(datum.field(), column_info.getCodecFlag(), ss);
-    }
-    return TiKVValue(ss.str());
-}
-
 template <typename T>
 inline T read(const char * s)
 {