ddl: flush data in local engine serially

ddl/index.go

@@ -23,6 +23,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -1595,6 +1596,7 @@ type addIndexIngestWorker struct {
 	writer           ingest.Writer
 	copReqSenderPool *copReqSenderPool
 	checkpointMgr    *ingest.CheckpointManager
+	flushLock        *sync.RWMutex
 
 	resultCh   chan *backfillResult
 	jobID      int64
@@ -1652,6 +1654,8 @@ func writeChunkToLocal(writer ingest.Writer,
 	handleDataBuf := make([]types.Datum, len(copCtx.handleOutputOffsets))
 	count := 0
 	var lastHandle kv.Handle
+	unlock := writer.LockForWrite()
+	defer unlock()
 	for row := iter.Begin(); row != iter.End(); row = iter.Next() {
 		idxDataBuf, handleDataBuf = idxDataBuf[:0], handleDataBuf[:0]
 		idxDataBuf = extractDatumByOffsets(row, copCtx.idxColOutputOffsets, copCtx.expColInfos, idxDataBuf)

ddl/ingest/backend.go

@@ -133,38 +133,47 @@ func (bc *litBackendCtx) Flush(indexID int64, force bool) (flushed, imported boo
 		return false, false, dbterror.ErrIngestFailed.FastGenByArgs("ingest engine not found")
 	}
 
-	bc.diskRoot.UpdateUsage()
-	shouldImport := bc.diskRoot.ShouldImport()
-	shouldFlush := force ||
-		shouldImport ||
-		time.Since(bc.timeOfLastFlush.Load()) >= bc.updateInterval
+	shouldFlush, _ := bc.ShouldSync(force)
+	if !shouldFlush {
+		return false, false, nil
+	}
+	ei.flushLock.Lock()
+	defer ei.flushLock.Unlock()
+	// Check for the second time to avoid duplicate import.
+	shouldFlush, shouldImport := bc.ShouldSync(force)
 	if !shouldFlush {
 		return false, false, nil
 	}
 
-	bc.timeOfLastFlush.Store(time.Now())
 	err = ei.Flush()
 	if err != nil {
 		return false, false, err
 	}
+	bc.timeOfLastFlush.Store(time.Now())
 
-	if force || shouldImport {
-		release := ei.acquireFlushLock()
-		if release == nil {
-			return true, false, nil
-		}
-		defer release()
-		logutil.BgLogger().Info(LitInfoUnsafeImport, zap.Int64("index ID", indexID),
+	if !shouldImport {
+		return true, false, nil
+	}
+	logutil.BgLogger().Info(LitInfoUnsafeImport, zap.Int64("index ID", indexID),
+		zap.String("usage info", bc.diskRoot.UsageInfo()))
+	err = bc.backend.UnsafeImportAndReset(bc.ctx, ei.uuid, int64(lightning.SplitRegionSize)*int64(lightning.MaxSplitRegionSizeRatio), int64(lightning.SplitRegionKeys))
+	if err != nil {
+		logutil.BgLogger().Error(LitErrIngestDataErr, zap.Int64("index ID", indexID),
 			zap.String("usage info", bc.diskRoot.UsageInfo()))
-		err = bc.backend.UnsafeImportAndReset(bc.ctx, ei.uuid, int64(lightning.SplitRegionSize)*int64(lightning.MaxSplitRegionSizeRatio), int64(lightning.SplitRegionKeys))
-		if err != nil {
-			logutil.BgLogger().Error(LitErrIngestDataErr, zap.Int64("index ID", indexID),
-				zap.String("usage info", bc.diskRoot.UsageInfo()))
-			return true, false, err
-		}
-		return true, true, nil
+		return true, false, err
+	}
+	return true, true, nil
+}
+
+func (bc *litBackendCtx) ShouldSync(force bool) (shouldFlush bool, shouldImport bool) {
+	if force {
+		return true, true
 	}
-	return true, false, nil
+	bc.diskRoot.UpdateUsage()
+	shouldImport = bc.diskRoot.ShouldImport()
+	shouldFlush = shouldImport ||
+		time.Since(bc.timeOfLastFlush.Load()) >= bc.updateInterval
+	return shouldFlush, shouldImport
 }
 
 // Done returns true if the lightning backfill is done.

ddl/ingest/disk_root.go

@@ -76,17 +76,19 @@ func (d *diskRootImpl) ShouldImport() bool {
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 	if d.bcUsed > variable.DDLDiskQuota.Load() {
-		logutil.BgLogger().Info("[ddl-ingest] disk usage is over quota, start importing",
+		logutil.BgLogger().Info("[ddl-ingest] disk usage is over quota",
 			zap.Uint64("quota", variable.DDLDiskQuota.Load()),
-			zap.String("usage", d.UsageInfo()))
+			zap.String("usage", d.usageInfo()))
 		return true
 	}
 	if d.used == 0 && d.capacity == 0 {
 		return false
 	}
 	if float64(d.used) >= float64(d.capacity)*capacityThreshold {
-		logutil.BgLogger().Warn("[ddl-ingest] available disk space is less than 10%, start importing, this may degrade the performance",
-			zap.String("usage", d.UsageInfo()))
+		logutil.BgLogger().Warn("[ddl-ingest] available disk space is less than 10%, "+
+			"this may degrade the performance, "+
+			"please make sure the disk available space is larger than @@tidb_ddl_disk_quota before adding index",
+			zap.String("usage", d.usageInfo()))
 		return true
 	}
 	return false
@@ -96,6 +98,10 @@ func (d *diskRootImpl) ShouldImport() bool {
 func (d *diskRootImpl) UsageInfo() string {
 	d.mu.RLock()
 	defer d.mu.RUnlock()
+	return d.usageInfo()
+}
+
+func (d *diskRootImpl) usageInfo() string {
 	return fmt.Sprintf("disk usage: %d/%d, backend usage: %d", d.used, d.capacity, d.bcUsed)
 }
 

ddl/ingest/engine.go

@@ -17,6 +17,7 @@ package ingest
 import (
 	"context"
 	"strconv"
+	"sync"
 	"sync/atomic"
 
 	"github.com/google/uuid"
@@ -41,6 +42,7 @@ type Engine interface {
 // Writer is the interface for the writer that can be used to write key-value pairs.
 type Writer interface {
 	WriteRow(key, idxVal []byte, handle tidbkv.Handle) error
+	LockForWrite() (unlock func())
 }
 
 // engineInfo is the engine for one index reorg task, each task will create several new writers under the
@@ -57,7 +59,7 @@ type engineInfo struct {
 	memRoot      MemRoot
 	diskRoot     DiskRoot
 	rowSeq       atomic.Int64
-	flushing     atomic.Bool
+	flushLock    *sync.RWMutex
 }
 
 // newEngineInfo create a new engineInfo struct.
@@ -74,6 +76,7 @@ func newEngineInfo(ctx context.Context, jobID, indexID int64, cfg *backend.Engin
 		writerCache:  generic.NewSyncMap[int, backend.EngineWriter](wCnt),
 		memRoot:      memRoot,
 		diskRoot:     diskRoot,
+		flushLock:    &sync.RWMutex{},
 	}
 }
 
@@ -88,17 +91,6 @@ func (ei *engineInfo) Flush() error {
 	return nil
 }
 
-// acquireFlushLock acquires the flush lock of the engine.
-func (ei *engineInfo) acquireFlushLock() (release func()) {
-	ok := ei.flushing.CompareAndSwap(false, true)
-	if !ok {
-		return nil
-	}
-	return func() {
-		ei.flushing.Store(false)
-	}
-}
-
 // Clean closes the engine and removes the local intermediate files.
 func (ei *engineInfo) Clean() {
 	if ei.openedEngine == nil {
@@ -170,6 +162,7 @@ type writerContext struct {
 	ctx    context.Context
 	unique bool
 	lWrite backend.EngineWriter
+	fLock  *sync.RWMutex
 }
 
 // CreateWriter creates a new writerContext.
@@ -216,6 +209,7 @@ func (ei *engineInfo) newWriterContext(workerID int, unique bool) (*writerContex
 		ctx:    ei.ctx,
 		unique: unique,
 		lWrite: lWrite,
+		fLock:  ei.flushLock,
 	}
 	return wc, nil
 }
@@ -247,3 +241,11 @@ func (wCtx *writerContext) WriteRow(key, idxVal []byte, handle tidbkv.Handle) er
 	row := kv.MakeRowsFromKvPairs(kvs)
 	return wCtx.lWrite.AppendRows(wCtx.ctx, nil, row)
 }
+
+// LockForWrite locks the local writer for write.
+func (wCtx *writerContext) LockForWrite() (unlock func()) {
+	wCtx.fLock.RLock()
+	return func() {
+		wCtx.fLock.RUnlock()
+	}
+}

ddl/ingest/mock.go

@@ -171,3 +171,8 @@ func (m *MockWriter) WriteRow(key, idxVal []byte, _ kv.Handle) error {
 	}
 	return txn.Set(key, idxVal)
 }
+
+// LockForWrite implements Writer.LockForWrite interface.
+func (*MockWriter) LockForWrite() func() {
+	return func() {}
+}