func updateTableSummary(ctx context.Context, db *sql.DB, instanceID, schema, table string, summaryInfo *tableSummaryInfo) error {
	if summaryInfo == nil {
		return nil
	}
	total, successNum, failedNum, ignoreNum := summaryInfo.get()
	if total == 0 {
		return nil
	}
	log.Info("summary info", zap.String("instance_id", instanceID), zap.String("schema", schema), zap.String("table", table), zap.Int64("chunk num", total), zap.Int64("success num", successNum), zap.Int64("failed num", failedNum), zap.Int64("ignore num", ignoreNum))
	checkedNum := successNum + failedNum + ignoreNum
	state := checkingState
	if checkedNum == 0 {
		state = notCheckedState
	} else if checkedNum == total {
		if total == successNum+ignoreNum {
			state = successState
		} else {
			state = failedState
		}
	}
	updateSQL := fmt.Sprintf("UPDATE `%s`.`%s` SET `chunk_num` = ?, `check_success_num` = ?, `check_failed_num` = ?, `check_ignore_num` = ?, `state` = ? WHERE `schema` = ? AND `table` = ?", checkpointSchemaName, summaryTableName)
	err := dbutil.ExecSQLWithRetry(ctx, db, updateSQL, total, successNum, failedNum, ignoreNum, state, schema, table)
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}
updateTableSummary gets summary info from `chunk` table, and then update `summary` table
package diff
import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"time"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb-tools/pkg/dbutil"
	"go.uber.org/zap"
)
var (
	successState = "success"
	failedState = "failed"
	errorState = "error"
	notCheckedState = "not_checked"
	checkingState = "checking"
	ignoreState = "ignore"
	checkpointSchemaName = "sync_diff_inspector"
	summaryTableName = "summary"
	chunkTableName = "chunk"
)
type tableSummaryInfo struct {
	sync.RWMutex
	totalNum   int64
	successNum int64
	failedNum  int64
	ignoreNum  int64
}
func newTableSummaryInfo(totalNum int64) *tableSummaryInfo {
	return &tableSummaryInfo{
		totalNum: totalNum,
	}
}
func (s *tableSummaryInfo) addSuccessNum() {
	s.Lock()
	s.successNum++
	s.Unlock()
}
func (s *tableSummaryInfo) addFailedNum() {
	s.Lock()
	s.failedNum++
	s.Unlock()
}
func (s *tableSummaryInfo) addIgnoreNum() {
	s.Lock()
	s.ignoreNum++
	s.Unlock()
}
func (s *tableSummaryInfo) get() (totalNum, successNum, failedNum, ignoreNum int64) {
	s.RLock()
	defer s.RUnlock()
	return s.totalNum, s.successNum, s.failedNum, s.ignoreNum
}
func saveChunk(ctx context.Context, db *sql.DB, chunkID int, instanceID, schema, table, checksum string, chunk *ChunkRange) error {
	chunkBytes, err := json.Marshal(chunk)
	if err != nil {
		return errors.Trace(err)
	}
	sql := fmt.Sprintf("REPLACE INTO `%s`.`%s` VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?);", checkpointSchemaName, chunkTableName)
	err = dbutil.ExecSQLWithRetry(ctx, db, sql, chunkID, instanceID, schema, table, chunk.Where, checksum, string(chunkBytes), chunk.State, time.Now())
	if err != nil {
		log.Error("save chunk info failed", zap.Error(err))
		return errors.Trace(err)
	}
	return nil
}
func initChunks(ctx context.Context, db *sql.DB, instanceID, schema, table string, chunks []*ChunkRange) error {
	beginTime := time.Now()
	batch := 100
	num := 0
	sqlPrefix := fmt.Sprintf("INSERT INTO `%s`.`%s` VALUES", checkpointSchemaName, chunkTableName)
	valuesPlaceholders := "(?, ?, ?, ?, ?, ?, ?, ?, ?)"
	valuesPlaceholdersArray := make([]string, 0, batch)
	values := make([]interface{}, 0, 9*batch)
	for i, chunk := range chunks {
		num++
		chunkBytes, err := json.Marshal(chunk)
		if err != nil {
			return errors.Trace(err)
		}
		values = append(values, chunk.ID, instanceID, schema, table, chunk.Where, "", string(chunkBytes), chunk.State, time.Now())
		valuesPlaceholdersArray = append(valuesPlaceholdersArray, valuesPlaceholders)
		if num >= batch || i == len(chunks)-1 {
			sql := fmt.Sprintf("%s%s", sqlPrefix, strings.Join(valuesPlaceholdersArray, ", "))
			err = dbutil.ExecSQLWithRetry(ctx, db, sql, values...)
			if err != nil {
				log.Error("save chunk info failed", zap.Error(err))
				return errors.Trace(err)
			}
			num = 0
			valuesPlaceholdersArray = valuesPlaceholdersArray[:0]
			values = values[:0]
		}
	}
	log.Info("initial chunks", zap.Duration("cost", time.Since(beginTime)))
	return nil
}
func getChunk(ctx context.Context, db *sql.DB, instanceID, schema, table string, chunkID int) (*ChunkRange, error) {
	query := fmt.Sprintf("SELECT `chunk_str` FROM `%s`.`%s` WHERE `instance_id` = ? AND `schema` = ? AND `table` = ? AND `chunk_id` = ? limit 1", checkpointSchemaName, chunkTableName)
	rows, err := db.QueryContext(ctx, query, instanceID, schema, table, chunkID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		fields, err1 := dbutil.ScanRow(rows)
		if err1 != nil {
			return nil, errors.Trace(err1)
		}
		chunkStr := fields["chunk_str"].Data
		chunk := new(ChunkRange)
		err := json.Unmarshal(chunkStr, &chunk)
		if err != nil {
			return nil, err
		}
		return chunk, nil
	}
	if rows.Err() != nil {
		return nil, errors.Trace(rows.Err())
	}
	return nil, errors.NotFoundf("instanceID %d, schema %s, table %s, chunk %d", instanceID, schema, table, chunkID)
}
func loadChunks(ctx context.Context, db *sql.DB, instanceID, schema, table string) ([]*ChunkRange, error) {
	chunks := make([]*ChunkRange, 0, 100)
	query := fmt.Sprintf("SELECT `chunk_str` FROM `%s`.`%s` WHERE `instance_id` = ? AND `schema` = ? AND `table` = ?", checkpointSchemaName, chunkTableName)
	rows, err := db.QueryContext(ctx, query, instanceID, schema, table)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		fields, err1 := dbutil.ScanRow(rows)
		if err1 != nil {
			return nil, errors.Trace(err1)
		}
		chunkStr := fields["chunk_str"].Data
		chunk := new(ChunkRange)
		err := json.Unmarshal(chunkStr, &chunk)
		if err != nil {
			return nil, err
		}
		chunk.updateColumnOffset()
		chunks = append(chunks, chunk)
	}
	return chunks, errors.Trace(rows.Err())
}
func getTableSummary(ctx context.Context, db *sql.DB, schema, table string) (total int64, success int64, failed int64, ignore int64, state string, err error) {
	query := fmt.Sprintf("SELECT `chunk_num`, `check_success_num`, `check_failed_num`, `check_ignore_num`, `state` FROM `%s`.`%s` WHERE `schema` = ? AND `table` = ? LIMIT 1",
		checkpointSchemaName, summaryTableName)
	rows, err := db.QueryContext(ctx, query, schema, table)
	if err != nil {
		return 0, 0, 0, 0, "", errors.Trace(err)
	}
	defer rows.Close()
	var totalNum, successNum, failedNum, ignoreNum sql.NullInt64
	var stateStr sql.NullString
	for rows.Next() {
		err1 := rows.Scan(&totalNum, &successNum, &failedNum, &ignoreNum, &stateStr)
		if err1 != nil {
			return 0, 0, 0, 0, "", errors.Trace(err1)
		}
		if !totalNum.Valid || !successNum.Valid || !failedNum.Valid || !ignoreNum.Valid || !stateStr.Valid {
			return 0, 0, 0, 0, "", errors.Errorf("some values are invalid, query: %s, args: %v", query, []interface{}{schema, table})
		}
		return totalNum.Int64, successNum.Int64, failedNum.Int64, ignoreNum.Int64, stateStr.String, nil
	}
	if rows.Err() != nil {
		return 0, 0, 0, 0, "", errors.Trace(rows.Err())
	}
	return 0, 0, 0, 0, "", errors.NotFoundf("schema %s, table %s summary info", schema, table)
}
func initTableSummary(ctx context.Context, db *sql.DB, schema, table string, configHash string) error {
	sql := fmt.Sprintf("REPLACE INTO `%s`.`%s`(`schema`, `table`, `state`, `config_hash`) VALUES(?, ?, ?, ?)", checkpointSchemaName, summaryTableName)
	err := dbutil.ExecSQLWithRetry(ctx, db, sql, schema, table, notCheckedState, configHash)
	if err != nil {
		log.Error("save summary info failed", zap.Error(err))
		return errors.Trace(err)
	}
	return nil
}