support setting decoder concurrency for TransactionPayloadEvent binlog (#1064)

Author: hjweddie

replication/binlogsyncer.go

@@ -134,6 +134,10 @@ type BinlogSyncerConfig struct {
 	// Only works with MariaDB flavor.
 	FillZeroLogPos bool
 
+	// PayloadDecoderConcurrency is used to control concurrency for decoding TransactionPayloadEvent.
+	// Default 0,  this will be set to GOMAXPROCS.
+	PayloadDecoderConcurrency int
+
 	// SynchronousEventHandler is used for synchronous event handling.
 	// This should not be used together with StartBackupWithHandler.
 	// If this is not nil, GetEvent does not need to be called.

replication/parser.go

@@ -40,6 +40,8 @@ type BinlogParser struct {
 	ignoreJSONDecodeErr      bool
 	verifyChecksum           bool
 
+	payloadDecoderConcurrency int
+
 	rowsEventDecodeFunc func(*RowsEvent, []byte) error
 
 	tableMapOptionalMetaDecodeFunc func([]byte) error
@@ -215,6 +217,10 @@ func (p *BinlogParser) SetFlavor(flavor string) {
 	p.flavor = flavor
 }
 
+func (p *BinlogParser) SetPayloadDecoderConcurrency(concurrency int) {
+	p.payloadDecoderConcurrency = concurrency
+}
+
 func (p *BinlogParser) SetRowsEventDecodeFunc(rowsEventDecodeFunc func(*RowsEvent, []byte) error) {
 	p.rowsEventDecodeFunc = rowsEventDecodeFunc
 }
@@ -456,6 +462,7 @@ func (p *BinlogParser) newRowsEvent(h *EventHeader) *RowsEvent {
 func (p *BinlogParser) newTransactionPayloadEvent() *TransactionPayloadEvent {
 	e := &TransactionPayloadEvent{}
 	e.format = *p.format
+	e.concurrency = p.payloadDecoderConcurrency
 
 	return e
 }

replication/transaction_payload_event.go

@@ -28,6 +28,7 @@ const (
 
 type TransactionPayloadEvent struct {
 	format           FormatDescriptionEvent
+	concurrency      int
 	Size             uint64
 	UncompressedSize uint64
 	CompressionType  uint64
@@ -103,7 +104,7 @@ func (e *TransactionPayloadEvent) decodePayload() error {
 			e.CompressionType, e.compressionType())
 	}
 
-	decoder, err := zstd.NewReader(nil, zstd.WithDecoderConcurrency(0))
+	decoder, err := zstd.NewReader(nil, zstd.WithDecoderConcurrency(e.concurrency))
 	if err != nil {
 		return err
 	}