cophandler

Documentation

Index

• Constants
• Variables
• func ExecutorListsToTree(exec []*tipb.Executor) *tipb.Executor
• func HandleCopRequest(dbReader *dbreader.DBReader, lockStore *lockstore.MemStore, ...) *coprocessor.Response
• func HandleCopRequestWithMPPCtx(dbReader *dbreader.DBReader, lockStore *lockstore.MemStore, ...) *coprocessor.Response
• func HandleMPPDAGReq(dbReader *dbreader.DBReader, req *coprocessor.Request, mppCtx *MPPCtx) *coprocessor.Response
• type ErrLocked
  • func BuildLockErr(key []byte, primaryKey []byte, startTS uint64, TTL uint64, lockType uint8) *ErrLocked
  • func (e *ErrLocked) Error() string
• type ExchangerTunnel
  • func (tunnel *ExchangerTunnel) RecvChunk() (tipbChunk *tipb.Chunk, err error)
• type MPPCtx
• type MPPTaskHandler
  • func (h *MPPTaskHandler) HandleEstablishConn(_ context.Context, req *mpp.EstablishMPPConnectionRequest) (*ExchangerTunnel, error)

Constants

const (
	// TableScan means reading from a table by table scan
	TableScan scanType = iota
	// IndexScan means reading from a table by index scan
	IndexScan
)

const (
	// MPPErrTunnelNotFound means you can't find an expected tunnel.
	MPPErrTunnelNotFound = iota
	// MPPErrEstablishConnMultiTimes means we receive the Establish requests at least twice.
	MPPErrEstablishConnMultiTimes
	// MPPErrMPPGatherIDMismatch means we get mismatched gather id, usually a bug in MPP coordinator
	MPPErrMPPGatherIDMismatch
)

const (
	// ErrExecutorNotSupportedMsg is the message for executor not supported.
	ErrExecutorNotSupportedMsg = "executor not supported: "
)

Variables

var (
	// DefaultBatchSize is the default batch size for newly allocated chunk during execution.
	DefaultBatchSize = 32
)

Functions

func ExecutorListsToTree

func ExecutorListsToTree(exec []*tipb.Executor) *tipb.Executor

ExecutorListsToTree converts a list of executors to a tree.

func HandleCopRequest

func HandleCopRequest(dbReader *dbreader.DBReader, lockStore *lockstore.MemStore, req *coprocessor.Request) *coprocessor.Response

HandleCopRequest handles coprocessor request.

func HandleCopRequestWithMPPCtx

func HandleCopRequestWithMPPCtx(dbReader *dbreader.DBReader, lockStore *lockstore.MemStore, req *coprocessor.Request, mppCtx *MPPCtx) *coprocessor.Response

HandleCopRequestWithMPPCtx handles coprocessor request, actually, this is the updated version for HandleCopRequest(after mpp test is supported), however, go does not support function overloading, I have to rename it to HandleCopRequestWithMPPCtx.

func HandleMPPDAGReq

func HandleMPPDAGReq(dbReader *dbreader.DBReader, req *coprocessor.Request, mppCtx *MPPCtx) *coprocessor.Response

HandleMPPDAGReq handles a cop request that is converted from mpp request. It returns nothing. Real data will return by stream rpc.

Types

type ErrLocked

type ErrLocked struct {
	Key      []byte
	Primary  []byte
	StartTS  uint64
	TTL      uint64
	LockType uint8
}

ErrLocked is returned when trying to Read/Write on a locked key. Client should backoff or cleanup the lock then retry.

func BuildLockErr

func BuildLockErr(key []byte, primaryKey []byte, startTS uint64, TTL uint64, lockType uint8) *ErrLocked

BuildLockErr generates ErrKeyLocked objects

func (*ErrLocked) Error

func (e *ErrLocked) Error() string

Error formats the lock to a string.

type ExchangerTunnel

type ExchangerTunnel struct {
	DataCh chan *tipb.Chunk

	ErrCh chan error
	// contains filtered or unexported fields
}

ExchangerTunnel contains a channel that can transfer data. Only One Sender and Receiver use this channel, so it's safe to close it by sender.

func (*ExchangerTunnel) RecvChunk

func (tunnel *ExchangerTunnel) RecvChunk() (tipbChunk *tipb.Chunk, err error)

RecvChunk receive tipb chunk

type MPPCtx

type MPPCtx struct {
	RPCClient   client.Client
	StoreAddr   string
	TaskHandler *MPPTaskHandler
	Ctx         context.Context
}

MPPCtx is the mpp execution context

type MPPTaskHandler

type MPPTaskHandler struct {
	TunnelSet map[int64]*ExchangerTunnel

	Meta      *mpp.TaskMeta
	RPCClient client.Client

	Status atomic.Int32
	Err    error
	// contains filtered or unexported fields
}

MPPTaskHandler exists in a single store.

func (*MPPTaskHandler) HandleEstablishConn

func (h *MPPTaskHandler) HandleEstablishConn(_ context.Context, req *mpp.EstablishMPPConnectionRequest) (*ExchangerTunnel, error)

HandleEstablishConn handles EstablishMPPConnectionRequest