db

package
v0.0.0-debug-20260702 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 2, 2026 License: Apache-2.0 Imports: 42 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	CronJobs_Name_GCTransferTable = "GC-Transfer-Table"
	CronJobs_Name_GCDisk          = "GC-Disk"
	CronJobs_Name_GCCheckpoint    = "GC-Checkpoint"
	CronJobs_Name_GCCatalogCache  = "GC-Catalog-Cache"
	CronJobs_Name_GCLogtail       = "GC-Logtail"
	CronJobs_Name_GCLockMerge     = "GC-Lock-Merge"

	CronJobs_Name_ReportStats = "Report-Stats"

	CronJobs_Name_Checker = "Checker"
)
View Source 
const (
	WALDir = "wal"

	Phase_Open = "open-tae"
)
View Source 
const (
	LockName string = "TAE"
)

Variables

View Source 
var (
	ErrTaskDuplicated = moerr.NewInternalErrorNoCtx("tae task: duplicated task found")
	ErrTaskNotFound   = moerr.NewInternalErrorNoCtx("tae task: task not found")
)
View Source 
var CronJobs_Open_ReplayMode = []string{
	CronJobs_Name_GCTransferTable,
	CronJobs_Name_GCCatalogCache,
	CronJobs_Name_GCLogtail,
	CronJobs_Name_ReportStats,
}
View Source 
var CronJobs_Open_WriteMode = []string{
	CronJobs_Name_GCTransferTable,
	CronJobs_Name_GCDisk,
	CronJobs_Name_GCCheckpoint,
	CronJobs_Name_GCCatalogCache,
	CronJobs_Name_GCLogtail,
	CronJobs_Name_GCLockMerge,
	CronJobs_Name_ReportStats,
}
View Source 
var CronJobs_Spec = map[string][]bool{
	CronJobs_Name_GCTransferTable: {true, true, true, true},
	CronJobs_Name_GCDisk:          {true, true, false, false},
	CronJobs_Name_GCCheckpoint:    {true, true, false, false},
	CronJobs_Name_GCCatalogCache:  {true, true, true, true},
	CronJobs_Name_GCLogtail:       {true, true, true, true},
	CronJobs_Name_GCLockMerge:     {true, true, true, false},
	CronJobs_Name_ReportStats:     {true, true, true, true},
	CronJobs_Name_Checker:         {true, false, true, false},
}

key(string): cron job name value(bool,bool,bool,bool): 1.bool,2.bool: can be in the write mode, must be in the write mode 3.bool,4.bool: can be in the replay mode, must be in the replay mode

View Source 
var ErrCancelReplayAnyway = moerr.NewInternalErrorNoCtx("terminate")
View Source 
var (
	ErrClosed = moerr.NewInternalErrorNoCtx("tae: closed")
)
View Source 
var ErrCronJobsOpen = moerr.NewInternalErrorNoCtx("cron jobs mock error")

Functions

func AddCronJob 

func AddCronJob(db *DB, name string, skipMode bool) (err error)

func AddCronJobs 

func AddCronJobs(db *DB) (err error)

func CanAddCronJob 

func CanAddCronJob(name string, isWriteModeDB, skipMode bool) bool

func CheckCronJobs 

func CheckCronJobs(db *DB, expectMode DBTxnMode) (err error)

func NewDBScanner 

func NewDBScanner(db *DB, errHandler ErrHandler) *dbScanner

func RemoveCronJob 

func RemoveCronJob(db *DB, name string)

func ScopeConflictCheck 

func ScopeConflictCheck(oldScope, newScope *common.ID) (err error)

Types

type ControlCmd 

type ControlCmd interface {
	WaitDone()
	Error() error
}

type ControlCmdType 

type ControlCmdType uint32
const (
	ControlCmd_Customized ControlCmdType = iota
	ControlCmd_ToReplayMode
	ControlCmd_ToWriteMode

	ControlCmd_ToStopGC
	ControlCmd_ToStartGC
)

type Controller 

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

func NewController 

func NewController(db *DB) *Controller

func (*Controller) AssembleDB 

func (c *Controller) AssembleDB(ctx context.Context) (err error)

func (*Controller) ScheduleCustomized 

func (c *Controller) ScheduleCustomized(
	ctx context.Context,
	fn func() error,
) (cmd ControlCmd, err error)

func (*Controller) Start 

func (c *Controller) Start()

func (*Controller) Stop 

func (c *Controller) Stop(fn func() error)

func (*Controller) SwitchTxnMode 

func (c *Controller) SwitchTxnMode(
	ctx context.Context,
	iarg int,
	sarg string,
) (err error)

type DB 

type DB struct {
	Dir        string
	TxnMode    atomic.Uint32
	Controller *Controller

	TxnServer rpc.TxnServer

	Opts *options.Options

	Catalog *catalog.Catalog

	TxnMgr *txnbase.TxnManager

	LogtailMgr *logtail.Manager
	Wal        wal.Store

	CronJobs *tasks.CancelableJobs

	BGCheckpointRunner checkpoint.Runner
	BGFlusher          checkpoint.Flusher

	MergeScheduler *merge.MergeScheduler

	DiskCleaner *gc2.DiskCleaner

	Runtime *dbutils.Runtime

	ReplayCtl *replayCtl

	DBLocker io.Closer

	Closed *atomic.Value
	// contains filtered or unexported fields
}

func Open 

func Open(
	ctx context.Context,
	dirname string,
	opts *options.Options,
	dbOpts ...DBOption,
) (db *DB, err error)

func (*DB) AddFaultPoint added in v1.0.0 

func (db *DB) AddFaultPoint(
	ctx context.Context, name string, freq string,
	action string, iarg int64, sarg string, constant bool,
) error

func (*DB) Close 

func (db *DB) Close() error

func (*DB) CollectCheckpointsInRange 

func (db *DB) CollectCheckpointsInRange(
	ctx context.Context, start, end types.TS,
) (ckpLoc string, lastEnd types.TS, err error)

func (*DB) CommitTxn 

func (db *DB) CommitTxn(txn txnif.AsyncTxn) (err error)

func (*DB) FlushTable added in v0.6.0 

func (db *DB) FlushTable(
	ctx context.Context,
	tenantID uint32,
	dbId, tableId uint64,
	ts types.TS,
) (err error)

func (*DB) ForceCheckpoint added in v0.8.0 

func (db *DB) ForceCheckpoint(
	ctx context.Context,
	ts types.TS,
) (err error)

func (*DB) ForceCheckpointForBackup added in v1.1.0 

func (db *DB) ForceCheckpointForBackup(
	ctx context.Context,
	ts types.TS,
) (location string, err error)

func (*DB) ForceFlush 

func (db *DB) ForceFlush(
	ctx context.Context, ts types.TS,
) (err error)

func (*DB) ForceGlobalCheckpoint added in v1.1.2 

func (db *DB) ForceGlobalCheckpoint(
	ctx context.Context,
	ts types.TS,
	historyRetention time.Duration,
) (err error)

func (*DB) GetOrCreateTxnWithMeta added in v0.6.0 

func (db *DB) GetOrCreateTxnWithMeta(
	info []byte,
	id []byte,
	ts types.TS,
) (txn txnif.AsyncTxn, err error)

func (*DB) GetTxnByID added in v0.8.0 

func (db *DB) GetTxnByID(id []byte) (txn txnif.AsyncTxn, err error)

func (*DB) GetTxnMode 

func (db *DB) GetTxnMode() DBTxnMode

func (*DB) GetUsageMemo added in v1.1.0 

func (db *DB) GetUsageMemo() *logtail.TNUsageMemo

func (*DB) IsReplayMode 

func (db *DB) IsReplayMode() bool

func (*DB) IsWriteMode 

func (db *DB) IsWriteMode() bool

func (*DB) ReplayWal 

func (db *DB) ReplayWal(
	ctx context.Context,
	maxTs types.TS,
	lsn uint64,
	onDone func(),
) (ctl *replayCtl, err error)

func (*DB) ResetTxnHeartbeat 

func (db *DB) ResetTxnHeartbeat()

func (*DB) RollbackTxn 

func (db *DB) RollbackTxn(txn txnif.AsyncTxn) error

func (*DB) StartTxn 

func (db *DB) StartTxn(info []byte) (txnif.AsyncTxn, error)

func (*DB) StartTxnWithStartTSAndSnapshotTS added in v1.0.0 

func (db *DB) StartTxnWithStartTSAndSnapshotTS(
	info []byte,
	ts types.TS,
) (txn txnif.AsyncTxn, err error)

func (*DB) StopTxnHeartbeat 

func (db *DB) StopTxnHeartbeat()

func (*DB) SwitchTxnMode 

func (db *DB) SwitchTxnMode(
	ctx context.Context,
	iarg int,
	sarg string,
) error

type DBOption 

type DBOption func(*DB)

func WithTxnMode 

func WithTxnMode(mode DBTxnMode) DBOption

type DBScanner 

type DBScanner interface {
	base.IHBHandle
	RegisterOp(ScannerOp)
}

type DBTxnMode 

type DBTxnMode uint32
const (
	DBTxnMode_Write DBTxnMode = iota
	DBTxnMode_Replay
)

func (DBTxnMode) IsReplayMode 

func (m DBTxnMode) IsReplayMode() bool

func (DBTxnMode) IsValid 

func (m DBTxnMode) IsValid() bool

func (DBTxnMode) IsWriteMode 

func (m DBTxnMode) IsWriteMode() bool

func (DBTxnMode) String 

func (m DBTxnMode) String() string

type ErrHandler 

type ErrHandler interface {
	OnObjectErr(entry *catalog.ObjectEntry, err error) error
	OnTombstoneErr(entry *catalog.ObjectEntry, err error) error
	OnTableErr(entry *catalog.TableEntry, err error) error
	OnDatabaseErr(entry *catalog.DBEntry, err error) error
}

type NoopErrHandler 

type NoopErrHandler struct{}

func (*NoopErrHandler) OnDatabaseErr 

func (h *NoopErrHandler) OnDatabaseErr(entry *catalog.DBEntry, err error) error

func (*NoopErrHandler) OnObjectErr added in v1.1.0 

func (h *NoopErrHandler) OnObjectErr(entry *catalog.ObjectEntry, err error) error

func (*NoopErrHandler) OnTableErr 

func (h *NoopErrHandler) OnTableErr(entry *catalog.TableEntry, err error) error

func (*NoopErrHandler) OnTombstoneErr 

func (h *NoopErrHandler) OnTombstoneErr(entry *catalog.ObjectEntry, err error) error

type ScannerOp 

type ScannerOp interface {
	catalog.Processor
	PreExecute() error
	PostExecute() error
}

type ScheduledTxnTask 

type ScheduledTxnTask struct {
	*tasks.BaseTask
	// contains filtered or unexported fields
}

func NewScheduledTxnTask 

func NewScheduledTxnTask(ctx *tasks.Context, db *DB, taskType tasks.TaskType, scopes []common.ID, factory tasks.TxnTaskFactory) (task *ScheduledTxnTask)

func (*ScheduledTxnTask) Execute 

func (task *ScheduledTxnTask) Execute(ctx context.Context) (err error)

func (*ScheduledTxnTask) Scope 

func (task *ScheduledTxnTask) Scope() *common.ID

func (*ScheduledTxnTask) Scopes 

func (task *ScheduledTxnTask) Scopes() []common.ID

type WalReplayer 

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

func (*WalReplayer) GetMaxTS 

func (replayer *WalReplayer) GetMaxTS() types.TS

func (*WalReplayer) MakeReplayHandle 

func (replayer *WalReplayer) MakeReplayHandle(
	sender chan<- *txnbase.TxnCmd,
) wal.ApplyHandle

func (*WalReplayer) MaxLSN 

func (replayer *WalReplayer) MaxLSN() uint64

func (*WalReplayer) OnReplayTxn 

func (replayer *WalReplayer) OnReplayTxn(cmd txnif.TxnCmd, lsn uint64)

func (*WalReplayer) OnTimeStamp 

func (replayer *WalReplayer) OnTimeStamp(ts types.TS)

func (*WalReplayer) PreReplayWal 

func (replayer *WalReplayer) PreReplayWal()

func (*WalReplayer) Schedule 

func (replayer *WalReplayer) Schedule(
	ctx context.Context,
	mode driver.ReplayMode,
	onDone func(),
) (
	ctl *replayCtl,
	err error,
)

Source Files

View all Source files

Directories

Path Synopsis
gc
v3
test
compatibility

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.