diff --git a/go.mod b/go.mod
index 01ebb6a..2554032 100644
--- a/go.mod
+++ b/go.mod
@@ -16,7 +16,6 @@ require (
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang-module/carbon v1.6.6
 	github.com/golang/mock v1.6.0
-	github.com/golang/protobuf v1.5.2
 	github.com/google/go-cmp v0.5.7
 	github.com/natefinch/lumberjack v2.0.0+incompatible
 	github.com/patrickmn/go-cache v2.1.0+incompatible
@@ -54,4 +53,5 @@ require (
 	github.com/tmc/grpc-websocket-proxy v0.0.0-20201229170055-e5319fda7802 // indirect
 	golang.org/x/sys v0.0.0-20220519141025-dcacdad47464 // indirect
 	golang.org/x/tools v0.1.10 // indirect
+	google.golang.org/protobuf v1.27.1
 )
diff --git a/pkg/dt/mysql_undo_executor.go b/pkg/dt/mysql_undo_executor.go
index 939a1c6..d1b9869 100644
--- a/pkg/dt/mysql_undo_executor.go
+++ b/pkg/dt/mysql_undo_executor.go
@@ -43,7 +43,7 @@ const (
 
 type BuildUndoSql func(undoLog undolog.SqlUndoLog) string
 
-func DeleteBuildUndoSql(undoLog *undolog.SqlUndoLog) string {
+func BuildDeleteUndoSql(undoLog *undolog.SqlUndoLog) string {
 	beforeImage := undoLog.BeforeImage
 	beforeImageRows := beforeImage.Rows
 
@@ -60,11 +60,11 @@ func DeleteBuildUndoSql(undoLog *undolog.SqlUndoLog) string {
 	var sbCols, sbVals strings.Builder
 	var size = len(fields)
 	for i, field := range fields {
-		fmt.Fprintf(&sbCols, "`%s`", field.Name)
-		fmt.Fprint(&sbVals, "?")
+		sbCols.WriteString(fmt.Sprintf("`%s`", field.Name))
+		sbVals.WriteByte('?')
 		if i < size-1 {
-			fmt.Fprint(&sbCols, ", ")
-			fmt.Fprint(&sbVals, ", ")
+			sbCols.WriteString(", ")
+			sbVals.WriteString(", ")
 		}
 	}
 	insertColumns := sbCols.String()
@@ -73,7 +73,7 @@ func DeleteBuildUndoSql(undoLog *undolog.SqlUndoLog) string {
 	return fmt.Sprintf(InsertSqlTemplate, undoLog.TableName, insertColumns, insertValues)
 }
 
-func InsertBuildUndoSql(undoLog *undolog.SqlUndoLog) string {
+func BuildInsertUndoSql(undoLog *undolog.SqlUndoLog) string {
 	afterImage := undoLog.AfterImage
 	afterImageRows := afterImage.Rows
 	if len(afterImageRows) == 0 {
@@ -84,7 +84,7 @@ func InsertBuildUndoSql(undoLog *undolog.SqlUndoLog) string {
 	return fmt.Sprintf(DeleteSqlTemplate, undoLog.TableName, pkField.Name)
 }
 
-func UpdateBuildUndoSql(undoLog *undolog.SqlUndoLog) string {
+func BuildUpdateUndoSql(undoLog *undolog.SqlUndoLog) string {
 	beforeImage := undoLog.BeforeImage
 	beforeImageRows := beforeImage.Rows
 
@@ -135,15 +135,15 @@ func (executor MysqlUndoExecutor) Execute(tx proto.Tx) error {
 	// DELETE FROM a WHERE pk = ?
 	switch executor.sqlUndoLog.SqlType {
 	case constant.SQLType_INSERT:
-		undoSql = InsertBuildUndoSql(executor.sqlUndoLog)
+		undoSql = BuildInsertUndoSql(executor.sqlUndoLog)
 		undoRows = *executor.sqlUndoLog.AfterImage
 
 	case constant.SQLType_DELETE:
-		undoSql = DeleteBuildUndoSql(executor.sqlUndoLog)
+		undoSql = BuildDeleteUndoSql(executor.sqlUndoLog)
 		undoRows = *executor.sqlUndoLog.BeforeImage
 
 	case constant.SQLType_UPDATE:
-		undoSql = UpdateBuildUndoSql(executor.sqlUndoLog)
+		undoSql = BuildUpdateUndoSql(executor.sqlUndoLog)
 		undoRows = *executor.sqlUndoLog.BeforeImage
 
 	default:
@@ -224,25 +224,64 @@ func (executor MysqlUndoExecutor) queryCurrentRecords(tx proto.Tx) (*schema.Tabl
 		pkValues = append(pkValues, field.Value)
 	}
 
+	if executor.sqlUndoLog.IsBinary {
+		selectSql := executor.buildCurrentRecordsForPrepareSql(tableMeta, pkName, pkValues)
+		dataTable, _, err := tx.ExecuteSql(context.Background(), selectSql, pkValues...)
+		if err != nil {
+			return nil, err
+		}
+		dt := dataTable.(*mysql.Result)
+		return schema.BuildBinaryRecords(tableMeta, dt), nil
+	} else {
+		selectSql := executor.buildCurrentRecordsForQuerySql(tableMeta, pkName, pkValues)
+		dataTable, _, err := tx.Query(context.Background(), selectSql)
+		if err != nil {
+			return nil, err
+		}
+		dt := dataTable.(*mysql.Result)
+		return schema.BuildTextRecords(tableMeta, dt), nil
+	}
+}
+
+func (executor MysqlUndoExecutor) buildCurrentRecordsForPrepareSql(tableMeta schema.TableMeta, pkColumn string, pkValues []interface{}) string {
 	var b strings.Builder
 	var i = 0
 	columnCount := len(tableMeta.Columns)
 	for _, columnName := range tableMeta.Columns {
-		fmt.Fprint(&b, misc.CheckAndReplace(columnName))
+		b.WriteString(misc.CheckAndReplace(columnName))
 		i = i + 1
 		if i < columnCount {
-			fmt.Fprint(&b, ",")
+			b.WriteByte(',')
 		} else {
-			fmt.Fprint(&b, " ")
+			b.WriteByte(' ')
 		}
 	}
-
 	inCondition := misc.MysqlAppendInParam(len(pkValues))
-	selectSql := fmt.Sprintf(SelectSqlTemplate, b.String(), tableMeta.TableName, pkName, inCondition)
-	dataTable, _, err := tx.ExecuteSql(context.Background(), selectSql, pkValues...)
-	if err != nil {
-		return nil, err
+	return fmt.Sprintf(SelectSqlTemplate, b.String(), tableMeta.TableName, pkColumn, inCondition)
+}
+
+func (executor MysqlUndoExecutor) buildCurrentRecordsForQuerySql(tableMeta schema.TableMeta, pkColumn string, pkValues []interface{}) string {
+	var columns strings.Builder
+	var inCondition strings.Builder
+	var i = 0
+	columnCount := len(tableMeta.Columns)
+	for _, columnName := range tableMeta.Columns {
+		columns.WriteString(misc.CheckAndReplace(columnName))
+		i = i + 1
+		if i < columnCount {
+			columns.WriteByte(',')
+		} else {
+			columns.WriteByte(' ')
+		}
+	}
+	inCondition.WriteByte('(')
+	for i, pk := range pkValues {
+		if i < len(pkValues)-1 {
+			inCondition.WriteString(fmt.Sprintf("'%s',", pk))
+		} else {
+			inCondition.WriteString(fmt.Sprintf("'%s'", pk))
+		}
 	}
-	dt := dataTable.(*mysql.Result)
-	return schema.BuildRecords(tableMeta, dt), nil
+	inCondition.WriteByte(')')
+	return fmt.Sprintf(SelectSqlTemplate, columns.String(), tableMeta.TableName, pkColumn, inCondition.String())
 }
diff --git a/pkg/dt/schema/table_records.go b/pkg/dt/schema/table_records.go
index 9cce78e..bc062b0 100644
--- a/pkg/dt/schema/table_records.go
+++ b/pkg/dt/schema/table_records.go
@@ -63,20 +63,20 @@ func BuildLockKey(lockKeyRecords *TableRecords) string {
 	}
 
 	var sb strings.Builder
-	fmt.Fprintf(&sb, lockKeyRecords.TableName)
-	fmt.Fprint(&sb, ":")
+	sb.WriteString(lockKeyRecords.TableName)
+	sb.WriteByte(':')
 	fields := lockKeyRecords.PKFields()
 	length := len(fields)
 	for i, field := range fields {
-		fmt.Fprint(&sb, field.Value)
+		sb.WriteString(fmt.Sprintf("%s", field.Value))
 		if i < length-1 {
-			fmt.Fprint(&sb, ",")
+			sb.WriteByte(',')
 		}
 	}
 	return sb.String()
 }
 
-func BuildRecords(meta TableMeta, result *mysql.Result) *TableRecords {
+func BuildBinaryRecords(meta TableMeta, result *mysql.Result) *TableRecords {
 	records := NewTableRecords(meta)
 	rs := make([]*Row, 0)
 
@@ -114,3 +114,42 @@ func BuildRecords(meta TableMeta, result *mysql.Result) *TableRecords {
 	records.Rows = rs
 	return records
 }
+
+func BuildTextRecords(meta TableMeta, result *mysql.Result) *TableRecords {
+	records := NewTableRecords(meta)
+	rs := make([]*Row, 0)
+
+	for {
+		row, err := result.Rows.Next()
+		if err != nil {
+			break
+		}
+
+		textRow := mysql.TextRow{Row: row}
+		values, err := textRow.Decode()
+		if err != nil {
+			break
+		}
+		fields := make([]*Field, 0, len(result.Fields))
+		for i, col := range result.Fields {
+			field := &Field{
+				Name: col.FiledName(),
+				Type: meta.AllColumns[col.FiledName()].DataType,
+			}
+			if values[i] != nil {
+				field.Value = values[i].Val
+			}
+			if strings.EqualFold(col.FiledName(), meta.GetPKName()) {
+				field.KeyType = PrimaryKey
+			}
+			fields = append(fields, field)
+		}
+		r := &Row{Fields: fields}
+		rs = append(rs, r)
+	}
+	if len(rs) == 0 {
+		return nil
+	}
+	records.Rows = rs
+	return records
+}
diff --git a/pkg/dt/undolog/protobuf_undo_log_parser.go b/pkg/dt/undolog/protobuf_undo_log_parser.go
index e6d8742..7c83509 100644
--- a/pkg/dt/undolog/protobuf_undo_log_parser.go
+++ b/pkg/dt/undolog/protobuf_undo_log_parser.go
@@ -23,7 +23,7 @@ import (
 	"reflect"
 	"time"
 
-	"github.com/golang/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 	"vimagination.zapto.org/byteio"
 
 	"github.com/cectc/dbpack/pkg/constant"
@@ -223,6 +223,7 @@ func convertPbTableRecords(pbRecords *PbTableRecords) *schema.TableRecords {
 
 func convertSqlUndoLog(undoLog *SqlUndoLog) *PbSqlUndoLog {
 	pbSqlUndoLog := &PbSqlUndoLog{
+		IsBinary:   undoLog.IsBinary,
 		SqlType:    int32(undoLog.SqlType),
 		SchemaName: undoLog.SchemaName,
 		TableName:  undoLog.TableName,
@@ -242,6 +243,7 @@ func convertSqlUndoLog(undoLog *SqlUndoLog) *PbSqlUndoLog {
 
 func convertPbSqlUndoLog(pbSqlUndoLog *PbSqlUndoLog) *SqlUndoLog {
 	sqlUndoLog := &SqlUndoLog{
+		IsBinary:   pbSqlUndoLog.IsBinary,
 		SqlType:    constant.SQLType(pbSqlUndoLog.SqlType),
 		SchemaName: pbSqlUndoLog.SchemaName,
 		TableName:  pbSqlUndoLog.TableName,
diff --git a/pkg/dt/undolog/protobuf_undo_log_parser_test.go b/pkg/dt/undolog/protobuf_undo_log_parser_test.go
index a5cb362..348ea4a 100644
--- a/pkg/dt/undolog/protobuf_undo_log_parser_test.go
+++ b/pkg/dt/undolog/protobuf_undo_log_parser_test.go
@@ -32,6 +32,7 @@ func getBranchUndoLog() *BranchUndoLog {
 		BranchID: 2000042936,
 		SqlUndoLogs: []*SqlUndoLog{
 			{
+				IsBinary:    true,
 				SqlType:     constant.SQLType_INSERT,
 				TableName:   "user",
 				BeforeImage: nil,
diff --git a/pkg/dt/undolog/table_records.pb.go b/pkg/dt/undolog/table_records.pb.go
index b166e36..d09b9be 100644
--- a/pkg/dt/undolog/table_records.pb.go
+++ b/pkg/dt/undolog/table_records.pb.go
@@ -1,212 +1,323 @@
-/*
- * Copyright 2022 CECTC, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+//
+// Copyright 2022 CECTC, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
 
 // Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.27.1
+// 	protoc        v3.11.4
 // source: table_records.proto
 
 package undolog
 
 import (
-	fmt "fmt"
-	math "math"
-
-	proto "github.com/golang/protobuf/proto"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
 )
 
-// Reference imports to suppress errors if they are not otherwise used.
-var _ = proto.Marshal
-var _ = fmt.Errorf
-var _ = math.Inf
-
-// This is a compile-time assertion to ensure that this generated file
-// is compatible with the proto package it is being compiled against.
-// A compilation error at this line likely means your copy of the
-// proto package needs to be updated.
-const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
 
 type PbField struct {
-	Name                 string   `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"`
-	KeyType              int32    `protobuf:"varint,2,opt,name=KeyType,proto3" json:"KeyType,omitempty"`
-	Type                 int32    `protobuf:"zigzag32,3,opt,name=Type,proto3" json:"Type,omitempty"`
-	Value                []byte   `protobuf:"bytes,4,opt,name=Value,proto3" json:"Value,omitempty"`
-	XXX_NoUnkeyedLiteral struct{} `json:"-"`
-	XXX_unrecognized     []byte   `json:"-"`
-	XXX_sizecache        int32    `json:"-"`
-}
-
-func (m *PbField) Reset()         { *m = PbField{} }
-func (m *PbField) String() string { return proto.CompactTextString(m) }
-func (*PbField) ProtoMessage()    {}
-func (*PbField) Descriptor() ([]byte, []int) {
-	return fileDescriptor_0f74341eba9e488d, []int{0}
-}
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
 
-func (m *PbField) XXX_Unmarshal(b []byte) error {
-	return xxx_messageInfo_PbField.Unmarshal(m, b)
-}
-func (m *PbField) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	return xxx_messageInfo_PbField.Marshal(b, m, deterministic)
+	Name    string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"`
+	KeyType int32  `protobuf:"varint,2,opt,name=KeyType,proto3" json:"KeyType,omitempty"`
+	Type    int32  `protobuf:"zigzag32,3,opt,name=Type,proto3" json:"Type,omitempty"`
+	Value   []byte `protobuf:"bytes,4,opt,name=Value,proto3" json:"Value,omitempty"`
 }
-func (m *PbField) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_PbField.Merge(m, src)
+
+func (x *PbField) Reset() {
+	*x = PbField{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_table_records_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
-func (m *PbField) XXX_Size() int {
-	return xxx_messageInfo_PbField.Size(m)
+
+func (x *PbField) String() string {
+	return protoimpl.X.MessageStringOf(x)
 }
-func (m *PbField) XXX_DiscardUnknown() {
-	xxx_messageInfo_PbField.DiscardUnknown(m)
+
+func (*PbField) ProtoMessage() {}
+
+func (x *PbField) ProtoReflect() protoreflect.Message {
+	mi := &file_table_records_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
 }
 
-var xxx_messageInfo_PbField proto.InternalMessageInfo
+// Deprecated: Use PbField.ProtoReflect.Descriptor instead.
+func (*PbField) Descriptor() ([]byte, []int) {
+	return file_table_records_proto_rawDescGZIP(), []int{0}
+}
 
-func (m *PbField) GetName() string {
-	if m != nil {
-		return m.Name
+func (x *PbField) GetName() string {
+	if x != nil {
+		return x.Name
 	}
 	return ""
 }
 
-func (m *PbField) GetKeyType() int32 {
-	if m != nil {
-		return m.KeyType
+func (x *PbField) GetKeyType() int32 {
+	if x != nil {
+		return x.KeyType
 	}
 	return 0
 }
 
-func (m *PbField) GetType() int32 {
-	if m != nil {
-		return m.Type
+func (x *PbField) GetType() int32 {
+	if x != nil {
+		return x.Type
 	}
 	return 0
 }
 
-func (m *PbField) GetValue() []byte {
-	if m != nil {
-		return m.Value
+func (x *PbField) GetValue() []byte {
+	if x != nil {
+		return x.Value
 	}
 	return nil
 }
 
 type PbRow struct {
-	Fields               []*PbField `protobuf:"bytes,1,rep,name=Fields,proto3" json:"Fields,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}   `json:"-"`
-	XXX_unrecognized     []byte     `json:"-"`
-	XXX_sizecache        int32      `json:"-"`
-}
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
 
-func (m *PbRow) Reset()         { *m = PbRow{} }
-func (m *PbRow) String() string { return proto.CompactTextString(m) }
-func (*PbRow) ProtoMessage()    {}
-func (*PbRow) Descriptor() ([]byte, []int) {
-	return fileDescriptor_0f74341eba9e488d, []int{1}
+	Fields []*PbField `protobuf:"bytes,1,rep,name=Fields,proto3" json:"Fields,omitempty"`
 }
 
-func (m *PbRow) XXX_Unmarshal(b []byte) error {
-	return xxx_messageInfo_PbRow.Unmarshal(m, b)
-}
-func (m *PbRow) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	return xxx_messageInfo_PbRow.Marshal(b, m, deterministic)
-}
-func (m *PbRow) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_PbRow.Merge(m, src)
+func (x *PbRow) Reset() {
+	*x = PbRow{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_table_records_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
-func (m *PbRow) XXX_Size() int {
-	return xxx_messageInfo_PbRow.Size(m)
+
+func (x *PbRow) String() string {
+	return protoimpl.X.MessageStringOf(x)
 }
-func (m *PbRow) XXX_DiscardUnknown() {
-	xxx_messageInfo_PbRow.DiscardUnknown(m)
+
+func (*PbRow) ProtoMessage() {}
+
+func (x *PbRow) ProtoReflect() protoreflect.Message {
+	mi := &file_table_records_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
 }
 
-var xxx_messageInfo_PbRow proto.InternalMessageInfo
+// Deprecated: Use PbRow.ProtoReflect.Descriptor instead.
+func (*PbRow) Descriptor() ([]byte, []int) {
+	return file_table_records_proto_rawDescGZIP(), []int{1}
+}
 
-func (m *PbRow) GetFields() []*PbField {
-	if m != nil {
-		return m.Fields
+func (x *PbRow) GetFields() []*PbField {
+	if x != nil {
+		return x.Fields
 	}
 	return nil
 }
 
 type PbTableRecords struct {
-	TableName            string   `protobuf:"bytes,1,opt,name=TableName,proto3" json:"TableName,omitempty"`
-	Rows                 []*PbRow `protobuf:"bytes,2,rep,name=Rows,proto3" json:"Rows,omitempty"`
-	XXX_NoUnkeyedLiteral struct{} `json:"-"`
-	XXX_unrecognized     []byte   `json:"-"`
-	XXX_sizecache        int32    `json:"-"`
-}
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
 
-func (m *PbTableRecords) Reset()         { *m = PbTableRecords{} }
-func (m *PbTableRecords) String() string { return proto.CompactTextString(m) }
-func (*PbTableRecords) ProtoMessage()    {}
-func (*PbTableRecords) Descriptor() ([]byte, []int) {
-	return fileDescriptor_0f74341eba9e488d, []int{2}
+	TableName string   `protobuf:"bytes,1,opt,name=TableName,proto3" json:"TableName,omitempty"`
+	Rows      []*PbRow `protobuf:"bytes,2,rep,name=Rows,proto3" json:"Rows,omitempty"`
 }
 
-func (m *PbTableRecords) XXX_Unmarshal(b []byte) error {
-	return xxx_messageInfo_PbTableRecords.Unmarshal(m, b)
-}
-func (m *PbTableRecords) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	return xxx_messageInfo_PbTableRecords.Marshal(b, m, deterministic)
-}
-func (m *PbTableRecords) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_PbTableRecords.Merge(m, src)
+func (x *PbTableRecords) Reset() {
+	*x = PbTableRecords{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_table_records_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
-func (m *PbTableRecords) XXX_Size() int {
-	return xxx_messageInfo_PbTableRecords.Size(m)
+
+func (x *PbTableRecords) String() string {
+	return protoimpl.X.MessageStringOf(x)
 }
-func (m *PbTableRecords) XXX_DiscardUnknown() {
-	xxx_messageInfo_PbTableRecords.DiscardUnknown(m)
+
+func (*PbTableRecords) ProtoMessage() {}
+
+func (x *PbTableRecords) ProtoReflect() protoreflect.Message {
+	mi := &file_table_records_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
 }
 
-var xxx_messageInfo_PbTableRecords proto.InternalMessageInfo
+// Deprecated: Use PbTableRecords.ProtoReflect.Descriptor instead.
+func (*PbTableRecords) Descriptor() ([]byte, []int) {
+	return file_table_records_proto_rawDescGZIP(), []int{2}
+}
 
-func (m *PbTableRecords) GetTableName() string {
-	if m != nil {
-		return m.TableName
+func (x *PbTableRecords) GetTableName() string {
+	if x != nil {
+		return x.TableName
 	}
 	return ""
 }
 
-func (m *PbTableRecords) GetRows() []*PbRow {
-	if m != nil {
-		return m.Rows
+func (x *PbTableRecords) GetRows() []*PbRow {
+	if x != nil {
+		return x.Rows
 	}
 	return nil
 }
 
-func init() {
-	proto.RegisterType((*PbField)(nil), "parser.PbField")
-	proto.RegisterType((*PbRow)(nil), "parser.PbRow")
-	proto.RegisterType((*PbTableRecords)(nil), "parser.PbTableRecords")
-}
-
-func init() { proto.RegisterFile("table_records.proto", fileDescriptor_0f74341eba9e488d) }
-
-var fileDescriptor_0f74341eba9e488d = []byte{
-	// 208 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x4c, 0x8f, 0x4d, 0x4b, 0xc4, 0x30,
-	0x10, 0x86, 0xc9, 0x6e, 0xdb, 0x65, 0xc7, 0x2f, 0x1c, 0x3d, 0xe4, 0xe0, 0x21, 0xf6, 0x62, 0x4e,
-	0x45, 0xf4, 0x3f, 0x78, 0x11, 0xa4, 0x0e, 0xc5, 0xab, 0x24, 0x76, 0x0e, 0x42, 0x25, 0x25, 0xa9,
-	0x94, 0xfe, 0x7b, 0xe9, 0xb4, 0xd2, 0xbd, 0xbd, 0x1f, 0xe1, 0xcd, 0x33, 0x70, 0x33, 0x38, 0xdf,
-	0xf1, 0x67, 0xe4, 0xaf, 0x10, 0xdb, 0x54, 0xf5, 0x31, 0x0c, 0x01, 0x8b, 0xde, 0xc5, 0xc4, 0xb1,
-	0x74, 0x70, 0xa8, 0xfd, 0xcb, 0x37, 0x77, 0x2d, 0x22, 0x64, 0x6f, 0xee, 0x87, 0xb5, 0x32, 0xca,
-	0x1e, 0x49, 0x34, 0x6a, 0x38, 0xbc, 0xf2, 0xd4, 0x4c, 0x3d, 0xeb, 0x9d, 0x51, 0x36, 0xa7, 0x7f,
-	0x3b, 0xbf, 0x96, 0x78, 0x6f, 0x94, 0xbd, 0x26, 0xd1, 0x78, 0x0b, 0xf9, 0x87, 0xeb, 0x7e, 0x59,
-	0x67, 0x46, 0xd9, 0x73, 0x5a, 0x4c, 0xf9, 0x08, 0x79, 0xed, 0x29, 0x8c, 0xf8, 0x00, 0x85, 0xfc,
-	0x94, 0xb4, 0x32, 0x7b, 0x7b, 0xf6, 0x74, 0x55, 0x2d, 0x10, 0xd5, 0x4a, 0x40, 0x6b, 0x5d, 0xbe,
-	0xc3, 0x65, 0xed, 0x9b, 0x99, 0x9a, 0x16, 0x68, 0xbc, 0x83, 0xa3, 0xf8, 0x13, 0xc0, 0x2d, 0xc0,
-	0x7b, 0xc8, 0x28, 0x8c, 0x49, 0xef, 0x64, 0xf6, 0x62, 0x9b, 0xa5, 0x30, 0x92, 0x54, 0xbe, 0x90,
-	0xb3, 0x9f, 0xff, 0x02, 0x00, 0x00, 0xff, 0xff, 0x95, 0xb3, 0xbd, 0x9d, 0x0d, 0x01, 0x00, 0x00,
+var File_table_records_proto protoreflect.FileDescriptor
+
+var file_table_records_proto_rawDesc = []byte{
+	0x0a, 0x13, 0x74, 0x61, 0x62, 0x6c, 0x65, 0x5f, 0x72, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x73, 0x2e,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x07, 0x75, 0x6e, 0x64, 0x6f, 0x6c, 0x6f, 0x67, 0x22, 0x61,
+	0x0a, 0x07, 0x50, 0x62, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x4e, 0x61, 0x6d,
+	0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a,
+	0x07, 0x4b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x07,
+	0x4b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x18,
+	0x03, 0x20, 0x01, 0x28, 0x11, 0x52, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x56,
+	0x61, 0x6c, 0x75, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x05, 0x56, 0x61, 0x6c, 0x75,
+	0x65, 0x22, 0x31, 0x0a, 0x05, 0x50, 0x62, 0x52, 0x6f, 0x77, 0x12, 0x28, 0x0a, 0x06, 0x46, 0x69,
+	0x65, 0x6c, 0x64, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x10, 0x2e, 0x75, 0x6e, 0x64,
+	0x6f, 0x6c, 0x6f, 0x67, 0x2e, 0x50, 0x62, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x06, 0x46, 0x69,
+	0x65, 0x6c, 0x64, 0x73, 0x22, 0x52, 0x0a, 0x0e, 0x50, 0x62, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x52,
+	0x65, 0x63, 0x6f, 0x72, 0x64, 0x73, 0x12, 0x1c, 0x0a, 0x09, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x4e,
+	0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x54, 0x61, 0x62, 0x6c, 0x65,
+	0x4e, 0x61, 0x6d, 0x65, 0x12, 0x22, 0x0a, 0x04, 0x52, 0x6f, 0x77, 0x73, 0x18, 0x02, 0x20, 0x03,
+	0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x75, 0x6e, 0x64, 0x6f, 0x6c, 0x6f, 0x67, 0x2e, 0x50, 0x62, 0x52,
+	0x6f, 0x77, 0x52, 0x04, 0x52, 0x6f, 0x77, 0x73, 0x42, 0x0b, 0x5a, 0x09, 0x2e, 0x3b, 0x75, 0x6e,
+	0x64, 0x6f, 0x6c, 0x6f, 0x67, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_table_records_proto_rawDescOnce sync.Once
+	file_table_records_proto_rawDescData = file_table_records_proto_rawDesc
+)
+
+func file_table_records_proto_rawDescGZIP() []byte {
+	file_table_records_proto_rawDescOnce.Do(func() {
+		file_table_records_proto_rawDescData = protoimpl.X.CompressGZIP(file_table_records_proto_rawDescData)
+	})
+	return file_table_records_proto_rawDescData
+}
+
+var file_table_records_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_table_records_proto_goTypes = []interface{}{
+	(*PbField)(nil),        // 0: undolog.PbField
+	(*PbRow)(nil),          // 1: undolog.PbRow
+	(*PbTableRecords)(nil), // 2: undolog.PbTableRecords
+}
+var file_table_records_proto_depIdxs = []int32{
+	0, // 0: undolog.PbRow.Fields:type_name -> undolog.PbField
+	1, // 1: undolog.PbTableRecords.Rows:type_name -> undolog.PbRow
+	2, // [2:2] is the sub-list for method output_type
+	2, // [2:2] is the sub-list for method input_type
+	2, // [2:2] is the sub-list for extension type_name
+	2, // [2:2] is the sub-list for extension extendee
+	0, // [0:2] is the sub-list for field type_name
+}
+
+func init() { file_table_records_proto_init() }
+func file_table_records_proto_init() {
+	if File_table_records_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_table_records_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*PbField); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_table_records_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*PbRow); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_table_records_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*PbTableRecords); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_table_records_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   3,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_table_records_proto_goTypes,
+		DependencyIndexes: file_table_records_proto_depIdxs,
+		MessageInfos:      file_table_records_proto_msgTypes,
+	}.Build()
+	File_table_records_proto = out.File
+	file_table_records_proto_rawDesc = nil
+	file_table_records_proto_goTypes = nil
+	file_table_records_proto_depIdxs = nil
 }
diff --git a/pkg/dt/undolog/undo_log.go b/pkg/dt/undolog/undo_log.go
index 9f67d3e..75a3370 100644
--- a/pkg/dt/undolog/undo_log.go
+++ b/pkg/dt/undolog/undo_log.go
@@ -22,6 +22,11 @@ import (
 )
 
 type SqlUndoLog struct {
+	// IsBinary binary protocol or text protocol, com_stmt_execute corresponds to
+	// binary protocol (prepared statement), com_query corresponds to text protocol
+	// (text statement).
+	IsBinary bool
+	// SqlType insert、delete、update
 	SqlType     constant.SQLType
 	SchemaName  string
 	TableName   string
diff --git a/pkg/dt/undolog/undo_log.pb.go b/pkg/dt/undolog/undo_log.pb.go
index e5eced4..7041be6 100644
--- a/pkg/dt/undolog/undo_log.pb.go
+++ b/pkg/dt/undolog/undo_log.pb.go
@@ -1,199 +1,309 @@
-/*
- * Copyright 2022 CECTC, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-// Code generated by protoc-gen-gogo. DO NOT EDIT.
+//
+// Copyright 2022 CECTC, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.27.1
+// 	protoc        v3.11.4
 // source: undo_log.proto
 
 package undolog
 
 import (
-	fmt "fmt"
-	math "math"
-
-	proto "github.com/gogo/protobuf/proto"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
 )
 
-// Reference imports to suppress errors if they are not otherwise used.
-var _ = proto.Marshal
-var _ = fmt.Errorf
-var _ = math.Inf
-
-// This is a compile-time assertion to ensure that this generated file
-// is compatible with the proto package it is being compiled against.
-// A compilation error at this line likely means your copy of the
-// proto package needs to be updated.
-const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
 
 type PbSqlUndoLog struct {
-	SqlType              int32           `protobuf:"varint,1,opt,name=SqlType,proto3" json:"SqlType,omitempty"`
-	SchemaName           string          `protobuf:"bytes,2,opt,name=SchemaName,proto3" json:"SchemaName,omitempty"`
-	TableName            string          `protobuf:"bytes,3,opt,name=TableName,proto3" json:"TableName,omitempty"`
-	LockKey              string          `protobuf:"bytes,4,opt,name=LockKey,proto3" json:"LockKey,omitempty"`
-	BeforeImage          *PbTableRecords `protobuf:"bytes,5,opt,name=BeforeImage,proto3" json:"BeforeImage,omitempty"`
-	AfterImage           *PbTableRecords `protobuf:"bytes,6,opt,name=AfterImage,proto3" json:"AfterImage,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}        `json:"-"`
-	XXX_unrecognized     []byte          `json:"-"`
-	XXX_sizecache        int32           `json:"-"`
-}
-
-func (m *PbSqlUndoLog) Reset()         { *m = PbSqlUndoLog{} }
-func (m *PbSqlUndoLog) String() string { return proto.CompactTextString(m) }
-func (*PbSqlUndoLog) ProtoMessage()    {}
-func (*PbSqlUndoLog) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5c8c28d37170efce, []int{0}
-}
-func (m *PbSqlUndoLog) XXX_Unmarshal(b []byte) error {
-	return xxx_messageInfo_PbSqlUndoLog.Unmarshal(m, b)
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	IsBinary    bool            `protobuf:"varint,1,opt,name=IsBinary,proto3" json:"IsBinary,omitempty"`
+	SqlType     int32           `protobuf:"varint,2,opt,name=SqlType,proto3" json:"SqlType,omitempty"`
+	SchemaName  string          `protobuf:"bytes,3,opt,name=SchemaName,proto3" json:"SchemaName,omitempty"`
+	TableName   string          `protobuf:"bytes,4,opt,name=TableName,proto3" json:"TableName,omitempty"`
+	LockKey     string          `protobuf:"bytes,5,opt,name=LockKey,proto3" json:"LockKey,omitempty"`
+	BeforeImage *PbTableRecords `protobuf:"bytes,6,opt,name=BeforeImage,proto3" json:"BeforeImage,omitempty"`
+	AfterImage  *PbTableRecords `protobuf:"bytes,7,opt,name=AfterImage,proto3" json:"AfterImage,omitempty"`
 }
-func (m *PbSqlUndoLog) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	return xxx_messageInfo_PbSqlUndoLog.Marshal(b, m, deterministic)
+
+func (x *PbSqlUndoLog) Reset() {
+	*x = PbSqlUndoLog{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_undo_log_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
-func (m *PbSqlUndoLog) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_PbSqlUndoLog.Merge(m, src)
+
+func (x *PbSqlUndoLog) String() string {
+	return protoimpl.X.MessageStringOf(x)
 }
-func (m *PbSqlUndoLog) XXX_Size() int {
-	return xxx_messageInfo_PbSqlUndoLog.Size(m)
+
+func (*PbSqlUndoLog) ProtoMessage() {}
+
+func (x *PbSqlUndoLog) ProtoReflect() protoreflect.Message {
+	mi := &file_undo_log_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
 }
-func (m *PbSqlUndoLog) XXX_DiscardUnknown() {
-	xxx_messageInfo_PbSqlUndoLog.DiscardUnknown(m)
+
+// Deprecated: Use PbSqlUndoLog.ProtoReflect.Descriptor instead.
+func (*PbSqlUndoLog) Descriptor() ([]byte, []int) {
+	return file_undo_log_proto_rawDescGZIP(), []int{0}
 }
 
-var xxx_messageInfo_PbSqlUndoLog proto.InternalMessageInfo
+func (x *PbSqlUndoLog) GetIsBinary() bool {
+	if x != nil {
+		return x.IsBinary
+	}
+	return false
+}
 
-func (m *PbSqlUndoLog) GetSqlType() int32 {
-	if m != nil {
-		return m.SqlType
+func (x *PbSqlUndoLog) GetSqlType() int32 {
+	if x != nil {
+		return x.SqlType
 	}
 	return 0
 }
 
-func (m *PbSqlUndoLog) GetSchemaName() string {
-	if m != nil {
-		return m.SchemaName
+func (x *PbSqlUndoLog) GetSchemaName() string {
+	if x != nil {
+		return x.SchemaName
 	}
 	return ""
 }
 
-func (m *PbSqlUndoLog) GetTableName() string {
-	if m != nil {
-		return m.TableName
+func (x *PbSqlUndoLog) GetTableName() string {
+	if x != nil {
+		return x.TableName
 	}
 	return ""
 }
 
-func (m *PbSqlUndoLog) GetLockKey() string {
-	if m != nil {
-		return m.LockKey
+func (x *PbSqlUndoLog) GetLockKey() string {
+	if x != nil {
+		return x.LockKey
 	}
 	return ""
 }
 
-func (m *PbSqlUndoLog) GetBeforeImage() *PbTableRecords {
-	if m != nil {
-		return m.BeforeImage
+func (x *PbSqlUndoLog) GetBeforeImage() *PbTableRecords {
+	if x != nil {
+		return x.BeforeImage
 	}
 	return nil
 }
 
-func (m *PbSqlUndoLog) GetAfterImage() *PbTableRecords {
-	if m != nil {
-		return m.AfterImage
+func (x *PbSqlUndoLog) GetAfterImage() *PbTableRecords {
+	if x != nil {
+		return x.AfterImage
 	}
 	return nil
 }
 
 type PbBranchUndoLog struct {
-	Xid                  string          `protobuf:"bytes,1,opt,name=Xid,proto3" json:"Xid,omitempty"`
-	BranchID             int64           `protobuf:"varint,2,opt,name=BranchID,proto3" json:"BranchID,omitempty"`
-	SqlUndoLogs          []*PbSqlUndoLog `protobuf:"bytes,3,rep,name=SqlUndoLogs,proto3" json:"SqlUndoLogs,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}        `json:"-"`
-	XXX_unrecognized     []byte          `json:"-"`
-	XXX_sizecache        int32           `json:"-"`
-}
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
 
-func (m *PbBranchUndoLog) Reset()         { *m = PbBranchUndoLog{} }
-func (m *PbBranchUndoLog) String() string { return proto.CompactTextString(m) }
-func (*PbBranchUndoLog) ProtoMessage()    {}
-func (*PbBranchUndoLog) Descriptor() ([]byte, []int) {
-	return fileDescriptor_5c8c28d37170efce, []int{1}
-}
-func (m *PbBranchUndoLog) XXX_Unmarshal(b []byte) error {
-	return xxx_messageInfo_PbBranchUndoLog.Unmarshal(m, b)
-}
-func (m *PbBranchUndoLog) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	return xxx_messageInfo_PbBranchUndoLog.Marshal(b, m, deterministic)
+	Xid         string          `protobuf:"bytes,1,opt,name=Xid,proto3" json:"Xid,omitempty"`
+	BranchID    int64           `protobuf:"varint,2,opt,name=BranchID,proto3" json:"BranchID,omitempty"`
+	SqlUndoLogs []*PbSqlUndoLog `protobuf:"bytes,3,rep,name=SqlUndoLogs,proto3" json:"SqlUndoLogs,omitempty"`
 }
-func (m *PbBranchUndoLog) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_PbBranchUndoLog.Merge(m, src)
+
+func (x *PbBranchUndoLog) Reset() {
+	*x = PbBranchUndoLog{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_undo_log_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
 }
-func (m *PbBranchUndoLog) XXX_Size() int {
-	return xxx_messageInfo_PbBranchUndoLog.Size(m)
+
+func (x *PbBranchUndoLog) String() string {
+	return protoimpl.X.MessageStringOf(x)
 }
-func (m *PbBranchUndoLog) XXX_DiscardUnknown() {
-	xxx_messageInfo_PbBranchUndoLog.DiscardUnknown(m)
+
+func (*PbBranchUndoLog) ProtoMessage() {}
+
+func (x *PbBranchUndoLog) ProtoReflect() protoreflect.Message {
+	mi := &file_undo_log_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
 }
 
-var xxx_messageInfo_PbBranchUndoLog proto.InternalMessageInfo
+// Deprecated: Use PbBranchUndoLog.ProtoReflect.Descriptor instead.
+func (*PbBranchUndoLog) Descriptor() ([]byte, []int) {
+	return file_undo_log_proto_rawDescGZIP(), []int{1}
+}
 
-func (m *PbBranchUndoLog) GetXid() string {
-	if m != nil {
-		return m.Xid
+func (x *PbBranchUndoLog) GetXid() string {
+	if x != nil {
+		return x.Xid
 	}
 	return ""
 }
 
-func (m *PbBranchUndoLog) GetBranchID() int64 {
-	if m != nil {
-		return m.BranchID
+func (x *PbBranchUndoLog) GetBranchID() int64 {
+	if x != nil {
+		return x.BranchID
 	}
 	return 0
 }
 
-func (m *PbBranchUndoLog) GetSqlUndoLogs() []*PbSqlUndoLog {
-	if m != nil {
-		return m.SqlUndoLogs
+func (x *PbBranchUndoLog) GetSqlUndoLogs() []*PbSqlUndoLog {
+	if x != nil {
+		return x.SqlUndoLogs
 	}
 	return nil
 }
 
-func init() {
-	proto.RegisterType((*PbSqlUndoLog)(nil), "undolog.PbSqlUndoLog")
-	proto.RegisterType((*PbBranchUndoLog)(nil), "undolog.PbBranchUndoLog")
-}
-
-func init() { proto.RegisterFile("undo_log.proto", fileDescriptor_5c8c28d37170efce) }
-
-var fileDescriptor_5c8c28d37170efce = []byte{
-	// 281 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x7c, 0x91, 0x4f, 0x4b, 0xc3, 0x40,
-	0x10, 0xc5, 0x89, 0xb1, 0xad, 0x99, 0x88, 0xca, 0x8a, 0xb8, 0x14, 0x91, 0xd0, 0x53, 0x4e, 0x39,
-	0xd4, 0x43, 0x11, 0x4f, 0x16, 0x2f, 0xc5, 0x22, 0x61, 0x53, 0x41, 0xbc, 0x94, 0xfc, 0x99, 0xa6,
-	0xe2, 0x26, 0xd3, 0x6e, 0x23, 0xd8, 0xef, 0xed, 0x07, 0x90, 0x6c, 0x1a, 0xb3, 0x27, 0x6f, 0x79,
-	0xf3, 0x7e, 0xf3, 0xe7, 0x65, 0xe1, 0xec, 0xab, 0xcc, 0x68, 0x29, 0x29, 0x0f, 0x36, 0x8a, 0x2a,
-	0x62, 0x83, 0x5a, 0x4b, 0xca, 0x87, 0x97, 0x55, 0x9c, 0x48, 0x5c, 0x2a, 0x4c, 0x49, 0x65, 0xbb,
-	0xc6, 0x1d, 0xfd, 0x58, 0x70, 0x1a, 0x26, 0xd1, 0x56, 0xbe, 0x96, 0x19, 0xcd, 0x29, 0x67, 0x1c,
-	0x06, 0xd1, 0x56, 0x2e, 0xf6, 0x1b, 0xe4, 0x96, 0x67, 0xf9, 0x3d, 0xd1, 0x4a, 0x76, 0x0b, 0x10,
-	0xa5, 0x6b, 0x2c, 0xe2, 0x97, 0xb8, 0x40, 0x7e, 0xe4, 0x59, 0xbe, 0x23, 0x8c, 0x0a, 0xbb, 0x01,
-	0x67, 0x51, 0x6f, 0xd0, 0xb6, 0xad, 0xed, 0xae, 0x50, 0xcf, 0x9d, 0x53, 0xfa, 0xf9, 0x8c, 0x7b,
-	0x7e, 0xac, 0xbd, 0x56, 0xb2, 0x7b, 0x70, 0xa7, 0xb8, 0x22, 0x85, 0xb3, 0x22, 0xce, 0x91, 0xf7,
-	0x3c, 0xcb, 0x77, 0xc7, 0xd7, 0xc1, 0xe1, 0xec, 0x20, 0x4c, 0xf4, 0x10, 0xd1, 0x9c, 0x2d, 0x4c,
-	0x96, 0x4d, 0x00, 0x1e, 0x57, 0x15, 0xaa, 0xa6, 0xb3, 0xff, 0x7f, 0xa7, 0x81, 0x8e, 0xbe, 0xe1,
-	0x3c, 0x4c, 0xa6, 0x2a, 0x2e, 0xd3, 0x75, 0x1b, 0xfc, 0x02, 0xec, 0xb7, 0x8f, 0x4c, 0x87, 0x76,
-	0x44, 0xfd, 0xc9, 0x86, 0x70, 0xd2, 0x20, 0xb3, 0x27, 0x1d, 0xd7, 0x16, 0x7f, 0x9a, 0x4d, 0xc0,
-	0xed, 0x7e, 0xda, 0x8e, 0xdb, 0x9e, 0xed, 0xbb, 0xe3, 0x2b, 0x63, 0x75, 0xe7, 0x0a, 0x93, 0x9c,
-	0xba, 0xef, 0x4e, 0xf0, 0x70, 0xc0, 0x92, 0xbe, 0x7e, 0x84, 0xbb, 0xdf, 0x00, 0x00, 0x00, 0xff,
-	0xff, 0x92, 0x0c, 0x65, 0x60, 0xb4, 0x01, 0x00, 0x00,
+var File_undo_log_proto protoreflect.FileDescriptor
+
+var file_undo_log_proto_rawDesc = []byte{
+	0x0a, 0x0e, 0x75, 0x6e, 0x64, 0x6f, 0x5f, 0x6c, 0x6f, 0x67, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x12, 0x07, 0x75, 0x6e, 0x64, 0x6f, 0x6c, 0x6f, 0x67, 0x1a, 0x13, 0x74, 0x61, 0x62, 0x6c, 0x65,
+	0x5f, 0x72, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x90,
+	0x02, 0x0a, 0x0c, 0x50, 0x62, 0x53, 0x71, 0x6c, 0x55, 0x6e, 0x64, 0x6f, 0x4c, 0x6f, 0x67, 0x12,
+	0x1a, 0x0a, 0x08, 0x49, 0x73, 0x42, 0x69, 0x6e, 0x61, 0x72, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x08, 0x52, 0x08, 0x49, 0x73, 0x42, 0x69, 0x6e, 0x61, 0x72, 0x79, 0x12, 0x18, 0x0a, 0x07, 0x53,
+	0x71, 0x6c, 0x54, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x07, 0x53, 0x71,
+	0x6c, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1e, 0x0a, 0x0a, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x4e,
+	0x61, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x53, 0x63, 0x68, 0x65, 0x6d,
+	0x61, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x4e, 0x61,
+	0x6d, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x4e,
+	0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x4c, 0x6f, 0x63, 0x6b, 0x4b, 0x65, 0x79, 0x18, 0x05,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x4c, 0x6f, 0x63, 0x6b, 0x4b, 0x65, 0x79, 0x12, 0x39, 0x0a,
+	0x0b, 0x42, 0x65, 0x66, 0x6f, 0x72, 0x65, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x18, 0x06, 0x20, 0x01,
+	0x28, 0x0b, 0x32, 0x17, 0x2e, 0x75, 0x6e, 0x64, 0x6f, 0x6c, 0x6f, 0x67, 0x2e, 0x50, 0x62, 0x54,
+	0x61, 0x62, 0x6c, 0x65, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x73, 0x52, 0x0b, 0x42, 0x65, 0x66,
+	0x6f, 0x72, 0x65, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x12, 0x37, 0x0a, 0x0a, 0x41, 0x66, 0x74, 0x65,
+	0x72, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x75,
+	0x6e, 0x64, 0x6f, 0x6c, 0x6f, 0x67, 0x2e, 0x50, 0x62, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x52, 0x65,
+	0x63, 0x6f, 0x72, 0x64, 0x73, 0x52, 0x0a, 0x41, 0x66, 0x74, 0x65, 0x72, 0x49, 0x6d, 0x61, 0x67,
+	0x65, 0x22, 0x78, 0x0a, 0x0f, 0x50, 0x62, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x55, 0x6e, 0x64,
+	0x6f, 0x4c, 0x6f, 0x67, 0x12, 0x10, 0x0a, 0x03, 0x58, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x03, 0x58, 0x69, 0x64, 0x12, 0x1a, 0x0a, 0x08, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68,
+	0x49, 0x44, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x52, 0x08, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68,
+	0x49, 0x44, 0x12, 0x37, 0x0a, 0x0b, 0x53, 0x71, 0x6c, 0x55, 0x6e, 0x64, 0x6f, 0x4c, 0x6f, 0x67,
+	0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x75, 0x6e, 0x64, 0x6f, 0x6c, 0x6f,
+	0x67, 0x2e, 0x50, 0x62, 0x53, 0x71, 0x6c, 0x55, 0x6e, 0x64, 0x6f, 0x4c, 0x6f, 0x67, 0x52, 0x0b,
+	0x53, 0x71, 0x6c, 0x55, 0x6e, 0x64, 0x6f, 0x4c, 0x6f, 0x67, 0x73, 0x42, 0x0b, 0x5a, 0x09, 0x2e,
+	0x3b, 0x75, 0x6e, 0x64, 0x6f, 0x6c, 0x6f, 0x67, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_undo_log_proto_rawDescOnce sync.Once
+	file_undo_log_proto_rawDescData = file_undo_log_proto_rawDesc
+)
+
+func file_undo_log_proto_rawDescGZIP() []byte {
+	file_undo_log_proto_rawDescOnce.Do(func() {
+		file_undo_log_proto_rawDescData = protoimpl.X.CompressGZIP(file_undo_log_proto_rawDescData)
+	})
+	return file_undo_log_proto_rawDescData
+}
+
+var file_undo_log_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_undo_log_proto_goTypes = []interface{}{
+	(*PbSqlUndoLog)(nil),    // 0: undolog.PbSqlUndoLog
+	(*PbBranchUndoLog)(nil), // 1: undolog.PbBranchUndoLog
+	(*PbTableRecords)(nil),  // 2: undolog.PbTableRecords
+}
+var file_undo_log_proto_depIdxs = []int32{
+	2, // 0: undolog.PbSqlUndoLog.BeforeImage:type_name -> undolog.PbTableRecords
+	2, // 1: undolog.PbSqlUndoLog.AfterImage:type_name -> undolog.PbTableRecords
+	0, // 2: undolog.PbBranchUndoLog.SqlUndoLogs:type_name -> undolog.PbSqlUndoLog
+	3, // [3:3] is the sub-list for method output_type
+	3, // [3:3] is the sub-list for method input_type
+	3, // [3:3] is the sub-list for extension type_name
+	3, // [3:3] is the sub-list for extension extendee
+	0, // [0:3] is the sub-list for field type_name
+}
+
+func init() { file_undo_log_proto_init() }
+func file_undo_log_proto_init() {
+	if File_undo_log_proto != nil {
+		return
+	}
+	file_table_records_proto_init()
+	if !protoimpl.UnsafeEnabled {
+		file_undo_log_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*PbSqlUndoLog); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_undo_log_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*PbBranchUndoLog); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_undo_log_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   2,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_undo_log_proto_goTypes,
+		DependencyIndexes: file_undo_log_proto_depIdxs,
+		MessageInfos:      file_undo_log_proto_msgTypes,
+	}.Build()
+	File_undo_log_proto = out.File
+	file_undo_log_proto_rawDesc = nil
+	file_undo_log_proto_goTypes = nil
+	file_undo_log_proto_depIdxs = nil
 }
diff --git a/pkg/dt/undolog/undo_log.proto b/pkg/dt/undolog/undo_log.proto
index d08da23..5b24619 100644
--- a/pkg/dt/undolog/undo_log.proto
+++ b/pkg/dt/undolog/undo_log.proto
@@ -22,12 +22,13 @@ import "table_records.proto";
 option go_package=".;undolog";
 
 message PbSqlUndoLog {
-    int32 SqlType = 1;
-    string SchemaName = 2;
-    string TableName = 3;
-    string LockKey = 4;
-    PbTableRecords BeforeImage = 5;
-    PbTableRecords AfterImage = 6;
+    bool IsBinary = 1;
+    int32 SqlType = 2;
+    string SchemaName = 3;
+    string TableName = 4;
+    string LockKey = 5;
+    PbTableRecords BeforeImage = 6;
+    PbTableRecords AfterImage = 7;
 }
 
 message PbBranchUndoLog {
diff --git a/pkg/filter/dt/exec/executor.go b/pkg/filter/dt/exec/executor.go
index 2d9d1fb..f0f2679 100644
--- a/pkg/filter/dt/exec/executor.go
+++ b/pkg/filter/dt/exec/executor.go
@@ -38,8 +38,14 @@ type Executable interface {
 	GetTableName() string
 }
 
-func BuildUndoItem(sqlType constant.SQLType, schemaName, tableName, lockKey string, beforeImage, afterImage *schema.TableRecords) *undolog.SqlUndoLog {
+func BuildUndoItem(
+	isBinary bool,
+	sqlType constant.SQLType,
+	schemaName, tableName,
+	lockKey string, beforeImage,
+	afterImage *schema.TableRecords) *undolog.SqlUndoLog {
 	sqlUndoLog := &undolog.SqlUndoLog{
+		IsBinary:    isBinary,
 		SqlType:     sqlType,
 		SchemaName:  schemaName,
 		TableName:   tableName,
diff --git a/pkg/filter/dt/exec/prepare_delete.go b/pkg/filter/dt/exec/prepare_delete.go
index e3f4ffd..819bc73 100644
--- a/pkg/filter/dt/exec/prepare_delete.go
+++ b/pkg/filter/dt/exec/prepare_delete.go
@@ -66,7 +66,7 @@ func (executor *prepareDeleteExecutor) BeforeImage(ctx context.Context) (*schema
 	if err != nil {
 		return nil, err
 	}
-	return schema.BuildRecords(tableMeta, result), nil
+	return schema.BuildBinaryRecords(tableMeta, result), nil
 }
 
 func (executor *prepareDeleteExecutor) AfterImage(ctx context.Context) (*schema.TableRecords, error) {
@@ -89,26 +89,28 @@ func (executor *prepareDeleteExecutor) GetTableName() string {
 
 func (executor *prepareDeleteExecutor) buildBeforeImageSql(tableMeta schema.TableMeta) string {
 	var b strings.Builder
-	fmt.Fprint(&b, "SELECT ")
+	b.WriteString("SELECT ")
 	var i = 0
 	columnCount := len(tableMeta.Columns)
 	for _, column := range tableMeta.Columns {
-		fmt.Fprint(&b, misc.CheckAndReplace(column))
+		b.WriteString(misc.CheckAndReplace(column))
 		i = i + 1
 		if i < columnCount {
-			fmt.Fprint(&b, ",")
+			b.WriteByte(',')
 		} else {
-			fmt.Fprint(&b, " ")
+			b.WriteByte(' ')
 		}
 	}
-	fmt.Fprintf(&b, " FROM %s WHERE ", executor.GetTableName())
-	fmt.Fprint(&b, executor.GetWhereCondition())
-	fmt.Fprint(&b, " FOR UPDATE")
+	b.WriteString(fmt.Sprintf(" FROM %s WHERE ", executor.GetTableName()))
+	b.WriteString(executor.GetWhereCondition())
+	b.WriteString(" FOR UPDATE")
 	return b.String()
 }
 
 func (executor *prepareDeleteExecutor) GetWhereCondition() string {
 	var sb strings.Builder
-	executor.stmt.Where.Restore(format.NewRestoreCtx(format.DefaultRestoreFlags, &sb))
+	if err := executor.stmt.Where.Restore(format.NewRestoreCtx(format.DefaultRestoreFlags, &sb)); err != nil {
+		log.Panic(err)
+	}
 	return sb.String()
 }
diff --git a/pkg/filter/dt/exec/prepare_global_lock.go b/pkg/filter/dt/exec/prepare_global_lock.go
index 7990377..7cc8665 100644
--- a/pkg/filter/dt/exec/prepare_global_lock.go
+++ b/pkg/filter/dt/exec/prepare_global_lock.go
@@ -122,25 +122,25 @@ func (executor *prepareGlobalLockExecutor) BeforeImage(ctx context.Context) (*sc
 	if err != nil {
 		return nil, err
 	}
-	return schema.BuildRecords(tableMeta, result), nil
+	return schema.BuildBinaryRecords(tableMeta, result), nil
 }
 
 func (executor *prepareGlobalLockExecutor) buildBeforeImageSql(tableMeta schema.TableMeta) string {
 	var b strings.Builder
-	fmt.Fprint(&b, "SELECT ")
+	b.WriteString("SELECT ")
 	var i = 0
 	columnCount := len(tableMeta.Columns)
 	for _, column := range tableMeta.Columns {
-		fmt.Fprint(&b, misc.CheckAndReplace(column))
+		b.WriteString(misc.CheckAndReplace(column))
 		i = i + 1
 		if i < columnCount {
-			fmt.Fprint(&b, ",")
+			b.WriteByte(',')
 		} else {
-			fmt.Fprint(&b, " ")
+			b.WriteByte(' ')
 		}
 	}
-	fmt.Fprintf(&b, " FROM %s WHERE ", executor.GetTableName())
-	fmt.Fprint(&b, executor.GetWhereCondition())
+	b.WriteString(fmt.Sprintf(" FROM %s WHERE ", executor.GetTableName()))
+	b.WriteString(executor.GetWhereCondition())
 	return b.String()
 }
 
diff --git a/pkg/filter/dt/exec/prepare_insert.go b/pkg/filter/dt/exec/prepare_insert.go
index 9e2b527..e3d03e2 100644
--- a/pkg/filter/dt/exec/prepare_insert.go
+++ b/pkg/filter/dt/exec/prepare_insert.go
@@ -101,27 +101,27 @@ func (executor *prepareInsertExecutor) buildTableRecords(ctx context.Context, pk
 	if err != nil {
 		return nil, err
 	}
-	return schema.BuildRecords(tableMeta, result), nil
+	return schema.BuildBinaryRecords(tableMeta, result), nil
 }
 
 func (executor *prepareInsertExecutor) buildAfterImageSql(tableMeta schema.TableMeta, pkValues []interface{}) string {
-	var sb strings.Builder
-	fmt.Fprint(&sb, "SELECT ")
+	var b strings.Builder
+	b.WriteString("SELECT ")
 	var i = 0
 	columnCount := len(tableMeta.Columns)
 	for _, column := range tableMeta.Columns {
-		fmt.Fprint(&sb, misc.CheckAndReplace(column))
+		b.WriteString(misc.CheckAndReplace(column))
 		i = i + 1
 		if i < columnCount {
-			fmt.Fprint(&sb, ",")
+			b.WriteByte(',')
 		} else {
-			fmt.Fprint(&sb, " ")
+			b.WriteByte(' ')
 		}
 	}
-	fmt.Fprintf(&sb, "FROM %s ", executor.GetTableName())
-	fmt.Fprintf(&sb, " WHERE `%s` IN ", tableMeta.GetPKName())
-	fmt.Fprint(&sb, misc.MysqlAppendInParam(len(pkValues)))
-	return sb.String()
+	b.WriteString(fmt.Sprintf("FROM %s ", executor.GetTableName()))
+	b.WriteString(fmt.Sprintf(" WHERE `%s` IN ", tableMeta.GetPKName()))
+	b.WriteString(misc.MysqlAppendInParam(len(pkValues)))
+	return b.String()
 }
 
 func (executor *prepareInsertExecutor) getPKValuesByColumn(ctx context.Context) ([]interface{}, error) {
diff --git a/pkg/filter/dt/exec/prepare_select_for_update.go b/pkg/filter/dt/exec/prepare_select_for_update.go
index 6536669..707a035 100644
--- a/pkg/filter/dt/exec/prepare_select_for_update.go
+++ b/pkg/filter/dt/exec/prepare_select_for_update.go
@@ -60,7 +60,7 @@ func (executor *prepareSelectForUpdateExecutor) Executable(ctx context.Context,
 	}
 
 	rlt := executor.result.(*mysql.Result)
-	selectPKRows := schema.BuildRecords(tableMeta, rlt)
+	selectPKRows := schema.BuildBinaryRecords(tableMeta, rlt)
 	lockKeys := schema.BuildLockKey(selectPKRows)
 	if lockKeys == "" {
 		return true, nil
diff --git a/pkg/filter/dt/exec/prepare_select_for_update_test.go b/pkg/filter/dt/exec/prepare_select_for_update_test.go
index 0790c77..88a5318 100644
--- a/pkg/filter/dt/exec/prepare_select_for_update_test.go
+++ b/pkg/filter/dt/exec/prepare_select_for_update_test.go
@@ -108,7 +108,7 @@ func getTableMetaPatch() *gomonkey.Patches {
 }
 
 func buildRecordsPatch() *gomonkey.Patches {
-	return gomonkey.ApplyFunc(schema.BuildRecords, func(meta schema.TableMeta, result *mysql.Result) *schema.TableRecords {
+	return gomonkey.ApplyFunc(schema.BuildBinaryRecords, func(meta schema.TableMeta, result *mysql.Result) *schema.TableRecords {
 		return &schema.TableRecords{
 			TableMeta: tableMeta,
 			TableName: "t",
diff --git a/pkg/filter/dt/exec/prepare_update.go b/pkg/filter/dt/exec/prepare_update.go
index cf7c3c0..005595a 100644
--- a/pkg/filter/dt/exec/prepare_update.go
+++ b/pkg/filter/dt/exec/prepare_update.go
@@ -69,7 +69,7 @@ func (executor *prepareUpdateExecutor) BeforeImage(ctx context.Context) (*schema
 	if err != nil {
 		return nil, err
 	}
-	return schema.BuildRecords(tableMeta, result), nil
+	return schema.BuildBinaryRecords(tableMeta, result), nil
 }
 
 func (executor *prepareUpdateExecutor) AfterImage(ctx context.Context) (*schema.TableRecords, error) {
@@ -91,7 +91,7 @@ func (executor *prepareUpdateExecutor) AfterImage(ctx context.Context) (*schema.
 	if err != nil {
 		return nil, err
 	}
-	return schema.BuildRecords(tableMeta, result), nil
+	return schema.BuildBinaryRecords(tableMeta, result), nil
 }
 
 func (executor *prepareUpdateExecutor) GetTableMeta(ctx context.Context) (schema.TableMeta, error) {
@@ -110,47 +110,49 @@ func (executor *prepareUpdateExecutor) GetTableName() string {
 
 func (executor *prepareUpdateExecutor) buildBeforeImageSql(tableMeta schema.TableMeta) string {
 	var b strings.Builder
-	fmt.Fprint(&b, "SELECT ")
+	b.WriteString("SELECT ")
 	var i = 0
 	columnCount := len(tableMeta.Columns)
 	for _, column := range tableMeta.Columns {
-		fmt.Fprint(&b, misc.CheckAndReplace(column))
+		b.WriteString(misc.CheckAndReplace(column))
 		i = i + 1
 		if i != columnCount {
-			fmt.Fprint(&b, ",")
+			b.WriteByte(',')
 		} else {
-			fmt.Fprint(&b, " ")
+			b.WriteByte(' ')
 		}
 	}
-	fmt.Fprintf(&b, " FROM %s WHERE ", executor.GetTableName())
-	fmt.Fprint(&b, executor.GetWhereCondition())
-	fmt.Fprint(&b, " FOR UPDATE")
+	b.WriteString(fmt.Sprintf(" FROM %s WHERE ", executor.GetTableName()))
+	b.WriteString(executor.GetWhereCondition())
+	b.WriteString(" FOR UPDATE")
 	return b.String()
 }
 
 func (executor *prepareUpdateExecutor) buildAfterImageSql(tableMeta schema.TableMeta, beforeImage *schema.TableRecords) string {
 	var b strings.Builder
-	fmt.Fprint(&b, "SELECT ")
+	b.WriteString("SELECT ")
 	var i = 0
 	columnCount := len(tableMeta.Columns)
 	for _, columnName := range tableMeta.Columns {
-		fmt.Fprint(&b, misc.CheckAndReplace(columnName))
+		b.WriteString(misc.CheckAndReplace(columnName))
 		i = i + 1
-		if i < columnCount {
-			fmt.Fprint(&b, ",")
+		if i != columnCount {
+			b.WriteByte(',')
 		} else {
-			fmt.Fprint(&b, " ")
+			b.WriteByte(' ')
 		}
 	}
-	fmt.Fprintf(&b, " FROM %s ", executor.GetTableName())
-	fmt.Fprintf(&b, "WHERE `%s` IN", tableMeta.GetPKName())
-	fmt.Fprint(&b, misc.MysqlAppendInParam(len(beforeImage.PKFields())))
+	b.WriteString(fmt.Sprintf(" FROM %s ", executor.GetTableName()))
+	b.WriteString(fmt.Sprintf("WHERE `%s` IN", tableMeta.GetPKName()))
+	b.WriteString(misc.MysqlAppendInParam(len(beforeImage.PKFields())))
 	return b.String()
 }
 
 func (executor *prepareUpdateExecutor) GetWhereCondition() string {
 	var sb strings.Builder
-	executor.stmt.Where.Restore(format.NewRestoreCtx(format.DefaultRestoreFlags, &sb))
+	if err := executor.stmt.Where.Restore(format.NewRestoreCtx(format.DefaultRestoreFlags, &sb)); err != nil {
+		log.Error(err)
+	}
 	return sb.String()
 }
 
diff --git a/pkg/filter/dt/exec/query_select_for_update.go b/pkg/filter/dt/exec/query_select_for_update.go
index 6b65c2b..9d73bcc 100644
--- a/pkg/filter/dt/exec/query_select_for_update.go
+++ b/pkg/filter/dt/exec/query_select_for_update.go
@@ -57,7 +57,7 @@ func (executor *querySelectForUpdateExecutor) Executable(ctx context.Context, lo
 	}
 
 	rlt := executor.result.(*mysql.Result)
-	selectPKRows := schema.BuildRecords(tableMeta, rlt)
+	selectPKRows := schema.BuildBinaryRecords(tableMeta, rlt)
 	lockKeys := schema.BuildLockKey(selectPKRows)
 	if lockKeys == "" {
 		return true, nil
diff --git a/pkg/filter/dt/exec/query_update.go b/pkg/filter/dt/exec/query_update.go
index 4d8fb09..71ac945 100644
--- a/pkg/filter/dt/exec/query_update.go
+++ b/pkg/filter/dt/exec/query_update.go
@@ -18,12 +18,14 @@ package exec
 
 import (
 	"context"
+	"fmt"
 	"strings"
 
 	"github.com/cectc/dbpack/pkg/driver"
 	"github.com/cectc/dbpack/pkg/dt/schema"
 	"github.com/cectc/dbpack/pkg/log"
 	"github.com/cectc/dbpack/pkg/meta"
+	"github.com/cectc/dbpack/pkg/misc"
 	"github.com/cectc/dbpack/pkg/resource"
 	"github.com/cectc/dbpack/third_party/parser/ast"
 	"github.com/cectc/dbpack/third_party/parser/format"
@@ -47,13 +49,34 @@ func NewQueryUpdateExecutor(
 }
 
 func (executor *queryUpdateExecutor) BeforeImage(ctx context.Context) (*schema.TableRecords, error) {
-	// todo
-	return nil, nil
+	tableMeta, err := executor.GetTableMeta(ctx)
+	if err != nil {
+		return nil, err
+	}
+	sql := executor.buildBeforeImageSql(tableMeta)
+	result, _, err := executor.conn.ExecuteWithWarningCount(sql, true)
+	if err != nil {
+		return nil, err
+	}
+	return schema.BuildTextRecords(tableMeta, result), nil
 }
 
 func (executor *queryUpdateExecutor) AfterImage(ctx context.Context) (*schema.TableRecords, error) {
-	// todo
-	return nil, nil
+	if executor.beforeImage == nil || len(executor.beforeImage.Rows) == 0 {
+		return nil, nil
+	}
+
+	tableMeta, err := executor.GetTableMeta(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	afterImageSql := executor.buildAfterImageSql(tableMeta, executor.beforeImage)
+	result, _, err := executor.conn.ExecuteWithWarningCount(afterImageSql, true)
+	if err != nil {
+		return nil, err
+	}
+	return schema.BuildTextRecords(tableMeta, result), nil
 }
 
 func (executor *queryUpdateExecutor) GetTableMeta(ctx context.Context) (schema.TableMeta, error) {
@@ -69,3 +92,68 @@ func (executor *queryUpdateExecutor) GetTableName() string {
 	}
 	return sb.String()
 }
+
+func (executor *queryUpdateExecutor) buildBeforeImageSql(tableMeta schema.TableMeta) string {
+	var b strings.Builder
+	b.WriteString("SELECT ")
+	var i = 0
+	columnCount := len(tableMeta.Columns)
+	for _, column := range tableMeta.Columns {
+		b.WriteString(misc.CheckAndReplace(column))
+		i = i + 1
+		if i != columnCount {
+			b.WriteByte(',')
+		} else {
+			b.WriteByte(' ')
+		}
+	}
+	b.WriteString(fmt.Sprintf(" FROM %s WHERE ", executor.GetTableName()))
+	b.WriteString(executor.GetWhereCondition())
+	b.WriteString(" FOR UPDATE")
+	return b.String()
+}
+
+func (executor *queryUpdateExecutor) buildAfterImageSql(tableMeta schema.TableMeta, beforeImage *schema.TableRecords) string {
+	var b strings.Builder
+	b.WriteString("SELECT ")
+	var i = 0
+	columnCount := len(tableMeta.Columns)
+	for _, column := range tableMeta.Columns {
+		b.WriteString(misc.CheckAndReplace(column))
+		i = i + 1
+		if i < columnCount {
+			b.WriteByte(',')
+		} else {
+			b.WriteByte(' ')
+		}
+	}
+	b.WriteString(fmt.Sprintf(" FROM %s ", executor.GetTableName()))
+	b.WriteString(fmt.Sprintf(" WHERE `%s` IN (", tableMeta.GetPKName()))
+	pkFields := beforeImage.PKFields()
+	for i, field := range pkFields {
+		if i < len(pkFields)-1 {
+			b.WriteString(fmt.Sprintf("'%s',", field.Value))
+		} else {
+			b.WriteString(fmt.Sprintf("'%s'", field.Value))
+		}
+	}
+	b.WriteByte(')')
+	return b.String()
+}
+
+func (executor *queryUpdateExecutor) GetWhereCondition() string {
+	var sb strings.Builder
+	if err := executor.stmt.Where.Restore(format.NewRestoreCtx(format.DefaultRestoreFlags, &sb)); err != nil {
+		log.Panic(err)
+	}
+	return sb.String()
+}
+
+func (executor *queryUpdateExecutor) GetUpdateColumns() []string {
+	columns := make([]string, 0)
+
+	for _, assignment := range executor.stmt.List {
+		columns = append(columns, assignment.Column.Name.String())
+	}
+	return columns
+}
diff --git a/pkg/filter/dt/filter_mysql_prepare.go b/pkg/filter/dt/filter_mysql_prepare.go
index 913cd58..898eaa8 100644
--- a/pkg/filter/dt/filter_mysql_prepare.go
+++ b/pkg/filter/dt/filter_mysql_prepare.go
@@ -103,7 +103,7 @@ func (f *_mysqlFilter) processAfterPrepareDelete(ctx context.Context, conn *driv
 
 	lockKeys := schema.BuildLockKey(biValue)
 	log.Debugf("delete, lockKey: %s", lockKeys)
-	undoLog := exec.BuildUndoItem(constant.SQLType_DELETE, schemaName, executor.GetTableName(), lockKeys, biValue, nil)
+	undoLog := exec.BuildUndoItem(true, constant.SQLType_DELETE, schemaName, executor.GetTableName(), lockKeys, biValue, nil)
 
 	branchID, err := f.registerBranchTransaction(ctx, xid, conn.DataSourceName(), lockKeys)
 	if err != nil {
@@ -132,7 +132,7 @@ func (f *_mysqlFilter) processAfterPrepareInsert(ctx context.Context, conn *driv
 
 	lockKeys := schema.BuildLockKey(afterImage)
 	log.Debugf("insert, lockKey: %s", lockKeys)
-	undoLog := exec.BuildUndoItem(constant.SQLType_INSERT, schemaName, executor.GetTableName(), lockKeys, nil, afterImage)
+	undoLog := exec.BuildUndoItem(true, constant.SQLType_INSERT, schemaName, executor.GetTableName(), lockKeys, nil, afterImage)
 
 	branchID, err := f.registerBranchTransaction(ctx, xid, conn.DataSourceName(), lockKeys)
 	if err != nil {
@@ -165,7 +165,7 @@ func (f *_mysqlFilter) processAfterPrepareUpdate(ctx context.Context, conn *driv
 
 	lockKeys := schema.BuildLockKey(afterImage)
 	log.Debugf("update, lockKey: %s", lockKeys)
-	undoLog := exec.BuildUndoItem(constant.SQLType_UPDATE, schemaName, executor.GetTableName(), lockKeys, beforeImage, afterImage)
+	undoLog := exec.BuildUndoItem(true, constant.SQLType_UPDATE, schemaName, executor.GetTableName(), lockKeys, beforeImage, afterImage)
 
 	branchID, err := f.registerBranchTransaction(ctx, xid, conn.DataSourceName(), lockKeys)
 	if err != nil {
diff --git a/pkg/filter/dt/filter_mysql_query.go b/pkg/filter/dt/filter_mysql_query.go
index 6a63530..e15f72c 100644
--- a/pkg/filter/dt/filter_mysql_query.go
+++ b/pkg/filter/dt/filter_mysql_query.go
@@ -102,7 +102,7 @@ func (f *_mysqlFilter) processAfterQueryDelete(ctx context.Context, conn *driver
 
 	lockKeys := schema.BuildLockKey(biValue)
 	log.Debugf("delete, lockKey: %s", lockKeys)
-	undoLog := exec.BuildUndoItem(constant.SQLType_DELETE, schemaName, executor.GetTableName(), lockKeys, biValue, nil)
+	undoLog := exec.BuildUndoItem(false, constant.SQLType_DELETE, schemaName, executor.GetTableName(), lockKeys, biValue, nil)
 
 	branchID, err := f.registerBranchTransaction(ctx, xid, conn.DataSourceName(), lockKeys)
 	if err != nil {
@@ -131,7 +131,7 @@ func (f *_mysqlFilter) processAfterQueryInsert(ctx context.Context, conn *driver
 
 	lockKeys := schema.BuildLockKey(afterImage)
 	log.Debugf("insert, lockKey: %s", lockKeys)
-	undoLog := exec.BuildUndoItem(constant.SQLType_INSERT, schemaName, executor.GetTableName(), lockKeys, nil, afterImage)
+	undoLog := exec.BuildUndoItem(false, constant.SQLType_INSERT, schemaName, executor.GetTableName(), lockKeys, nil, afterImage)
 
 	branchID, err := f.registerBranchTransaction(ctx, xid, conn.DataSourceName(), lockKeys)
 	if err != nil {
@@ -163,7 +163,7 @@ func (f *_mysqlFilter) processAfterQueryUpdate(ctx context.Context, conn *driver
 
 	lockKeys := schema.BuildLockKey(afterImage)
 	log.Debugf("update, lockKey: %s", lockKeys)
-	undoLog := exec.BuildUndoItem(constant.SQLType_UPDATE, schemaName, executor.GetTableName(), lockKeys, beforeImage, afterImage)
+	undoLog := exec.BuildUndoItem(false, constant.SQLType_UPDATE, schemaName, executor.GetTableName(), lockKeys, beforeImage, afterImage)
 
 	branchID, err := f.registerBranchTransaction(ctx, xid, conn.DataSourceName(), lockKeys)
 	if err != nil {
diff --git a/test/sdb/distributed_transaction_test.go b/test/sdb/distributed_transaction_test.go
index 4c1a041..01b2c25 100644
--- a/test/sdb/distributed_transaction_test.go
+++ b/test/sdb/distributed_transaction_test.go
@@ -36,7 +36,8 @@ import (
 )
 
 const (
-	phiEmployeeDSN = `root:123456@tcp(127.0.0.1:3306)/employees?timeout=1s&readTimeout=1s&writeTimeout=1s&parseTime=true&loc=Local&charset=utf8mb4,utf8`
+	dataSourceName2 = "dksl:123456@tcp(127.0.0.1:13306)/employees?interpolateParams=true&timeout=10s&readTimeout=10s&writeTimeout=10s&parseTime=true&loc=Local&charset=utf8mb4,utf8"
+	phiEmployeeDSN  = `root:123456@tcp(127.0.0.1:3306)/employees?timeout=1s&readTimeout=1s&writeTimeout=1s&parseTime=true&loc=Local&charset=utf8mb4,utf8`
 
 	insertEmployeeForDT   = `INSERT INTO employees ( id, emp_no, birth_date, first_name, last_name, gender, hire_date ) VALUES (?, ?, ?, ?, ?, ?, ?)`
 	insertDepartmentForDT = `INSERT INTO departments ( id, dept_no, dept_name ) VALUES (?, ?, ?)`
@@ -51,7 +52,8 @@ const (
 
 type _DistributedTransactionSuite struct {
 	suite.Suite
-	db *sql.DB
+	db  *sql.DB
+	db2 *sql.DB
 }
 
 func TestDistributedTransaction(t *testing.T) {
@@ -97,10 +99,17 @@ func (suite *_DistributedTransactionSuite) SetupSuite() {
 		db.Exec(insertDepartmentForDT, 1, 1002, "sales")
 		db.Exec(insertDeptEmpForDT, 1, 100002, 1002, "2020-01-01", "2022-01-01")
 		db.Exec(insertSalariesForDT, 1, 100002, 8000, "2020-01-01", "2025-01-01")
+
+		db.Exec(insertSalariesForDT, 2, 100003, 8000, "2020-01-01", "2025-01-01")
+	}
+
+	db2, err := sql.Open(driverName, dataSourceName2)
+	if suite.NoErrorf(err, "connection error: %v", err) {
+		suite.db2 = db2
 	}
 }
 
-func (suite *_DistributedTransactionSuite) TestDistributedTransaction() {
+func (suite *_DistributedTransactionSuite) TestDistributedTransactionPrepareRequest() {
 	transactionManager := dt.GetDistributedTransactionManager()
 	xid, err := transactionManager.Begin(context.Background(), "test-distributed-transaction", 60000)
 	if suite.NoErrorf(err, "begin global transaction error: %v", err) {
@@ -167,6 +176,73 @@ func (suite *_DistributedTransactionSuite) TestDistributedTransaction() {
 	}
 }
 
+func (suite *_DistributedTransactionSuite) TestDistributedTransactionQueryRequest() {
+	transactionManager := dt.GetDistributedTransactionManager()
+	xid, err := transactionManager.Begin(context.Background(), "test-distributed-transaction", 60000)
+	if suite.NoErrorf(err, "begin global transaction error: %v", err) {
+		tx, err := suite.db2.Begin()
+		if suite.NoErrorf(err, "begin tx error: %v", err) {
+			//insertSql := fmt.Sprintf(`INSERT /*+ XID('%s') */ INTO dept_manager ( id, emp_no, dept_no, from_date, to_date ) VALUES (?, ?, ?, ?, ?)`, xid)
+			//result, err := tx.Exec(insertSql, 1, 100002, 1002, "2022-01-01", "2024-01-01")
+			//if suite.NoErrorf(err, "insert row error: %v", err) {
+			//	affected, err := result.RowsAffected()
+			//	if suite.NoErrorf(err, "insert row error: %v", err) {
+			//		suite.Equal(int64(1), affected)
+			//	}
+			//}
+			//
+			//deleteSql := fmt.Sprintf(`DELETE /*+ XID('%s') */ FROM dept_emp WHERE emp_no = ? and dept_no = ?`, xid)
+			//result, err = tx.Exec(deleteSql, 100002, 1002)
+			//if suite.NoErrorf(err, "delete row error: %v", err) {
+			//	affected, err := result.RowsAffected()
+			//	if suite.NoErrorf(err, "delete row error: %v", err) {
+			//		suite.Equal(int64(1), affected)
+			//	}
+			//}
+
+			updateSql := fmt.Sprintf(`UPDATE /*+ XID('%s') */ salaries SET salary = ? WHERE emp_no = ?`, xid)
+			result, err := tx.Exec(updateSql, 20000, 100003)
+			if suite.NoErrorf(err, "update row error: %v", err) {
+				affected, err := result.RowsAffected()
+				if suite.NoErrorf(err, "update row error: %v", err) {
+					suite.Equal(int64(1), affected)
+				}
+			}
+
+			err = tx.Commit()
+			if suite.NoErrorf(err, "commit local transaction error: %v", err) {
+				status, err := transactionManager.Rollback(context.Background(), xid)
+				if suite.NoErrorf(err, "rollback global transaction err: %v", err) {
+					suite.Equal(api.Rollbacking, status)
+				}
+				time.Sleep(5 * time.Second)
+
+				//checkDeptEmp := `SELECT 1 FROM dept_emp WHERE emp_no = ? and dept_no = ?`
+				//rows, err := suite.db.Query(checkDeptEmp, 100002, 1002)
+				//if suite.NoErrorf(err, "check dept_emp error: %v", err) {
+				//	var exists int
+				//	if rows.Next() {
+				//		err := rows.Scan(&exists)
+				//		suite.NoError(err)
+				//	}
+				//	suite.Equal(1, exists)
+				//}
+
+				checkSalaries := `SELECT 1 FROM salaries WHERE emp_no = ? and salary = ?`
+				rows, err := suite.db2.Query(checkSalaries, 100002, 8000)
+				if suite.NoErrorf(err, "check salaries error: %v", err) {
+					var exists int
+					if rows.Next() {
+						err := rows.Scan(&exists)
+						suite.NoError(err)
+					}
+					suite.Equal(1, exists)
+				}
+			}
+		}
+	}
+}
+
 func (suite *_DistributedTransactionSuite) TearDownSuite() {
 	suite.db.Exec(deleteEmployeeForDT, 1)
 	suite.db.Exec(deleteDepartmentForDT, 1)