ddl: change finishedWritingNeedImport to importStarted

pkg/ddl/backfilling.go

@@ -593,7 +593,7 @@ func (dc *ddlCtx) writePhysicalTableRecord(
 	}
 	defer scheduler.close(true)
 	if lit, ok := scheduler.(*ingestBackfillScheduler); ok {
-		if lit.finishedWritingNeedImport() {
+		if lit.importStarted() {
 			return nil
 		}
 	}

pkg/ddl/backfilling_scheduler.go

@@ -370,13 +370,13 @@ func newIngestBackfillScheduler(
 	}, nil
 }
 
-func (b *ingestBackfillScheduler) finishedWritingNeedImport() bool {
+func (b *ingestBackfillScheduler) importStarted() bool {
 	job := b.reorgInfo.Job
 	bc, ok := ingest.LitBackCtxMgr.Load(job.ID)
 	if !ok {
 		return false
 	}
-	return bc.FinishedWritingNeedImport()
+	return bc.ImportStarted()
 }
 
 func (b *ingestBackfillScheduler) setupWorkers() error {

pkg/ddl/ingest/backend.go

@@ -59,11 +59,11 @@ type BackendCtx interface {
 	// TODO(lance6716): refine the interface to let caller don't need to pass the
 	// indexID, and unify with CollectRemoteDuplicateRows.
 	FinishImport(indexID int64, unique bool, tbl table.Table) error
-	// FinishedWritingNeedImport returns true only when all the engines are finished
-	// writing and only need import. Considering the calling usage of FinishImport,
-	// it will return true after a successful call of FinishImport and may return
-	// true after a failed call of FinishImport.
-	FinishedWritingNeedImport() bool
+	// ImportStarted returns true only when all the engines are finished writing and
+	// import is started by FinishImport. Considering the calling usage of
+	// FinishImport, it will return true after a successful call of FinishImport and
+	// may return true after a failed call of FinishImport.
+	ImportStarted() bool
 
 	CollectRemoteDuplicateRows(indexID int64, tbl table.Table) error
 	FlushController

pkg/ddl/ingest/engine_mgr.go

@@ -111,13 +111,13 @@ func (bc *litBackendCtx) UnregisterEngines() {
 	bc.memRoot.Release(numIdx * (structSizeEngineInfo + engineCacheSize))
 }
 
-// FinishedWritingNeedImport implements BackendCtx.
-func (bc *litBackendCtx) FinishedWritingNeedImport() bool {
+// ImportStarted implements BackendCtx.
+func (bc *litBackendCtx) ImportStarted() bool {
 	if len(bc.engines) == 0 {
 		return false
 	}
 	for _, ei := range bc.engines {
-		if ei.closedEngine != nil {
+		if ei.openedEngine == nil {
 			return true
 		}
 	}

pkg/ddl/ingest/mock.go

@@ -119,8 +119,8 @@ func (*MockBackendCtx) UnregisterEngines() {
 	logutil.DDLIngestLogger().Info("mock backend ctx unregister")
 }
 
-// FinishedWritingNeedImport implements BackendCtx interface.
-func (*MockBackendCtx) FinishedWritingNeedImport() bool {
+// ImportStarted implements BackendCtx interface.
+func (*MockBackendCtx) ImportStarted() bool {
 	return false
 }
 

tests/realtikvtest/addindextest4/ingest_test.go

@@ -376,6 +376,7 @@ func TestAddIndexFinishImportError(t *testing.T) {
 	tk.MustExec("create database addindexlit;")
 	tk.MustExec("use addindexlit;")
 	tk.MustExec(`set global tidb_ddl_enable_fast_reorg=on;`)
+	tk.MustExec("set global tidb_enable_dist_task = off;")
 
 	tk.MustExec("create table t (a int primary key, b int);")
 	for i := 0; i < 4; i++ {