executor

Documentation

Index

• Constants
• func CreateScalarSource(ID parser.NodeID, params ScalarParams, options transform.Options) (parser.Source, *transform.Controller)
• func CreateSource(ID parser.NodeID, params SourceParams, storage storage.Storage, ...) (parser.Source, *transform.Controller)
• func CreateTransform(ID parser.NodeID, params transform.Params, options transform.Options) (transform.OpNode, *transform.Controller)
• type Engine
  • func NewEngine(store storage.Storage) *Engine
  • func (e *Engine) Close() error
  • func (e *Engine) Execute(ctx context.Context, query *storage.FetchQuery, opts *EngineOptions, ...)
  • func (e *Engine) ExecuteExpr(ctx context.Context, parser parser.Parser, opts *EngineOptions, ...)
• type EngineOptions
• type ExecutionState
  • func GenerateExecutionState(pplan plan.PhysicalPlan, storage storage.Storage) (*ExecutionState, error)
  • func (s *ExecutionState) Execute(ctx context.Context) error
  • func (s *ExecutionState) String() string
• type Query
• type QueryStatistics
• type QueryTask
• type Result
• type ResultChan
• type ResultNode
  • func (r *ResultNode) Process(ID parser.NodeID, block block.Block) error
  • func (r *ResultNode) ResultChan() chan ResultChan
• type ScalarParams
• type SourceParams
• type TaskStatus
  • func (t TaskStatus) String() string
• type Tracker
  • func NewTracker() *Tracker
  • func (t *Tracker) Close() error
  • func (t *Tracker) DetachQuery(qid uint64) error
  • func (t *Tracker) KillQuery(qid uint64) error
  • func (t *Tracker) Track(query storage.Query, connClosed <-chan bool) (*QueryTask, error)

Constants

const (
	// DefaultQueryTimeout is the default timeout for executing a query.
	// A value of zero will have no query timeout.
	DefaultQueryTimeout = time.Duration(0)
)

Functions

func CreateScalarSource

func CreateScalarSource(
	ID parser.NodeID,
	params ScalarParams,
	options transform.Options,
) (parser.Source, *transform.Controller)

CreateScalarSource creates a scalar source node

func CreateSource

func CreateSource(
	ID parser.NodeID,
	params SourceParams, storage storage.Storage,
	options transform.Options,
) (parser.Source, *transform.Controller)

CreateSource creates a source node

func CreateTransform

func CreateTransform(
	ID parser.NodeID,
	params transform.Params,
	options transform.Options,
) (transform.OpNode, *transform.Controller)

CreateTransform creates a transform node which works on functions and contains state

Types

type Engine

type Engine struct {
	Stats *QueryStatistics
	// contains filtered or unexported fields
}

Engine executes a Query.

func NewEngine

func NewEngine(store storage.Storage) *Engine

NewEngine returns a new instance of QueryExecutor.

func (*Engine) Close

func (e *Engine) Close() error

Close kills all running queries and prevents new queries from being attached.

func (*Engine) Execute

func (e *Engine) Execute(ctx context.Context, query *storage.FetchQuery, opts *EngineOptions, closing <-chan bool, results chan *storage.QueryResult)

Execute runs the query and closes the results channel once done

func (*Engine) ExecuteExpr

func (e *Engine) ExecuteExpr(ctx context.Context, parser parser.Parser, opts *EngineOptions, params models.RequestParams, results chan Query)

ExecuteExpr runs the query DAG and closes the results channel once done nolint: unparam

type EngineOptions

type EngineOptions struct {
	// AbortCh is a channel that signals when results are no longer desired by the caller.
	AbortCh <-chan bool
}

EngineOptions can be used to pass custom flags to engine

type ExecutionState

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

ExecutionState represents the execution hierarchy

func GenerateExecutionState

func GenerateExecutionState(
	pplan plan.PhysicalPlan,
	storage storage.Storage,
) (*ExecutionState, error)

GenerateExecutionState creates an execution state from the physical plan

func (*ExecutionState) Execute

func (s *ExecutionState) Execute(ctx context.Context) error

Execute the sources in parallel and return the first error

func (*ExecutionState) String

func (s *ExecutionState) String() string

String representation of the state

type Query

type Query struct {
	Err    error
	Result Result
}

Query is the result after execution

type QueryStatistics

type QueryStatistics struct {
	ActiveQueries          int64
	ExecutedQueries        int64
	FinishedQueries        int64
	QueryExecutionDuration int64
}

QueryStatistics keeps statistics related to the QueryExecutor.

type QueryTask

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

QueryTask is the internal data structure for managing queries. For the public use data structure that gets returned, see QueryTask.

type Result

type Result interface {
	ResultChan() chan ResultChan
	// contains filtered or unexported methods
}

Result provides the execution results

type ResultChan

type ResultChan struct {
	Block block.Block
	Err   error
}

ResultChan has the result from a block

type ResultNode

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

ResultNode is used to provide the results to the caller from the query execution

func (*ResultNode) Process

func (r *ResultNode) Process(ID parser.NodeID, block block.Block) error

Process the block

func (*ResultNode) ResultChan

func (r *ResultNode) ResultChan() chan ResultChan

ResultChan return a channel to stream back resultChan to the client

type ScalarParams

type ScalarParams interface {
	parser.Params
	Node(
		controller *transform.Controller,
		options transform.Options,
	) parser.Source
}

ScalarParams are defined by sources

type SourceParams

type SourceParams interface {
	parser.Params
	Node(
		controller *transform.Controller,
		storage storage.Storage,
		options transform.Options,
	) parser.Source
}

SourceParams are defined by sources

type TaskStatus

type TaskStatus int

TaskStatus is the status of a query task

const (
	// RunningTask is set when the task is running.
	RunningTask TaskStatus = iota

	// KilledTask is set when the task is killed, but resources are still
	// being used.
	KilledTask
)

func (TaskStatus) String

func (t TaskStatus) String() string

type Tracker

type Tracker struct {
	// Log queries if they are slower than this time.
	// If zero, slow queries will never be logged.
	LogQueriesAfter time.Duration

	// Maximum number of concurrent queries.
	MaxConcurrentQueries int
	// contains filtered or unexported fields
}

Tracker tracks the query state

func NewTracker

func NewTracker() *Tracker

NewTracker creates a new tracker.

func (*Tracker) Close

func (t *Tracker) Close() error

Close kills all running queries and prevents new queries from being attached.

func (*Tracker) DetachQuery

func (t *Tracker) DetachQuery(qid uint64) error

DetachQuery removes a query from the query table. If the query is not in the killed state, this will also close the related channel.

func (*Tracker) KillQuery

func (t *Tracker) KillQuery(qid uint64) error

KillQuery enters a query into the killed state and closes the channel from the TaskManager. This method can be used to forcefully terminate a running query.

func (*Tracker) Track

func (t *Tracker) Track(query storage.Query, connClosed <-chan bool) (*QueryTask, error)

Track is used to add a new query to tracker