Skip to content

Commit ad9a6e6

Browse files
author
Chungeun Choi
authored
Add function to parse 'extradata’ in rows event (#817)
* feature - parse extra data * fixed - parse extradata * fix typos and add comments * fix according to coding style * Fix typos and change to formatting * Fix Incorrect Byte Parsing in NDB Info from Extra Data * Add Test Code to Validate 'extradata' Parsing in 'rows event' * fixed conding style * Add event type to rows event data parsing for partition ID * Add event type to check partition Id in rows event
1 parent 09a16f2 commit ad9a6e6

File tree

3 files changed

+161
-3
lines changed

3 files changed

+161
-3
lines changed

‎replication/const.go

+5
Original file line numberDiff line numberDiff line change
@@ -250,3 +250,8 @@ const (
250250
LAST_INSERT_ID
251251
INSERT_ID
252252
)
253+
254+
const (
255+
ENUM_EXTRA_ROW_INFO_TYPECODE_NDB byte = iota
256+
ENUM_EXTRA_ROW_INFO_TYPECODE_PARTITION
257+
)

‎replication/row_event.go

+36-3
Original file line numberDiff line numberDiff line change
@@ -878,7 +878,12 @@ type RowsEvent struct {
878878
Flags uint16
879879

880880
// if version == 2
881-
ExtraData []byte
881+
// Use when DataLen value is greater than 2
882+
NdbFormat byte
883+
NdbData []byte
884+
885+
PartitionId uint16
886+
SourcePartitionId uint16
882887

883888
// lenenc_int
884889
ColumnCount uint64
@@ -955,8 +960,12 @@ func (e *RowsEvent) DecodeHeader(data []byte) (int, error) {
955960
if e.Version == 2 {
956961
dataLen := binary.LittleEndian.Uint16(data[pos:])
957962
pos += 2
958-
959-
e.ExtraData = data[pos : pos+int(dataLen-2)]
963+
if dataLen > 2 {
964+
err := e.decodeExtraData(data[pos:])
965+
if err != nil {
966+
return 0, err
967+
}
968+
}
960969
pos += int(dataLen - 2)
961970
}
962971

@@ -985,6 +994,29 @@ func (e *RowsEvent) DecodeHeader(data []byte) (int, error) {
985994
return pos, nil
986995
}
987996

997+
func (e *RowsEvent) decodeExtraData(data []byte) (err2 error) {
998+
pos := 0
999+
extraDataType := data[pos]
1000+
pos += 1
1001+
switch extraDataType {
1002+
case ENUM_EXTRA_ROW_INFO_TYPECODE_NDB:
1003+
var ndbLength int = int(data[pos])
1004+
pos += 1
1005+
e.NdbFormat = data[pos]
1006+
pos += 1
1007+
e.NdbData = data[pos : pos+ndbLength-2]
1008+
case ENUM_EXTRA_ROW_INFO_TYPECODE_PARTITION:
1009+
if e.eventType == UPDATE_ROWS_EVENTv1 || e.eventType == UPDATE_ROWS_EVENTv2 || e.eventType == PARTIAL_UPDATE_ROWS_EVENT {
1010+
e.PartitionId = binary.LittleEndian.Uint16(data[pos:])
1011+
pos += 2
1012+
e.SourcePartitionId = binary.LittleEndian.Uint16(data[pos:])
1013+
} else {
1014+
e.PartitionId = binary.LittleEndian.Uint16(data[pos:])
1015+
}
1016+
}
1017+
return nil
1018+
}
1019+
9881020
func (e *RowsEvent) DecodeData(pos int, data []byte) (err2 error) {
9891021
// Rows_log_event::print_verbose()
9901022

@@ -1742,6 +1774,7 @@ func (e *RowsEvent) Dump(w io.Writer) {
17421774
fmt.Fprintf(w, "TableID: %d\n", e.TableID)
17431775
fmt.Fprintf(w, "Flags: %d\n", e.Flags)
17441776
fmt.Fprintf(w, "Column count: %d\n", e.ColumnCount)
1777+
fmt.Fprintf(w, "NDB data: %s\n", e.NdbData)
17451778

17461779
fmt.Fprintf(w, "Values:\n")
17471780
for _, rows := range e.Rows {

‎replication/row_event_test.go

+120
Original file line numberDiff line numberDiff line change
@@ -1056,6 +1056,126 @@ func TestTableMapOptMetaVisibility(t *testing.T) {
10561056
}
10571057
}
10581058

1059+
func TestRowsDataExtraData(t *testing.T) {
1060+
// Only after mysql 8.0.16 version can be parsed from extradata to 'partition info' and 'ndb info'
1061+
testcases := []struct {
1062+
data []byte
1063+
tableData []byte
1064+
eventType EventType
1065+
expectPartitionId uint16
1066+
expectSourcePartitionId uint16
1067+
expectNdbFormat byte
1068+
expectNdbData []byte
1069+
}{
1070+
/*
1071+
mysql-cluster 8.0.32
1072+
1073+
+-------+------+------+-----+---------+-------+
1074+
| Field | Type | Null | Key | Default | Extra |
1075+
+-------+------+------+-----+---------+-------+
1076+
| p | int | NO | PRI | NULL | |
1077+
| c | int | YES | UNI | NULL | |
1078+
+-------+------+------+-----+---------+-------+
1079+
1080+
CREATE TABLE t (
1081+
p INT PRIMARY KEY,
1082+
c INT,
1083+
UNIQUE KEY u (c)
1084+
) ENGINE NDB;
1085+
1086+
INSERT INTO t VALUES (1,1), (2,2), (3,3), (4,4), (5,5);
1087+
*/
1088+
{
1089+
data: []byte("s\x00\x00\x00\x00\x00\x01\x00\x0f\x00\x00\f\x00\x01\x00\x00\x04\x80\x00\x04\x00\x00\x00\x02\xff\x00\x01\x00\x00\x00\x01\x00\x00\x00\x00\x02\x00\x00\x00\x02\x00\x00\x00\x00\x04\x00\x00\x00\x04\x00\x00\x00\x00\x03\x00\x00\x00\x03\x00\x00\x00\x00\x05\x00\x00\x00\x05\x00\x00\x00"),
1090+
tableData: []byte("s\x00\x00\x00\x00\x00\x01\x00\abdteste\x00\x01t\x00\x02\x03\x03\x00\x02\x01\x01\x00"),
1091+
eventType: WRITE_ROWS_EVENTv2,
1092+
expectPartitionId: 0x0,
1093+
expectSourcePartitionId: 0x0,
1094+
expectNdbFormat: 0x0,
1095+
expectNdbData: []byte("\x01\x00\x00\x04\x80\x00\x04\x00\x00\x00"),
1096+
},
1097+
/*
1098+
mysql 8.0.16
1099+
1100+
+-------+------+------+-----+---------+-------+
1101+
| Field | Type | Null | Key | Default | Extra |
1102+
+-------+------+------+-----+---------+-------+
1103+
| id | int | YES | | NULL | |
1104+
+-------+------+------+-----+---------+-------+
1105+
1106+
CREATE TABLE test (id INTEGER)
1107+
PARTITION BY RANGE (id) (
1108+
PARTITION p0 VALUES LESS THAN (1),
1109+
PARTITION p1 VALUES LESS THAN (2),
1110+
PARTITION p2 VALUES LESS THAN (3),
1111+
PARTITION p3 VALUES LESS THAN (4),
1112+
PARTITION p4 VALUES LESS THAN (5)
1113+
);
1114+
1115+
INSERT INTO test (id) VALUES(3);
1116+
UPDATE test set id = 1 WHERE id = 3;
1117+
*/
1118+
{
1119+
data: []byte("p\x03\x00\x00\x00\x00\x01\x00\x05\x00\x01\x03\x00\x01\xff\x00\x03\x00\x00\x00"),
1120+
tableData: []byte("p\x03\x00\x00\x00\x00\x01\x00\x04test\x00\x04test\x00\x01\x03\x00\x01\x01\x01\x00"),
1121+
eventType: WRITE_ROWS_EVENTv2,
1122+
expectPartitionId: 0x3,
1123+
expectSourcePartitionId: 0x0,
1124+
expectNdbFormat: 0x0,
1125+
expectNdbData: []byte(nil),
1126+
},
1127+
{
1128+
data: []byte("p\x03\x00\x00\x00\x00\x01\x00\a\x00\x01\x01\x00\x03\x00\x01\xff\xff\x00\x03\x00\x00\x00\x00\x01\x00\x00\x00"),
1129+
tableData: []byte("p\x03\x00\x00\x00\x00\x01\x00\x04test\x00\x04test\x00\x01\x03\x00\x01\x01\x01\x00"),
1130+
eventType: UPDATE_ROWS_EVENTv2,
1131+
expectPartitionId: 0x1,
1132+
expectSourcePartitionId: 0x3,
1133+
expectNdbFormat: 0x0,
1134+
expectNdbData: []byte(nil),
1135+
},
1136+
// mysql 5.7 and mariadb 14(15) does not surpot extra data
1137+
{
1138+
data: []byte("m\x00\x00\x00\x00\x00\x01\x00\x02\x00\x01\xff\xfe\x03\x00\x00\x00"),
1139+
tableData: []byte("m\x00\x00\x00\x00\x00\x01\x00\x04test\x00\x04test\x00\x01\x03\x00\x01"),
1140+
eventType: WRITE_ROWS_EVENTv2,
1141+
expectPartitionId: 0x0,
1142+
expectSourcePartitionId: 0x0,
1143+
expectNdbFormat: 0x0,
1144+
expectNdbData: []byte(nil),
1145+
},
1146+
{
1147+
data: []byte("m\x00\x00\x00\x00\x00\x01\x00\x02\x00\x01\xff\xff\xfe\x03\x00\x00\x00\xfe\x01\x00\x00\x00"),
1148+
tableData: []byte("m\x00\x00\x00\x00\x00\x01\x00\x04test\x00\x04test\x00\x01\x03\x00\x01"),
1149+
eventType: UPDATE_ROWS_EVENTv2,
1150+
expectPartitionId: 0x0,
1151+
expectSourcePartitionId: 0x0,
1152+
expectNdbFormat: 0x0,
1153+
expectNdbData: []byte(nil),
1154+
},
1155+
}
1156+
1157+
for _, tc := range testcases {
1158+
tableMapEvent := new(TableMapEvent)
1159+
tableMapEvent.tableIDSize = 6
1160+
err := tableMapEvent.Decode(tc.tableData)
1161+
require.NoError(t, err)
1162+
1163+
rowsEvent := new(RowsEvent)
1164+
rowsEvent.tableIDSize = 6
1165+
rowsEvent.tables = make(map[uint64]*TableMapEvent)
1166+
rowsEvent.tables[tableMapEvent.TableID] = tableMapEvent
1167+
rowsEvent.Version = 2
1168+
rowsEvent.eventType = tc.eventType
1169+
1170+
err = rowsEvent.Decode(tc.data)
1171+
require.NoError(t, err)
1172+
require.Equal(t, tc.expectPartitionId, rowsEvent.PartitionId)
1173+
require.Equal(t, tc.expectSourcePartitionId, rowsEvent.SourcePartitionId)
1174+
require.Equal(t, tc.expectNdbFormat, rowsEvent.NdbFormat)
1175+
require.Equal(t, tc.expectNdbData, rowsEvent.NdbData)
1176+
}
1177+
}
1178+
10591179
func TestTableMapHelperMaps(t *testing.T) {
10601180
/*
10611181
CREATE TABLE `_types` (

0 commit comments

Comments
 (0)