Decoding of compressed binlog events #773

dveeden · 2023-02-21T21:01:37Z

Closes: #533

dveeden · 2023-02-21T21:14:52Z

With this PR and this code:

package main

import (
	"context"
	"os"

	"github.com/go-mysql-org/go-mysql/mysql"
	"github.com/go-mysql-org/go-mysql/replication"
)

func main() {
	cfg := replication.BinlogSyncerConfig{
		ServerID: 100,
		Flavor:   "mysql",
		Host:     "127.0.0.1",
		Port:     8032,
		User:     "root",
		Password: "msandbox",
	}
	syncer := replication.NewBinlogSyncer(cfg)

	gtidSet, _ := mysql.ParseMysqlGTIDSet("")
	streamer, _ := syncer.StartSyncGTID(gtidSet)

	for {
		ev, _ := streamer.GetEvent(context.Background())
		ev.Dump(os.Stdout)
	}
}

And then on MySQL 8.0.32 with compression enabled:

INSERT INTO t1 VALUES(111, "AAAAAAAAAAABBBBBBBBBBBBBBBCCCCCCCCCCCCCCC");

mysql> SHOW BINLOG EVENTS;
+---------------+------+---------------------+-----------+-------------+----------------------------------------------------------------------------------+
| Log_name      | Pos  | Event_type          | Server_id | End_log_pos | Info                                                                             |
+---------------+------+---------------------+-----------+-------------+----------------------------------------------------------------------------------+

...
| binlog.000001 | 1371 | Gtid                |         1 |        1450 | SET @@SESSION.GTID_NEXT= '00008032-0000-0000-0000-000000008032:6'                |
| binlog.000001 | 1450 | Transaction_payload |         1 |        1629 | compression='ZSTD', decompressed_size=227 bytes                                  |
| binlog.000001 | 1629 | Query               |         1 |        1629 | BEGIN                                                                            |
| binlog.000001 | 1629 | Table_map           |         1 |        1629 | table_id: 239 (test.t1)                                                          |
| binlog.000001 | 1629 | Write_rows          |         1 |        1629 | table_id: 239 flags: STMT_END_F                                                  |
| binlog.000001 | 1629 | Xid                 |         1 |        1629 | COMMIT /* xid=11002 */                                                           |

Then this is the output I get:

=== GTIDEvent ===
Date: 2023-02-21 21:59:22
Log position: 1450
Event size: 79
Commit flag: 0
GTID_NEXT: 00008032-0000-0000-0000-000000008032:6
LAST_COMMITTED: 5
SEQUENCE_NUMBER: 6
Immediate commmit timestamp: 1677013162351965 (2023-02-21T21:59:22.351965+01:00)
Orignal commmit timestamp: 1677013162351965 (2023-02-21T21:59:22.351965+01:00)
Transaction length: 258
Immediate server version: 80032
Orignal server version: 80032

=== TransactionPlayloadEvent ===
Date: 2023-02-21 21:59:22
Log position: 1629
Event size: 179
Payload Size: 146
Payload Uncompressed Size: 227
Payload CompressionType: ZSTD
Payload Body: 
00000000  28 b5 2f fd 00 58 34 04  00 32 88 1a 24 90 29 e9  |(./..X4..2..$.).|
00000010  b0 8b 40 ee ee 6e 6f 7e  77 a7 05 bf 81 8c 99 23  |[email protected]~w......#|
00000020  6a 7b 5e bb ac 81 81 1e  b1 3d 9b 4e 3f 26 64 cb  |j{^......=.N?&d.|
00000030  14 ff af c0 5c f8 2b 48  b7 da c3 71 9c f8 2b 3f  |....\.+H...q..+?|
00000040  1c 25 7f 0f e9 56 5b 14  6f 31 65 30 90 70 06 43  |.%...V[.o1e0.p.C|
00000050  75 bf dc f8 4b 48 b7 5a  04 02 3d ce 2c a2 b0 df  |u...KH.Z..=.,...|
00000060  e0 f7 fb a9 09 8b 12 e0  ff b1 8c e8 86 57 37 79  |.............W7y|
00000070  33 f4 17 d3 ad 16 0b 00  80 70 20 33 20 05 0e 4e  |3........p 3 ..N|
00000080  57 35 47 52 e2 40 68 05  a4 c1 3d d0 1e 06 e6 01  |W5GR.@h...=.....|
00000090  00 00                                             |..|
Decompressed: 
00000000  aa 30 f5 63 02 01 00 00  00 47 00 00 00 00 00 00  |.0.c.....G......|
00000010  00 08 00 63 08 00 00 00  00 00 00 04 00 00 1d 00  |...c............|
00000020  00 00 00 00 00 01 20 00  a0 45 00 00 00 00 06 03  |...... ..E......|
00000030  73 74 64 04 ff 00 ff 00  ff 00 12 ff 00 74 65 73  |std..........tes|
00000040  74 00 42 45 47 49 4e aa  30 f5 63 13 01 00 00 00  |t.BEGIN.0.c.....|
00000050  32 00 00 00 00 00 00 00  00 00 ef 00 00 00 00 00  |2...............|
00000060  01 00 04 74 65 73 74 00  02 74 31 00 02 03 0f 02  |...test..t1.....|
00000070  fc 03 02 01 01 00 02 01  2d aa 30 f5 63 1e 01 00  |........-.0.c...|
00000080  00 00 4f 00 00 00 00 00  00 00 00 00 ef 00 00 00  |..O.............|
00000090  00 00 01 00 02 00 02 ff  00 6f 00 00 00 29 00 41  |.........o...).A|
000000a0  41 41 41 41 41 41 41 41  41 41 42 42 42 42 42 42  |AAAAAAAAAABBBBBB|
000000b0  42 42 42 42 42 42 42 42  42 43 43 43 43 43 43 43  |BBBBBBBBBCCCCCCC|
000000c0  43 43 43 43 43 43 43 43  aa 30 f5 63 10 01 00 00  |CCCCCCCC.0.c....|
000000d0  00 1b 00 00 00 00 00 00  00 00 00 fa 2a 00 00 00  |............*...|
000000e0  00 00 00                                          |...|

From mysqlbinlog:

# at 1371
#230221 21:59:22 server id 1  end_log_pos 1450 CRC32 0xc9682e44 	GTID	last_committed=5	sequence_number=6	rbr_only=yes	original_committed_timestamp=1677013162351965	immediate_commit_timestamp=1677013162351965	transaction_length=258
/*!50718 SET TRANSACTION ISOLATION LEVEL READ COMMITTED*//*!*/;
# original_commit_timestamp=1677013162351965 (2023-02-21 21:59:22.351965 CET)
# immediate_commit_timestamp=1677013162351965 (2023-02-21 21:59:22.351965 CET)
/*!80001 SET @@session.original_commit_timestamp=1677013162351965*//*!*/;
/*!80014 SET @@session.original_server_version=80032*//*!*/;
/*!80014 SET @@session.immediate_server_version=80032*//*!*/;
SET @@SESSION.GTID_NEXT= '00008032-0000-0000-0000-000000008032:6'/*!*/;
# at 1450
#230221 21:59:22 server id 1  end_log_pos 1629 CRC32 0xe045fa5c 	Transaction_Payload		payload_size=146	compression_type=ZSTD	uncompressed_size=227
# Start of compressed events!
# at 1629
#230221 21:59:22 server id 1  end_log_pos 1629 CRC32 0xad7e8ae5 	Query	thread_id=2147	exec_time=0	error_code=0
SET TIMESTAMP=1677013162/*!*/;
BEGIN
/*!*/;
# at 1629
#230221 21:59:22 server id 1  end_log_pos 1629 CRC32 0xe9e840a7 	Table_map: `test`.`t1` mapped to number 239
# has_generated_invisible_primary_key=0
# at 1629
#230221 21:59:22 server id 1  end_log_pos 1629 CRC32 0x56c735d9 	Write_rows: table id 239 flags: STMT_END_F

BINLOG '
qjD1YxMBAAAANgAAAAAAAAAAAO8AAAAAAAEABHRlc3QAAnQxAAIDDwL8AwIBAQACAS2nQOjp
qjD1Yx4BAAAAUwAAAAAAAAAAAO8AAAAAAAEAAgAC/wBvAAAAKQBBQUFBQUFBQUFBQUJCQkJCQkJC
QkJCQkJCQkNDQ0NDQ0NDQ0NDQ0NDQ9k1x1Y=
'/*!*/;
# at 1629
#230221 21:59:22 server id 1  end_log_pos 1629 CRC32 0xe9449ff5 	Xid = 11002
COMMIT/*!*/;
# End of compressed events!

dveeden · 2023-02-22T20:47:48Z

With b29c29f it now decodes the events that are embedded in the payload of TransactionPayloadEvent.

=== GTIDEvent ===
Date: 2023-02-21 21:59:22
Log position: 1450
Event size: 79
Commit flag: 0
GTID_NEXT: 00008032-0000-0000-0000-000000008032:6
LAST_COMMITTED: 5
SEQUENCE_NUMBER: 6
Immediate commmit timestamp: 1677013162351965 (2023-02-21T21:59:22.351965+01:00)
Orignal commmit timestamp: 1677013162351965 (2023-02-21T21:59:22.351965+01:00)
Transaction length: 258
Immediate server version: 80032
Orignal server version: 80032

=== TransactionPlayloadEvent ===
Date: 2023-02-21 21:59:22
Log position: 1629
Event size: 179
Payload Size: 146
Payload Uncompressed Size: 227
Payload CompressionType: ZSTD
Payload Body: 
00000000  28 b5 2f fd 00 58 34 04  00 32 88 1a 24 90 29 e9  |(./..X4..2..$.).|
00000010  b0 8b 40 ee ee 6e 6f 7e  77 a7 05 bf 81 8c 99 23  |[email protected]~w......#|
00000020  6a 7b 5e bb ac 81 81 1e  b1 3d 9b 4e 3f 26 64 cb  |j{^......=.N?&d.|
00000030  14 ff af c0 5c f8 2b 48  b7 da c3 71 9c f8 2b 3f  |....\.+H...q..+?|
00000040  1c 25 7f 0f e9 56 5b 14  6f 31 65 30 90 70 06 43  |.%...V[.o1e0.p.C|
00000050  75 bf dc f8 4b 48 b7 5a  04 02 3d ce 2c a2 b0 df  |u...KH.Z..=.,...|
00000060  e0 f7 fb a9 09 8b 12 e0  ff b1 8c e8 86 57 37 79  |.............W7y|
00000070  33 f4 17 d3 ad 16 0b 00  80 70 20 33 20 05 0e 4e  |3........p 3 ..N|
00000080  57 35 47 52 e2 40 68 05  a4 c1 3d d0 1e 06 e6 01  |W5GR.@h...=.....|
00000090  00 00                                             |..|
Decompressed: 
00000000  aa 30 f5 63 02 01 00 00  00 47 00 00 00 00 00 00  |.0.c.....G......|
00000010  00 08 00 63 08 00 00 00  00 00 00 04 00 00 1d 00  |...c............|
00000020  00 00 00 00 00 01 20 00  a0 45 00 00 00 00 06 03  |...... ..E......|
00000030  73 74 64 04 ff 00 ff 00  ff 00 12 ff 00 74 65 73  |std..........tes|
00000040  74 00 42 45 47 49 4e aa  30 f5 63 13 01 00 00 00  |t.BEGIN.0.c.....|
00000050  32 00 00 00 00 00 00 00  00 00 ef 00 00 00 00 00  |2...............|
00000060  01 00 04 74 65 73 74 00  02 74 31 00 02 03 0f 02  |...test..t1.....|
00000070  fc 03 02 01 01 00 02 01  2d aa 30 f5 63 1e 01 00  |........-.0.c...|
00000080  00 00 4f 00 00 00 00 00  00 00 00 00 ef 00 00 00  |..O.............|
00000090  00 00 01 00 02 00 02 ff  00 6f 00 00 00 29 00 41  |.........o...).A|
000000a0  41 41 41 41 41 41 41 41  41 41 42 42 42 42 42 42  |AAAAAAAAAABBBBBB|
000000b0  42 42 42 42 42 42 42 42  42 43 43 43 43 43 43 43  |BBBBBBBBBCCCCCCC|
000000c0  43 43 43 43 43 43 43 43  aa 30 f5 63 10 01 00 00  |CCCCCCCC.0.c....|
000000d0  00 1b 00 00 00 00 00 00  00 00 00 fa 2a 00 00 00  |............*...|
000000e0  00 00 00                                          |...|
=== QueryEvent ===
Date: 2023-02-21 21:59:22
Log position: 0
Event size: 71
Slave proxy ID: 2147
Execution time: 0
Error code: 0
Schema: test
Query: BEGIN

=== TableMapEvent ===
Date: 2023-02-21 21:59:22
Log position: 0
Event size: 50
TableID: 239
TableID size: 6
Flags: 1
Schema: test
Table: t1
Column count: 2
Column type: 
00000000  03 0f                                             |..|
NULL bitmap: 
00000000  02                                                |.|
Signedness bitmap: 
00000000  00                                                |.|
Default charset: [45]
Column charset: []
Set str value: []
Enum str value: []
Column name: []
Geometry type: []
Primary key: []
Primary key prefix: []
Enum/set default charset: []
Enum/set column charset: []
UnsignedMap: map[int]bool{0:false}
CollationMap: map[int]uint64{1:0x2d}
EnumSetCollationMap: map[int]uint64(nil)
EnumStrValueMap: map[int][]string(nil)
SetStrValueMap: map[int][]string(nil)
GeometryTypeMap: map[int]uint64(nil)
Columns: 
  <n/a>  type=3    unsigned=no   null=no 
  <n/a>  type=15   collation=45   null=yes

=== WriteRowsEventV2 ===
Date: 2023-02-21 21:59:22
Log position: 0
Event size: 79
TableID: 239
Flags: 1
Column count: 2
Values:
--
0:111
1:"AAAAAAAAAAABBBBBBBBBBBBBBBCCCCCCCCCCCCCCC"

=== XIDEvent ===
Date: 2023-02-21 21:59:22
Log position: 0
Event size: 27
XID: 11002

dveeden · 2023-02-22T20:50:05Z

@lance6716 @atercattus what would you suggest for what the next step is?

One problem is that the github.com/klauspost/compress/zstd dependency is causing failures on older Go versions.
update: In f9c9016 I've switched to github.com/DataDog/zstd

…zstd

lance6716

rest lgtm

replication/const.go

replication/transaction_payload_event.go

replication/const.go

replication/transaction_payload_event.go

lance6716 · 2023-02-24T03:40:12Z

@lance6716 @atercattus what would you suggest for what the next step is?

I have no suggestion for TransactionPayloadEvent, this PR is great 👍 And I can't remember more incompatibility with MySQL 8.0 for now.

Co-authored-by: lance6716 <[email protected]>

atercattus · 2023-02-24T08:56:13Z

And I can't remember more incompatibility with MySQL 8.0 for now.

I'm working on support for PARTIAL_UPDATE_ROWS_EVENT. Except this, I also don't know any mysql8 incompatibilities.

dveeden · 2023-02-24T09:02:26Z

@lance6716 I think I've addressed all your comments. PTAL

@atercattus feel free to poke me if there is something I can do to help with PARTIAL_UPDATE_ROWS_EVENT

atercattus · 2023-02-24T09:21:21Z

@dveeden, thanks. I hope, I can do this myself :) Also, this PR is not directly related to PARTIAL_UPDATE_ROWS_EVENT.

Decoding of compressed binlog events

29ab7e7

dveeden force-pushed the compressed_binlog_events branch from 2e66968 to 29ab7e7 Compare February 21, 2023 21:22

Decode the uncompressed payload from TransactionPayloadEvent

b29c29f

dveeden force-pushed the compressed_binlog_events branch from e10c715 to b29c29f Compare February 22, 2023 20:47

dveeden added 2 commits February 22, 2023 21:59

Fix typo

7dbc07b

Switch from github.com/klauspost/compress/zstd to github.com/DataDog/…

f9c9016

…zstd

dveeden force-pushed the compressed_binlog_events branch from 41d5375 to f9c9016 Compare February 22, 2023 21:15

Fix issue found by linter: Remove fieldTypeName

846ca9c

dveeden marked this pull request as ready for review February 22, 2023 21:18

dveeden added 3 commits February 23, 2023 16:42

Move decoding from Dump() to Decode()

eb4c15e

Slight change in naming

f0abf29

decodePayload, doesn't use the data argument

ede7ac9

lance6716 reviewed Feb 24, 2023

View reviewed changes

dveeden and others added 5 commits February 24, 2023 08:24

Update replication/const.go

c7a5476

Co-authored-by: lance6716 <[email protected]>

Update based on review

527b681

A few more checks

1a0cfc7

Only use format, not the whole parser

e5c8adc

Merge branch 'master' into compressed_binlog_events

f60b982

atercattus approved these changes Feb 24, 2023

View reviewed changes

lance6716 approved these changes Feb 25, 2023

View reviewed changes

lance6716 merged commit e320144 into go-mysql-org:master Feb 25, 2023

froot mentioned this pull request May 8, 2024

Fix bug in handling sub events of replication.TransactionPayloadEvent #875

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Decoding of compressed binlog events #773

Decoding of compressed binlog events #773

dveeden commented Feb 21, 2023 •

edited

Loading

dveeden commented Feb 21, 2023 •

edited

Loading

dveeden commented Feb 22, 2023

dveeden commented Feb 22, 2023 •

edited

Loading

lance6716 left a comment

lance6716 commented Feb 24, 2023

atercattus commented Feb 24, 2023

dveeden commented Feb 24, 2023

atercattus commented Feb 24, 2023

Decoding of compressed binlog events #773

Decoding of compressed binlog events #773

Conversation

dveeden commented Feb 21, 2023 • edited Loading

dveeden commented Feb 21, 2023 • edited Loading

dveeden commented Feb 22, 2023

dveeden commented Feb 22, 2023 • edited Loading

lance6716 left a comment

Choose a reason for hiding this comment

lance6716 commented Feb 24, 2023

atercattus commented Feb 24, 2023

dveeden commented Feb 24, 2023

atercattus commented Feb 24, 2023

dveeden commented Feb 21, 2023 •

edited

Loading

dveeden commented Feb 21, 2023 •

edited

Loading

dveeden commented Feb 22, 2023 •

edited

Loading