Refactor merge policy

pkg/cnservice/server_task.go

@@ -273,11 +273,11 @@ func (s *service) registerExecutorsLocked() {
 				stats := objectio.ObjectStats(b)
 				objs[i] = stats.ObjectName().String()
 			}
-			sql := fmt.Sprintf("select mo_ctl('DN', 'MERGEOBJECTS', '%s.%s:%s')",
-				mergeTask.DbName, mergeTask.TableName, strings.Join(objs, ","))
+			sql := fmt.Sprintf("select mo_ctl('DN', 'MERGEOBJECTS', '.%d.%d:%s')",
+				mergeTask.TblId, mergeTask.AccountId, strings.Join(objs, ","))
 			ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
 			defer cancel()
-			opts := executor.Options{}.WithAccountID(mergeTask.AccountId).WithWaitCommittedLogApplied()
+			opts := executor.Options{}.WithWaitCommittedLogApplied()
 			_, err = s.sqlExecutor.Exec(ctx, sql, opts)
 			return err
 		},

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

@@ -967,65 +967,67 @@ func (r *runner) tryCompactTree(entry *logtail.DirtyTreeEntry, force bool) {
 		dirtyTree := entry.GetTree().GetTable(table.ID)
 		_, endTs := entry.GetTimeRange()
 
-		stats := table.Stats
-		stats.Lock()
-		defer stats.Unlock()
+		func() {
+			stats := table.Stats
+			stats.Lock()
+			defer stats.Unlock()
+
+			if force {
+				logutil.Infof("[flushtabletail] force flush %v-%s", table.ID, table.GetLastestSchemaLocked().Name)
+				if err := r.fireFlushTabletail(table, dirtyTree, endTs); err == nil {
+					stats.ResetDeadlineWithLock()
+				}
+				return
+			}
 
-		if force {
-			logutil.Infof("[flushtabletail] force flush %v-%s", table.ID, table.GetLastestSchemaLocked().Name)
-			if err := r.fireFlushTabletail(table, dirtyTree, endTs); err == nil {
+			if stats.LastFlush.IsEmpty() {
+				// first boot, just bail out, and never enter this branch again
+				stats.LastFlush = stats.LastFlush.Next()
 				stats.ResetDeadlineWithLock()
+				return
 			}
-			continue
-		}
-
-		if stats.LastFlush.IsEmpty() {
-			// first boot, just bail out, and never enter this branch again
-			stats.LastFlush = stats.LastFlush.Next()
-			stats.ResetDeadlineWithLock()
-			continue
-		}
 
-		flushReady := func() bool {
-			if !table.IsActive() {
-				count++
-				if pressure < 0.5 || count < 200 {
+			flushReady := func() bool {
+				if !table.IsActive() {
+					count++
+					if pressure < 0.5 || count < 200 {
+						return true
+					}
+					return false
+				}
+				if stats.FlushDeadline.Before(time.Now()) {
+					return true
+				}
+				if asize+dsize > stats.FlushMemCapacity {
+					return true
+				}
+				if asize < common.Const1MBytes && dsize > 2*common.Const1MBytes+common.Const1MBytes/2 {
+					return true
+				}
+				if asize > common.Const1MBytes && rand.Float64() < pressure {
 					return true
 				}
 				return false
 			}
-			if stats.FlushDeadline.Before(time.Now()) {
-				return true
-			}
-			if asize+dsize > stats.FlushMemCapacity {
-				return true
-			}
-			if asize < common.Const1MBytes && dsize > 2*common.Const1MBytes+common.Const1MBytes/2 {
-				return true
-			}
-			if asize > common.Const1MBytes && rand.Float64() < pressure {
-				return true
-			}
-			return false
-		}
 
-		ready := flushReady()
-		// debug log, delete later
-		if !stats.LastFlush.IsEmpty() && asize+dsize > 2*1000*1024 {
-			logutil.Infof("[flushtabletail] %v(%v) %v dels  FlushCountDown %v, flushReady %v",
-				table.GetLastestSchemaLocked().Name,
-				common.HumanReadableBytes(asize+dsize),
-				common.HumanReadableBytes(dsize),
-				time.Until(stats.FlushDeadline),
-				ready,
-			)
-		}
+			ready := flushReady()
+			// debug log, delete later
+			if !stats.LastFlush.IsEmpty() && asize+dsize > 2*1000*1024 {
+				logutil.Infof("[flushtabletail] %v(%v) %v dels  FlushCountDown %v, flushReady %v",
+					table.GetLastestSchemaLocked().Name,
+					common.HumanReadableBytes(asize+dsize),
+					common.HumanReadableBytes(dsize),
+					time.Until(stats.FlushDeadline),
+					ready,
+				)
+			}
 
-		if ready {
-			if err := r.fireFlushTabletail(table, dirtyTree, endTs); err == nil {
-				stats.ResetDeadlineWithLock()
+			if ready {
+				if err := r.fireFlushTabletail(table, dirtyTree, endTs); err == nil {
+					stats.ResetDeadlineWithLock()
+				}
 			}
-		}
+		}()
 	}
 }
 

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

@@ -64,6 +64,7 @@ type DB struct {
 	BGScanner          wb.IHeartbeater
 	BGCheckpointRunner checkpoint.Runner
 	MergeHandle        *MergeTaskBuilder
+	mergeScheduler     *merge.Scheduler
 
 	DiskCleaner *gc2.DiskCleaner
 

pkg/vm/engine/tae/db/merge/executor.go

@@ -17,17 +17,17 @@ package merge
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"sync"
 	"sync/atomic"
 
+	"github.com/KimMachineGun/automemlimit/memlimit"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/objectio"
 	"github.com/matrixorigin/matrixone/pkg/pb/api"
 	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
-
-	"github.com/KimMachineGun/automemlimit/memlimit"
-	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/catalog"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/db/dbutils"
@@ -36,6 +36,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/tasks"
 	"github.com/shirou/gopsutil/v3/cpu"
 	"github.com/shirou/gopsutil/v3/mem"
+	"go.uber.org/zap"
 )
 
 type activeTaskStats map[uint64]struct {
@@ -51,8 +52,8 @@ type MergeExecutor struct {
 	memAvail            int
 	memSpare            int // 10% of total memory or container memory limit
 	cpuPercent          float64
-	activeMergeBlkCount int32
-	activeEstimateBytes int64
+	activeMergeBlkCount atomic.Int32
+	activeEstimateBytes atomic.Int64
 	taskConsume         struct {
 		sync.Mutex
 		o map[objectio.ObjectId]struct{}
@@ -98,22 +99,22 @@ func (e *MergeExecutor) RefreshMemInfo() {
 }
 
 func (e *MergeExecutor) PrintStats() {
-	cnt := atomic.LoadInt32(&e.activeMergeBlkCount)
+	cnt := e.activeMergeBlkCount.Load()
 	if cnt == 0 && e.MemAvailBytes() > 512*common.Const1MBytes {
 		return
 	}
 
 	logutil.Infof(
-		"Mergeblocks avail mem: %v(%v reserved), active mergeing size: %v, active merging blk cnt: %d",
+		"[Mergeblocks] avail mem: %v(%v reserved), active mergeing size: %v, active merging blk cnt: %d",
 		common.HumanReadableBytes(e.memAvail),
 		common.HumanReadableBytes(e.memSpare),
-		common.HumanReadableBytes(int(atomic.LoadInt64(&e.activeEstimateBytes))), cnt,
+		common.HumanReadableBytes(int(e.activeEstimateBytes.Load())), cnt,
 	)
 }
 
 func (e *MergeExecutor) AddActiveTask(taskId uint64, blkn, esize int) {
-	atomic.AddInt64(&e.activeEstimateBytes, int64(esize))
-	atomic.AddInt32(&e.activeMergeBlkCount, int32(blkn))
+	e.activeEstimateBytes.Add(int64(esize))
+	e.activeMergeBlkCount.Add(int32(blkn))
 	e.taskConsume.Lock()
 	if e.taskConsume.m == nil {
 		e.taskConsume.m = make(activeTaskStats)
@@ -133,11 +134,42 @@ func (e *MergeExecutor) OnExecDone(v any) {
 	delete(e.taskConsume.m, task.ID())
 	e.taskConsume.Unlock()
 
-	atomic.AddInt32(&e.activeMergeBlkCount, -int32(stat.blk))
-	atomic.AddInt64(&e.activeEstimateBytes, -int64(stat.estBytes))
+	e.activeMergeBlkCount.Add(-int32(stat.blk))
+	e.activeEstimateBytes.Add(-int64(stat.estBytes))
+}
+
+func (e *MergeExecutor) ExecuteSingleObjMerge(entry *catalog.TableEntry, mobjs []*catalog.ObjectEntry, kind TaskHostKind) {
+	e.tableName = fmt.Sprintf("%v-%v", entry.ID, entry.GetLastestSchema().Name)
+
+	if ActiveCNObj.CheckOverlapOnCNActive(mobjs) {
+		return
+	}
+
+	osize, _, _ := estimateMergeConsume(mobjs)
+	blkCnt := 0
+	for _, obj := range mobjs {
+		blkCnt += obj.BlockCnt()
+	}
+
+	for _, obj := range mobjs {
+		factory := func(ctx *tasks.Context, txn txnif.AsyncTxn) (tasks.Task, error) {
+			return jobs.NewMergeObjectsTask(ctx, txn, []*catalog.ObjectEntry{obj}, e.rt, common.DefaultMaxOsizeObjMB*common.Const1MBytes)
+		}
+		task, err := e.rt.Scheduler.ScheduleMultiScopedTxnTaskWithObserver(nil, tasks.DataCompactionTask, []common.ID{*obj.AsCommonID()}, factory, e)
+		if err != nil {
+			if !errors.Is(err, tasks.ErrScheduleScopeConflict) {
+				logutil.Info("[Mergeblocks] Schedule error", zap.Error(err))
+			}
+			return
+		}
+		singleOSize, singleESize, _ := estimateSingleObjMergeConsume(obj)
+		e.AddActiveTask(task.ID(), obj.BlockCnt(), singleESize)
+		logSingleObjMergeTask(e.tableName, task.ID(), obj, obj.BlockCnt(), singleOSize, singleESize)
+	}
+	entry.Stats.AddMerge(osize, len(mobjs), blkCnt)
 }
 
-func (e *MergeExecutor) ExecuteFor(entry *catalog.TableEntry, mobjs []*catalog.ObjectEntry, kind TaskHostKind) {
+func (e *MergeExecutor) ExecuteMultiObjMerge(entry *catalog.TableEntry, mobjs []*catalog.ObjectEntry, kind TaskHostKind) {
 	e.tableName = fmt.Sprintf("%v-%v", entry.ID, entry.GetLastestSchema().Name)
 
 	if ActiveCNObj.CheckOverlapOnCNActive(mobjs) {
@@ -176,7 +208,7 @@ func (e *MergeExecutor) ExecuteFor(entry *catalog.TableEntry, mobjs []*catalog.O
 			ActiveCNObj.AddActiveCNObj(mobjs)
 			logMergeTask(e.tableName, math.MaxUint64, mobjs, blkCnt, osize, esize)
 		} else {
-			logutil.Warnf("mergeblocks send to cn error: %v", err)
+			logutil.Error("[Mergeblocks] Send to cn error", zap.Error(err))
 			return
 		}
 	} else {
@@ -191,8 +223,8 @@ func (e *MergeExecutor) ExecuteFor(entry *catalog.TableEntry, mobjs []*catalog.O
 		}
 		task, err := e.rt.Scheduler.ScheduleMultiScopedTxnTaskWithObserver(nil, tasks.DataCompactionTask, scopes, factory, e)
 		if err != nil {
-			if err != tasks.ErrScheduleScopeConflict {
-				logutil.Infof("[Mergeblocks] Schedule error info=%v", err)
+			if !errors.Is(err, tasks.ErrScheduleScopeConflict) {
+				logutil.Info("[Mergeblocks] Schedule error", zap.Error(err))
 			}
 			return
 		}
@@ -204,7 +236,7 @@ func (e *MergeExecutor) ExecuteFor(entry *catalog.TableEntry, mobjs []*catalog.O
 }
 
 func (e *MergeExecutor) MemAvailBytes() int {
-	merging := int(atomic.LoadInt64(&e.activeEstimateBytes))
+	merging := int(e.activeEstimateBytes.Load())
 	avail := e.memAvail - e.memSpare - merging
 	if avail < 0 {
 		avail = 0
@@ -216,6 +248,28 @@ func (e *MergeExecutor) CPUPercent() int64 {
 	return int64(e.cpuPercent)
 }
 
+func logSingleObjMergeTask(name string, taskId uint64, obj *catalog.ObjectEntry, blkn, osize, esize int) {
+	rows := obj.GetRemainingRows()
+	infoBuf := &bytes.Buffer{}
+	infoBuf.WriteString(fmt.Sprintf(" %d(%s)", rows, common.ShortObjId(obj.ID)))
+	platform := fmt.Sprintf("t%d", taskId)
+	if taskId == math.MaxUint64 {
+		platform = "CN"
+		v2.TaskCNMergeScheduledByCounter.Inc()
+		v2.TaskCNMergedSizeCounter.Add(float64(osize))
+	} else {
+		v2.TaskDNMergeScheduledByCounter.Inc()
+		v2.TaskDNMergedSizeCounter.Add(float64(osize))
+	}
+	logutil.Infof(
+		"[Mergeblocks] Scheduled %v [%v|on1,bn%d|%s,%s], merged(%v): %s", name,
+		platform, blkn,
+		common.HumanReadableBytes(osize), common.HumanReadableBytes(esize),
+		rows,
+		infoBuf.String(),
+	)
+}
+
 func logMergeTask(name string, taskId uint64, merges []*catalog.ObjectEntry, blkn, osize, esize int) {
 	rows := 0
 	infoBuf := &bytes.Buffer{}

pkg/vm/engine/tae/db/merge/mod.go

@@ -200,13 +200,19 @@ const (
 )
 
 type Policy interface {
-	OnObject(obj *catalog.ObjectEntry, force bool)
-	Revise(cpu, mem int64, largeFirst bool) ([]*catalog.ObjectEntry, TaskHostKind)
+	OnObject(obj *catalog.ObjectEntry)
+	Revise(cpu, mem int64) ([]*catalog.ObjectEntry, TaskHostKind)
 	ResetForTable(*catalog.TableEntry)
 	SetConfig(*catalog.TableEntry, func() txnif.AsyncTxn, any)
 	GetConfig(*catalog.TableEntry) any
 }
 
+type policy interface {
+	Revise(cpu, mem int64) ([]*catalog.ObjectEntry, TaskHostKind)
+	OnObject(*catalog.ObjectEntry)
+	ResetForTable(*catalog.TableEntry)
+}
+
 func NewUpdatePolicyReq(c *BasicPolicyConfig) *api.AlterTableReq {
 	return &api.AlterTableReq{
 		Kind: api.AlterKind_UpdatePolicy,