operator

Documentation

Index

• type AggregateOp
  • func (p *AggregateOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, ...) interface{}
• type AnalyticFuncsOp
  • func (p *AnalyticFuncsOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, ...) (got interface{})
• type EmitterOp
  • func (p *EmitterOp) Apply(ctx api.StreamContext, data any, _ *xsql.FunctionValuer, ...) any
• type FilterOp
  • func (p *FilterOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, ...) interface{}
• type FuncOp
  • func (p *FuncOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, ...) interface{}
• type HavingOp
  • func (p *HavingOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, ...) interface{}
• type JoinOp
  • func (jp *JoinOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, ...) interface{}
• type OrderOp
  • func (p *OrderOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, ...) interface{}
• type Preprocessor
  • func NewPreprocessor(isSchemaless bool, fields map[string]*ast.JsonStreamField, _ bool, _ []string, ...) (*Preprocessor, error)
  • func (p *Preprocessor) Apply(ctx api.StreamContext, data interface{}, _ *xsql.FunctionValuer, ...) interface{}
• type ProjectOp
  • func (pp *ProjectOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, ...) interface{}
• type ProjectSetOperator
  • func (ps *ProjectSetOperator) Apply(ctx api.StreamContext, data interface{}, _ *xsql.FunctionValuer, ...) interface{}
• type WindowFuncOperator
  • func (wf *WindowFuncOperator) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, ...) interface{}

Types

type AggregateOp

type AggregateOp struct {
	Dimensions ast.Dimensions
}

func (*AggregateOp) Apply

func (p *AggregateOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) interface{}

Apply

 input: Collection
*  output: Collection

type AnalyticFuncsOp

type AnalyticFuncsOp struct {
	Funcs      []*ast.Call
	FieldFuncs []*ast.Call
}

func (*AnalyticFuncsOp) Apply

func (p *AnalyticFuncsOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) (got interface{})

type EmitterOp

type EmitterOp struct {
	Emitter string
}

EmitterOp set the emitter to the stream name It is only planned after shared connection node in which the emitter is not determined in the source side

func (*EmitterOp) Apply

func (p *EmitterOp) Apply(ctx api.StreamContext, data any, _ *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) any

type FilterOp

type FilterOp struct {
	Condition  ast.Expr
	StateFuncs []*ast.Call
}

func (*FilterOp) Apply

func (p *FilterOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) interface{}

Apply the filter operator to each message in the stream The input data could be a xsql.Row or a xsql.Collection For xsql.Row, apply the condition to the row and return the row if the condition is true For xsql.Collection, apply the condition to each row and return the rows that meet the condition If error happens, return the error

type FuncOp

type FuncOp struct {
	IsAgg    bool
	CallExpr *ast.Call
	Name     string
}

func (*FuncOp) Apply

func (p *FuncOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, afv *xsql.AggregateFunctionValuer) interface{}

type HavingOp

type HavingOp struct {
	Condition  ast.Expr
	StateFuncs []*ast.Call
}

func (*HavingOp) Apply

func (p *HavingOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, afv *xsql.AggregateFunctionValuer) interface{}

type JoinOp

type JoinOp struct {
	From  *ast.Table
	Joins ast.Joins
}

JoinOp TODO join expr should only be the equal op between 2 streams like tb1.id = tb2.id

func (*JoinOp) Apply

func (jp *JoinOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) interface{}

Apply JoinOp to join two streams. If running in continuous query, the inner join will always return empty result because there is only one stream data.

type OrderOp

type OrderOp struct {
	SortFields ast.SortFields
}

func (*OrderOp) Apply

func (p *OrderOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, afv *xsql.AggregateFunctionValuer) interface{}

type Preprocessor

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

Preprocessor only planned when 1. eventTime, to convert the timestamp field 2. schema validate and convert, when strict_validation is on and field type is not binary Do not convert types

func NewPreprocessor

func NewPreprocessor(isSchemaless bool, fields map[string]*ast.JsonStreamField, _ bool, _ []string, iet bool, timestampField string, timestampFormat string, isBinary bool, strictValidation bool) (*Preprocessor, error)

func (*Preprocessor) Apply

func (p *Preprocessor) Apply(ctx api.StreamContext, data interface{}, _ *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) interface{}

Apply the preprocessor to the tuple

	input: *xsql.Tuple
 *	output: *xsql.Tuple

type ProjectOp

type ProjectOp struct {
	ColNames         [][]string // list of [col, table]
	AliasNames       []string   // list of alias name
	ExprNames        []string   // list of expr name
	ExceptNames      []string   // list of except name
	WindowFuncNames  map[string]ast.Field
	AllWildcard      bool
	WildcardEmitters map[string]bool
	AliasFields      ast.Fields
	ExprFields       ast.Fields
	IsAggregate      bool // Whether the project is used in an aggregate context. This is set by planner by analyzing the SQL query
	EnableLimit      bool
	LimitCount       int

	SendMeta bool
	// contains filtered or unexported fields
}

func (*ProjectOp) Apply

func (pp *ProjectOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, afv *xsql.AggregateFunctionValuer) interface{}

Apply

input: *xsql.Tuple| xsql.Collection

output: []map[string]interface{}

type ProjectSetOperator

type ProjectSetOperator struct {
	SrfMapping  map[string]struct{}
	EnableLimit bool
	LimitCount  int
}

func (*ProjectSetOperator) Apply

func (ps *ProjectSetOperator) Apply(ctx api.StreamContext, data interface{}, _ *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) interface{}

Apply implement UnOperation ProjectSetOperator will extract the results from the set-returning-function into multi rows by aligning other columns For tuple, ProjectSetOperator will do the following transform: {"a":[1,2],"b":3} => {"a":1,"b":3},{"a":2,"b":3} For Collection, ProjectSetOperator will do the following transform: [{"a":[1,2],"b":3},{"a":[1,2],"b":4}] = > [{"a":"1","b":3},{"a":"2","b":3},{"a":"1","b":4},{"a":"2","b":4}]

type WindowFuncOperator

type WindowFuncOperator struct {
	WindowFuncField ast.Field
}

func (*WindowFuncOperator) Apply

func (wf *WindowFuncOperator) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, afv *xsql.AggregateFunctionValuer) interface{}