sorter

Documentation

Index

• func BlockCacheAccess() *prometheus.GaugeVec
• func CompactionDuration() *prometheus.HistogramVec
• func InMemoryDataSize() *prometheus.GaugeVec
• func InitMetrics(registry *prometheus.Registry)
• func IterReadDuration() *prometheus.HistogramVec
• func IteratorGauge() *prometheus.GaugeVec
• func LevelCount() *prometheus.GaugeVec
• func OnDiskDataSize() *prometheus.GaugeVec
• func RangeCleanCount() prometheus.Counter
• func WriteBytes() *prometheus.HistogramVec
• func WriteDelayCount() *prometheus.GaugeVec
• func WriteDuration() *prometheus.HistogramVec
• type EventIterator
• type MountedEventIter
  • func NewMountedEventIter(changefeedID model.ChangeFeedID, iter EventIterator, mg entry.MounterGroup, ...) *MountedEventIter
  • func (i *MountedEventIter) Close() error
  • func (i *MountedEventIter) Next(ctx context.Context) (event *model.PolymorphicEvent, txnFinished Position, err error)
• type Position
  • func GenCommitFence(commitTs model.Ts) Position
  • func (p Position) Compare(q Position) int
  • func (p Position) IsCommitFence() bool
  • func (p Position) Next() Position
  • func (p Position) Prev() Position
  • func (p Position) Valid() bool
• type SortEngine
• type TableStats

Functions

func BlockCacheAccess

func BlockCacheAccess() *prometheus.GaugeVec

BlockCacheAccess returns dbBlockCacheAccess.

func CompactionDuration

func CompactionDuration() *prometheus.HistogramVec

CompactionDuration returns sorterCompactDurationHistogram.

func InMemoryDataSize

func InMemoryDataSize() *prometheus.GaugeVec

InMemoryDataSize returns inMemoryDataSizeGauge.

func InitMetrics

func InitMetrics(registry *prometheus.Registry)

InitMetrics registers all metrics in this file

func IterReadDuration

func IterReadDuration() *prometheus.HistogramVec

IterReadDuration returns sorterIterReadDurationHistogram.

func IteratorGauge

func IteratorGauge() *prometheus.GaugeVec

IteratorGauge returns dbIteratorGauge.

func LevelCount

func LevelCount() *prometheus.GaugeVec

LevelCount returns dbLevelCount.

func OnDiskDataSize

func OnDiskDataSize() *prometheus.GaugeVec

OnDiskDataSize returns onDiskDataSizeGauge.

func RangeCleanCount

func RangeCleanCount() prometheus.Counter

RangeCleanCount returns dbRangeCleanCount.

func WriteBytes

func WriteBytes() *prometheus.HistogramVec

WriteBytes returns sorterWriteBytesHistogram.

func WriteDelayCount

func WriteDelayCount() *prometheus.GaugeVec

WriteDelayCount returns dbWriteDelayCount.

func WriteDuration

func WriteDuration() *prometheus.HistogramVec

WriteDuration returns sorterWriteDurationHistogram.

Types

type EventIterator

type EventIterator interface {
	// Next is used to fetch one event. nil indicates it reaches the stop point.
	//
	// txnFinished indicates whether all events in the current transaction are
	// fetched or not. Users should keep fetching events until txnFinished.Valid()
	// returns true.
	//
	// NOTE: event.IsResolved() will always be false.
	Next() (event *model.PolymorphicEvent, txnFinished Position, err error)

	// Close closes the iterator.
	Close() error
}

EventIterator is an iterator to fetch events from SortEngine. It's unnecessary to be thread-safe.

type MountedEventIter

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

MountedEventIter is just like EventIterator, but returns mounted events.

func NewMountedEventIter

func NewMountedEventIter(
	changefeedID model.ChangeFeedID,
	iter EventIterator,
	mg entry.MounterGroup,
	maxBatchSize int,
	quota *memquota.MemQuota,
) *MountedEventIter

NewMountedEventIter creates a MountedEventIter instance.

func (*MountedEventIter) Close

func (i *MountedEventIter) Close() error

Close implements sorter.EventIterator.

func (*MountedEventIter) Next

func (i *MountedEventIter) Next(ctx context.Context) (event *model.PolymorphicEvent, txnFinished Position, err error)

Next returns the next mounted event.

type Position

type Position struct {
	StartTs  model.Ts
	CommitTs model.Ts
}

Position is used to

1. fetch or clear events from an engine, for example, see SortEngine.FetchByTable.
2. calculate the next position with method Next.

func GenCommitFence

func GenCommitFence(commitTs model.Ts) Position

GenCommitFence generates a Position which is a commit fence. CommitFence indicates all transactions with same CommitTs are less than the position.

func (Position) Compare

func (p Position) Compare(q Position) int

Compare compares 2 Position, just like strcmp in C.

func (Position) IsCommitFence

func (p Position) IsCommitFence() bool

IsCommitFence indicates all transactions with same CommitTs are less than the position.

func (Position) Next

func (p Position) Next() Position

Next can only be called on a valid Position.

func (Position) Prev

func (p Position) Prev() Position

Prev can only be called on a valid Position.

func (Position) Valid

func (p Position) Valid() bool

Valid indicates whether the position is valid or not.

type SortEngine

type SortEngine interface {
	// IsTableBased tells whether the sort engine is based on table or not.
	// If it's based on table, fetching events by table is preferred.
	IsTableBased() bool

	// AddTable adds the table into the engine.
	AddTable(span tablepb.Span, startTs model.Ts)

	// RemoveTable removes the table from the engine.
	RemoveTable(span tablepb.Span)

	// Add adds the given events into the sort engine.
	//
	// NOTE: it's an asynchronous interface. To get the notification of when
	// events are available for fetching, OnResolve is what you want.
	Add(span tablepb.Span, events ...*model.PolymorphicEvent)

	// OnResolve pushes action into SortEngine's hook list, which
	// will be called after any events are resolved.
	OnResolve(action func(tablepb.Span, model.Ts))

	// FetchByTable creates an iterator to fetch events from the given table.
	// lowerBound is inclusive and only resolved events can be retrieved.
	//
	// NOTE: FetchByTable is always available even if IsTableBased returns false.
	FetchByTable(span tablepb.Span, lowerBound, upperBound Position) EventIterator

	// FetchAllTables creates an iterator to fetch events from all tables.
	// lowerBound is inclusive and only resolved events can be retrieved.
	//
	// NOTE: It's only available if IsTableBased returns false.
	FetchAllTables(lowerBound Position) EventIterator

	// CleanByTable tells the engine events of the given table in the given range
	// (unlimited, upperBound] are committed and not necessary any more.
	// The SortEngine instance can GC them later.
	//
	// NOTE: CleanByTable is always available even if IsTableBased returns false.
	CleanByTable(span tablepb.Span, upperBound Position) error

	// CleanAllTables tells the engine events of all tables in the given range
	// (unlimited, upperBound] are committed and not necessary any more.
	// The SortEngine instance can GC them later.
	//
	// NOTE: It's only available if IsTableBased returns false.
	CleanAllTables(upperBound Position) error

	// GetStatsByTable gets the statistics of the given table.
	GetStatsByTable(span tablepb.Span) TableStats

	// Close closes the engine. All data written by this instance can be deleted.
	//
	// NOTE: it leads an undefined behavior to close an engine with active iterators.
	Close() error

	// SlotsAndHasher returns how many slots contained by the Engine, and
	// a hasher for table spans.
	// The hasher should return a slot index for the given table span.
	SlotsAndHasher() (slotCount int, hasher func(tablepb.Span, int) int)
}

SortEngine is a storage engine to store and sort CDC events. Every changefeed will have one SortEngine instance. NOTE: All interfaces are thread-safe.

type TableStats

type TableStats struct {
	ReceivedMaxCommitTs   model.Ts
	ReceivedMaxResolvedTs model.Ts
}

TableStats of a sort engine.