local

Documentation

Index

• func NewLocalBackend(ctx context.Context, tls *common.TLS, cfg *config.Config, g glue.Glue, ...) (backend.Backend, error)
• func VerifyRLimit(estimateMaxFiles Rlim_t) error
• type DupKVStream
• type DuplicateManager
  • func NewDuplicateManager(tbl table.Table, tableName string, splitCli restore.SplitClient, ...) (*DuplicateManager, error)
  • func (m *DuplicateManager) CollectDuplicateRowsFromDupDB(ctx context.Context, dupDB *pebble.DB, keyAdapter KeyAdapter) error
  • func (m *DuplicateManager) CollectDuplicateRowsFromTiKV(ctx context.Context, importClientFactory ImportClientFactory) error
  • func (m *DuplicateManager) RecordDataConflictError(ctx context.Context, stream DupKVStream) error
  • func (m *DuplicateManager) RecordIndexConflictError(ctx context.Context, stream DupKVStream, tableID int64, ...) error
• type Engine
  • func (e *Engine) Cleanup(dataDir string) error
  • func (e *Engine) Close() error
  • func (e *Engine) Exist(dataDir string) error
• type ImportClientFactory
• type KeyAdapter
• type LocalDupKVStream
  • func NewLocalDupKVStream(dupDB *pebble.DB, keyAdapter KeyAdapter, keyRange tidbkv.KeyRange) *LocalDupKVStream
  • func (s *LocalDupKVStream) Close() error
  • func (s *LocalDupKVStream) Next() (key, val []byte, err error)
• type Range
• type RangePropertiesCollector
  • func (c *RangePropertiesCollector) Add(key pebble.InternalKey, value []byte) error
  • func (c *RangePropertiesCollector) Finish(userProps map[string]string) error
  • func (c *RangePropertiesCollector) Name() string
• type RemoteDupKVStream
  • func NewRemoteDupKVStream(ctx context.Context, region *restore.RegionInfo, keyRange tidbkv.KeyRange, ...) (*RemoteDupKVStream, error)
  • func (s *RemoteDupKVStream) Close() error
  • func (s *RemoteDupKVStream) Next() (key, val []byte, err error)
• type Rlim_t
  • func GetSystemRLimit() (Rlim_t, error)
• type Writer
  • func (w *Writer) AppendRows(ctx context.Context, tableName string, columnNames []string, rows kv.Rows) error
  • func (w *Writer) Close(ctx context.Context) (backend.ChunkFlushStatus, error)
  • func (w *Writer) IsSynced() bool

Functions

func NewLocalBackend

func NewLocalBackend(
	ctx context.Context,
	tls *common.TLS,
	cfg *config.Config,
	g glue.Glue,
	maxOpenFiles int,
	errorMgr *errormanager.ErrorManager,
) (backend.Backend, error)

NewLocalBackend creates new connections to tikv.

func VerifyRLimit

func VerifyRLimit(estimateMaxFiles Rlim_t) error

VerifyRLimit checks whether the open-file limit is large enough. In Local-backend, we need to read and write a lot of L0 SST files, so we need to check system max open files limit.

Types

type DupKVStream

type DupKVStream interface {
	// Next returns the next key-value pair or any error it encountered.
	// At the end of the stream, the error is io.EOF.
	Next() (key, val []byte, err error)
	// Close closes the stream.
	Close() error
}

DupKVStream is a streaming interface for collecting duplicate key-value pairs.

type DuplicateManager

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

DuplicateManager provides methods to collect and decode duplicated KV pairs into row data. The results are stored into the errorMgr.

func NewDuplicateManager

func NewDuplicateManager(
	tbl table.Table,
	tableName string,
	splitCli restore.SplitClient,
	tikvCli *tikv.KVStore,
	errMgr *errormanager.ErrorManager,
	sessOpts *kv.SessionOptions,
	concurrency int,
	hasDupe *atomic.Bool,
) (*DuplicateManager, error)

NewDuplicateManager creates a new DuplicateManager.

func (*DuplicateManager) CollectDuplicateRowsFromDupDB

func (m *DuplicateManager) CollectDuplicateRowsFromDupDB(ctx context.Context, dupDB *pebble.DB, keyAdapter KeyAdapter) error

CollectDuplicateRowsFromDupDB collects duplicates from the duplicate DB and records all duplicate row info into errorMgr.

func (*DuplicateManager) CollectDuplicateRowsFromTiKV

func (m *DuplicateManager) CollectDuplicateRowsFromTiKV(ctx context.Context, importClientFactory ImportClientFactory) error

CollectDuplicateRowsFromTiKV collects duplicates from the remote TiKV and records all duplicate row info into errorMgr.

func (*DuplicateManager) RecordDataConflictError

func (m *DuplicateManager) RecordDataConflictError(ctx context.Context, stream DupKVStream) error

RecordDataConflictError records data conflicts to errorMgr. The key received from stream must be a row key.

func (*DuplicateManager) RecordIndexConflictError

func (m *DuplicateManager) RecordIndexConflictError(ctx context.Context, stream DupKVStream, tableID int64, indexInfo *model.IndexInfo) error

RecordIndexConflictError records index conflicts to errorMgr. The key received from stream must be an index key.

type Engine

type Engine struct {
	UUID uuid.UUID
	// contains filtered or unexported fields
}

func (*Engine) Cleanup

func (e *Engine) Cleanup(dataDir string) error

Cleanup remove meta and db files

func (*Engine) Close

func (e *Engine) Close() error

func (*Engine) Exist

func (e *Engine) Exist(dataDir string) error

Exist checks if db folder existing (meta sometimes won't flush before lightning exit)

type ImportClientFactory

type ImportClientFactory interface {
	Create(ctx context.Context, storeID uint64) (sst.ImportSSTClient, error)
	Close()
}

ImportClientFactory is factory to create new import client for specific store.

type KeyAdapter

type KeyAdapter interface {
	// Encode encodes the key with its corresponding rowID. It appends the encoded key to dst and returns the
	// resulting slice. The encoded key is guaranteed to be in ascending order for comparison.
	Encode(dst []byte, key []byte, rowID int64) []byte

	// Decode decodes the original key to dst. It appends the encoded key to dst and returns the resulting slice.
	Decode(dst []byte, data []byte) ([]byte, error)

	// EncodedLen returns the encoded key length.
	EncodedLen(key []byte) int
}

KeyAdapter is used to encode and decode keys.

type LocalDupKVStream

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

LocalDupKVStream implements the interface of DupKVStream. It collects duplicate key-value pairs from a pebble.DB.

func NewLocalDupKVStream

func NewLocalDupKVStream(dupDB *pebble.DB, keyAdapter KeyAdapter, keyRange tidbkv.KeyRange) *LocalDupKVStream

NewLocalDupKVStream creates a new LocalDupKVStream with the given duplicate db and key range.

func (*LocalDupKVStream) Close

func (s *LocalDupKVStream) Close() error

func (*LocalDupKVStream) Next

func (s *LocalDupKVStream) Next() (key, val []byte, err error)

type Range

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

Range record start and end key for localStoreDir.DB so we can write it to tikv in streaming

type RangePropertiesCollector

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

func (*RangePropertiesCollector) Add

func (c *RangePropertiesCollector) Add(key pebble.InternalKey, value []byte) error

Add implements `pebble.TablePropertyCollector`. Add implements `TablePropertyCollector.Add`.

func (*RangePropertiesCollector) Finish

func (c *RangePropertiesCollector) Finish(userProps map[string]string) error

func (*RangePropertiesCollector) Name

func (c *RangePropertiesCollector) Name() string

type RemoteDupKVStream

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

RemoteDupKVStream implements the interface of DupKVStream. It collects duplicate key-value pairs from a TiKV region.

func NewRemoteDupKVStream

func NewRemoteDupKVStream(
	ctx context.Context,
	region *restore.RegionInfo,
	keyRange tidbkv.KeyRange,
	importClientFactory ImportClientFactory,
) (*RemoteDupKVStream, error)

NewRemoteDupKVStream creates a new RemoteDupKVStream.

func (*RemoteDupKVStream) Close

func (s *RemoteDupKVStream) Close() error

func (*RemoteDupKVStream) Next

func (s *RemoteDupKVStream) Next() (key, val []byte, err error)

type Rlim_t

type Rlim_t = uint64

func GetSystemRLimit

func GetSystemRLimit() (Rlim_t, error)

type Writer

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

func (*Writer) AppendRows

func (w *Writer) AppendRows(ctx context.Context, tableName string, columnNames []string, rows kv.Rows) error

func (*Writer) Close

func (w *Writer) Close(ctx context.Context) (backend.ChunkFlushStatus, error)

func (*Writer) IsSynced

func (w *Writer) IsSynced() bool