get full table data as of scn

Author: LordofAvernus

driver/mysql/applier.go

@@ -327,8 +327,9 @@ func (a *Applier) Run() {
 	}
 	a.ai.OnError = a.onError
 
-	go a.updateGtidLoop()
-
+	if sourceType == "mysql" {
+		go a.updateGtidLoop()
+	}
 	if a.stage != JobFullCopy {
 		a.stage = JobFullCopy
 		a.sendEvent(JobFullCopy)

driver/oracle/extractor/dumper.go

@@ -1,6 +1,11 @@
 package extractor
 
 import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"time"
+
 	"github.com/actiontech/dtle/driver/common"
 	"github.com/actiontech/dtle/driver/oracle/config"
 
@@ -9,15 +14,20 @@ import (
 
 type dumper struct {
 	*common.Dumper
-	db *config.OracleDB
+	db          *config.OracleDB
+	snapshotSCN int64
 }
 
 func NewDumper(db *config.OracleDB, table *common.Table, chunkSize int64,
-	logger g.LoggerType, memory *int64) *dumper {
-	return &dumper{
+	logger g.LoggerType, memory *int64, scn int64) *dumper {
+	d := &dumper{
 		common.NewDumper(table, chunkSize, logger, memory),
 		db,
+		scn,
 	}
+	d.PrepareForDumping = d.prepareForDumping
+	d.GetChunkData = d.getChunkData
+	return d
 }
 
 func (d *dumper) prepareForDumping() error {
@@ -39,12 +49,95 @@ func (d *dumper) getChunkData() (nRows int64, err error) {
 		entry.TableSchema = d.Table.TableSchemaRename
 	}
 
-	// build query sql  (SELECT * FROM x.x LIMIT x OFFSET x AS OF SCN 123)
+	// build query sql
+	// SELECT * FROM %s.%s AS OF SCN %d where ROWNUM <= %d
+	// minus
+	// SELECT * FROM %s.%s AS OF SCN %d where ROWNUM < %d
+	defer func() {
+		if err != nil {
+			entry.Err = err.Error()
+		}
+		if err == nil && len(entry.ValuesX) == 0 {
+			return
+		}
+
+		keepGoing := true
+		timer := time.NewTicker(time.Second)
+		defer timer.Stop()
+		for keepGoing {
+			select {
+			case <-d.ShutdownCh:
+				keepGoing = false
+			case d.ResultsChannel <- entry:
+				atomic.AddInt64(d.Memory, int64(entry.Size()))
+				//d.logger.Debug("*** memory", "memory", atomic.LoadInt64(d.memory))
+				keepGoing = false
+			case <-timer.C:
+				d.Logger.Debug("resultsChannel full. waiting and ping conn")
+				errPing := d.db.MetaDataConn.PingContext(context.TODO())
+				if errPing != nil {
+					d.Logger.Debug("ping query row got error.", "err", errPing)
+				}
+			}
+		}
+		d.Logger.Debug("resultsChannel", "n", len(d.ResultsChannel))
+	}()
+
+	query := fmt.Sprintf(
+		`SELECT * FROM %s.%s AS OF SCN %d where ROWNUM <= %d
+		minus 
+		SELECT * FROM %s.%s AS OF SCN %d where ROWNUM < %d `,
+		d.TableSchema,
+		d.TableName,
+		d.snapshotSCN,
+		(d.Iteration+1)*d.ChunkSize,
+		d.TableSchema,
+		d.TableName,
+		d.snapshotSCN,
+		(d.Iteration*d.ChunkSize)+1,
+	)
 
-	// offset
+	d.Logger.Debug("getChunkData.", "query", query)
+
+	// this must be increased after building query
 	d.Iteration += 1
-	// exec sql
-	// d.db.MetaDataConn.QueryRowContext()
+	rows, err := d.db.MetaDataConn.QueryContext(context.TODO(), query)
+	if err != nil {
+		newErr := fmt.Errorf("error at select chunk. err: %v", err)
+		d.Logger.Error(newErr.Error())
+		return 0, err
+	}
+
+	columns, err := rows.Columns()
+	if err != nil {
+		return 0, err
+	}
+
+	scanArgs := make([]interface{}, len(columns)) // tmp use, for casting `values` to `[]interface{}`
+
+	for rows.Next() {
+		rowValuesRaw := make([]*[]byte, len(columns))
+		for i := range rowValuesRaw {
+			scanArgs[i] = &rowValuesRaw[i]
+		}
+
+		err = rows.Scan(scanArgs...)
+		if err != nil {
+			return 0, err
+		}
+
+		entry.ValuesX = append(entry.ValuesX, rowValuesRaw)
+	}
+
+	nRows = int64(len(entry.ValuesX))
+	d.Logger.Debug("getChunkData.", "n_row", nRows)
+
+	if d.Table.TableRename != "" {
+		entry.TableName = d.Table.TableRename
+	}
+	if d.Table.TableSchemaRename != "" {
+		entry.TableSchema = d.Table.TableSchemaRename
+	}
 
 	return nRows, nil
 }

driver/oracle/extractor/extractor_oracle.go

@@ -237,21 +237,21 @@ func (e *ExtractorOracle) Run() {
 			e.onError(common.TaskStateDead, err)
 			return
 		}
-	}
 
-	{
-		startSCN, committedSCN, err := e.calculateSCNPos()
-		if err != nil {
-			e.onError(common.TaskStateDead, errors.Wrap(err, "calculateSCNPos"))
-			return
-		}
-		e.LogMinerStream = NewLogMinerStream(e.oracleDB, e.logger, e.mysqlContext.ReplicateDoDb, e.mysqlContext.ReplicateIgnoreDb,
-			startSCN, committedSCN, 100000)
-		e.logger.Debug("start .initiateStreaming before")
-		if err := e.initiateStreaming(); err != nil {
-			e.logger.Error("error at initiateStreaming", "err", err)
-			e.onError(common.TaskStateDead, err)
-			return
+		{
+			startSCN, committedSCN, err := e.calculateSCNPos()
+			if err != nil {
+				e.onError(common.TaskStateDead, errors.Wrap(err, "calculateSCNPos"))
+				return
+			}
+			e.LogMinerStream = NewLogMinerStream(e.oracleDB, e.logger, e.mysqlContext.ReplicateDoDb, e.mysqlContext.ReplicateIgnoreDb,
+				startSCN, committedSCN, 100000)
+			e.logger.Debug("start .initiateStreaming before")
+			if err := e.initiateStreaming(); err != nil {
+				e.logger.Error("error at initiateStreaming", "err", err)
+				e.onError(common.TaskStateDead, err)
+				return
+			}
 		}
 	}
 }
@@ -872,7 +872,7 @@ func (e *ExtractorOracle) oracleDump() error {
 	}
 
 	// step 2 : get current scn for d.Dump()
-	_, err := (&LogMinerStream{oracleDB: e.oracleDB}).GetCurrentSnapshotSCN()
+	currentSCN, err := (&LogMinerStream{oracleDB: e.oracleDB}).GetCurrentSnapshotSCN()
 	if err != nil {
 		return err
 	}
@@ -935,7 +935,7 @@ func (e *ExtractorOracle) oracleDump() error {
 	// todo need merged dumper with mysql dumper
 	for _, db := range e.replicateDoDb {
 		for _, t := range db.Tables {
-			d := NewDumper(e.oracleDB, t, e.mysqlContext.ChunkSize, e.logger.ResetNamed("dumper"), e.memory1)
+			d := NewDumper(e.oracleDB, t, e.mysqlContext.ChunkSize, e.logger.ResetNamed("dumper"), e.memory1, currentSCN)
 			if err := d.Dump(); err != nil {
 				e.onError(common.TaskStateDead, err)
 			}