splitter

Documentation

Index

• Constants
• func GetSplitFields(table *model.TableInfo, splitFields []string) ([]*model.ColumnInfo, error)
• type BucketIterator
  • func NewBucketIterator(ctx context.Context, progressID string, table *common.TableDiff, ...) (*BucketIterator, error)
  • func NewBucketIteratorWithCheckpoint(ctx context.Context, progressID string, table *common.TableDiff, ...) (*BucketIterator, error)
  • func (s *BucketIterator) Close()
  • func (s *BucketIterator) GetIndexID() int64
  • func (s *BucketIterator) Next() (*chunk.Range, error)
• type ChunkIterator
• type LimitIterator
  • func NewLimitIterator(ctx context.Context, progressID string, table *common.TableDiff, ...) (*LimitIterator, error)
  • func NewLimitIteratorWithCheckpoint(ctx context.Context, progressID string, table *common.TableDiff, ...) (*LimitIterator, error)
  • func (lmt *LimitIterator) Close()
  • func (lmt *LimitIterator) GetIndexID() int64
  • func (lmt *LimitIterator) Next() (*chunk.Range, error)
• type RandomIterator
  • func NewRandomIterator(ctx context.Context, progressID string, table *common.TableDiff, ...) (*RandomIterator, error)
  • func NewRandomIteratorWithCheckpoint(ctx context.Context, progressID string, table *common.TableDiff, ...) (*RandomIterator, error)
  • func (s *RandomIterator) Close()
  • func (s *RandomIterator) Next() (*chunk.Range, error)
• type RangeInfo
  • func FromNode(n *checkpoints.Node) *RangeInfo
  • func (r *RangeInfo) Copy() *RangeInfo
  • func (r *RangeInfo) GetBucketIndexLeft() int
  • func (r *RangeInfo) GetBucketIndexRight() int
  • func (r *RangeInfo) GetChunk() *chunk.Range
  • func (r *RangeInfo) GetChunkIndex() int
  • func (r *RangeInfo) GetTableIndex() int
  • func (r *RangeInfo) ToNode() *checkpoints.Node
  • func (r *RangeInfo) Update(column, lower, upper string, updateLower, updateUpper bool, ...)

Constants

const DefaultChannelBuffer = 1024

const (
	SplitThreshold = 1000
)

Functions

func GetSplitFields

func GetSplitFields(table *model.TableInfo, splitFields []string) ([]*model.ColumnInfo, error)

GetSplitFields returns fields to split chunks, order by pk, uk, index, columns.

Types

type BucketIterator

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

func NewBucketIterator

func NewBucketIterator(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB) (*BucketIterator, error)

func NewBucketIteratorWithCheckpoint

func NewBucketIteratorWithCheckpoint(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB, startRange *RangeInfo, checkThreadCount int) (*BucketIterator, error)

func (*BucketIterator) Close

func (s *BucketIterator) Close()

func (*BucketIterator) GetIndexID

func (s *BucketIterator) GetIndexID() int64

func (*BucketIterator) Next

func (s *BucketIterator) Next() (*chunk.Range, error)

type ChunkIterator

type ChunkIterator interface {
	// Next seeks the next chunk, return nil if seeks to end.
	Next() (*chunk.Range, error)
	Close()
}

ChunkIterator generate next chunk for only one table lazily.

type LimitIterator

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

func NewLimitIterator

func NewLimitIterator(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB) (*LimitIterator, error)

func NewLimitIteratorWithCheckpoint

func NewLimitIteratorWithCheckpoint(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB, startRange *RangeInfo) (*LimitIterator, error)

func (*LimitIterator) Close

func (lmt *LimitIterator) Close()

func (*LimitIterator) GetIndexID

func (lmt *LimitIterator) GetIndexID() int64

func (*LimitIterator) Next

func (lmt *LimitIterator) Next() (*chunk.Range, error)

type RandomIterator

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

func NewRandomIterator

func NewRandomIterator(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB) (*RandomIterator, error)

func NewRandomIteratorWithCheckpoint

func NewRandomIteratorWithCheckpoint(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB, startRange *RangeInfo) (*RandomIterator, error)

func (*RandomIterator) Close

func (s *RandomIterator) Close()

func (*RandomIterator) Next

func (s *RandomIterator) Next() (*chunk.Range, error)

type RangeInfo

type RangeInfo struct {
	ChunkRange *chunk.Range `json:"chunk-range"`
	// for bucket checkpoint
	IndexID int64 `json:"index-id"`

	ProgressID string `json:"progress-id"`
}

RangeInfo represents the unit of a process chunk. It's the only entrance of checkpoint.

func FromNode

func FromNode(n *checkpoints.Node) *RangeInfo

func (*RangeInfo) Copy

func (r *RangeInfo) Copy() *RangeInfo

func (*RangeInfo) GetBucketIndexLeft

func (r *RangeInfo) GetBucketIndexLeft() int

func (*RangeInfo) GetBucketIndexRight

func (r *RangeInfo) GetBucketIndexRight() int

func (*RangeInfo) GetChunk

func (r *RangeInfo) GetChunk() *chunk.Range

func (*RangeInfo) GetChunkIndex

func (r *RangeInfo) GetChunkIndex() int

func (*RangeInfo) GetTableIndex

func (r *RangeInfo) GetTableIndex() int

GetTableIndex return the index of table diffs. IMPORTANT!!! We need to keep the tables order during checkpoint. So we should have to save the config info to checkpoint file too

func (*RangeInfo) ToNode

func (r *RangeInfo) ToNode() *checkpoints.Node

func (*RangeInfo) Update

func (r *RangeInfo) Update(column, lower, upper string, updateLower, updateUpper bool, collation, limits string)