Merge branch 'master' into clear

Author: lance6716

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.
@@ -208,6 +212,7 @@ func NewBinlogSyncer(cfg BinlogSyncerConfig) *BinlogSyncer {
 	b.parser.SetUseDecimal(b.cfg.UseDecimal)
 	b.parser.SetUseFloatWithTrailingZero(b.cfg.UseFloatWithTrailingZero)
 	b.parser.SetVerifyChecksum(b.cfg.VerifyChecksum)
+	b.parser.SetPayloadDecoderConcurrency(cfg.PayloadDecoderConcurrency)
 	b.parser.SetRowsEventDecodeFunc(b.cfg.RowsEventDecodeFunc)
 	b.parser.SetTableMapOptionalMetaDecodeFunc(b.cfg.TableMapOptionalMetaDecodeFunc)
 	b.running = false

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
 	}