remove DumpEntry.Err

Author: ffffwh

driver/common/dumper.go

@@ -27,7 +27,10 @@ type Dumper struct {
 	Table              *Table
 	Iteration          int64
 	Columns            string
+	// ResultsChannel should be closed after writing all entries.
 	ResultsChannel     chan *DumpEntry
+	// Set Err (if there is) before closing ResultsChannel.
+	Err                error
 	shutdown           bool
 	ShutdownCh         chan struct{}
 	shutdownLock       sync.Mutex
@@ -71,6 +74,7 @@ func (d *Dumper) Dump() error {
 
 			nRows, err := d.GetChunkData()
 			if err != nil {
+				d.Err = err
 				d.Logger.Error("error at dump", "err", err)
 				break
 			}

driver/common/type.schema

@@ -7,7 +7,6 @@ struct DumpEntry {
 	TbSQL           []string
 	ValuesX    [][]*[]byte
 	TotalCount int64
-	Err        string
 	Table      []byte
 	ColumnMapTo []string
 }

driver/common/type.schema.gen.go

@@ -21,7 +21,6 @@ type DumpEntry struct {
 	TbSQL           []string
 	ValuesX         [][]*[]byte
 	TotalCount      int64
-	Err             string
 	Table           []byte
 	ColumnMapTo     []string
 }
@@ -228,21 +227,6 @@ func (d *DumpEntry) Size() (s uint64) {
 		}
 
 	}
-	{
-		l := uint64(len(d.Err))
-
-		{
-
-			t := l
-			for t >= 0x80 {
-				t >>= 7
-				s++
-			}
-			s++
-
-		}
-		s += l
-	}
 	{
 		l := uint64(len(d.Table))
 
@@ -560,25 +544,6 @@ func (d *DumpEntry) Marshal(buf []byte) ([]byte, error) {
 		buf[i+7+0] = byte(d.TotalCount >> 56)
 
 	}
-	{
-		l := uint64(len(d.Err))
-
-		{
-
-			t := uint64(l)
-
-			for t >= 0x80 {
-				buf[i+8] = byte(t) | 0x80
-				t >>= 7
-				i++
-			}
-			buf[i+8] = byte(t)
-			i++
-
-		}
-		copy(buf[i+8:], d.Err)
-		i += l
-	}
 	{
 		l := uint64(len(d.Table))
 
@@ -924,26 +889,6 @@ func (d *DumpEntry) Unmarshal(buf []byte) (uint64, error) {
 	{
 		l := uint64(0)
 
-		{
-
-			bs := uint8(7)
-			t := uint64(buf[i+8] & 0x7F)
-			for buf[i+8]&0x80 == 0x80 {
-				i++
-				t |= uint64(buf[i+8]&0x7F) << bs
-				bs += 7
-			}
-			i++
-
-			l = t
-
-		}
-		d.Err = string(buf[i+8 : i+8+l])
-		i += l
-	}
-	{
-		l := uint64(0)
-
 		{
 
 			bs := uint8(7)

driver/mysql/dumper.go

@@ -161,7 +161,7 @@ func (d *dumper) getChunkData() (nRows int64, err error) {
 	}
 	defer func() {
 		if err != nil {
-			entry.Err = err.Error()
+			return
 		}
 		if err == nil && len(entry.ValuesX) == 0 {
 			return

driver/mysql/extractor.go

@@ -1383,12 +1383,9 @@ func (e *Extractor) mysqlDump() error {
 	e.logger.Info("Step: scanning contents of x tables", "n", step, "x", e.tableCount)
 	startScan := g.CurrentTimeMillis()
 	counter := 0
-	//pool := models.NewPool(10)
 	for _, db := range e.replicateDoDb {
 		for _, tbCtx := range db.TableMap {
 			t := tbCtx.Table
-			//pool.Add(1)
-			//go func(t *config.Table) {
 			counter++
 			// Obtain a record maker for this table, which knows about the schema ...
 			// Choose how we create statements based on the # of rows ...
@@ -1402,34 +1399,29 @@ func (e *Extractor) mysqlDump() error {
 			e.dumpers = append(e.dumpers, d)
 			// Scan the rows in the table ...
 			for entry := range d.ResultsChannel {
-				if entry.Err != "" {
-					e.onError(common.TaskStateDead, fmt.Errorf(entry.Err))
-				} else {
-					memSize := int64(entry.Size())
-					if !d.sentTableDef {
-						tableBs, err := common.EncodeTable(d.Table)
-						if err != nil {
-							err = errors.Wrap(err, "full copy: EncodeTable")
-							e.onError(common.TaskStateDead, err)
-							return err
-						} else {
-							entry.Table = tableBs
-							d.sentTableDef = true
-						}
-					}
-					if err = e.encodeAndSendDumpEntry(entry); err != nil {
+				memSize := int64(entry.Size())
+				if !d.sentTableDef {
+					tableBs, err := common.EncodeTable(d.Table)
+					if err != nil {
+						err = errors.Wrap(err, "full copy: EncodeTable")
 						e.onError(common.TaskStateDead, err)
+						return err
+					} else {
+						entry.Table = tableBs
+						d.sentTableDef = true
 					}
-					atomic.AddInt64(&e.TotalRowsCopied, int64(len(entry.ValuesX)))
-					atomic.AddInt64(d.Memory, -memSize)
 				}
+				if err = e.encodeAndSendDumpEntry(entry); err != nil {
+					e.onError(common.TaskStateDead, err)
+				}
+				atomic.AddInt64(&e.TotalRowsCopied, int64(len(entry.ValuesX)))
+				atomic.AddInt64(d.Memory, -memSize)
+			}
+			if d.Err != nil {
+				e.onError(common.TaskStateDead, d.Err)
 			}
-
-			//pool.Done()
-			//}(tb)
 		}
 	}
-	//pool.Wait()
 	step++
 
 	// We've copied all of the tables, but our buffer holds onto the very last record.

driver/oracle/extractor/dumper.go

@@ -55,7 +55,7 @@ func (d *dumper) getChunkData() (nRows int64, err error) {
 	// SELECT * FROM %s.%s AS OF SCN %d where ROWNUM < %d
 	defer func() {
 		if err != nil {
-			entry.Err = err.Error()
+			return
 		}
 		if err == nil && len(entry.ValuesX) == 0 {
 			return

driver/oracle/extractor/extractor_oracle.go

@@ -964,14 +964,13 @@ func (e *ExtractorOracle) oracleDump() error {
 
 			// Scan the rows in the table ...
 			for entry := range d.ResultsChannel {
-				if entry.Err != "" {
-					e.onError(common.TaskStateDead, fmt.Errorf(entry.Err))
-				} else {
-					if err := e.encodeAndSendDumpEntry(entry); err != nil {
-						e.onError(common.TaskStateDead, err)
-					}
+				if err := e.encodeAndSendDumpEntry(entry); err != nil {
+					e.onError(common.TaskStateDead, err)
 				}
 			}
+			if d.Err != nil {
+				e.onError(common.TaskStateDead, d.Err)
+			}
 		}
 	}
 	e.fullCopyCoordinates = &common.OracleCoordinates{