process

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: 48 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ExecStop = iota
	ExecNext
	ExecHasMore
)

Variables

View Source 
var (
	// PipelineCleanupWaitTimeout bounds cleanup-only waits for terminal
	// pipeline signals. Normal query execution does not use this timeout.
	PipelineCleanupWaitTimeout = 30 * time.Second

	// PipelineSignalSendTimeout bounds cleanup-only terminal signal sends.
	PipelineSignalSendTimeout = 30 * time.Second

	// ErrPipelineEndSignalDeliveryFailed marks a successful cleanup path that
	// could not enqueue its normal End signal and had to fall back to abort.
	ErrPipelineEndSignalDeliveryFailed = moerr.NewInternalErrorNoCtx("pipeline end signal delivery failed")

	// ErrPipelineTerminalWithoutCause marks a failure/abort terminal event whose
	// caller did not provide the original cause. Error and Abort terminal events
	// must always carry a non-nil error so receivers cannot confuse failure with
	// graceful End.
	ErrPipelineTerminalWithoutCause = moerr.NewInternalErrorNoCtx("pipeline terminal event missing cause")
)
View Source 
var (
	NormalEndRegisterMessage = NewRegMsg(nil)
)

Functions

func EnumLogLevel2ZapLogLevel 

func EnumLogLevel2ZapLogLevel(level pipeline.SessionLoggerInfo_LogLevel) zapcore.Level

func GetQueryCtxFromProc 

func GetQueryCtxFromProc(proc *Process) (context.Context, context.CancelFunc)

GetQueryCtxFromProc returns the query context and its cancel function. just for easy access.

func MockProcessInfoWithPro 

func MockProcessInfoWithPro(
	sql string,
	pro any,
) (pipeline.ProcessInfo, error)

func NormalizePipelineCleanupError 

func NormalizePipelineCleanupError(pipelineFailed bool, err error) error

func ReplacePipelineCtx 

func ReplacePipelineCtx(proc *Process, ctx context.Context, cancel context.CancelCauseFunc)

ReplacePipelineCtx replaces the pipeline context and cancel function for the process. It's a very dangerous operation, should be used with caution. And we only use it for the newly built pipeline by the pipeline's ParallelRun method.

func SendPipelineSignalWithContext 

func SendPipelineSignalWithContext(ctx context.Context, reg *WaitRegister, signal PipelineSignal) bool

func SendPipelineSignalWithTimeout 

func SendPipelineSignalWithTimeout(reg *WaitRegister, signal PipelineSignal, timeout time.Duration) bool

func TrySendPipelineSignal 

func TrySendPipelineSignal(reg *WaitRegister, signal PipelineSignal) bool

func WaitRegisterChannelState 

func WaitRegisterChannelState(reg *WaitRegister) (int, int)

func WarnPipelineCleanupf 

func WarnPipelineCleanupf(proc *Process, key string, format string, args ...any)

Types

type Analyzer 

type Analyzer interface {
	Start()
	Stop()
	ChildrenCallStop(time time.Time)
	Alloc(int64)
	SetMemUsed(int64)
	Spill(int64)
	SpillRows(int64)
	Input(batch *batch.Batch)
	Output(*batch.Batch)
	WaitStop(time.Time)
	Network(*batch.Batch)
	AddWrittenRows(int64)
	AddDeletedRows(int64)
	AddScanTime(t time.Time)
	AddInsertTime(t time.Time)
	AddIncrementTime(t time.Time)
	AddWaitLockTime(t time.Time)
	AddS3RequestCount(counter *perfcounter.CounterSet)
	AddFileServiceCacheInfo(counter *perfcounter.CounterSet)
	AddDiskIO(counter *perfcounter.CounterSet)
	AddReadSizeInfo(counter *perfcounter.CounterSet)
	AddParquetProfile(stats ParquetProfileStats)

	GetOpCounterSet() *perfcounter.CounterSet
	GetOpStats() *OperatorStats

	InputBlock()
	ScanBytes(*batch.Batch)

	Reset()
}

Analyze analyzes information for operator

func NewAnalyzer 

func NewAnalyzer(idx int, isFirst bool, isLast bool, operatorName string) Analyzer

NewAnalyzer is used to provide resource statistics services for physical plann operators

func NewTempAnalyzer 

func NewTempAnalyzer() Analyzer

NewTempAnalyzer is used to provide resource statistics services for non operator logic

type BaseProcess 

type BaseProcess struct {
	LoadTag bool

	StmtProfile *StmtProfile
	// Id, query id.
	Id  string
	Lim Limitation

	// unix timestamp
	UnixTime         int64
	TxnClient        client.TxnClient
	SessionInfo      SessionInfo
	FileService      fileservice.FileService
	LockService      lockservice.LockService
	TaskService      taskservice.TaskService
	PartitionService partitionservice.PartitionService
	IncrService      incrservice.AutoIncrementService

	LastInsertID    *uint64
	LoadLocalReader *io.PipeReader
	Aicm            *defines.AutoIncrCacheManager

	QueryClient qclient.QueryClient
	Hakeeper    logservice.CNHAKeeperClient
	UdfService  udf.Service
	WaitPolicy  lock.WaitPolicy

	TxnOperator      client.TxnOperator
	CloneTxnOperator client.TxnOperator

	// post dml sqls run right after all pipelines finished.
	PostDmlSqlList *threadsafe.Slice[string]

	// stage cache to avoid to run same stage SQL repeatedly
	StageCache *threadsafe.Map[string, stage.StageDef]

	// DivByZeroErrorMode caches whether division by zero should error (true) or return NULL (false)
	// -1: not initialized, 0: return NULL, 1: return error
	DivByZeroErrorMode int32

	// IsFrontend reports whether this proc is attached to a frontend
	// client session (mysql client query or the in-frontend backSession
	// that pkg/frontend/back_exec.go drives). Defaults false — every
	// other proc (internal SQL executor invocations from idxcron,
	// ProcessInitSQL, bootstrap, cron jobs, task service, …) is
	// background. pkg/sql/compile/sql_executor.go's NewTopProcess sets
	// this from opts.IsFrontend(); the two frontend proc-construction
	// sites in pkg/frontend (mysql_cmd_executor, back_exec) set it
	// directly. This is the canonical signal for code that needs to
	// distinguish "have a session" from "don't" — relying on
	// proc.resolveVariableFunc being nil is unreliable because
	// background paths also attach resolvers (idxcron via the task's
	// captured Metadata, ProcessInitSQL via executor.DefaultResolveVariable).
	IsFrontend bool
	// contains filtered or unexported fields
}

func (*BaseProcess) GetContextBase 

func (bp *BaseProcess) GetContextBase() *QueryBaseContext

type ExecStatus added in v1.0.0 

type ExecStatus int

type Limitation 

type Limitation struct {
	// Size, memory threshold for operator.
	Size int64
	// BatchRows, max rows for batch.
	BatchRows int64
	// BatchSize, max size for batch.
	BatchSize int64
	// PartitionRows, max rows for partition.
	PartitionRows int64
	// ReaderSize, memory threshold for storage's reader
	ReaderSize int64
	// MaxMessageSize max size for read messages from dn
	MaxMsgSize uint64
}

Limitation specifies the maximum resources that can be used in one query.

func ConvertToProcessLimitation 

func ConvertToProcessLimitation(
	lim pipeline.ProcessLimitation,
) Limitation

convert pipeline.ProcessLimitation to process.Limitation

type MetricType 

type MetricType int
const (
	OpScanTime      MetricType = 0
	OpInsertTime    MetricType = 1
	OpIncrementTime MetricType = 2
	OpWaitLockTime  MetricType = 3
)

type OperatorStats 

type OperatorStats struct {
	OperatorName     string `json:"-"`
	CallNum          int    `json:"CallCount,omitempty"`
	TimeConsumed     int64  `json:"TimeConsumed,omitempty"`
	WaitTimeConsumed int64  `json:"WaitTimeConsumed,omitempty"`
	MemorySize       int64  `json:"MemorySize,omitempty"`
	SpillSize        int64  `json:"SpillSize,omitempty"`
	SpillRows        int64  `json:"SpillRows,omitempty"`
	InputRows        int64  `json:"InputRows,omitempty"`
	InputSize        int64  `json:"InputSize,omitempty"`
	OutputRows       int64  `json:"OutputRows,omitempty"`
	OutputSize       int64  `json:"OutputSize,omitempty"`
	NetworkIO        int64  `json:"NetworkIO,omitempty"`
	DiskIO           int64  `json:"DiskIO,omitempty"`

	InputBlocks  int64 `json:"-"`
	ScanBytes    int64 `json:"-"`
	ReadSize     int64 `json:"ReadSize,omitempty"`     // ReadSize: actual bytes read from storage layer (excluding rowid tombstone)
	S3ReadSize   int64 `json:"S3ReadSize,omitempty"`   // S3ReadSize: actual bytes read from S3 (excluding rowid tombstone)
	DiskReadSize int64 `json:"DiskReadSize,omitempty"` // DiskReadSize: actual bytes read from disk cache (excluding rowid tombstone)
	WrittenRows  int64 `json:"WrittenRows,omitempty"`  // WrittenRows Used to estimate S3input
	DeletedRows  int64 `json:"DeletedRows,omitempty"`  // DeletedRows Used to estimate S3input

	S3List      int64 `json:"S3List,omitempty"`
	S3Head      int64 `json:"S3Head,omitempty"`
	S3Put       int64 `json:"S3Put,omitempty"`
	S3Get       int64 `json:"S3Get,omitempty"`
	S3Delete    int64 `json:"S3Delete,omitempty"`
	S3DeleteMul int64 `json:"S3DeleteMul,omitempty"`

	CacheRead       int64 `json:"CacheRead,omitempty"`
	CacheHit        int64 `json:"CacheHit,omitempty"`
	CacheMemoryRead int64 `json:"CacheMemoryRead,omitempty"`
	CacheMemoryHit  int64 `json:"CacheMemoryHit,omitempty"`
	CacheDiskRead   int64 `json:"CacheDiskRead,omitempty"`
	CacheDiskHit    int64 `json:"CacheDiskHit,omitempty"`
	CacheRemoteRead int64 `json:"CacheRemoteRead,omitempty"`
	CacheRemoteHit  int64 `json:"CacheRemoteHit,omitempty"`

	ParquetFiles          int64 `json:"ParquetFiles,omitempty"`
	ParquetRowGroups      int64 `json:"ParquetRowGroups,omitempty"`
	ParquetRowsRead       int64 `json:"ParquetRowsRead,omitempty"`
	ParquetBytesRead      int64 `json:"ParquetBytesRead,omitempty"`
	ParquetPrefetchBytes  int64 `json:"ParquetPrefetchBytes,omitempty"`
	ParquetOpenTime       int64 `json:"ParquetOpenTime,omitempty"`
	ParquetReadPageTime   int64 `json:"ParquetReadPageTime,omitempty"`
	ParquetMapTime        int64 `json:"ParquetMapTime,omitempty"`
	ParquetRowModeTime    int64 `json:"ParquetRowModeTime,omitempty"`
	ParquetPeakBatchBytes int64 `json:"ParquetPeakBatchBytes,omitempty"`

	OperatorMetrics map[MetricType]int64 `json:"OperatorMetrics,omitempty"`

	BackgroundQueries []*plan.Query `json:"BackgroundQueries,omitempty"`
}

func NewOperatorStats 

func NewOperatorStats(operatorName string) *OperatorStats

func (*OperatorStats) AddOpMetric 

func (ps *OperatorStats) AddOpMetric(key MetricType, value int64)

func (*OperatorStats) GetMetricByKey 

func (ps *OperatorStats) GetMetricByKey(metricType MetricType) int64

func (*OperatorStats) Reset 

func (ps *OperatorStats) Reset()

func (*OperatorStats) String 

func (ps *OperatorStats) String() string

type ParquetProfileStats 

type ParquetProfileStats struct {
	Files          int64
	RowGroups      int64
	RowsRead       int64
	BytesRead      int64
	PrefetchBytes  int64
	OpenTime       int64
	ReadPageTime   int64
	MapTime        int64
	RowModeTime    int64
	PeakBatchBytes int64
}

func (*ParquetProfileStats) Add 

func (stats *ParquetProfileStats) Add(delta ParquetProfileStats)

func (ParquetProfileStats) Empty 

func (stats ParquetProfileStats) Empty() bool

type PipelineActionType 

type PipelineActionType uint8
const (
	GetFromIndex PipelineActionType = iota
	GetDirectly
)

type PipelineEdge 

type PipelineEdge struct {
	// Ch2 is the underlying data+terminal signal channel.
	// Exposed for direct select compatibility with PipelineSignalReceiver.
	Ch2 chan PipelineSignal

	// NilBatchCnt is the number of legacy nil-batches or typed End signals this
	// channel must receive before it is considered done. 0 defaults to 1.
	NilBatchCnt int
	// contains filtered or unexported fields
}

PipelineEdge is an explicit pipeline edge abstraction with typed lifecycle events. It owns both the signal channel and the idempotent terminal state.

Invariants:

  1. Terminal state (End/Error/Abort) is a protocol event, not an implicit nil batch.
  2. End is counted per expected sender; Error and Abort are fatal first-wins terminals that consume the edge's remaining sender count.
  3. Done() provides an observable whole-edge terminal signal.
  4. Every send/receive is cancelable via context, or bounded by the edge timeout configuration.

func NewPipelineEdge 

func NewPipelineEdge(channelBufferSize int, nilBatchCnt int) *PipelineEdge

NewPipelineEdge creates a new PipelineEdge. channelBufferSize is the buffer size for Ch2. nilBatchCnt is the NilBatchCnt value (0 defaults to 1, same as WaitRegister).

func NewPipelineEdgeFromReg 

func NewPipelineEdgeFromReg(reg *WaitRegister) *PipelineEdge

NewPipelineEdgeFromReg returns the same edge object behind a WaitRegister name. If reg is nil, a new edge with buffer size 1 is created.

func (*PipelineEdge) Abort 

func (e *PipelineEdge) Abort(err error) bool

Abort marks the edge aborted and tries to enqueue an Abort signal.

func (*PipelineEdge) Aborted 

func (e *PipelineEdge) Aborted() <-chan struct{}

Aborted returns a channel that is closed when the edge is aborted (cancellation, remote failure, etc.). Unlike Done, this does NOT close on normal End.

func (*PipelineEdge) AsWaitRegister 

func (e *PipelineEdge) AsWaitRegister() *WaitRegister

AsWaitRegister returns the same edge object under the historical type name.

func (*PipelineEdge) ChannelState 

func (e *PipelineEdge) ChannelState() (int, int)

ChannelState returns (len, cap) for the underlying channel.

func (*PipelineEdge) Done 

func (e *PipelineEdge) Done() <-chan struct{}

Done returns a channel that is closed once the whole edge is terminal: all expected End signals have been delivered, or the edge receives Error/Abort. It never blocks.

func (*PipelineEdge) Err 

func (e *PipelineEdge) Err() error

Err returns the terminal error, or nil if the edge ended without error.

func (*PipelineEdge) ResetForReuse 

func (e *PipelineEdge) ResetForReuse(channelBufferSize int, nilBatchCnt int)

ResetForReuse reinitializes the edge before it is wired into a newly compiled pipeline. It must not be called while senders or receivers can still access the edge.

func (*PipelineEdge) SendData 

func (e *PipelineEdge) SendData(ctx context.Context, spool *pSpool.PipelineSpool, idx int) bool

SendData sends a data batch via the edge. It returns true if the signal was successfully sent, false if the context was cancelled.

func (*PipelineEdge) SendDataDirect 

func (e *PipelineEdge) SendDataDirect(ctx context.Context, bat *batch.Batch, mp *mpool.MPool) bool

SendDataDirect sends a batch directly (not via spool) through the edge.

func (*PipelineEdge) SendEnd 

func (e *PipelineEdge) SendEnd() bool

SendEnd sends one sender's End signal. Done closes after the edge has delivered the expected number of End signals.

func (*PipelineEdge) SendError 

func (e *PipelineEdge) SendError(err error) bool

SendError marks the edge failed and tries to enqueue an Error signal.

func (*PipelineEdge) SendSignalWithContext 

func (e *PipelineEdge) SendSignalWithContext(ctx context.Context, signal PipelineSignal) bool

SendSignalWithContext sends a signal to the edge's channel with context.

func (*PipelineEdge) SendSignalWithTimeout 

func (e *PipelineEdge) SendSignalWithTimeout(signal PipelineSignal, timeout time.Duration) bool

SendSignalWithTimeout sends a signal to the edge's channel with optional timeout.

func (*PipelineEdge) SetNilBatchCntForReuse 

func (e *PipelineEdge) SetNilBatchCntForReuse(nilBatchCnt int)

SetNilBatchCntForReuse updates the legacy nil-batch count while preserving the channel buffer. It also drains buffered stale signals and resets terminal state. It is compile/reuse-time only and must not race with live senders or receivers.

func (*PipelineEdge) TryAbort 

func (e *PipelineEdge) TryAbort(err error) bool

TryAbort attempts a non-blocking Abort. Returns true if delivered.

func (*PipelineEdge) TrySendEnd 

func (e *PipelineEdge) TrySendEnd() bool

TrySendEnd attempts a non-blocking End send. Returns true if delivered.

func (*PipelineEdge) TrySendError 

func (e *PipelineEdge) TrySendError(err error) bool

TrySendError attempts a non-blocking Error send. Returns true if delivered.

func (*PipelineEdge) TrySendSignal 

func (e *PipelineEdge) TrySendSignal(signal PipelineSignal) bool

TrySendSignal tries a non-blocking send to the edge's channel.

type PipelineEventType 

type PipelineEventType uint8

PipelineEventType is the typed terminal event for pipeline edges. It makes the termination protocol explicit instead of relying on nil-batch conventions and NilBatchCnt counting. 

const (
	// EventData carries a normal data batch. Batch may be nil for empty
	// intermediate results; nil alone is NOT a terminal signal.
	EventData PipelineEventType = iota
	// EventEnd signals graceful pipeline completion with no error.
	EventEnd
	// EventError signals pipeline completion due to an error.
	EventError
	// EventAbort signals forced pipeline teardown (cancellation, remote failure, etc.).
	EventAbort
)

func (PipelineEventType) IsTerminal 

func (t PipelineEventType) IsTerminal() bool

func (PipelineEventType) String 

func (t PipelineEventType) String() string

type PipelineSignal 

type PipelineSignal struct {

	// EventType is set for typed terminal events (EventEnd/EventError/EventAbort).
	// When EventType != EventData, this signal carries a terminal event, not a data batch.
	EventType PipelineEventType
	// contains filtered or unexported fields
}

func BuildCleanupSignal 

func BuildCleanupSignal(pipelineFailed bool, err error) PipelineSignal

BuildCleanupSignal returns the appropriate terminal signal for pipeline cleanup. EventError is used when pipelineFailed=true or err!=nil; EventEnd otherwise.

func NewAbortSignal 

func NewAbortSignal(err error) PipelineSignal

NewAbortSignal returns a PipelineSignal carrying an explicit EventAbort. It is idempotent: receivers must treat multiple Abort signals safely.

func NewEndSignal 

func NewEndSignal() PipelineSignal

NewEndSignal returns a PipelineSignal carrying an explicit EventEnd. It is idempotent: receivers must treat multiple End signals safely.

func NewErrorSignal 

func NewErrorSignal(err error) PipelineSignal

NewErrorSignal returns a PipelineSignal carrying an explicit EventError. It is idempotent: receivers must treat multiple Error signals safely.

func NewPipelineSignalToDirectly 

func NewPipelineSignalToDirectly(data *batch.Batch, err error, mp *mpool.MPool) PipelineSignal

NewPipelineSignalToDirectly return a signal indicates the receiver to get data from signal directly.

func NewPipelineSignalToGetFromSpool 

func NewPipelineSignalToGetFromSpool(source *pSpool.PipelineSpool, index int) PipelineSignal

NewPipelineSignalToGetFromSpool return a signal indicate the receiver to get data from source by index.

func (PipelineSignal) Action 

func (signal PipelineSignal) Action() (data *batch.Batch, info error)

Action will get the input batch from one place according to which type this signal is.

For terminal events (EventEnd/EventError/EventAbort), nil batch is returned. The result batch of this function is an READ-ONLY one.

func (PipelineSignal) IsTerminal 

func (signal PipelineSignal) IsTerminal() bool

IsTerminal returns true for End, Error, or Abort signals.

func (PipelineSignal) TerminalErr 

func (signal PipelineSignal) TerminalErr() error

TerminalErr returns the error carried by EventError or EventAbort.

type PipelineSignalReceiver 

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

func InitPipelineSignalReceiver 

func InitPipelineSignalReceiver(runningCtx context.Context, regs []*WaitRegister) *PipelineSignalReceiver

func (*PipelineSignalReceiver) GetNextBatch 

func (receiver *PipelineSignalReceiver) GetNextBatch(
	analyzer Analyzer) (content *batch.Batch, info error)

func (*PipelineSignalReceiver) State 

func (receiver *PipelineSignalReceiver) State() PipelineSignalReceiverState

func (*PipelineSignalReceiver) WaitingEnd 

func (receiver *PipelineSignalReceiver) WaitingEnd()

func (*PipelineSignalReceiver) WaitingEndWithTimeout 

func (receiver *PipelineSignalReceiver) WaitingEndWithTimeout(timeout time.Duration) bool

WaitingEndWithTimeout is cleanup-only. It intentionally uses a detached cleanup context because Pipeline.Cleanup cancels the process context before operator Reset runs, but cleanup still needs a bounded chance to drain terminal pipeline signals and release spool references.

type PipelineSignalReceiverState 

type PipelineSignalReceiverState struct {
	Alive      int
	NilBatches []int
	ChannelLen []int
	ChannelCap []int
}

type Process 

type Process struct {
	// BaseProcess is the common part of one process, and it's shared by all its children processes.
	Base *BaseProcess
	Reg  Register

	// Ctx and Cancel are pipeline's context and cancel function.
	// Every pipeline has its own context, and the lifecycle of the pipeline is controlled by the context.
	Ctx     context.Context
	Cancel  context.CancelCauseFunc
	Session Session
}

Process contains context used in query execution one or more pipeline will be generated for one query, and one pipeline has one process instance.

func NewTopProcess 

func NewTopProcess(
	topContext context.Context,
	mp *mpool.MPool,
	txnClient client.TxnClient, txnOperator client.TxnOperator,
	fileService fileservice.FileService,
	lockService lockservice.LockService,
	queryClient qclient.QueryClient, HAKeeper logservice.CNHAKeeperClient,
	udfService udf.Service,
	autoIncrease *defines.AutoIncrCacheManager,
	taskService taskservice.TaskService) *Process

NewTopProcess creates a new top process for the query. It is used to store all the query information, including pool, txn, file service, lock service, etc.

Each developer should watch that, do not call this function twice during the query execution. Use Process.NewContextChildProc() and Process.NewNoContextChildProc() to create a new child process instead.

The returning Process will hold a top context, which is session-client level context. It can be modified by calling Process.ReplaceTopCtx() method, but should be careful to avoid modifying it after the query starts.

There should be announced that the returning Process owns a hack query context field and pipeline context to avoid nil pointer panic. These two filed will be really created and refreshed when the pipeline was going to start, by calling process.BuildQueryCtx() and process.BuildPipelineContext() method.

func (*Process) AllocVectorOfRows added in v0.6.0 

func (proc *Process) AllocVectorOfRows(typ types.Type, nele int, nsp *nulls.Nulls) (*vector.Vector, error)

func (*Process) BuildPipelineContext 

func (proc *Process) BuildPipelineContext(parentContext context.Context) context.Context

BuildPipelineContext cleans the old pipeline context and creates a new one from the input parent context.

func (*Process) BuildProcessInfo 

func (proc *Process) BuildProcessInfo(
	sql string,
) (pipeline.ProcessInfo, error)

func (*Process) Debug added in v1.2.1 

func (proc *Process) Debug(ctx context.Context, msg string, fields ...zap.Field)

func (*Process) DebugBreakDump 

func (proc *Process) DebugBreakDump(cond bool)

func (*Process) Debugf added in v1.2.1 

func (proc *Process) Debugf(ctx context.Context, msg string, args ...any)

func (*Process) Error added in v1.2.1 

func (proc *Process) Error(ctx context.Context, msg string, fields ...zap.Field)

func (*Process) Errorf added in v1.2.1 

func (proc *Process) Errorf(ctx context.Context, msg string, args ...any)

func (*Process) Fatal added in v1.2.1 

func (proc *Process) Fatal(ctx context.Context, msg string, fields ...zap.Field)

func (*Process) Fatalf added in v1.2.1 

func (proc *Process) Fatalf(ctx context.Context, msg string, args ...any)

func (*Process) Free 

func (proc *Process) Free()

Free do memory clean for the process.

func (*Process) GetBaseProcessRunningStatus 

func (proc *Process) GetBaseProcessRunningStatus() bool

func (*Process) GetCloneTxnOperator added in v1.2.1 

func (proc *Process) GetCloneTxnOperator() client.TxnOperator

func (*Process) GetFileService 

func (proc *Process) GetFileService() fileservice.FileService

func (*Process) GetHaKeeper 

func (proc *Process) GetHaKeeper() logservice.CNHAKeeperClient

func (*Process) GetIncrService 

func (proc *Process) GetIncrService() incrservice.AutoIncrementService

func (*Process) GetLastInsertID added in v0.7.0 

func (proc *Process) GetLastInsertID() uint64

func (*Process) GetLim 

func (proc *Process) GetLim() Limitation

func (*Process) GetLoadLocalReader 

func (proc *Process) GetLoadLocalReader() *io.PipeReader

func (*Process) GetLockService 

func (proc *Process) GetLockService() lockservice.LockService

func (*Process) GetMPool added in v0.6.0 

func (proc *Process) GetMPool() *mpool.MPool

func (*Process) GetMessageBoard 

func (proc *Process) GetMessageBoard() *message.MessageBoard

func (*Process) GetPartitionService 

func (proc *Process) GetPartitionService() partitionservice.PartitionService

func (*Process) GetPostDmlSqlList 

func (proc *Process) GetPostDmlSqlList() *threadsafe.Slice[string]

func (*Process) GetPrepareParams added in v0.8.0 

func (proc *Process) GetPrepareParams() *vector.Vector

func (*Process) GetPrepareParamsAt added in v0.8.0 

func (proc *Process) GetPrepareParamsAt(i int) ([]byte, error)

func (*Process) GetQueryClient 

func (proc *Process) GetQueryClient() qclient.QueryClient

func (*Process) GetQueryContextError 

func (proc *Process) GetQueryContextError() error

GetQueryContextError return error once top context or query context with error.

func (*Process) GetResolveVariableFunc added in v0.8.0 

func (proc *Process) GetResolveVariableFunc() func(varName string, isSystemVar, isGlobalVar bool) (interface{}, error)

func (*Process) GetService 

func (proc *Process) GetService() string

func (*Process) GetSession 

func (proc *Process) GetSession() Session

func (*Process) GetSessionInfo added in v0.8.0 

func (proc *Process) GetSessionInfo() *SessionInfo

func (*Process) GetSpillFileService 

func (proc *Process) GetSpillFileService() (fileservice.MutableFileService, error)

func (*Process) GetStageCache 

func (proc *Process) GetStageCache() *threadsafe.Map[string, stage.StageDef]

func (*Process) GetStmtProfile added in v1.0.0 

func (proc *Process) GetStmtProfile() *StmtProfile

func (*Process) GetTaskService 

func (proc *Process) GetTaskService() taskservice.TaskService

func (*Process) GetTopContext 

func (proc *Process) GetTopContext() context.Context

func (*Process) GetTxnOperator 

func (proc *Process) GetTxnOperator() client.TxnOperator

func (*Process) GetUnixTime 

func (proc *Process) GetUnixTime() int64

func (*Process) GetWaitPolicy 

func (proc *Process) GetWaitPolicy() lock.WaitPolicy

func (*Process) Info added in v1.2.1 

func (proc *Process) Info(ctx context.Context, msg string, fields ...zap.Field)

func (*Process) Infof added in v1.2.1 

func (proc *Process) Infof(ctx context.Context, msg string, args ...any)

func (*Process) InitSeq added in v0.8.0 

func (proc *Process) InitSeq()

func (*Process) Mp 

func (proc *Process) Mp() *mpool.MPool

func (*Process) NewBatchFromSrc added in v1.2.0 

func (proc *Process) NewBatchFromSrc(src *batch.Batch, preAllocSize int) (*batch.Batch, error)

func (*Process) NewContextChildProc 

func (proc *Process) NewContextChildProc(dataEntryCount int) *Process

NewContextChildProc make a new child and init its context field. This is used for parallel execution, which will make a new child process to run a pipeline directly. todo: I will remove this method next day, it's a waste to create a new context.

func (*Process) NewNoContextChildProc 

func (proc *Process) NewNoContextChildProc(dataEntryCount int) *Process

NewNoContextChildProc make a new child process without a context field. This is used for the compile-process, which doesn't need to pass the context.

func (*Process) NewNoContextChildProcWithChannel 

func (proc *Process) NewNoContextChildProcWithChannel(dataEntryCount int, channelBufferSize []int32, nilbatchCnt []int32) *Process

NewNoContextChildProcWithChannel make a new child process without a context field. channelBufferSize and nilbatchCnt is the extra information for Reg.

func (*Process) OperatorOutofMemory added in v0.6.0 

func (proc *Process) OperatorOutofMemory(size int64) bool

func (*Process) QueryId added in v0.6.0 

func (proc *Process) QueryId() string

func (*Process) ReplaceTopCtx 

func (proc *Process) ReplaceTopCtx(topCtx context.Context)

ReplaceTopCtx sets the new top context.

func (*Process) ResetCloneTxnOperator 

func (proc *Process) ResetCloneTxnOperator()

ResetCloneTxnOperator cleans the clone txn operator for process reuse.

func (*Process) ResetQueryContext 

func (proc *Process) ResetQueryContext()

ResetQueryContext cleans the context and cancel function for process reuse.

func (*Process) SaveToTopContext 

func (proc *Process) SaveToTopContext(key, value any) context.Context

SaveToTopContext for easy access to change the top context. it's same to a combined operator list like `GetTopContext() + ReplaceTopCtx() + GetTopContext()`.

func (*Process) SetBaseProcessRunningStatus 

func (proc *Process) SetBaseProcessRunningStatus(status bool)

func (*Process) SetCacheForAutoCol added in v0.8.0 

func (proc *Process) SetCacheForAutoCol(name string)

func (*Process) SetCloneTxnOperator added in v1.2.1 

func (proc *Process) SetCloneTxnOperator(op client.TxnOperator)

func (*Process) SetFileService 

func (proc *Process) SetFileService(fs fileservice.FileService)

func (*Process) SetLastInsertID added in v0.7.0 

func (proc *Process) SetLastInsertID(num uint64)

func (*Process) SetMPool 

func (proc *Process) SetMPool(mp *mpool.MPool)

func (*Process) SetMessageBoard 

func (proc *Process) SetMessageBoard(mb *message.MessageBoard)

func (*Process) SetPrepareParams added in v0.8.0 

func (proc *Process) SetPrepareParams(prepareParams *vector.Vector)

func (*Process) SetQueryId added in v0.6.0 

func (proc *Process) SetQueryId(id string)

func (*Process) SetResolveVariableFunc added in v0.8.0 

func (proc *Process) SetResolveVariableFunc(f func(varName string, isSystemVar, isGlobalVar bool) (interface{}, error))

func (*Process) SetStmtProfile added in v1.0.0 

func (proc *Process) SetStmtProfile(sp *StmtProfile)

func (*Process) Warn added in v1.2.1 

func (proc *Process) Warn(ctx context.Context, msg string, fields ...zap.Field)

func (*Process) Warnf added in v1.2.1 

func (proc *Process) Warnf(ctx context.Context, msg string, args ...any)

type ProcessCodecService 

type ProcessCodecService interface {
	Encode(
		proc *Process,
		sql string,
	) ([]byte, error)

	Decode(
		ctx context.Context,
		data pipeline.ProcessInfo,
	) (*Process, error)
}

func GetCodecService 

func GetCodecService(service string) ProcessCodecService

func NewCodecService 

func NewCodecService(
	txnClient client.TxnClient,
	fileService fileservice.FileService,
	lockService lockservice.LockService,
	partitionService partitionservice.PartitionService,
	queryClient qclient.QueryClient,
	hakeeper logservice.CNHAKeeperClient,
	udfService udf.Service,
	engine engine.Engine,
) ProcessCodecService

type QueryBaseContext 

type QueryBaseContext struct {

	// TODO: this is a hack here for resolving pipeline loop of recursive cte.
	// 	we do a special clean-up for the first return pipeline in this pipeline-loop to get rid of deadlock.
	//
	// if pipelineLoopBreak is true, this means no need to do special cleanup yet.
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*QueryBaseContext) BuildQueryCtx 

func (qbCtx *QueryBaseContext) BuildQueryCtx(parent context.Context) context.Context

BuildQueryCtx refreshes the query context and cancellation method after the outer context was ready to run the query.

func (*QueryBaseContext) DoSpecialCleanUp 

func (qbCtx *QueryBaseContext) DoSpecialCleanUp(isMergeCTE bool) bool

func (*QueryBaseContext) SaveToQueryContext 

func (qbCtx *QueryBaseContext) SaveToQueryContext(key, value any) context.Context

SaveToQueryContext saves the key-value pair to the query context. Every pipeline context can access the key-value pair by calling its own context.Value() method. But should be careful to avoid adding key-value pairs after the pipeline context has been created.

func (*QueryBaseContext) WithCounterSetToQueryContext 

func (qbCtx *QueryBaseContext) WithCounterSetToQueryContext(sets ...*perfcounter.CounterSet) context.Context

WithCounterSetToQueryContext sets the counter set to the query context.

type Register 

type Register struct {
	// MergeReceivers, receives result of multi previous operators from other pipelines
	// e.g. merge operator.
	MergeReceivers []*WaitRegister
}

Register used in execution pipeline and shared with all operators of the same pipeline.

type RegisterMessage added in v1.2.1 

type RegisterMessage struct {
	Batch *batch.Batch
	Err   error
}

RegisterMessage channel data Err == nil means pipeline finish with error Batch == nil means pipeline finish without error Batch != nil means pipeline is running

func NewRegMsg added in v1.2.1 

func NewRegMsg(bat *batch.Batch) *RegisterMessage

type RemotePipelineInformationChannel 

type RemotePipelineInformationChannel chan *WrapCs

RemotePipelineInformationChannel used to deliver remote receiver pipeline's information.

remote run Server will use this channel to send information to dispatch operator.

type Session 

type Session interface {
	GetTempTable(dbName, alias string) (string, bool)
	AddTempTable(dbName, alias, realName string)
	RemoveTempTable(dbName, alias string)
	RemoveTempTableByRealName(realName string)
	// GetSqlModeNoAutoValueOnZero reports whether sql_mode contains NO_AUTO_VALUE_ON_ZERO.
	// ok=false means the session doesn't support the cache.
	GetSqlModeNoAutoValueOnZero() (bool, bool)
}

type SessionInfo added in v0.5.0 

type SessionInfo struct {
	Account              string
	User                 string
	Host                 string
	Role                 string
	ConnectionID         uint64
	LastInsertID         uint64
	Database             string
	Version              string
	TimeZone             *time.Location
	LockWaitTimeout      int64
	StorageEngine        engine.Engine
	QueryId              []string
	ResultColTypes       []types.Type
	SeqCurValues         map[uint64]string
	SeqDeleteKeys        []uint64
	SeqAddValues         map[uint64]string
	SeqLastValue         []string
	SqlHelper            sqlHelper
	Buf                  *buffer.Buffer
	SourceInMemScanBatch []*kafka.Message
	LogLevel             zapcore.Level
	SessionId            uuid.UUID
}

SessionInfo session information

func ConvertToProcessSessionInfo 

func ConvertToProcessSessionInfo(
	sei pipeline.SessionInfo,
) (SessionInfo, error)

convert pipeline.SessionInfo to process.SessionInfo

func (*SessionInfo) GetCharset added in v0.5.0 

func (si *SessionInfo) GetCharset() string

func (*SessionInfo) GetCollation added in v0.5.0 

func (si *SessionInfo) GetCollation() string

func (*SessionInfo) GetConnectionID added in v0.5.0 

func (si *SessionInfo) GetConnectionID() uint64

func (*SessionInfo) GetDatabase added in v0.5.0 

func (si *SessionInfo) GetDatabase() string

func (*SessionInfo) GetHost added in v0.5.0 

func (si *SessionInfo) GetHost() string

func (*SessionInfo) GetRole added in v0.5.0 

func (si *SessionInfo) GetRole() string

func (*SessionInfo) GetUser added in v0.5.0 

func (si *SessionInfo) GetUser() string

func (*SessionInfo) GetUserHost added in v0.5.0 

func (si *SessionInfo) GetUserHost() string

func (*SessionInfo) GetVersion added in v0.5.0 

func (si *SessionInfo) GetVersion() string

type StmtProfile added in v1.0.0 

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

StmtProfile will be clear for every statement

func NewStmtProfile added in v1.2.1 

func NewStmtProfile(txnId, stmtId uuid.UUID) *StmtProfile

func (*StmtProfile) Clear added in v1.0.0 

func (sp *StmtProfile) Clear()

func (*StmtProfile) GetDivByZeroIgnore 

func (sp *StmtProfile) GetDivByZeroIgnore() bool

func (*StmtProfile) GetDivByZeroRuntimeProfile 

func (sp *StmtProfile) GetDivByZeroRuntimeProfile() (stmtType, queryType string, ignore bool)

func (*StmtProfile) GetQueryStart added in v1.0.0 

func (sp *StmtProfile) GetQueryStart() time.Time

func (*StmtProfile) GetQueryType added in v1.0.0 

func (sp *StmtProfile) GetQueryType() string

func (*StmtProfile) GetSqlOfStmt added in v1.0.0 

func (sp *StmtProfile) GetSqlOfStmt() string

func (*StmtProfile) GetSqlSourceType added in v1.0.0 

func (sp *StmtProfile) GetSqlSourceType() string

func (*StmtProfile) GetStmtId added in v1.0.0 

func (sp *StmtProfile) GetStmtId() uuid.UUID

func (*StmtProfile) GetStmtType added in v1.0.0 

func (sp *StmtProfile) GetStmtType() string

func (*StmtProfile) GetTxnId added in v1.0.0 

func (sp *StmtProfile) GetTxnId() uuid.UUID

func (*StmtProfile) SetDivByZeroRuntimeProfile 

func (sp *StmtProfile) SetDivByZeroRuntimeProfile(stmtType, queryType string, ignore bool)

func (*StmtProfile) SetQueryStart added in v1.0.0 

func (sp *StmtProfile) SetQueryStart(t time.Time)

func (*StmtProfile) SetQueryType added in v1.0.0 

func (sp *StmtProfile) SetQueryType(qt string)

func (*StmtProfile) SetSqlOfStmt added in v1.0.0 

func (sp *StmtProfile) SetSqlOfStmt(sot string)

func (*StmtProfile) SetSqlSourceType added in v1.0.0 

func (sp *StmtProfile) SetSqlSourceType(st string)

func (*StmtProfile) SetStmtId added in v1.0.0 

func (sp *StmtProfile) SetStmtId(id uuid.UUID)

func (*StmtProfile) SetStmtType added in v1.0.0 

func (sp *StmtProfile) SetStmtType(st string)

func (*StmtProfile) SetTxnId added in v1.0.0 

func (sp *StmtProfile) SetTxnId(id []byte)

type WaitRegister 

type WaitRegister = PipelineEdge

WaitRegister is the historical name for a pipeline edge.

Keep the old type name at API boundaries, but do not keep a second state object. Ch2, the nil-batch counter, and typed terminal state all live in PipelineEdge.

type WrapCs added in v0.7.0 

type WrapCs struct {
	sync.RWMutex
	ReceiverDone bool
	MsgId        uint64
	Uid          uuid.UUID
	Cs           morpc.ClientSession
	Err          chan error
}

WrapCs record information about pipeline's remote receiver.

Source Files

View all Source files

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.