ddl: flush data in local engine serially

br/pkg/lightning/backend/local/engine.go

@@ -1222,6 +1222,7 @@ func (w *Writer) createSSTWriter() (*sstWriter, error) {
 		return nil, err
 	}
 	sw := &sstWriter{sstMeta: &sstMeta{path: path}, writer: writer, logger: w.engine.logger}
+	sw.logger.Info("create sst writer", zap.String("path", path))
 	return sw, nil
 }
 
@@ -1398,6 +1399,7 @@ func (i dbSSTIngester) mergeSSTs(metas []*sstMeta, dir string) (*sstMeta, error)
 	for _, p := range metas {
 		f, err := os.Open(p.path)
 		if err != nil {
+			i.e.logger.Error("open sst file failed", zap.String("path", p.path), zap.Error(err))
 			return nil, errors.Trace(err)
 		}
 		reader, err := sstable.NewReader(f, sstable.ReaderOptions{})
@@ -1492,6 +1494,8 @@ func (i dbSSTIngester) mergeSSTs(metas []*sstMeta, dir string) (*sstMeta, error)
 			totalSize += m.fileSize
 			if err := os.Remove(m.path); err != nil {
 				i.e.logger.Warn("async cleanup sst file failed", zap.Error(err))
+			} else {
+				i.e.logger.Warn("async cleanup sst file", zap.String("path", m.path))
 			}
 		}
 		// decrease the pending size after clean up

ddl/ingest/backend.go

@@ -142,18 +142,19 @@ func (bc *litBackendCtx) Flush(indexID int64, force bool) (flushed, imported boo
 		return false, false, nil
 	}
 
+	release := ei.acquireFlushLock()
+	if release == nil {
+		return false, false, nil
+	}
+	defer release()
+
 	bc.timeOfLastFlush.Store(time.Now())
 	err = ei.Flush()
 	if err != nil {
 		return false, false, err
 	}
 
 	if force || shouldImport {
-		release := ei.acquireFlushLock()
-		if release == nil {
-			return true, false, nil
-		}
-		defer release()
 		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))

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)
 }