dump: getChunkData: fetch all rows and split

misc: logs

Author: ffffwh

driver/mysql/dumper.go

@@ -9,6 +9,7 @@ package mysql
 import (
 	"context"
 	"fmt"
+	"github.com/pkg/errors"
 	"os"
 	"strings"
 	"sync/atomic"
@@ -182,10 +183,8 @@ func (d *dumper) getChunkData() (nRows int64, err error) {
 	d.Iteration += 1
 	rows, err := d.db.Query(query)
 	if err != nil {
-		d.Logger.Debug("error at select chunk. query: ", query)
-		newErr := fmt.Errorf("error at select chunk. err: %v", err)
-		d.Logger.Error(newErr.Error())
-		return 0, err
+		d.Logger.Error("error at select chunk", "query", query)
+		return 0, errors.Wrap(err, "select chunk")
 	}
 
 	columns, err := rows.Columns()
@@ -264,9 +263,9 @@ func (d *dumper) getChunkData() (nRows int64, err error) {
 	entrySize := 0
 	var valuesX [][]*[]byte
 	scanArgs := make([]interface{}, len(columns)) // tmp use, for casting `values` to `[]interface{}`
+	splitPoints := []int{0}
 	hasRow := rows.Next()
 	for hasRow {
-		nRows += 1
 		rowValuesRaw := make([]*[]byte, len(columns))
 		for i := range rowValuesRaw {
 			scanArgs[i] = &rowValuesRaw[i]
@@ -283,14 +282,20 @@ func (d *dumper) getChunkData() (nRows int64, err error) {
 		entrySize += getRowSize(rowValuesRaw)
 
 		if !hasRow || entrySize >= d.dumpEntryLimit {
-			d.Logger.Debug("reach DUMP_ENTRY_LIMIT.", "size", entrySize, "n_row", len(valuesX))
-
-			err = handleEntry(valuesX, !hasRow)
-			if err != nil {
-				return 0, err
-			}
+			d.Logger.Debug("reach dumpEntryLimit.", "size", entrySize, "point", len(valuesX))
+			splitPoints = append(splitPoints, len(valuesX))
 			entrySize = 0
-			valuesX = nil
+		}
+	}
+	if err = rows.Err(); err != nil {
+		return 0, err
+	}
+	nRows = int64(len(valuesX))
+
+	for i := 1; i < len(splitPoints); i++ {
+		err = handleEntry(valuesX[splitPoints[i-1]:splitPoints[i]], i == len(splitPoints)-1)
+		if err != nil {
+			return 0, err
 		}
 	}
 

driver/mysql/extractor.go

@@ -1060,17 +1060,14 @@ func (e *Extractor) StreamEvents() error {
 // retryOperation attempts up to `count` attempts at running given function,
 // exiting as soon as it returns with non-error.
 // gno: only for logging
-func (e *Extractor) publish(subject string, txMsg []byte, gno int64) (err error) {
-	msgLen := len(txMsg)
-
-	data := txMsg
+func (e *Extractor) publish(subject string, data []byte, gno int64) (err error) {
 	lenData := len(data)
 
 	// lenData < NatsMaxMsg: 1 msg
 	// lenData = k * NatsMaxMsg + b, where k >= 1 && b >= 0: (k+1) msg
 	// b could be 0. we send a zero-len msg as a sign of termination.
 	nSeg := lenData/g.NatsMaxMsg + 1
-	e.logger.Debug("publish. msg", "subject", subject, "gno", gno, "nSeg", nSeg, "spanLen", lenData, "msgLen", msgLen)
+	e.logger.Debug("publish. msg", "subject", subject, "gno", gno, "nSeg", nSeg, "spanLen", lenData)
 	bak := make([]byte, 4)
 	if nSeg > 1 {
 		// ensure there are 4 bytes to save iSeg
@@ -1439,10 +1436,12 @@ func (e *Extractor) encodeAndSendDumpEntry(entry *common.DumpEntry) error {
 	if err != nil {
 		return err
 	}
+	e.logger.Debug("encodeAndSendDumpEntry. after Marshal", "size", len(bs))
 	txMsg, err := common.Compress(bs)
 	if err != nil {
 		return errors.Wrap(err, "common.Compress")
 	}
+	e.logger.Debug("encodeAndSendDumpEntry. after Compress", "size", len(txMsg))
 	if err := e.publish(fmt.Sprintf("%s_full", e.subject), txMsg, 0); err != nil {
 		return err
 	}