merge

Documentation

Index

• Variables
• func CleanUpUselessFiles(entry *api.MergeCommitEntry, fs fileservice.FileService)
• func NewUpdatePolicyReq(c *BasicPolicyConfig) *api.AlterTableReq
• type ActiveCNObjMap
  • func (e *ActiveCNObjMap) AddActiveCNObj(entries []*catalog.ObjectEntry)
  • func (e *ActiveCNObjMap) CheckOverlapOnCNActive(entries []*catalog.ObjectEntry) bool
  • func (e *ActiveCNObjMap) Prune(id uint64, ago time.Duration)
  • func (e *ActiveCNObjMap) RemoveActiveCNObj(ids []objectio.ObjectId)
  • func (e *ActiveCNObjMap) String() string
• type BasicPolicyConfig
  • func (c *BasicPolicyConfig) String() string
• type CNMergeScheduler
  • func NewTaskServiceGetter(getter taskservice.Getter) CNMergeScheduler
• type MergeExecutor
  • func NewMergeExecutor(rt *dbutils.Runtime, sched CNMergeScheduler) *MergeExecutor
  • func (e *MergeExecutor) AddActiveTask(taskId uint64, blkn, esize int)
  • func (e *MergeExecutor) CPUPercent() int64
  • func (e *MergeExecutor) ExecuteFor(entry *catalog.TableEntry, mobjs []*catalog.ObjectEntry, kind TaskHostKind)
  • func (e *MergeExecutor) MemAvailBytes() int
  • func (e *MergeExecutor) OnExecDone(v any)
  • func (e *MergeExecutor) PrintStats()
  • func (e *MergeExecutor) RefreshMemInfo()
  • func (e *MergeExecutor) TransferPageSizeLimit() uint64
• type Policy
  • func NewBasicPolicy() Policy
• type TaskHostKind

Variables

var DisableDeltaLocMerge atomic.Bool

var StopMerge atomic.Bool

Functions

func CleanUpUselessFiles

func CleanUpUselessFiles(entry *api.MergeCommitEntry, fs fileservice.FileService)

func NewUpdatePolicyReq

func NewUpdatePolicyReq(c *BasicPolicyConfig) *api.AlterTableReq

Types

type ActiveCNObjMap

type ActiveCNObjMap struct {
	sync.Mutex
	// contains filtered or unexported fields
}

var ActiveCNObj ActiveCNObjMap = ActiveCNObjMap{
	// contains filtered or unexported fields
}

func (*ActiveCNObjMap) AddActiveCNObj

func (e *ActiveCNObjMap) AddActiveCNObj(entries []*catalog.ObjectEntry)

func (*ActiveCNObjMap) CheckOverlapOnCNActive

func (e *ActiveCNObjMap) CheckOverlapOnCNActive(entries []*catalog.ObjectEntry) bool

func (*ActiveCNObjMap) Prune

func (e *ActiveCNObjMap) Prune(id uint64, ago time.Duration)

func (*ActiveCNObjMap) RemoveActiveCNObj

func (e *ActiveCNObjMap) RemoveActiveCNObj(ids []objectio.ObjectId)

func (*ActiveCNObjMap) String

func (e *ActiveCNObjMap) String() string

type BasicPolicyConfig

type BasicPolicyConfig struct {
	MergeMaxOneRun    int
	ObjectMinOsize    uint32
	MaxOsizeMergedObj uint32
	MinCNMergeSize    uint64
	FromUser          bool
	MergeHints        []api.MergeHint
	// contains filtered or unexported fields
}

func (*BasicPolicyConfig) String

func (c *BasicPolicyConfig) String() string

type CNMergeScheduler

type CNMergeScheduler interface {
	SendMergeTask(ctx context.Context, task *api.MergeTaskEntry) error
}

func NewTaskServiceGetter

func NewTaskServiceGetter(getter taskservice.Getter) CNMergeScheduler

type MergeExecutor

type MergeExecutor struct {
	// contains filtered or unexported fields
}

MergeExecutor consider resources to decide to merge or not.

func NewMergeExecutor

func NewMergeExecutor(rt *dbutils.Runtime, sched CNMergeScheduler) *MergeExecutor

func (*MergeExecutor) AddActiveTask

func (e *MergeExecutor) AddActiveTask(taskId uint64, blkn, esize int)

func (*MergeExecutor) CPUPercent

func (e *MergeExecutor) CPUPercent() int64

func (*MergeExecutor) ExecuteFor

func (e *MergeExecutor) ExecuteFor(entry *catalog.TableEntry, mobjs []*catalog.ObjectEntry, kind TaskHostKind)

func (*MergeExecutor) MemAvailBytes

func (e *MergeExecutor) MemAvailBytes() int

func (*MergeExecutor) OnExecDone

func (e *MergeExecutor) OnExecDone(v any)

func (*MergeExecutor) PrintStats

func (e *MergeExecutor) PrintStats()

func (*MergeExecutor) RefreshMemInfo

func (e *MergeExecutor) RefreshMemInfo()

func (*MergeExecutor) TransferPageSizeLimit

func (e *MergeExecutor) TransferPageSizeLimit() uint64

type Policy

type Policy interface {
	OnObject(obj *catalog.ObjectEntry, force bool)
	Revise(cpu, mem int64) ([]*catalog.ObjectEntry, TaskHostKind)
	ResetForTable(*catalog.TableEntry)
	SetConfig(*catalog.TableEntry, func() txnif.AsyncTxn, any)
	GetConfig(*catalog.TableEntry) any
}

func NewBasicPolicy

func NewBasicPolicy() Policy

type TaskHostKind

type TaskHostKind int

const (
	TaskHostCN TaskHostKind = iota
	TaskHostDN
)