func (engine *ClosedEngine) Cleanup(ctx context.Context) error {
	task := engine.logger.Begin(zap.InfoLevel, "cleanup")
	err := engine.backend.CleanupEngine(ctx, engine.uuid)
	task.End(zap.WarnLevel, err)
	return err
}
Cleanup deletes the intermediate data from target.
package backend
import (
	"context"
	"fmt"
	"sort"
	"time"
	"github.com/google/uuid"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/parser/model"
	"github.com/pingcap/tidb/store/tikv/oracle"
	"github.com/pingcap/tidb/table"
	"github.com/pingcap/tidb/types"
	"go.uber.org/zap"
	"github.com/pingcap/tidb-lightning/lightning/common"
	"github.com/pingcap/tidb-lightning/lightning/log"
	"github.com/pingcap/tidb-lightning/lightning/metric"
	"github.com/pingcap/tidb-lightning/lightning/verification"
)
const (
	maxRetryTimes = 3  
)
func makeTag(tableName string, engineID int32) string {
	return fmt.Sprintf("%s:%d", tableName, engineID)
}
func makeLogger(tag string, engineUUID uuid.UUID) log.Logger {
	return log.With(
		zap.String("engineTag", tag),
		zap.Stringer("engineUUID", engineUUID),
	)
}
func MakeUUID(tableName string, engineID int32) (string, uuid.UUID) {
	tag := makeTag(tableName, engineID)
	engineUUID := uuid.NewSHA1(engineNamespace, []byte(tag))
	return tag, engineUUID
}
var engineNamespace = uuid.MustParse("d68d6abe-c59e-45d6-ade8-e2b0ceb7bedf")
type EngineFileSize struct {
	UUID uuid.UUID
	DiskSize int64
	MemSize int64
	IsImporting bool
}
type AbstractBackend interface {
	Close()
	MakeEmptyRows() Rows
	RetryImportDelay() time.Duration
	ShouldPostProcess() bool
	NewEncoder(tbl table.Table, options *SessionOptions) (Encoder, error)
	OpenEngine(ctx context.Context, engineUUID uuid.UUID) error
	CloseEngine(ctx context.Context, engineUUID uuid.UUID) error
	ImportEngine(ctx context.Context, engineUUID uuid.UUID) error
	CleanupEngine(ctx context.Context, engineUUID uuid.UUID) error
	CheckRequirements(ctx context.Context) error
	FetchRemoteTableModels(ctx context.Context, schemaName string) ([]*model.TableInfo, error)
	FlushEngine(ctx context.Context, engineUUID uuid.UUID) error
	FlushAllEngines(ctx context.Context) error
	EngineFileSizes() []EngineFileSize
	ResetEngine(ctx context.Context, engineUUID uuid.UUID) error
	LocalWriter(ctx context.Context, engineUUID uuid.UUID, maxCacheSize int64) (EngineWriter, error)
}
func fetchRemoteTableModelsFromTLS(ctx context.Context, tls *common.TLS, schema string) ([]*model.TableInfo, error) {
	var tables []*model.TableInfo
	err := tls.GetJSON(ctx, "/schema/"+schema, &tables)
	if err != nil {
		return nil, errors.Annotatef(err, "cannot read schema '%s' from remote", schema)
	}
	return tables, nil
}
type Backend struct {
	abstract AbstractBackend
}
type engine struct {
	backend AbstractBackend
	logger  log.Logger
	uuid    uuid.UUID
}
type OpenedEngine struct {
	engine
	tableName string
	ts        uint64
}
type ClosedEngine struct {
	engine
}
type LocalEngineWriter struct {
	writer    EngineWriter
	tableName string
	ts        uint64
}
func MakeBackend(ab AbstractBackend) Backend {
	return Backend{abstract: ab}
}
func (be Backend) Close() {
	be.abstract.Close()
}
func (be Backend) MakeEmptyRows() Rows {
	return be.abstract.MakeEmptyRows()
}
func (be Backend) NewEncoder(tbl table.Table, options *SessionOptions) (Encoder, error) {
	return be.abstract.NewEncoder(tbl, options)
}
func (be Backend) ShouldPostProcess() bool {
	return be.abstract.ShouldPostProcess()
}
func (be Backend) CheckRequirements(ctx context.Context) error {
	return be.abstract.CheckRequirements(ctx)
}
func (be Backend) FetchRemoteTableModels(ctx context.Context, schemaName string) ([]*model.TableInfo, error) {
	return be.abstract.FetchRemoteTableModels(ctx, schemaName)
}
func (be Backend) FlushAll(ctx context.Context) error {
	return be.abstract.FlushAllEngines(ctx)
}
func (be Backend) CheckDiskQuota(quota int64) (
	largeEngines []uuid.UUID,
	inProgressLargeEngines int,
	totalDiskSize int64,
	totalMemSize int64,
) {
	sizes := be.abstract.EngineFileSizes()
	sort.Slice(sizes, func(i, j int) bool {
		a, b := &sizes[i], &sizes[j]
		if a.IsImporting != b.IsImporting {
			return a.IsImporting
		}
		return a.DiskSize+a.MemSize < b.DiskSize+b.MemSize
	})
	for _, size := range sizes {
		totalDiskSize += size.DiskSize
		totalMemSize += size.MemSize
		if totalDiskSize+totalMemSize > quota {
			if size.IsImporting {
				inProgressLargeEngines++
			} else {
				largeEngines = append(largeEngines, size.UUID)
			}
		}
	}
	return
}
func (be Backend) UnsafeImportAndReset(ctx context.Context, engineUUID uuid.UUID) error {
	closedEngine := ClosedEngine{
		engine: engine{
			backend: be.abstract,
			logger:  makeLogger("<import-and-reset>", engineUUID),
			uuid:    engineUUID,
		},
	}
	if err := closedEngine.Import(ctx); err != nil {
		return err
	}
	return be.abstract.ResetEngine(ctx, engineUUID)
}
func (be Backend) OpenEngine(ctx context.Context, tableName string, engineID int32) (*OpenedEngine, error) {
	tag, engineUUID := MakeUUID(tableName, engineID)
	logger := makeLogger(tag, engineUUID)
	if err := be.abstract.OpenEngine(ctx, engineUUID); err != nil {
		return nil, err
	}
	openCounter := metric.ImporterEngineCounter.WithLabelValues("open")
	openCounter.Inc()
	logger.Info("open engine")
	failpoint.Inject("FailIfEngineCountExceeds", func(val failpoint.Value) {
		closedCounter := metric.ImporterEngineCounter.WithLabelValues("closed")
		openCount := metric.ReadCounter(openCounter)
		closedCount := metric.ReadCounter(closedCounter)
		if injectValue := val.(int); openCount-closedCount > float64(injectValue) {
			panic(fmt.Sprintf("forcing failure due to FailIfEngineCountExceeds: %v - %v >= %d", openCount, closedCount, injectValue))
		}
	})
	return &OpenedEngine{
		engine: engine{
			backend: be.abstract,
			logger:  logger,
			uuid:    engineUUID,
		},
		tableName: tableName,
		ts:        oracle.ComposeTS(time.Now().Unix()*1000, 0),
	}, nil
}
func (engine *OpenedEngine) Close(ctx context.Context) (*ClosedEngine, error) {
	closedEngine, err := engine.unsafeClose(ctx)
	if err == nil {
		metric.ImporterEngineCounter.WithLabelValues("closed").Inc()
	}
	return closedEngine, err
}
func (engine *OpenedEngine) Flush(ctx context.Context) error {
	return engine.backend.FlushEngine(ctx, engine.uuid)
}
func (engine *OpenedEngine) WriteRows(ctx context.Context, columnNames []string, rows Rows) error {
	writer, err := engine.backend.LocalWriter(ctx, engine.uuid, LocalMemoryTableSize)
	if err != nil {
		return err
	}
	if err = writer.AppendRows(ctx, engine.tableName, columnNames, engine.ts, rows); err != nil {
		writer.Close()
		return err
	}
	return writer.Close()
}
func (engine *OpenedEngine) LocalWriter(ctx context.Context, maxCacheSize int64) (*LocalEngineWriter, error) {
	w, err := engine.backend.LocalWriter(ctx, engine.uuid, maxCacheSize)
	if err != nil {
		return nil, err
	}
	return &LocalEngineWriter{writer: w, ts: engine.ts, tableName: engine.tableName}, nil
}
func (w *LocalEngineWriter) WriteRows(ctx context.Context, columnNames []string, rows Rows) error {
	return w.writer.AppendRows(ctx, w.tableName, columnNames, w.ts, rows)
}
func (w *LocalEngineWriter) Close() error {
	return w.writer.Close()
}
func (be Backend) UnsafeCloseEngine(ctx context.Context, tableName string, engineID int32) (*ClosedEngine, error) {
	tag, engineUUID := MakeUUID(tableName, engineID)
	return be.UnsafeCloseEngineWithUUID(ctx, tag, engineUUID)
}
func (be Backend) UnsafeCloseEngineWithUUID(ctx context.Context, tag string, engineUUID uuid.UUID) (*ClosedEngine, error) {
	return engine{
		backend: be.abstract,
		logger:  makeLogger(tag, engineUUID),
		uuid:    engineUUID,
	}.unsafeClose(ctx)
}
func (en engine) unsafeClose(ctx context.Context) (*ClosedEngine, error) {
	task := en.logger.Begin(zap.InfoLevel, "engine close")
	err := en.backend.CloseEngine(ctx, en.uuid)
	task.End(zap.ErrorLevel, err)
	if err != nil {
		return nil, err
	}
	return &ClosedEngine{engine: en}, nil
}
func (engine *ClosedEngine) Import(ctx context.Context) error {
	var err error
	for i := 0; i < maxRetryTimes; i++ {
		task := engine.logger.With(zap.Int("retryCnt", i)).Begin(zap.InfoLevel, "import")
		err = engine.backend.ImportEngine(ctx, engine.uuid)
		if !common.IsRetryableError(err) {
			task.End(zap.ErrorLevel, err)
			return err
		}
		task.Warn("import spuriously failed, going to retry again", log.ShortError(err))
		time.Sleep(engine.backend.RetryImportDelay())
	}
	return errors.Annotatef(err, "[%s] import reach max retry %d and still failed", engine.uuid, maxRetryTimes)
}