flushing objects in chunks (#22319)

- Process appendable objects in controlled chunks to maintain stable memory usage.
- A chunk means 400 appendable data objects

Approved by: @XuPeng-SH

Author: aptend

pkg/vm/engine/tae/db/checkpoint/flusher.go

@@ -540,30 +540,68 @@ func (flusher *flushImpl) fireFlushTabletail(
 		return nil
 	}
 
-	// freeze all append
-	scopes := make([]common.ID, 0, len(metas))
-	for _, meta := range metas {
-		if !meta.GetObjectData().PrepareCompact() {
-			logutil.Info("[FlushTabletail] data prepareCompact false", zap.String("table", tableDesc), zap.String("obj", meta.ID().String()))
-			return moerr.GetOkExpectedEOB()
-		}
-		scopes = append(scopes, *meta.AsCommonID())
+	if len(metas) == 0 && len(tombstoneMetas) > 0 {
+		metas = append(metas, nil) // make it a non-empty chunk
 	}
-	for _, meta := range tombstoneMetas {
-		if !meta.GetObjectData().PrepareCompact() {
-			logutil.Info("[FlushTabletail] tomb prepareCompact false", zap.String("table", tableDesc), zap.String("obj", meta.ID().String()))
-			return moerr.GetOkExpectedEOB()
+
+	metaChunks := slices.Chunk(metas, 400)
+	firstChunk := true
+
+nextChunk:
+	for chunk := range metaChunks {
+		if len(chunk) == 1 && chunk[0] == nil {
+			chunk = chunk[:0] // remove the placeholder
+		}
+		scopes := make([]common.ID, 0, len(chunk))
+		for _, meta := range chunk {
+			if !meta.GetObjectData().PrepareCompact() {
+				logutil.Info("[FlushTabletail] data prepareCompact false",
+					zap.String("table", tableDesc),
+					zap.String("obj", meta.ID().String()),
+				)
+				break nextChunk
+			}
+			scopes = append(scopes, *meta.AsCommonID())
+		}
+		if firstChunk {
+			for _, meta := range tombstoneMetas {
+				if !meta.GetObjectData().PrepareCompact() {
+					logutil.Info("[FlushTabletail] tomb prepareCompact false",
+						zap.String("table", tableDesc),
+						zap.String("obj", meta.ID().String()),
+					)
+					// As we treat tombstone as a monolithic chunk,
+					// skip this fire and wait for the next run if freeze fails.
+					return moerr.GetOkExpectedEOB()
+				}
+				scopes = append(scopes, *meta.AsCommonID())
+			}
+		}
+		var factory tasks.TxnTaskFactory
+		if firstChunk {
+			factory = jobs.FlushTableTailTaskFactory(chunk, tombstoneMetas, flusher.rt)
+			firstChunk = false
+		} else {
+			logutil.Info("[FlushTabletail] fire chunk",
+				zap.String("table", tableDesc),
+				zap.Int("chunk-size", len(chunk)),
+			)
+			factory = jobs.FlushTableTailTaskFactory(chunk, nil, flusher.rt)
 		}
-		scopes = append(scopes, *meta.AsCommonID())
-	}
 
-	factory := jobs.FlushTableTailTaskFactory(metas, tombstoneMetas, flusher.rt)
-	if _, err := flusher.rt.Scheduler.ScheduleMultiScopedTxnTask(nil, tasks.FlushTableTailTask, scopes, factory); err != nil {
-		if err != tasks.ErrScheduleScopeConflict {
-			logutil.Error("[FlushTabletail] Sched Failure", zap.String("table", tableDesc), zap.Error(err))
+		_, err := flusher.rt.Scheduler.ScheduleMultiScopedTxnTask(
+			nil, tasks.FlushTableTailTask, scopes, factory)
+		if err != nil {
+			if err != tasks.ErrScheduleScopeConflict {
+				logutil.Error("[FlushTabletail] Sched Failure",
+					zap.String("table", tableDesc),
+					zap.Error(err),
+				)
+			}
+			return moerr.GetOkExpectedEOB()
 		}
-		return moerr.GetOkExpectedEOB()
 	}
+
 	return nil
 }