// Copyright 2018 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package diff

import (
	"container/heap"
	"context"
	"crypto/md5"
	"database/sql"
	"encoding/json"
	"fmt"
	"github.com/godror/godror"
	"github.com/pingcap/parser/mysql"
	"math/rand"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pingcap/enterprise-toolkit/o2t-sync-diff/pkg/dbutil"
	"github.com/pingcap/enterprise-toolkit/o2t-sync-diff/pkg/utils"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb/parser/model"
	"go.uber.org/zap"
)

var (
	// cancel the context for `Equal`, only used in test
	cancelEqualFunc context.CancelFunc
)

// TableInstance record a table instance
type TableInstance struct {
	Conn       *sql.DB `json:"-"`
	Schema     string  `json:"schema"`
	Table      string  `json:"table"`
	InstanceID string  `json:"instance-id"`
	DBType     string  `json:"db_type"`
	info       *model.TableInfo
	Charset    string `json:"charset"`
}

// TableDiff saves config for diff table
type TableDiff struct {
	// source tables
	SourceTables []*TableInstance `json:"source-tables"`
	// target table
	TargetTable *TableInstance `json:"target-table"`

	// columns be ignored
	IgnoreColumns []string `json:"-"`

	// fields only be checked in table
	OnlyCheckColumns []string `json:"-"`

	//Use IgnoreColumns or OnlyCheckColumns to get this map
	IgnoreColsMap map[string]*model.ColumnInfo `json:"-"`

	//Use IgnoreColumns or OnlyCheckColumns or PK/UK to get this map
	PrintIgnoreColsMap map[string]*model.ColumnInfo `json:"-"`

	// field should be the primary key, unique key or field with index
	Fields string `json:"fields"`

	// select range, for example: "age > 10 AND age < 20"
	Range string `json:"range"`

	//for Oracle db, this range scope should equal to Range scope, for example date scope
	OracleRange string `json:"oracle-range"`

	//oracle hint
	OracleHint string `json:"oracle-hint"`
	//tidb hint
	TidbHint string `json:"tidb-hint"`

	OracleSCN string `json:"oracle-scn"`

	// db type for fix sql. support Oracle, Tidb. Default value is Tidb
	FixTarget string `json:"-"`

	EmptyStringNullCompareSwitch bool `json:"-"`

	Ascii0Switch bool `json:"-"`

	// for example, the whole data is [1...100]
	// we can split these data to [1...10], [11...20], ..., [91...100]
	// the [1...10] is a chunk, and it's chunk size is 10
	// size of the split chunk
	ChunkSize int `json:"chunk-size"`

	// sampling check percent, for example 10 means only check 10% data
	Sample int `json:"sample"`

	// how many goroutines are created to check data
	CheckThreadCount int `json:"-"`

	// set false if want to comapre the data directly
	UseChecksum bool `json:"-"`

	// set true if just want compare data by checksum, will skip select data when checksum is not equal
	OnlyUseChecksum bool `json:"-"`

	// collation config in mysql/tidb, should corresponding to charset.
	Collation string `json:"collation"`

	// ignore check table's struct
	IgnoreStructCheck bool `json:"-"`

	// ignore check table's data
	IgnoreDataCheck bool `json:"-"`

	// set true will continue check from the latest checkpoint
	UseCheckpoint bool `json:"-"`

	// get tidb statistics information from which table instance. if is nil, will split chunk by random.
	TiDBStatsSource *TableInstance `json:"tidb-stats-source"`

	sqlCh chan string

	wg sync.WaitGroup

	configHash string

	CpDB *sql.DB `json:"-"`

	// 1 means true, 0 means false
	checkpointLoaded int32

	// create after all chunks is splited, or load from checkpoint
	summaryInfo *tableSummaryInfo
}

func (t *TableDiff) setConfigHash() error {
	jsonBytes, err := json.Marshal(t)
	if err != nil {
		return errors.Trace(err)
	}

	t.configHash = fmt.Sprintf("%x", md5.Sum(jsonBytes))
	log.Debug("sync-diff-inspector config", zap.ByteString("config", jsonBytes), zap.String("hash", t.configHash))

	return nil
}

func (t *TableDiff) havePKOrUK() bool {
	for _, index := range t.TargetTable.info.Indices {
		if index.Primary || index.Unique {
			return true
		}
	}
	return false
}

// Equal tests whether two database have same data and schema.
func (t *TableDiff) Equal(ctx context.Context, writeFixSQL func(string) error) (bool, bool, error) {
	t.adjustConfig()
	t.sqlCh = make(chan string)

	err := t.getTableInfo(ctx)
	if err != nil {
		return false, false, errors.Trace(err)
	}

	if len(t.IgnoreColumns) > 0 || len(t.OnlyCheckColumns) > 0 {
		//if !t.havePKOrUK() {
		//	return false,false, errors.Errorf("%s table have no PK or UK", t.TargetTable.Table)
		//}
		err := t.ignoreColumns()
		if err != nil {
			return false, false, errors.Trace(err)
		}
	}

	structEqual := true
	dataEqual := true

	if !t.IgnoreStructCheck {
		structEqual, err = t.CheckTableStruct(ctx)
		if err != nil {
			return false, false, errors.Trace(err)
		}

		if !structEqual {
			return false, false, nil
		}
	}

	if !t.IgnoreDataCheck {
		stopWriteSqlsCh := t.WriteSqls(ctx, writeFixSQL)
		stopUpdateSummaryCh := t.UpdateSummaryInfo(ctx)

		dataEqual, err = t.CheckTableData(ctx)
		if err != nil {
			return structEqual, false, errors.Trace(err)
		}

		select {
		case <-ctx.Done():
		case stopWriteSqlsCh <- true:
		}

		select {
		case <-ctx.Done():
		case stopUpdateSummaryCh <- true:
		}
	}

	t.wg.Wait()
	return structEqual, dataEqual, nil
}

// CheckTableStruct checks table's struct
func (t *TableDiff) CheckTableStruct(ctx context.Context) (bool, error) {
	for _, sourceTable := range t.SourceTables {
		eq, msg := dbutil.EqualTableInfo(sourceTable.info, t.TargetTable.info)
		if !eq {
			log.Warn("table struct is not equal", zap.String("reason", msg))
			return false, nil
		}
		log.Info("table struct is equal", zap.Reflect("source", sourceTable.info), zap.Reflect("target", t.TargetTable.info))
	}

	return true, nil
}

func (t *TableDiff) adjustConfig() {
	if t.ChunkSize <= 0 {
		log.Warn("chunk size is less than 0, will use default value 1000", zap.Int("chunk size", t.ChunkSize))
		t.ChunkSize = 1000
	}

	if t.ChunkSize < 1000 || t.ChunkSize > 10000 {
		log.Warn("chunk size is recommend in range [1000, 10000]", zap.Int("chunk size", t.ChunkSize))
	}

	if len(t.Range) == 0 {
		t.Range = "1=1"
	}

	if len(t.OracleRange) == 0 {
		t.OracleRange = "1=1"
	}

	if t.Sample <= 0 {
		t.Sample = 100
	}

	if t.CheckThreadCount <= 0 {
		t.CheckThreadCount = 4
	}
}

func (t *TableDiff) getTableInfo(ctx context.Context) error {
	tableInfo, err := dbutil.GetTableInfo(ctx, t.TargetTable.Conn, t.TargetTable.Schema, t.TargetTable.Table)
	if err != nil {
		return errors.Trace(err)
	}
	//t.TargetTable.info = ignoreColumns(tableInfo, t.IgnoreColumns)
	t.TargetTable.info = tableInfo

	//if source db is oracle, no need to get table info like tidb.
	if t.SourceTables[0].DBType == dbutil.TypeOracle {
		return nil
	}
	for _, sourceTable := range t.SourceTables {
		tableInfo, err := dbutil.GetTableInfo(ctx, sourceTable.Conn, sourceTable.Schema, sourceTable.Table)
		if err != nil {
			return errors.Trace(err)
		}
		//sourceTable.info = ignoreColumns(tableInfo, t.IgnoreColumns)
		sourceTable.info = tableInfo
	}

	return nil
}

// CheckTableData checks table's data
func (t *TableDiff) CheckTableData(ctx context.Context) (equal bool, err error) {
	table := t.TargetTable

	useTiDB := false
	if t.TiDBStatsSource != nil {
		table = t.TiDBStatsSource
		useTiDB = true
	}

	fromCheckpoint := true
	chunks, err := t.LoadCheckpoint(ctx)
	if err != nil {
		return false, errors.Trace(err)
	}

	if len(chunks) == 0 {
		log.Info("don't have checkpoint info, or the last check success, or config changed, will split chunks")

		fromCheckpoint = false
		chunks, err = SplitChunks(ctx, table, t.Fields, t.Range, t.OracleRange, t.ChunkSize, t.Collation, useTiDB, t.CpDB)
		if err != nil {
			return false, errors.Trace(err)
		}
	}

	if len(chunks) == 0 {
		log.Warn("get 0 chunks, table is not checked", zap.String("table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)))
		return true, nil
	}

	t.summaryInfo = newTableSummaryInfo(int64(len(chunks)))

	checkResultCh := make(chan bool, t.CheckThreadCount)
	defer close(checkResultCh)
	pool := utils.NewWorkerPool(uint(t.CheckThreadCount), "worker-pool")

	go func() {
		for _, chunk := range chunks {
			pChunk := chunk
			pool.Apply(func() {
				t.newCheckChunksDataEqual(ctx, t.Sample < 100 && !fromCheckpoint, pChunk, checkResultCh)
			})
		}
	}()

	checkedNum := 0
	equal = true

CheckResult:
	for {
		select {
		case eq := <-checkResultCh:
			checkedNum++
			if !eq {
				equal = false
			}
			if len(chunks) == checkedNum {
				break CheckResult
			}
		case <-ctx.Done():
			equal = false
			break CheckResult
		}
	}
	pool.WaitFinished()

	return equal, nil
}

// LoadCheckpoint do some prepare work before check data, like adjust config and create checkpoint table
func (t *TableDiff) LoadCheckpoint(ctx context.Context) ([]*ChunkRange, error) {
	ctx1, cancel1 := context.WithTimeout(ctx, 5*dbutil.DefaultTimeout)
	defer cancel1()

	err := t.setConfigHash()
	if err != nil {
		return nil, errors.Trace(err)
	}

	err = createCheckpointTable(ctx1, t.CpDB)
	if err != nil {
		return nil, errors.Trace(err)
	}

	if t.UseCheckpoint {
		useCheckpoint, err := loadFromCheckPoint(ctx1, t.CpDB, t.TargetTable.Schema, t.TargetTable.Table, t.configHash)
		if err != nil {
			return nil, errors.Trace(err)
		}

		if useCheckpoint {
			log.Info("use checkpoint to load chunks")
			chunks, err := loadChunks(ctx1, t.CpDB, t.TargetTable.InstanceID, t.TargetTable.Schema, t.TargetTable.Table)
			if err != nil {
				log.Error("load chunks info", zap.Error(err))
				return nil, errors.Trace(err)
			}

			atomic.StoreInt32(&t.checkpointLoaded, 1)
			return chunks, nil
		}
	}

	// clean old checkpoint information, and initial table summary
	err = cleanCheckpoint(ctx1, t.CpDB, t.TargetTable.Schema, t.TargetTable.Table)
	if err != nil {
		return nil, errors.Trace(err)
	}

	err = initTableSummary(ctx1, t.CpDB, t.TargetTable.Schema, t.TargetTable.Table, t.configHash)
	if err != nil {
		return nil, errors.Trace(err)
	}

	atomic.StoreInt32(&t.checkpointLoaded, 1)
	return nil, nil
}

func (t *TableDiff) checkChunksDataEqual(ctx context.Context, filterByRand bool, chunks chan *ChunkRange, resultCh chan bool) {
	var err error
	for {
		select {
		case chunk, ok := <-chunks:
			if !ok {
				return
			}
			eq := false
			if chunk.State == successState || chunk.State == ignoreState {
				eq = true
				if chunk.State == ignoreState {
					t.summaryInfo.addIgnoreNum()
				} else if chunk.State == successState {
					t.summaryInfo.addSuccessNum()
				}
			} else {
				eq, err = t.checkChunkDataEqual(ctx, filterByRand, chunk)
				if err != nil {
					log.Error("check chunk data equal failed", zap.String("chunk", chunk.String()), zap.Error(err))
					eq = false
				} else if !eq {
					log.Warn("check chunk data not equal", zap.String("chunk", chunk.String()))
				}
			}

			select {
			case resultCh <- eq:
			case <-ctx.Done():
				return
			}
		case <-ctx.Done():
			return
		}
	}
}

func (t *TableDiff) newCheckChunksDataEqual(ctx context.Context, filterByRand bool, chunk *ChunkRange, resultCh chan bool) {
	var err error
	eq := false
	if chunk.State == successState || chunk.State == ignoreState {
		eq = true
		if chunk.State == ignoreState {
			t.summaryInfo.addIgnoreNum()
		} else if chunk.State == successState {
			t.summaryInfo.addSuccessNum()
		}
	} else {
		eq, err = t.checkChunkDataEqual(ctx, filterByRand, chunk)
		if err != nil {
			log.Error("check chunk data equal failed", zap.String("chunk", chunk.String()), zap.Error(err))
			eq = false
		} else if !eq {
			log.Warn("check chunk data not equal", zap.String("chunk", chunk.String()))
		}
	}

	select {
	case resultCh <- eq:
	case <-ctx.Done():
		return
	}
}

func (t *TableDiff) checkChunkDataEqual(ctx context.Context, filterByRand bool, chunk *ChunkRange) (equal bool, err error) {
	failpoint.Inject("CancelCheckChunkDataEqual", func(val failpoint.Value) {
		chunkID := val.(int)
		if chunkID != chunk.ID {
			failpoint.Return(false, nil)
		}

		log.Info("check chunk data equal failed", zap.String("failpoint", "CancelCheckChunkDataEqual"))
		cancelEqualFunc()
	})

	update := func() {
		ctx1, cancel1 := context.WithTimeout(ctx, dbutil.DefaultTimeout)
		defer cancel1()

		err1 := saveChunk(ctx1, t.CpDB, chunk.ID, t.TargetTable.InstanceID, t.TargetTable.Schema, t.TargetTable.Table, "", chunk)
		if err1 != nil {
			log.Warn("update chunk info", zap.Error(err1))
		}
	}

	defer func() {
		if chunk.State == ignoreState {
			t.summaryInfo.addIgnoreNum()
		} else {
			if err != nil {
				chunk.State = errorState
				t.summaryInfo.addFailedNum()
			} else {
				if equal {
					chunk.State = successState
					t.summaryInfo.addSuccessNum()
				} else {
					chunk.State = failedState
					t.summaryInfo.addFailedNum()
				}
			}
		}
		update()
	}()

	if filterByRand {
		rand.Seed(time.Now().UnixNano())
		r := rand.Intn(100)
		if r > t.Sample {
			chunk.State = ignoreState
			return true, nil
		}
	}

	chunk.State = checkingState
	update()

	if t.UseChecksum {
		// first check the checksum is equal or not
		equal, err = t.compareChecksum(ctx, chunk)
		if err != nil {
			return false, errors.Trace(err)
		}
		if equal {
			return true, nil
		}
	}

	if t.UseChecksum && t.OnlyUseChecksum {
		return false, nil
	}

	// if checksum is not equal or don't need compare checksum, compare the data
	log.Info("select data and then check data", zap.String("table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)), zap.String("where", dbutil.ReplacePlaceholder(chunk.Where, chunk.Args)))
	sourceDBType := t.SourceTables[0].DBType
	if sourceDBType == dbutil.TypeOracle && (t.TargetTable.DBType == dbutil.TypeTidb || t.TargetTable.DBType == dbutil.TypeMysql) {
		equal, err = t.compareRowsByCRC32(ctx, chunk)
	} else {
		equal, err = t.compareRows(ctx, chunk)
	}
	if err != nil {
		return false, errors.Trace(err)
	}

	return equal, nil
}

// checksumInfo save some information about checksum
type checksumInfo struct {
	checksum int64
	err      error
	cost     time.Duration
	tp       string
}

func (t *TableDiff) ignoreColumns() error {
	excludeColumns := make(map[string]*model.ColumnInfo)
	printExcludeColumns := make(map[string]*model.ColumnInfo)
	if len(t.OnlyCheckColumns) > 0 {
		colsMap := columnsSliceToMap(t.TargetTable.info.Columns)
		for i := 0; i < len(t.OnlyCheckColumns); i++ {
			if _, ok := colsMap[t.OnlyCheckColumns[i]]; !ok {
				return errors.Errorf("column[%s] in only-check-columns does not exist in table[%s].", t.OnlyCheckColumns[i], t.TargetTable.Table)
			}
		}
		checkColMap := utils.SliceToMap(t.OnlyCheckColumns)
		columns := t.TargetTable.info.Columns
		for i := 0; i < len(columns); i++ {
			if _, ok := checkColMap[columns[i].Name.O]; !ok {
				excludeColumns[columns[i].Name.O] = columns[i]
				printExcludeColumns[columns[i].Name.O] = columns[i]
			}
		}
	}

	if len(t.IgnoreColumns) > 0 {
		excludeCount := 0
		colsMap := columnsSliceToMap(t.TargetTable.info.Columns)
		for i := 0; i < len(t.IgnoreColumns); i++ {
			if col, ok := colsMap[t.IgnoreColumns[i]]; !ok {
				return errors.Errorf("column[%s] in ignore-columns does not exist in table[%s].", t.IgnoreColumns[i], t.TargetTable.Table)
			} else {
				excludeColumns[col.Name.O] = col
				printExcludeColumns[col.Name.O] = col
				excludeCount++
			}
		}
		if excludeCount == len(t.TargetTable.info.Columns) {
			return errors.Errorf("in table [%s],all columns are excluded by user.", t.TargetTable.Table)
		}
	}

	t.IgnoreColsMap = excludeColumns
	t.PrintIgnoreColsMap = printExcludeColumns
	return nil
}

func columnsSliceToMap(columns []*model.ColumnInfo) map[string]*model.ColumnInfo {
	colsMap := make(map[string]*model.ColumnInfo)
	for _, col := range columns {
		colsMap[col.Name.O] = col
	}
	return colsMap
}

// check the checksum is equal or not
func (t *TableDiff) compareChecksum(ctx context.Context, chunk *ChunkRange) (bool, error) {
	ctx1, cancel1 := context.WithCancel(ctx)
	defer cancel1()

	var (
		getSourceChecksumDuration, getTargetChecksumDuration time.Duration
		sourceChecksum, targetChecksum                       int64
		checksumInfoCh                                       = make(chan checksumInfo)
		firstErr                                             error
	)
	defer close(checksumInfoCh)

	getChecksum := func(db *sql.DB, sourceDbType, targetDbType, schema, table string, chunk *ChunkRange, tbInfo *model.TableInfo, ignoreColumns map[string]*model.ColumnInfo, args []interface{}, tp string, charset string) {
		beginTime := time.Now()
		var (
			checksum int64
			err      error
		)
		if sourceDbType == dbutil.TypeOracle && (targetDbType == dbutil.TypeTidb || targetDbType == dbutil.TypeMysql) {
			if tp == "source" {
				checksum, chunk.sourceCount, err = dbutil.GetOracleSumCRC32Checksum(ctx1, db, schema, table, tbInfo, ignoreColumns, chunk.OracleWhere, charset, t.OracleSCN, t.OracleHint, t.Ascii0Switch)
			} else {
				checksum, chunk.targetCount, err = dbutil.GetTiDBSumCRC32Checksum(ctx1, db, schema, table, tbInfo, ignoreColumns, chunk.Where, args, t.TidbHint, t.EmptyStringNullCompareSwitch, t.Ascii0Switch)
			}

		} else {
			checksum, err = dbutil.GetCRC32Checksum(ctx1, db, schema, table, tbInfo, chunk.Where, args)
		}
		cost := time.Since(beginTime)

		checksumInfoCh <- checksumInfo{
			checksum: checksum,
			err:      err,
			cost:     cost,
			tp:       tp,
		}
	}

	args := utils.StringsToInterfaces(chunk.Args)
	for _, sourceTable := range t.SourceTables {
		go getChecksum(sourceTable.Conn, sourceTable.DBType, t.TargetTable.DBType, sourceTable.Schema, sourceTable.Table, chunk, t.TargetTable.info, t.IgnoreColsMap, args, "source", sourceTable.Charset)
	}
	sourceDBType := t.SourceTables[0].DBType
	go getChecksum(t.TargetTable.Conn, sourceDBType, t.TargetTable.DBType, t.TargetTable.Schema, t.TargetTable.Table, chunk, t.TargetTable.info, t.IgnoreColsMap, args, "target", "")

	for i := 0; i < len(t.SourceTables)+1; i++ {
		checksumInfo := <-checksumInfoCh
		if checksumInfo.err != nil {
			// only need to return the first error, others are context cancel error
			if firstErr == nil {
				firstErr = checksumInfo.err
				cancel1()
			}

			continue
		}

		if checksumInfo.tp == "source" {
			sourceChecksum = sourceChecksum + checksumInfo.checksum
			if checksumInfo.cost > getSourceChecksumDuration {
				getSourceChecksumDuration = checksumInfo.cost
				chunk.sourceTime = getSourceChecksumDuration.String()
			}
		} else {
			targetChecksum = checksumInfo.checksum
			getTargetChecksumDuration = checksumInfo.cost
			chunk.targetTime = getTargetChecksumDuration.String()
		}
	}

	if firstErr != nil {
		return false, errors.Trace(firstErr)
	}

	if sourceChecksum == targetChecksum {
		log.Debug("checksum is equal", zap.String("table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)), zap.String("where", dbutil.ReplacePlaceholder(chunk.Where, chunk.Args)), zap.Int64("checksum", sourceChecksum), zap.Duration("get source checksum cost", getSourceChecksumDuration), zap.Duration("get target checksum cost", getTargetChecksumDuration))
		return true, nil
	}

	log.Warn("checksum is not equal", zap.String("table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)), zap.String("where", dbutil.ReplacePlaceholder(chunk.Where, chunk.Args)), zap.Int64("source checksum", sourceChecksum), zap.Int64("target checksum", targetChecksum), zap.Duration("get source checksum cost", getSourceChecksumDuration), zap.Duration("get target checksum cost", getTargetChecksumDuration))

	return false, nil
}

func (t *TableDiff) compareRowsByCRC32(ctx context.Context, chunk *ChunkRange) (bool, error) {
	beginTime := time.Now()

	args := utils.StringsToInterfaces(chunk.Args)

	targetRows, orderKeyCols, err := getTidbChunkCRC32Rows(ctx, t.TargetTable.Conn, t.TargetTable.Schema, t.TargetTable.Table, t.TargetTable.info, t.IgnoreColsMap, chunk.Where, args, t.TidbHint, t.EmptyStringNullCompareSwitch, t.Ascii0Switch)
	if err != nil {
		return false, errors.Trace(err)
	}
	defer targetRows.Close()

	sourceTable := t.SourceTables[0]
	sourceRows, _, sourceErr := getOracleChunkCRC32Rows(ctx, sourceTable.Conn, sourceTable.Schema, sourceTable.Table, t.TargetTable.info, t.IgnoreColsMap, chunk.OracleWhere, sourceTable.Charset, t.OracleSCN, t.OracleHint, t.Ascii0Switch)
	if sourceErr != nil {
		return false, errors.Trace(sourceErr)
	}
	defer sourceRows.Close()
	getRowCRC32Data := func(rows *sql.Rows) (rowData map[string]*dbutil.ColumnData, err error) {
		for rows.Next() {
			rowData, err = dbutil.UpperCaseKeyScanRow(rows)
			return
		}
		return
	}
	ignoreColsComment := genIgnoreColsComment(t.PrintIgnoreColsMap)

	var lastSourceData, lastTargetData map[string]*dbutil.ColumnData
	equal := true

	var (
		equalNum   = 0
		noEqualNum = 0
	)
	defer func() {
		log.Info("compare row in one chunk stat",
			zap.String("target table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)),
			zap.Int("equal num", equalNum), zap.Int("no equal num", noEqualNum),
			zap.String("where", dbutil.ReplacePlaceholder(chunk.Where, chunk.Args)))
	}()
	for {
		if lastSourceData == nil {
			lastSourceData, err = getRowCRC32Data(sourceRows)
			if err != nil {
				return false, err
			}
		}
		if lastTargetData == nil {
			lastTargetData, err = getRowCRC32Data(targetRows)
			if err != nil {
				return false, err
			}
		}

		if lastSourceData == nil {
			// don't have source data, so all the targetRows's data is redundant, should be deleted
			for lastTargetData != nil {
				targetRow, targetRowErr := getTidbRowByOrderKey(lastTargetData, orderKeyCols, ctx, t.TargetTable.Conn, t.TargetTable.Schema, t.TargetTable.Table, t.TargetTable.info, t.PrintIgnoreColsMap)
				if targetRowErr != nil {
					log.Error("get target row by order key failed.", zap.String("schema", t.TargetTable.Schema), zap.String("table", t.TargetTable.Table))
					return false, errors.Trace(targetRowErr)
				}
				targetRowMap, targetRowMapErr := getRowCRC32Data(targetRow)
				targetRow.Close()
				if targetRowMapErr != nil {
					return false, errors.Trace(targetRowErr)
				}
				sql := ""
				if targetRowMap == nil {
					log.Error("generate fix sql failed, getTidbRowByOrderKey get nothing", zap.String("schema", t.TargetTable.Schema), zap.String("table", t.TargetTable.Table), zap.String("order key", getOrderKeyData(lastTargetData, orderKeyCols)))
				} else {
					if dbutil.TypeOracle == t.FixTarget {
						sql = generateOracleCommentAndDML("insert", nil, targetRowMap, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
						log.Info("[insert]", zap.String("sql", sql))
					} else {
						sql = generateOracleCommentAndDML("delete", targetRowMap, nil, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
						log.Info("[delete]", zap.String("sql", sql))
					}
				}

				select {
				case t.sqlCh <- sql:
				case <-ctx.Done():
					return false, nil
				}
				equal = false
				noEqualNum++
				printCompareNumInfo(noEqualNum, equalNum, t)
				lastTargetData, err = getRowCRC32Data(targetRows)
				if err != nil {
					return false, err
				}
			}
			break
		}

		if lastTargetData == nil {
			// target lack some data, should insert the last source data
			for lastSourceData != nil {
				sourceRow, sourceRowErr := getOracleRowByOrderKey(lastSourceData, orderKeyCols, ctx, sourceTable.Conn, sourceTable.Schema, sourceTable.Table, t.TargetTable.info, t.PrintIgnoreColsMap, t.OracleSCN)
				if sourceRowErr != nil {
					log.Error("get source row by order key failed.", zap.String("schema", sourceTable.Schema), zap.String("table", sourceTable.Table))
					return false, errors.Trace(sourceRowErr)
				}
				sourceRowMap, sourceRowMapErr := getRowCRC32Data(sourceRow)
				sourceRow.Close()
				if sourceRowMapErr != nil {
					return false, errors.Trace(sourceRowMapErr)
				}
				sql := ""
				if sourceRowMap == nil {
					log.Error("generate fix sql failed, getOracleRowByOrderKey get nothing", zap.String("schema", sourceTable.Schema), zap.String("table", sourceTable.Table), zap.String("order key", getOrderKeyData(lastSourceData, orderKeyCols)))
				} else {
					if dbutil.TypeOracle == t.FixTarget {
						sql = generateOracleCommentAndDML("delete", sourceRowMap, nil, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
						log.Debug("[delete]", zap.String("sql", sql))
					} else {
						sql = generateOracleCommentAndDML("insert", nil, sourceRowMap, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
						log.Debug("[insert]", zap.String("sql", sql))
					}
				}

				select {
				case t.sqlCh <- sql:
				case <-ctx.Done():
					return false, nil
				}
				equal = false
				noEqualNum++
				printCompareNumInfo(noEqualNum, equalNum, t)
				lastSourceData, err = getRowCRC32Data(sourceRows)
				if err != nil {
					return false, err
				}
			}
			break
		}

		eq, cmp, err := compareCRC32Row(lastSourceData, lastTargetData, orderKeyCols)
		if err != nil {
			return false, errors.Trace(err)
		}
		if eq {
			lastSourceData = nil
			lastTargetData = nil
			equalNum++
			printCompareNumInfo(noEqualNum, equalNum, t)
			continue
		}

		equal = false
		noEqualNum++
		printCompareNumInfo(noEqualNum, equalNum, t)
		sql := ""
		log.Debug("not equal and generate sql")
		switch cmp {
		case 1:
			// delete
			targetRow, targetRowErr := getTidbRowByOrderKey(lastTargetData, orderKeyCols, ctx, t.TargetTable.Conn, t.TargetTable.Schema, t.TargetTable.Table, t.TargetTable.info, t.PrintIgnoreColsMap)
			if targetRowErr != nil {
				log.Error("get target row by order key failed.", zap.String("schema", t.TargetTable.Schema), zap.String("table", t.TargetTable.Table))
				return false, errors.Trace(targetRowErr)
			}
			targetRowMap, targetRowMapErr := getRowCRC32Data(targetRow)
			targetRow.Close()
			if targetRowMapErr != nil {
				return false, errors.Trace(targetRowErr)
			}
			if targetRowMap == nil {
				log.Error("generate fix sql failed, getTidbRowByOrderKey get nothing", zap.String("schema", t.TargetTable.Schema), zap.String("table", t.TargetTable.Table), zap.String("order key", getOrderKeyData(lastTargetData, orderKeyCols)))
			} else {
				if dbutil.TypeOracle == t.FixTarget {
					sql = generateOracleCommentAndDML("insert", nil, targetRowMap, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
					log.Debug("[insert]", zap.String("sql", sql))
				} else {
					sql = generateOracleCommentAndDML("delete", targetRowMap, nil, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
					log.Debug("[delete]", zap.String("sql", sql))
				}
			}
			lastTargetData = nil
		case -1:
			// insert
			sourceRow, sourceRowErr := getOracleRowByOrderKey(lastSourceData, orderKeyCols, ctx, sourceTable.Conn, sourceTable.Schema, sourceTable.Table, t.TargetTable.info, t.PrintIgnoreColsMap, t.OracleSCN)
			if sourceRowErr != nil {
				log.Error("get source row by order key failed.", zap.String("schema", sourceTable.Schema), zap.String("table", sourceTable.Table))
				return false, errors.Trace(sourceRowErr)
			}
			sourceRowMap, sourceRowMapErr := getRowCRC32Data(sourceRow)
			sourceRow.Close()
			if sourceRowMapErr != nil {
				return false, errors.Trace(sourceRowMapErr)
			}
			if sourceRowMap == nil {
				log.Error("generate fix sql failed, getOracleRowByOrderKey get nothing", zap.String("schema", sourceTable.Schema), zap.String("table", sourceTable.Table), zap.String("order key", getOrderKeyData(lastSourceData, orderKeyCols)))
			} else {
				if dbutil.TypeOracle == t.FixTarget {
					sql = generateOracleCommentAndDML("delete", sourceRowMap, nil, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
					log.Debug("[delete]", zap.String("sql", sql))
				} else {
					sql = generateOracleCommentAndDML("insert", nil, sourceRowMap, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
					log.Debug("[insert]", zap.String("sql", sql))
				}
			}
			lastSourceData = nil
		case 0:
			// update
			sourceRow, sourceRowErr := getOracleRowByOrderKey(lastSourceData, orderKeyCols, ctx, sourceTable.Conn, sourceTable.Schema, sourceTable.Table, t.TargetTable.info, t.PrintIgnoreColsMap, t.OracleSCN)
			targetRow, targetRowErr := getTidbRowByOrderKey(lastTargetData, orderKeyCols, ctx, t.TargetTable.Conn, t.TargetTable.Schema, t.TargetTable.Table, t.TargetTable.info, t.PrintIgnoreColsMap)
			if sourceRowErr != nil {
				log.Error("get source row by order key failed.", zap.String("schema", sourceTable.Schema), zap.String("table", sourceTable.Table))
				return false, errors.Trace(sourceRowErr)
			}
			if targetRowErr != nil {
				log.Error("get target row by order key failed.", zap.String("schema", t.TargetTable.Schema), zap.String("table", t.TargetTable.Table))
				return false, errors.Trace(sourceRowErr)
			}
			sourceRowMap, sourceRowMapErr := getRowCRC32Data(sourceRow)
			sourceRow.Close()
			targetRowMap, targetRowMapErr := getRowCRC32Data(targetRow)
			targetRow.Close()
			if sourceRowMapErr != nil {
				return false, errors.Trace(sourceRowMapErr)
			}
			if targetRowMapErr != nil {
				return false, errors.Trace(targetRowMapErr)
			}
			if sourceRowMap == nil || targetRowMap == nil {
				if sourceRowMap == nil {
					log.Error("generate fix sql failed, getOracleRowByOrderKey get nothing", zap.String("schema", sourceTable.Schema), zap.String("table", sourceTable.Table), zap.String("order key", getOrderKeyData(lastSourceData, orderKeyCols)))
				}
				if targetRowMap == nil {
					log.Error("generate fix sql failed, getTidbRowByOrderKey get nothing", zap.String("schema", t.TargetTable.Schema), zap.String("table", t.TargetTable.Table), zap.String("order key", getOrderKeyData(lastTargetData, orderKeyCols)))
				}
			} else {
				if dbutil.TypeOracle == t.FixTarget {
					sql = generateOracleCommentAndDML("update", sourceRowMap, targetRowMap, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
					log.Debug("[update]", zap.String("sql", sql))
				} else {
					sql = generateOracleCommentAndDML("update", targetRowMap, sourceRowMap, t.TargetTable.info, t.TargetTable.Schema, sourceTable.Schema, sourceTable.Table, ignoreColsComment, orderKeyCols, t.FixTarget)
					log.Debug("[update]", zap.String("sql", sql))
				}
			}
			lastSourceData = nil
			lastTargetData = nil
		}

		select {
		case t.sqlCh <- sql:
		case <-ctx.Done():
			return false, nil
		}
	}

	if equal {
		log.Info("rows is equal", zap.String("table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)), zap.String("where", dbutil.ReplacePlaceholder(chunk.Where, chunk.Args)), zap.Duration("cost", time.Since(beginTime)))
	} else {
		log.Warn("rows is not equal", zap.String("table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)), zap.String("where", dbutil.ReplacePlaceholder(chunk.Where, chunk.Args)), zap.Duration("cost", time.Since(beginTime)))
	}

	return equal, nil
}

func getOrderKeyData(data map[string]*dbutil.ColumnData, orderKeyCols []*model.ColumnInfo) string {
	builder := new(strings.Builder)
	for _, key := range orderKeyCols {
		colValue, ok := data[strings.ToUpper(key.Name.O)]
		if !ok {
			log.Error(fmt.Sprintf("order key %s does not exist in columns of tidb crc32 row", strings.ToUpper(key.Name.O)))
		}
		if colValue.IsNull {
			builder.WriteString(fmt.Sprintf("[%s=NULL]", key.Name.O))
		} else {
			builder.WriteString(fmt.Sprintf("[%s=%s]", key.Name.O, string(colValue.Data)))
		}
	}
	return builder.String()
}

func printCompareNumInfo(noEqualNum int, equalNum int, t *TableDiff) {
	if (noEqualNum+equalNum)%1000 == 0 {
		log.Debug("stat of rows compare in one chunk", zap.String("target table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)),
			zap.Int("noEqualNum", noEqualNum), zap.Int("equalNum", equalNum))
	}
}

func (t *TableDiff) compareRows(ctx context.Context, chunk *ChunkRange) (bool, error) {
	beginTime := time.Now()

	sourceRows := make(map[int]*sql.Rows)
	sourceHaveData := make(map[int]bool)
	args := utils.StringsToInterfaces(chunk.Args)

	targetRows, orderKeyCols, err := getChunkRows(ctx, t.TargetTable.Conn, t.TargetTable.Schema, t.TargetTable.Table, t.TargetTable.info, chunk.Where, args, t.Collation)
	if err != nil {
		return false, errors.Trace(err)
	}
	defer targetRows.Close()

	var (
		rows            *sql.Rows
		getChunkRowsErr error
	)
	for i, sourceTable := range t.SourceTables {
		if sourceTable.DBType == dbutil.TypeOracle {
			rows, _, getChunkRowsErr = getOracleChunkRows(ctx, sourceTable.Conn, sourceTable.Schema, sourceTable.Table, t.TargetTable.info, chunk.OracleWhere)
		} else {
			rows, _, getChunkRowsErr = getChunkRows(ctx, sourceTable.Conn, sourceTable.Schema, sourceTable.Table, sourceTable.info, chunk.Where, args, t.Collation)
		}

		if getChunkRowsErr != nil {
			return false, errors.Trace(getChunkRowsErr)
		}
		defer rows.Close()

		sourceRows[i] = rows
		sourceHaveData[i] = false
	}

	sourceRowDatas := &RowDatas{
		Rows:         make([]RowData, 0, len(sourceRows)),
		OrderKeyCols: orderKeyCols,
	}
	heap.Init(sourceRowDatas)

	getRowData := func(rows *sql.Rows) (rowData map[string]*dbutil.ColumnData, err error) {
		for rows.Next() {
			rowData, err = dbutil.UpperCaseKeyScanRow(rows)
			return
		}
		return
	}

	// getSourceRow gets one row from all the sources, it should be the smallest.
	// first get rows from every source, and then push them to the heap, and then pop to get the smallest one
	getSourceRow := func() (map[string]*dbutil.ColumnData, error) {
		if len(sourceHaveData) == 0 {
			return nil, nil
		}

		needDeleteSource := make([]int, 0, 1)
		for i, haveData := range sourceHaveData {
			if !haveData {
				rowData, err := getRowData(sourceRows[i])
				if err != nil {
					return nil, err
				}

				if rowData != nil {
					sourceHaveData[i] = true
					heap.Push(sourceRowDatas, RowData{
						Data:   rowData,
						Source: i,
					})
				}
			}

			if !sourceHaveData[i] {
				if sourceRows[i].Err() != nil {
					return nil, sourceRows[i].Err()

				}
				// still don't have data, means the rows is read to the end, so delete the source
				needDeleteSource = append(needDeleteSource, i)
			}
		}

		for _, i := range needDeleteSource {
			delete(sourceHaveData, i)
		}

		// all the sources had read to the end, no data to return
		if len(sourceRowDatas.Rows) == 0 {
			return nil, nil
		}

		rowData := heap.Pop(sourceRowDatas).(RowData)
		sourceHaveData[rowData.Source] = false

		return rowData.Data, nil
	}

	var lastSourceData, lastTargetData map[string]*dbutil.ColumnData
	equal := true

	columnsMap := make(map[string]*model.ColumnInfo)
	for _, col := range t.TargetTable.info.Columns {
		columnsMap[strings.ToUpper(col.Name.O)] = col
	}

	for {
		if lastSourceData == nil {
			lastSourceData, err = getSourceRow()
			if err != nil {
				return false, err
			}
		}

		if lastTargetData == nil {
			lastTargetData, err = getRowData(targetRows)
			if err != nil {
				return false, err
			}
		}

		if lastSourceData == nil {
			// don't have source data, so all the targetRows's data is redundant, should be deleted
			for lastTargetData != nil {
				sql := generateDML("delete", lastTargetData, t.TargetTable.info, t.TargetTable.Schema)
				log.Info("[delete]", zap.String("sql", sql))

				select {
				case t.sqlCh <- sql:
				case <-ctx.Done():
					return false, nil
				}
				equal = false

				lastTargetData, err = getRowData(targetRows)
				if err != nil {
					return false, err
				}
			}
			break
		}

		if lastTargetData == nil {
			// target lack some data, should insert the last source datas
			for lastSourceData != nil {
				sql := generateDML("replace", lastSourceData, t.TargetTable.info, t.TargetTable.Schema)
				log.Info("[insert]", zap.String("sql", sql))

				select {
				case t.sqlCh <- sql:
				case <-ctx.Done():
					return false, nil
				}
				equal = false

				lastSourceData, err = getSourceRow()
				if err != nil {
					return false, err
				}
			}
			break
		}

		eq, cmp, err := compareData(lastSourceData, lastTargetData, orderKeyCols, columnsMap)
		if err != nil {
			return false, errors.Trace(err)
		}
		if eq {
			lastSourceData = nil
			lastTargetData = nil
			continue
		}

		equal = false
		sql := ""

		switch cmp {
		case 1:
			// delete
			sql = generateDML("delete", lastTargetData, t.TargetTable.info, t.TargetTable.Schema)
			log.Info("[delete]", zap.String("sql", sql))
			lastTargetData = nil
		case -1:
			// insert
			sql = generateDML("replace", lastSourceData, t.TargetTable.info, t.TargetTable.Schema)
			log.Info("[insert]", zap.String("sql", sql))
			lastSourceData = nil
		case 0:
			// update
			sql = generateDML("replace", lastSourceData, t.TargetTable.info, t.TargetTable.Schema)
			log.Info("[update]", zap.String("sql", sql))
			lastSourceData = nil
			lastTargetData = nil
		}

		select {
		case t.sqlCh <- sql:
		case <-ctx.Done():
			return false, nil
		}
	}

	if equal {
		log.Info("rows is equal", zap.String("table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)), zap.String("where", dbutil.ReplacePlaceholder(chunk.Where, chunk.Args)), zap.Duration("cost", time.Since(beginTime)))
	} else {
		log.Warn("rows is not equal", zap.String("table", dbutil.TableName(t.TargetTable.Schema, t.TargetTable.Table)), zap.String("where", dbutil.ReplacePlaceholder(chunk.Where, chunk.Args)), zap.Duration("cost", time.Since(beginTime)))
	}

	return equal, nil
}

// WriteSqls write sqls to file
func (t *TableDiff) WriteSqls(ctx context.Context, writeFixSQL func(string) error) chan bool {
	t.wg.Add(1)
	stopWriteCh := make(chan bool)

	go func() {
		defer t.wg.Done()

		stop := false
		for {
			select {
			case dml, ok := <-t.sqlCh:
				if !ok {
					return
				}

				err := writeFixSQL(fmt.Sprintf("%s\n", dml))
				if err != nil {
					log.Error("write sql failed", zap.String("sql", dml), zap.Error(err))
				}

			case <-stopWriteCh:
				stop = true
			case <-ctx.Done():
				return
			default:
				if stop {
					return
				}
				time.Sleep(100 * time.Millisecond)
			}
		}
	}()

	return stopWriteCh
}

// UpdateSummaryInfo updaets summary infomation
func (t *TableDiff) UpdateSummaryInfo(ctx context.Context) chan bool {
	t.wg.Add(1)
	stopUpdateCh := make(chan bool)

	go func() {
		update := func() {
			ctx1, cancel1 := context.WithTimeout(ctx, dbutil.DefaultTimeout)
			defer cancel1()

			err := updateTableSummary(ctx1, t.CpDB, t.TargetTable.InstanceID, t.TargetTable.Schema, t.TargetTable.Table, t.summaryInfo)
			if err != nil {
				log.Warn("save table summary info failed", zap.String("schema", t.TargetTable.Schema), zap.String("table", t.TargetTable.Table), zap.Error(err))
			}
		}
		defer func() {
			update()
			t.wg.Done()
		}()

		ticker := time.NewTicker(10 * time.Second)
		defer ticker.Stop()

		for {
			select {
			case <-ctx.Done():
				return
			case <-stopUpdateCh:
				return
			case <-ticker.C:
				if atomic.LoadInt32(&t.checkpointLoaded) == 1 {
					update()
				}
			}
		}
	}()

	return stopUpdateCh
}

func generateDML(tp string, data map[string]*dbutil.ColumnData, table *model.TableInfo, schema string) (sql string) {
	switch tp {
	case "replace":
		colNames := make([]string, 0, len(table.Columns))
		values := make([]string, 0, len(table.Columns))
		for _, col := range table.Columns {
			if col.IsGenerated() {
				continue
			}

			colNames = append(colNames, dbutil.ColumnName(col.Name.O))
			if data[strings.ToUpper(col.Name.O)].IsNull {
				values = append(values, "NULL")
				continue
			}

			if needQuotes(col.FieldType) {
				values = append(values, fmt.Sprintf("'%s'", strings.Replace(string(data[strings.ToUpper(col.Name.O)].Data), "'", "\\'", -1)))
			} else {
				values = append(values, string(data[strings.ToUpper(col.Name.O)].Data))
			}
		}

		sql = fmt.Sprintf("REPLACE INTO %s(%s) VALUES (%s);", dbutil.TableName(schema, table.Name.O), strings.Join(colNames, ","), strings.Join(values, ","))
	case "delete":
		kvs := make([]string, 0, len(table.Columns))
		for _, col := range table.Columns {
			if col.IsGenerated() {
				continue
			}

			if data[strings.ToUpper(col.Name.O)].IsNull {
				kvs = append(kvs, fmt.Sprintf("%s is NULL", dbutil.ColumnName(col.Name.O)))
				continue
			}

			if needQuotes(col.FieldType) {
				kvs = append(kvs, fmt.Sprintf("%s = '%s'", dbutil.ColumnName(col.Name.O), strings.Replace(string(data[strings.ToUpper(col.Name.O)].Data), "'", "\\'", -1)))
			} else {
				kvs = append(kvs, fmt.Sprintf("%s = %s", dbutil.ColumnName(col.Name.O), string(data[strings.ToUpper(col.Name.O)].Data)))
			}
		}
		sql = fmt.Sprintf("DELETE FROM %s WHERE %s;", dbutil.TableName(schema, table.Name.O), strings.Join(kvs, " AND "))
	default:
		log.Error("unknown sql type", zap.String("type", tp))
	}

	return
}

func generateOracleCommentAndDML(tp string, targetData, sourceData map[string]*dbutil.ColumnData, table *model.TableInfo,
	targetSchema, sourceSchema, sourceTableName string, ignoreColsComment string, orderKeyCols []*model.ColumnInfo, fixTarget string) (content string) {
	switch tp {
	case "update":
		content = genUpdateCommentAndSQL(table, sourceData, targetData, orderKeyCols, targetSchema, sourceSchema, sourceTableName, ignoreColsComment, fixTarget)
	case "insert":
		content = genInsertCommentAndSQL(table, sourceData, targetSchema, sourceSchema, sourceTableName, ignoreColsComment, fixTarget)
	case "delete":
		content = genDeleteCommentAndSQL(table, targetData, orderKeyCols, targetSchema, sourceSchema, sourceTableName, ignoreColsComment, fixTarget)
	default:
		log.Error("unknown sql type", zap.String("type", tp))
	}
	return
}

func genInsertCommentAndSQL(table *model.TableInfo, sourceData map[string]*dbutil.ColumnData, targetSchema string, sourceSchema string,
	sourceTable string, ignoreColsComment string, fixTarget string) (content string) {
	colNames := make([]string, 0, len(table.Columns))
	values := make([]string, 0, len(table.Columns))
	oracleValues := make([]string, 0, len(table.Columns))
	targetTableName := dbutil.TableName(targetSchema, table.Name.O)
	sourceTableName := dbutil.TableName(sourceSchema, sourceTable)
	if dbutil.TypeOracle == fixTarget {
		targetTableName = dbutil.OracleTableName(targetSchema, table.Name.O)
		sourceTableName = dbutil.OracleTableName(sourceSchema, sourceTable)
	}
	for _, col := range table.Columns {
		if col.IsGenerated() {
			continue
		}
		upperColName := strings.ToUpper(col.Name.O)
		if dbutil.TypeOracle == fixTarget {
			colNames = append(colNames, dbutil.OracleColumnName(col.Name.O))
			//source data come from Tidb
			if sourceData[upperColName].IsNull {
				values = append(values, "NULL")
				oracleValues = append(oracleValues, "NULL")
				continue
			}
		} else {
			colNames = append(colNames, dbutil.ColumnName(col.Name.O))
			//source data come from Oracle
			if sourceData[upperColName].IsNull || string(sourceData[upperColName].Data) == "" {
				values = append(values, "NULL")
				continue
			}
		}

		if needQuotes(col.FieldType) {
			values = append(values, fmt.Sprintf("'%s'", processQuoteStringValue(sourceData[upperColName].Data)))
		} else {
			values = append(values, string(sourceData[upperColName].Data))
		}
		if dbutil.TypeOracle == fixTarget {
			oracleValues = append(oracleValues, genOracleValues(col, string(sourceData[upperColName].Data)))
		}
	}
	comment := generateComment(colNames, nil, values, targetTableName, sourceTableName, ignoreColsComment)
	sql := ""
	if dbutil.TypeTidb == fixTarget {
		sql = fmt.Sprintf("INSERT INTO %s(%s) VALUES (%s);", targetTableName, strings.Join(colNames, ","), strings.Join(values, ","))
	} else if dbutil.TypeOracle == fixTarget {
		sql = fmt.Sprintf("INSERT INTO %s(%s) VALUES (%s);", sourceTableName, strings.Join(colNames, ","), strings.Join(oracleValues, ","))
	} else {
		log.Error(fmt.Sprintf("gen fix sql failed, fix-target[%s] unknown", fixTarget))
		sql = fmt.Sprintf("gen fix sql failed, fix-target[%s] unknown", fixTarget)
	}
	content = fmt.Sprintf("%s\n%s", comment, sql)
	return content
}

func genDeleteCommentAndSQL(table *model.TableInfo, targetData map[string]*dbutil.ColumnData, orderKeyCols []*model.ColumnInfo,
	targetSchema string, sourceSchema string, sourceTable string, ignoreColsComment string, fixTarget string) (content string) {
	whereStatements := make([]string, 0, len(table.Columns))
	colNames := make([]string, 0, len(table.Columns))
	targetValues := make([]string, 0, len(table.Columns))
	targetTableName := dbutil.TableName(targetSchema, table.Name.O)
	sourceTableName := dbutil.TableName(sourceSchema, sourceTable)
	sqlTableName := targetTableName
	limitStr := "LIMIT 1"
	if dbutil.TypeOracle == fixTarget {
		targetTableName = dbutil.OracleTableName(targetSchema, table.Name.O)
		sourceTableName = dbutil.OracleTableName(sourceSchema, sourceTable)
		sqlTableName = sourceTableName
		limitStr = "AND ROWNUM <= 1"
	}
	for _, col := range table.Columns {
		if col.IsGenerated() {
			continue
		}
		if dbutil.TypeOracle == fixTarget {
			colNames = append(colNames, dbutil.OracleColumnName(col.Name.O))
		} else {
			colNames = append(colNames, dbutil.ColumnName(col.Name.O))
		}
		targetValue, whereStatement := genWhereStatementAndValue(targetData, col, orderKeyCols, fixTarget)
		targetValues = append(targetValues, targetValue)
		if whereStatement != "" {
			whereStatements = append(whereStatements, whereStatement)
		}
	}
	comment := generateComment(colNames, targetValues, nil, targetTableName, sourceTableName, ignoreColsComment)
	sql := fmt.Sprintf("DELETE FROM %s WHERE %s %s;", sqlTableName, strings.Join(whereStatements, " AND "), limitStr)
	content = fmt.Sprintf("%s\n%s", comment, sql)
	return content
}

func genUpdateCommentAndSQL(table *model.TableInfo, sourceData map[string]*dbutil.ColumnData, targetData map[string]*dbutil.ColumnData, orderKeyCols []*model.ColumnInfo,
	targetSchema string, sourceSchema string, sourceTable string, ignoreColsComment string, fixTarget string) string {
	colNames := make([]string, 0, len(table.Columns))
	sourceValues := make([]string, 0, len(table.Columns))
	targetValues := make([]string, 0, len(table.Columns))
	setStatements := make([]string, 0, len(table.Columns))
	whereStatements := make([]string, 0, len(table.Columns))
	targetTableName := dbutil.TableName(targetSchema, table.Name.O)
	sourceTableName := dbutil.TableName(sourceSchema, sourceTable)
	sqlTableName := targetTableName
	limitStr := "LIMIT 1"
	if dbutil.TypeOracle == fixTarget {
		targetTableName = dbutil.OracleTableName(targetSchema, table.Name.O)
		sourceTableName = dbutil.OracleTableName(sourceSchema, sourceTable)
		sqlTableName = sourceTableName
		limitStr = "AND ROWNUM <= 1"
	}
	for _, col := range table.Columns {
		if col.IsGenerated() {
			continue
		}
		if dbutil.TypeOracle == fixTarget {
			colNames = append(colNames, dbutil.OracleColumnName(col.Name.O))
		} else {
			colNames = append(colNames, dbutil.ColumnName(col.Name.O))
		}
		//generate set part and source value in comment
		sourceValue, setStatement := genUpdateSetStatementAndValue(sourceData, col, fixTarget)
		sourceValues = append(sourceValues, sourceValue)
		setStatements = append(setStatements, setStatement)

		//generate where part and target value in comment
		targetValue, whereStatement := genWhereStatementAndValue(targetData, col, orderKeyCols, fixTarget)
		targetValues = append(targetValues, targetValue)
		if whereStatement != "" {
			whereStatements = append(whereStatements, whereStatement)
		}
	}
	comment := generateComment(colNames, targetValues, sourceValues, targetTableName, sourceTableName, ignoreColsComment)
	sql := fmt.Sprintf("UPDATE %s SET %s WHERE %s %s;", sqlTableName, strings.Join(setStatements, ", "), strings.Join(whereStatements, " AND "), limitStr)
	return fmt.Sprintf("%s\n%s", comment, sql)
}

func genWhereStatementAndValue(targetData map[string]*dbutil.ColumnData, col *model.ColumnInfo,
	orderKeyCols []*model.ColumnInfo, fixTarget string) (targetValue string, whereStatement string) {
	colName := ""
	upperColName := strings.ToUpper(col.Name.O)
	//process scenario that column is null
	if dbutil.TypeOracle == fixTarget {
		colName = dbutil.OracleColumnName(col.Name.O)
		if targetData[upperColName].IsNull || string(targetData[upperColName].Data) == "" {
			targetValue = "NULL"
			if isOrderKey(col.Name.O, orderKeyCols) {
				whereStatement = fmt.Sprintf("%s is NULL", colName)
			}
			return
		}
	} else {
		colName = dbutil.ColumnName(col.Name.O)
		if targetData[upperColName].IsNull {
			targetValue = "NULL"
			if isOrderKey(col.Name.O, orderKeyCols) {
				whereStatement = fmt.Sprintf("%s is NULL", colName)
			}
			return
		}
	}
	//process scenario that column is not null
	val := ""
	if needQuotes(col.FieldType) {
		val = processQuoteStringValue(targetData[upperColName].Data)
		targetValue = fmt.Sprintf("'%s'", val)
		if isOrderKey(col.Name.O, orderKeyCols) {
			if dbutil.TypeOracle == fixTarget {
				whereStatement = fmt.Sprintf("%s = %s", colName, genOracleValues(col, val))
			} else {
				whereStatement = fmt.Sprintf("%s = '%s'", colName, val)
			}
		}
		return
	}
	val = string(targetData[upperColName].Data)
	targetValue = val
	if isOrderKey(col.Name.O, orderKeyCols) {
		whereStatement = fmt.Sprintf("%s = %s", colName, val)
	}
	return
}

func genUpdateSetStatementAndValue(sourceData map[string]*dbutil.ColumnData, col *model.ColumnInfo, fixTarget string) (sourceValue string, setStatement string) {
	colName := ""
	upperColName := strings.ToUpper(col.Name.O)
	//process scenario that column is null
	if dbutil.TypeOracle == fixTarget {
		colName = dbutil.OracleColumnName(col.Name.O)
		if sourceData[upperColName].IsNull {
			sourceValue = "NULL"
			setStatement = fmt.Sprintf("%s = NULL", colName)
			return
		}
	} else {
		colName = dbutil.ColumnName(col.Name.O)
		if sourceData[upperColName].IsNull || string(sourceData[upperColName].Data) == "" {
			sourceValue = "NULL"
			setStatement = fmt.Sprintf("%s = NULL", colName)
			return
		}
	}
	//process scenario that column is not null
	val := ""
	if needQuotes(col.FieldType) {
		val = processQuoteStringValue(sourceData[upperColName].Data)
		sourceValue = fmt.Sprintf("'%s'", val)
		if dbutil.TypeOracle == fixTarget {
			setStatement = fmt.Sprintf("%s = %s", colName, genOracleValues(col, val))
		} else {
			setStatement = fmt.Sprintf("%s = '%s'", colName, val)
		}
		return
	}
	val = string(sourceData[upperColName].Data)
	sourceValue = val
	setStatement = fmt.Sprintf("%s = %s", colName, val)
	return
}

func processQuoteStringValue(data []byte) string {
	return strings.Replace(string(data), "'", "\\'", -1)
}

func isOrderKey(name string, orderKeyCols []*model.ColumnInfo) bool {
	for _, col := range orderKeyCols {
		if col.Name.O == name {
			return true
		}
	}
	return false
}

func generateComment(colNames, targetValues, sourceValues []string, targetTableName, sourceTableName string, ignoreColsComment string) string {
	source := fmt.Sprintf("/*\nSource: %s  ", sourceTableName)
	target := fmt.Sprintf("Target: %s  ", targetTableName)
	for i := 0; i < len(colNames); i++ {
		if targetValues != nil {
			target = target + fmt.Sprintf("[%s:%s]  ", colNames[i], targetValues[i])
		}
		if sourceValues != nil {
			source = source + fmt.Sprintf("[%s:%s]  ", colNames[i], sourceValues[i])
		}
	}
	if ignoreColsComment == "" {
		return fmt.Sprintf("%s\n%s\n*/", source, target)
	} else {
		return fmt.Sprintf("%s\n%s\n%s\n*/", source, target, ignoreColsComment)
	}

}

func genOracleValues(column *model.ColumnInfo, value string) string {
	if value == "" {
		return "''"
	}
	switch column.GetType() {
	case mysql.TypeDate:
		return fmt.Sprintf("TO_DATE('%s', 'yyyy-mm-dd')", value)
	case mysql.TypeDatetime:
		if column.GetDecimal() == 0 {
			return fmt.Sprintf("TO_DATE('%s', 'yyyy-mm-dd hh24:mi:ss')", value)
		}
		return fmt.Sprintf("TO_TIMESTAMP('%s', 'yyyy-mm-dd hh24:mi:ss.ff%d')", value, column.GetDecimal())
	case mysql.TypeTimestamp:
		if column.GetDecimal() == 0 {
			return fmt.Sprintf("TO_TIMESTAMP('%v', 'yyyy-mm-dd hh24:mi:ss')", value)
		}
		return fmt.Sprintf("TO_TIMESTAMP('%s', 'yyyy-mm-dd hh24:mi:ss.ff%d')", value, column.GetDecimal())
	case mysql.TypeDuration:
		return fmt.Sprintf("TO_DATE('%s', 'hh24:mi:ss')", value)
	case mysql.TypeTiny, mysql.TypeShort, mysql.TypeLong, mysql.TypeLonglong, mysql.TypeInt24,
		mysql.TypeYear, mysql.TypeFloat, mysql.TypeDouble, mysql.TypeNewDecimal:
		return fmt.Sprintf("%s", value)
	default:
		return fmt.Sprintf("'%s'", value)
	}
}

func genIgnoreColsComment(printIgnoreColumns map[string]*model.ColumnInfo) string {
	if len(printIgnoreColumns) == 0 {
		return ""
	}
	names := make([]string, 0, 0)
	for key, _ := range printIgnoreColumns {
		names = append(names, key)
	}
	return fmt.Sprintf("Ignore Columns: [%s] ", strings.Join(names, ","))
}

func compareCRC32Row(map1, map2 map[string]*dbutil.ColumnData, orderKeyCols []*model.ColumnInfo) (equal bool, cmp int32, err error) {
	var (
		data1, data2 *dbutil.ColumnData
		key          string
		ok           bool
	)

	equal = true

	defer func() {
		if equal || err != nil {
			return
		}

		if cmp == 0 {
			log.Warn("find different row", zap.String("column", key), zap.String("row1", rowToString(map1)), zap.String("row2", rowToString(map2)))
		} else if cmp > 0 {
			log.Warn("target had superfluous data", zap.String("row", rowToString(map2)))
		} else {
			log.Warn("target lack data", zap.String("row", rowToString(map1)))
		}
	}()

	key = "CHECKSUM"
	if string(map1[key].Data) == string(map2[key].Data) {
		equal = true
		return
	} else {
		equal = false
		log.Debug("value of CHECKSUM column is not equal.", zap.String("column", key), zap.String("value1", string(map1[key].Data)), zap.String("value2", string(map2[key].Data)))
	}

	for _, col := range orderKeyCols {
		if data1, ok = map1[strings.ToUpper(col.Name.O)]; !ok {
			err = errors.Errorf("don't have key %s", col.Name.O)
			return
		}
		if data2, ok = map2[strings.ToUpper(col.Name.O)]; !ok {
			err = errors.Errorf("don't have key %s", col.Name.O)
			return
		}
		if needQuotes(col.FieldType) {
			strData1 := string(data1.Data)
			strData2 := string(data2.Data)
			if len(strData1) == len(strData2) && strData1 == strData2 {
				continue
			}

			if strData1 < strData2 {
				cmp = -1
				equal = false
				log.Debug("compare string column, value1 < value2, comp=-1", zap.String("column", col.Name.O), zap.String("value1", strData1), zap.String("value2", strData2))
			} else if strData1 > strData2 {
				cmp = 1
				equal = false
				log.Debug("compare string column, value1 > value2, comp=1", zap.String("column", col.Name.O), zap.String("value1", strData1), zap.String("value2", strData2))
			}
			break

		} else {
			var (
				num1, num2 float64
				err1, err2 error
			)
			if !data1.IsNull && !data2.IsNull {
				num1, err1 = strconv.ParseFloat(string(data1.Data), 64)
				num2, err2 = strconv.ParseFloat(string(data2.Data), 64)
				if err1 != nil || err2 != nil {
					err = errors.Errorf("value of column %s convert %s, %s to float failed, err1: %v, err2: %v", key, string(data1.Data), string(data2.Data), err1, err2)
					equal = false
					return
				}
				if num1 == num2 {
					continue
				}
				if num1 < num2 {
					cmp = -1
					equal = false
					log.Debug("compare number column, value1 < value2, cmp=-1", zap.String("column", col.Name.O), zap.Float64("value1", num1), zap.Float64("value2", num2))
				} else if num1 > num2 {
					cmp = 1
					equal = false
					log.Debug("compare number column, value1 > value2, cmp=1", zap.String("column", col.Name.O), zap.Float64("value1", num1), zap.Float64("value2", num2))
				}
			} else if data1.IsNull && data2.IsNull {
				//they are null, so equal
				continue
			} else { // data1 is null or data2 is null
				if data1.IsNull {
					//target lack data
					cmp = -1
					log.Debug("compare number column, value1 is null", zap.String("column", col.Name.O), zap.Float64("value2", num2))
				} else {
					// target had superfluous data
					cmp = 1
					log.Debug("compare number column, value2 is null", zap.String("column", col.Name.O), zap.Float64("value1", num1))
				}
				equal = false
			}
			break
		}
	}

	return
}

func compareData(map1, map2 map[string]*dbutil.ColumnData, orderKeyCols []*model.ColumnInfo, columnsMap map[string]*model.ColumnInfo) (equal bool, cmp int32, err error) {
	var (
		data1, data2 *dbutil.ColumnData
		key          string
		ok           bool
	)

	equal = true

	defer func() {
		if equal || err != nil {
			return
		}

		if cmp == 0 {
			log.Warn("find different row", zap.String("column", key), zap.String("row1", rowToString(map1)), zap.String("row2", rowToString(map2)))
		} else if cmp > 0 {
			log.Warn("target had superfluous data", zap.String("row", rowToString(map2)))
		} else {
			log.Warn("target lack data", zap.String("row", rowToString(map1)))
		}
	}()

	for key, data1 = range map1 {
		if data2, ok = map2[key]; !ok {
			return false, 0, errors.Errorf("don't have key %s", key)
		}
		if dbutil.IsNumberOrFloatType(columnsMap[key].GetType()) {
			var (
				num1, num2 float64
				err1, err2 error
			)
			if !data1.IsNull && !data2.IsNull {
				num1, err1 = strconv.ParseFloat(string(data1.Data), 64)
				num2, err2 = strconv.ParseFloat(string(data2.Data), 64)
				if err1 != nil || err2 != nil {
					err = errors.Errorf("value of column %s convert %s, %s to float failed, err1: %v, err2: %v", key, string(data1.Data), string(data2.Data), err1, err2)
					equal = false
					return
				}
				if num1 == num2 {
					continue
				}
			} else if data1.IsNull && data2.IsNull {
				//they are null, so equal
				continue
			}

		} else {
			//for other column type, compare them as a string
			if string(data1.Data) == string(data2.Data) {
				continue
			}
		}
		equal = false

		break
	}
	if equal {
		return
	}

	for _, col := range orderKeyCols {
		if data1, ok = map1[strings.ToUpper(col.Name.O)]; !ok {
			err = errors.Errorf("don't have key %s", col.Name.O)
			return
		}
		if data2, ok = map2[strings.ToUpper(col.Name.O)]; !ok {
			err = errors.Errorf("don't have key %s", col.Name.O)
			return
		}
		if needQuotes(col.FieldType) {
			strData1 := string(data1.Data)
			strData2 := string(data2.Data)

			if len(strData1) == len(strData2) && strData1 == strData2 {
				continue
			}

			if strData1 < strData2 {
				cmp = -1
			} else if strData1 > strData2 {
				cmp = 1
			}
			break

		} else {
			var (
				num1, num2 float64
				err1, err2 error
			)
			if !data1.IsNull && !data2.IsNull {
				num1, err1 = strconv.ParseFloat(string(data1.Data), 64)
				num2, err2 = strconv.ParseFloat(string(data2.Data), 64)
				if err1 != nil || err2 != nil {
					err = errors.Errorf("value of column %s convert %s, %s to float failed, err1: %v, err2: %v", key, string(data1.Data), string(data2.Data), err1, err2)
					equal = false
					return
				}
				if num1 == num2 {
					continue
				}
				if num1 < num2 {
					cmp = -1
				} else if num1 > num2 {
					cmp = 1
				}

			} else if data1.IsNull && data2.IsNull {
				//they are null, so equal
				continue
			} else { // data1 is null or data2 is null
				if data1.IsNull {
					// target had superfluous data
					cmp = 1
				} else {
					//target lack data
					cmp = -1
				}
			}
			break
		}
	}

	return
}

func getChunkRows(ctx context.Context, db *sql.DB, schema, table string, tableInfo *model.TableInfo, where string,
	args []interface{}, collation string) (*sql.Rows, []*model.ColumnInfo, error) {
	orderKeys, orderKeyCols := dbutil.SelectUniqueOrderKey(tableInfo)

	columnNames := make([]string, 0, len(tableInfo.Columns))
	for _, col := range tableInfo.Columns {
		if dbutil.IsTimeType(col.GetType()) {
			columnNames = append(columnNames, "DATE_FORMAT("+dbutil.ColumnName(col.Name.O)+",'%Y-%m-%d %H:%i:%s') as "+dbutil.ColumnName(col.Name.O))
			continue
		}
		columnNames = append(columnNames, dbutil.ColumnName(col.Name.O))
	}
	columns := strings.Join(columnNames, ", ")

	if collation != "" {
		collation = fmt.Sprintf(" COLLATE \"%s\"", collation)
	}

	for i, key := range orderKeys {
		orderKeys[i] = dbutil.ColumnName(key)
	}

	query := fmt.Sprintf("SELECT /*!40001 SQL_NO_CACHE */ %s FROM %s WHERE %s ORDER BY %s%s",
		columns, dbutil.TableName(schema, table), where, strings.Join(orderKeys, ","), collation)

	log.Debug("select data", zap.String("sql", query), zap.Reflect("args", args))
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}

	return rows, orderKeyCols, nil
}

//get chunk rows from oracle
func getOracleChunkRows(ctx context.Context, db *sql.DB, schema, table string, tableInfo *model.TableInfo, where string) (*sql.Rows, []*model.ColumnInfo, error) {
	orderKeys, orderKeyCols := dbutil.SelectUniqueOrderKey(tableInfo)

	columnNames := make([]string, 0, len(tableInfo.Columns))
	for _, col := range tableInfo.Columns {
		if dbutil.IsTimeType(col.GetType()) {
			columnNames = append(columnNames, fmt.Sprintf("TO_CHAR(%s,'yyyy-mm-dd hh24:mi:ss') as %s", dbutil.OracleColumnName(col.Name.O), dbutil.OracleColumnName(col.Name.O)))
			continue
		}
		if dbutil.IsCharType(col.GetType()) {
			columnNames = append(columnNames, fmt.Sprintf("rtrim(%s) as %s", dbutil.OracleColumnName(col.Name.O), dbutil.OracleColumnName(col.Name.O)))
			continue
		}

		columnNames = append(columnNames, dbutil.OracleColumnName(col.Name.O))
	}
	columns := strings.Join(columnNames, ", ")

	for i, key := range orderKeys {
		orderKeys[i] = dbutil.OracleColumnName(key)
	}

	query := fmt.Sprintf("SELECT %s FROM %s WHERE %s ORDER BY %s",
		columns, dbutil.OracleTableName(schema, table), where, strings.Join(orderKeys, ","))

	log.Debug("select data", zap.String("sql", query))
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}

	return rows, orderKeyCols, nil
}

func getTidbChunkCRC32Rows(ctx context.Context, db *sql.DB, schemaName, tableName string, tbInfo *model.TableInfo, ignoreColumns map[string]*model.ColumnInfo, where string,
	args []interface{}, tidbHint string, emptyStringNullCompareSwitch, ascii0Switch bool) (*sql.Rows, []*model.ColumnInfo, error) {
	orderKeys, orderKeyCols := dbutil.SelectOrderKey(tbInfo, ignoreColumns)

	columnNames := make([]string, 0, len(tbInfo.Columns))
	columnIsNull := make([]string, 0, len(tbInfo.Columns))
	colName := ""
	newColName := ""
	for _, col := range tbInfo.Columns {
		if _, ok := ignoreColumns[col.Name.O]; ok || col.IsGenerated() {
			continue
		}
		colName = dbutil.ColumnName(col.Name.O)
		if dbutil.IsFloatOrDoubleType(col.GetType()) {
			columnNames = append(columnNames, fmt.Sprintf("cast(%s as char)", colName))
		} else if dbutil.IsStringType(col.GetType()) {
			newColName = colName
			if ascii0Switch {
				newColName = fmt.Sprintf("REPLACE(%s, char(0), '')",newColName)
			}
			if emptyStringNullCompareSwitch {
				newColName = fmt.Sprintf("case %s when '' then null else %s end", newColName, newColName)
			}
			columnNames = append(columnNames, newColName)
		} else {
			columnNames = append(columnNames, colName)
		}
		columnIsNull = append(columnIsNull, fmt.Sprintf("ISNULL(%s)", colName))
	}

	orderkeyColNames := make([]string, 0, len(orderKeyCols))
	for i, key := range orderKeys {
		if dbutil.IsFloatOrDoubleType(orderKeyCols[i].GetType()) {
			orderkeyColNames = append(orderkeyColNames, fmt.Sprintf("cast(%s as char) as %s", dbutil.ColumnName(key), dbutil.ColumnName(key)))
		} else {
			orderkeyColNames = append(orderkeyColNames, dbutil.ColumnName(key))
		}
		orderKeys[i] = dbutil.ColumnName(key)
	}

	query := fmt.Sprintf("SELECT %s %s, CAST(CRC32(CONCAT_WS(',', %s, CONCAT(%s))) AS UNSIGNED) AS checksum FROM %s WHERE %s ORDER BY %s ;",
		tidbHint, strings.Join(orderkeyColNames, ", "), strings.Join(columnNames, ", "), strings.Join(columnIsNull, ", "),
		dbutil.TableName(schemaName, tableName), where, strings.Join(orderKeys, ","))

	log.Debug("get tidb chunk crc32 rows", zap.String("sql", query), zap.Reflect("args", args))
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		log.Error("get tidb chunk crc32 rows failed", zap.String("schema", schemaName),
			zap.String("table", tableName), zap.String("sql", query))
		return nil, nil, errors.Trace(err)
	}

	return rows, orderKeyCols, nil
}

func getOracleChunkCRC32Rows(ctx context.Context, db *sql.DB, schemaName, tableName string, tbInfo *model.TableInfo, ignoreColumns map[string]*model.ColumnInfo,
	where string, charset string, oracleSCN string, oracleHint string, ascii0Switch bool) (*sql.Rows, []*model.ColumnInfo, error) {
	_, orderKeyCols := dbutil.SelectOrderKey(tbInfo, ignoreColumns)

	columnNvl2 := make([]string, 0, len(tbInfo.Columns))
	columnNames := make([]string, 0, len(tbInfo.Columns))
	for _, col := range tbInfo.Columns {
		if _, ok := ignoreColumns[col.Name.O]; ok || col.IsGenerated() {
			continue
		}
		oracleColName := dbutil.OracleColumnName(col.Name.O)
		if dbutil.IsTimeType(col.GetType()) {
			columnNvl2 = append(columnNvl2, fmt.Sprintf("NVL2(%s,to_char(%s,'%s')||',',NULL)", oracleColName, oracleColName, dbutil.ProcessOracleDateTimeFormat(col)))
			columnNames = append(columnNames, fmt.Sprintf("NVL2(%s,0,1)", oracleColName))
			continue
		}
		if dbutil.IsCharType(col.GetType()) {
			if ascii0Switch {
				columnNvl2 = append(columnNvl2, fmt.Sprintf("NVL2(replace(rtrim(%s),chr(0),''),replace(rtrim(%s),chr(0),'')||',',NULL)", oracleColName, oracleColName))
			} else {
				columnNvl2 = append(columnNvl2, fmt.Sprintf("NVL2(rtrim(%s),rtrim(%s)||',',NULL)", oracleColName, oracleColName))
			}
			columnNames = append(columnNames, fmt.Sprintf("NVL2(%s,0,1)", oracleColName))
			continue
		} else if dbutil.IsStringType(col.GetType()) {
			if ascii0Switch {
				columnNvl2 = append(columnNvl2, fmt.Sprintf("NVL2(replace(%s,chr(0),''),replace(%s,chr(0),'')||',',NULL)", oracleColName, oracleColName))
				columnNames = append(columnNames, fmt.Sprintf("NVL2(%s,0,1)", oracleColName))
				continue
			}
		}
		if dbutil.IsDecimalType(col.GetType()) {
			columnNvl2 = append(columnNvl2, fmt.Sprintf("NVL2(%s,to_char(%s,'%s')||',',NULL)", oracleColName, oracleColName, dbutil.ProcessOracleFloatFormat(col)))
			columnNames = append(columnNames, fmt.Sprintf("NVL2(%s,0,1)", oracleColName))
			continue
		}
		if dbutil.IsFloatOrDoubleType(col.GetType()) {
			columnNvl2 = append(columnNvl2, fmt.Sprintf("NVL2(%s,to_char(to_number(%s))||',',NULL)", oracleColName, oracleColName))
			columnNames = append(columnNames, fmt.Sprintf("NVL2(%s,0,1)", oracleColName))
			continue
		}
		columnNvl2 = append(columnNvl2, fmt.Sprintf("NVL2(%s,%s||',',NULL)", oracleColName, oracleColName))
		columnNames = append(columnNames, fmt.Sprintf("NVL2(%s,0,1)", oracleColName))
	}
	orderkeyColNames := make([]string, 0, len(orderKeyCols))
	orderbyOrderKeyColNames := make([]string, 0, len(orderKeyCols))
	for _, orderKeyCol := range orderKeyCols {
		if dbutil.IsCharType(orderKeyCol.GetType()) {
			orderkeyColNames = append(orderkeyColNames, fmt.Sprintf("rtrim(%s) as %s", dbutil.OracleColumnName(orderKeyCol.Name.O), dbutil.OracleColumnName(orderKeyCol.Name.O)))
		} else if dbutil.IsTimeType(orderKeyCol.GetType()) {
			orderkeyColNames = append(orderkeyColNames, fmt.Sprintf("TO_CHAR(%s,'%s') as %s", dbutil.OracleColumnName(orderKeyCol.Name.O), dbutil.ProcessOracleDateTimeFormat(orderKeyCol), dbutil.OracleColumnName(orderKeyCol.Name.O)))
		} else {
			orderkeyColNames = append(orderkeyColNames, dbutil.OracleColumnName(orderKeyCol.Name.O))
		}
		orderbyOrderKeyColNames = append(orderbyOrderKeyColNames, dbutil.OracleColumnName(orderKeyCol.Name.O))
	}
	oracleTableName := dbutil.OracleTableName(schemaName, tableName)
	if oracleSCN != "" {
		oracleTableName = fmt.Sprintf("%s as of scn %s", oracleTableName, oracleSCN)
	}
	var query string
	if charset != "" {
		query = fmt.Sprintf("SELECT %s %s, CRC32(CONVERT(%s || %s, 'al32utf8','zhs16gbk')) AS checksum FROM %s WHERE %s ORDER BY %s",
			oracleHint, strings.Join(orderkeyColNames, ", "), strings.Join(columnNvl2, "||"), strings.Join(columnNames, "||"), oracleTableName,
			where, strings.Join(orderbyOrderKeyColNames, ", "))
	} else {
		query = fmt.Sprintf("SELECT %s %s, CRC32(%s || %s) AS checksum FROM %s WHERE %s ORDER BY %s",
			oracleHint, strings.Join(orderkeyColNames, ", "), strings.Join(columnNvl2, "||"), strings.Join(columnNames, "||"), oracleTableName,
			where, strings.Join(orderbyOrderKeyColNames, ", "))
	}

	log.Debug("get oracle chunk crc32 rows", zap.String("sql", query))
	rows, err := db.QueryContext(ctx, query, godror.PrefetchCount(1000), godror.FetchArraySize(1000))
	if err != nil {
		log.Error("get oracle chunk crc32 rows failed", zap.String("schema", schemaName),
			zap.String("table", tableName), zap.String("sql", query))
		return nil, nil, errors.Trace(err)
	}

	return rows, orderKeyCols, nil
}

func getTidbRowByOrderKey(data map[string]*dbutil.ColumnData, orderKeyCols []*model.ColumnInfo, ctx context.Context, db *sql.DB,
	schemaName, tableName string, tableInfo *model.TableInfo, printIgnoreColumns map[string]*model.ColumnInfo) (*sql.Rows, error) {

	columnNames := make([]string, 0, len(tableInfo.Columns))
	for _, col := range tableInfo.Columns {
		if col.IsGenerated() {
			continue
		}
		if dbutil.IsFloatOrDoubleType(col.GetType()) {
			columnNames = append(columnNames, fmt.Sprintf("cast(%s as char) as %s", dbutil.ColumnName(col.Name.O), dbutil.ColumnName(col.Name.O)))
			continue
		}
		columnNames = append(columnNames, dbutil.ColumnName(col.Name.O))
	}
	columns := strings.Join(columnNames, ", ")

	condition := make([]string, 0, len(orderKeyCols))
	for _, key := range orderKeyCols {
		column, ok := data[strings.ToUpper(key.Name.O)]
		if !ok {
			return nil, errors.NotFoundf("order key %s does not exist in columns of tidb crc32 row", strings.ToUpper(key.Name.O))
		}
		if column.IsNull {
			condition = append(condition, fmt.Sprintf("%s is NULL", dbutil.ColumnName(key.Name.O)))
			continue
		}
		if dbutil.IsNumberOrFloatType(key.GetType()) {
			if dbutil.IsFloatOrDoubleType(key.GetType()) {
				condition = append(condition, fmt.Sprintf("cast(%s as char) = '%s'", dbutil.ColumnName(key.Name.O), string(column.Data)))
			} else {
				condition = append(condition, fmt.Sprintf("%s = %s", dbutil.ColumnName(key.Name.O), string(column.Data)))
			}
		} else {
			condition = append(condition, fmt.Sprintf("%s = '%s'", dbutil.ColumnName(key.Name.O), string(column.Data)))
		}
	}

	query := fmt.Sprintf("SELECT %s FROM %s WHERE %s;",
		columns, dbutil.TableName(schemaName, tableName), strings.Join(condition, " AND "))

	log.Debug("get tidb one row by order key", zap.String("sql", query))
	log.Debug("db stats", zap.Reflect("info ", db.Stats()))
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		log.Error("get tidb one row by order key failed.", zap.String("schema", schemaName),
			zap.String("table", tableName), zap.String("sql", query))
		return nil, errors.Trace(err)
	}
	return rows, nil
}

func getOracleRowByOrderKey(data map[string]*dbutil.ColumnData, orderKeyCols []*model.ColumnInfo, ctx context.Context, db *sql.DB,
	schemaName, tableName string, tableInfo *model.TableInfo, printIgnoreColumns map[string]*model.ColumnInfo, oracleSCN string) (*sql.Rows, error) {

	columnNames := make([]string, 0, len(tableInfo.Columns))
	for _, col := range tableInfo.Columns {
		if col.IsGenerated() {
			continue
		}
		if dbutil.IsTimeType(col.GetType()) {
			columnNames = append(columnNames, fmt.Sprintf("TO_CHAR(%s,'%s') as %s", dbutil.OracleColumnName(col.Name.O), dbutil.ProcessOracleDateTimeFormat(col), dbutil.OracleColumnName(col.Name.O)))
			continue
		}
		if dbutil.IsCharType(col.GetType()) {
			columnNames = append(columnNames, fmt.Sprintf("rtrim(%s) as %s", dbutil.OracleColumnName(col.Name.O), dbutil.OracleColumnName(col.Name.O)))
			continue
		}
		if dbutil.IsDecimalType(col.GetType()) {
			columnNames = append(columnNames, fmt.Sprintf("TO_CHAR(%s,'%s') as %s", dbutil.OracleColumnName(col.Name.O), dbutil.ProcessOracleFloatFormat(col), dbutil.OracleColumnName(col.Name.O)))
			continue
		}
		if dbutil.IsFloatOrDoubleType(col.GetType()) {
			columnNames = append(columnNames, fmt.Sprintf("TO_CHAR(TO_NUMBER(%s)) as %s", dbutil.OracleColumnName(col.Name.O), dbutil.OracleColumnName(col.Name.O)))
			continue
		}
		columnNames = append(columnNames, dbutil.OracleColumnName(col.Name.O))
	}
	columns := strings.Join(columnNames, ", ")

	condition := make([]string, 0, len(orderKeyCols))

	for _, key := range orderKeyCols {
		column, ok := data[strings.ToUpper(key.Name.O)]
		if !ok {
			return nil, errors.NotFoundf("order key %s does not exist in columns of oracle crc32 row", strings.ToUpper(key.Name.O))
		}

		if column.IsNull || "" == string(column.Data) {
			condition = append(condition, fmt.Sprintf("((%s is NULL) or (%s = ''))", dbutil.OracleColumnName(key.Name.O), dbutil.OracleColumnName(key.Name.O)))
			continue
		}
		if dbutil.IsNumberOrFloatType(key.GetType()) {
			condition = append(condition, fmt.Sprintf("%s = %s", dbutil.OracleColumnName(key.Name.O), string(column.Data)))
		} else if dbutil.IsTimeType(key.GetType()) {
			condition = append(condition, fmt.Sprintf("%s = %s", dbutil.OracleColumnName(key.Name.O), dbutil.GenOracleTimeValueByColumnType(key, string(column.Data))))
		} else {
			condition = append(condition, fmt.Sprintf("%s = '%s'", dbutil.OracleColumnName(key.Name.O), string(column.Data)))
		}
	}
	oracleTableName := dbutil.OracleTableName(schemaName, tableName)
	if oracleSCN != "" {
		oracleTableName = fmt.Sprintf("%s as of scn %s", oracleTableName, oracleSCN)
	}
	query := fmt.Sprintf("SELECT %s FROM %s WHERE %s",
		columns, oracleTableName, strings.Join(condition, " AND "))

	log.Debug("get oracle one row by order key", zap.String("sql", query))
	log.Debug("db stats", zap.Reflect("info ", db.Stats()))
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		log.Error("get oracle row by order key failed.", zap.String("schema", schemaName),
			zap.String("table", tableName), zap.String("sql", query))
		return nil, errors.Trace(err)
	}

	return rows, nil
}

// GetBetterIndex returns the index more dinstict.
// If the index is primary key or unique, it can be return directly.
// Otherwise select the index which has higher value of `COUNT(DISTINCT a)/COUNT(*)`.
func GetBetterIndex(ctx context.Context, db *sql.DB, schema, table string, tableInfo *model.TableInfo) ([]*model.IndexInfo, error) {
	// SELECT COUNT(DISTINCT city)/COUNT(*) FROM `schema`.`table`;
	indices := dbutil.FindAllIndex(tableInfo)
	for _, index := range indices {
		if index.Primary || index.Unique {
			return []*model.IndexInfo{index}, nil
		}
	}
	sels := make([]float64, len(indices))
	for _, index := range indices {
		column := GetColumnsFromIndex(index, tableInfo)[0]
		selectivity, err := GetSelectivity(ctx, db, schema, table, column.Name.O, tableInfo)
		if err != nil {
			return indices, errors.Trace(err)
		}
		log.Debug("index selectivity", zap.String("table", dbutil.TableName(schema, table)), zap.Float64("selectivity", selectivity))
		sels = append(sels, selectivity)
	}
	sort.Slice(indices, func(i, j int) bool {
		return sels[i] > sels[j]
	})
	return indices, nil
}

// GetColumnsFromIndex returns `ColumnInfo`s of the specified index.
func GetColumnsFromIndex(index *model.IndexInfo, tableInfo *model.TableInfo) []*model.ColumnInfo {
	indexColumns := make([]*model.ColumnInfo, 0, len(index.Columns))
	for _, indexColumn := range index.Columns {
		indexColumns = append(indexColumns, tableInfo.Columns[indexColumn.Offset])
	}

	return indexColumns
}

// GetSelectivity returns the value of `COUNT(DISTINCT col)/COUNT(1)` SQL.
func GetSelectivity(ctx context.Context, db *sql.DB, schemaName, tableName, columnName string, tbInfo *model.TableInfo) (float64, error) {
	query := fmt.Sprintf("SELECT COUNT(DISTINCT %s)/COUNT(1) as SEL FROM %s;", dbutil.ColumnName(columnName), dbutil.TableName(schemaName, tableName))
	var selectivity sql.NullFloat64
	args := []interface{}{}
	err := db.QueryRowContext(ctx, query, args...).Scan(&selectivity)
	if err != nil {
		log.Warn("execute get selectivity query fail", zap.String("query", query))
		return 0.0, errors.Trace(err)
	}
	if !selectivity.Valid {
		// if don't have any data, the checksum will be `NULL`
		log.Warn("get empty count or checksum", zap.String("sql", query))
		return 0.0, nil
	}
	return selectivity.Float64, nil
}