physical

Documentation

Index

• func ExplainExpr(expr Expression, withTypeInfo bool) *graph.Node
• func ExplainNode(node Node, withTypeInfo bool) *graph.Node
• func VariableNameMatchesField(varName, fieldName string) bool
• type Aggregate
• type AggregateDescriptor
• type AggregateDetails
• type And
• type Cast
• type Coalesce
• type Constant
• type CountingTrigger
• type Database
• type Datasource
  • func (node *Datasource) PushDownPredicates(newPredicates, pushedDownPredicates []Expression) (rejected []Expression, pushedDown []Expression, changed bool)
• type DatasourceImplementation
• type DatasourceRepository
  • func (dr *DatasourceRepository) GetDatasource(ctx context.Context, name string) (DatasourceImplementation, Schema, error)
• type Distinct
• type EndOfStreamTrigger
• type Environment
  • func (env Environment) WithRecordSchema(schema Schema) Environment
• type Expression
  • func RenameVariablesExpr(oldToNew map[string]string, expr Expression) Expression
  • func (expr *Expression) Materialize(ctx context.Context, env Environment) (execution.Expression, error)
  • func (expr Expression) SplitByAnd() []Expression
  • func (expr Expression) VariablesUsed() []string
• type ExpressionType
  • func (t ExpressionType) String() string
• type Filter
• type FunctionCall
• type FunctionDescriptor
• type FunctionDetails
• type GroupBy
• type LookupJoin
• type Map
• type MultiTrigger
• type Node
  • func RenameVariables(oldToNew map[string]string, node Node) Node
  • func (node *Node) Materialize(ctx context.Context, env Environment) (execution.Node, error)
• type NodeType
  • func (t NodeType) String() string
• type Or
• type OrderBy
• type QueryExpression
• type Requalifier
• type Schema
  • func NewSchema(fields []SchemaField, timeField int) Schema
• type SchemaField
• type StreamJoin
• type TableValuedFunction
• type TableValuedFunctionArgument
  • func (arg *TableValuedFunctionArgument) String() string
• type TableValuedFunctionArgumentDescriptor
• type TableValuedFunctionArgumentExpression
• type TableValuedFunctionArgumentTable
• type TableValuedFunctionArgumentType
• type TableValuedFunctionDescriptor
• type Transformers
  • func (t *Transformers) TransformExpr(expr Expression) Expression
  • func (t *Transformers) TransformNode(node Node) Node
• type Trigger
  • func (t *Trigger) Materialize(ctx context.Context, env Environment) func() execution.Trigger
• type TriggerType
  • func (t TriggerType) String() string
• type Tuple
• type TypeAssertion
• type Unnest
• type Variable
• type VariableContext
  • func (varCtx *VariableContext) WithRecordSchema(schema Schema) *VariableContext
• type WatermarkTrigger

Functions

func ExplainExpr

func ExplainExpr(expr Expression, withTypeInfo bool) *graph.Node

func ExplainNode

func ExplainNode(node Node, withTypeInfo bool) *graph.Node

func VariableNameMatchesField

func VariableNameMatchesField(varName, fieldName string) bool

Types

type Aggregate

type Aggregate struct {
	Name                string
	OutputType          octosql.Type
	AggregateDescriptor AggregateDescriptor
}

type AggregateDescriptor

type AggregateDescriptor struct {
	ArgumentType octosql.Type
	OutputType   octosql.Type
	TypeFn       func(octosql.Type) (octosql.Type, bool)
	Prototype    func() nodes.Aggregate
}

type AggregateDetails

type AggregateDetails struct {
	Description string
	Descriptors []AggregateDescriptor
}

type And

type And struct {
	Arguments []Expression
}

type Cast

type Cast struct {
	Expression Expression
	TargetType octosql.Type
}

type Coalesce

type Coalesce struct {
	Arguments []Expression
}

type Constant

type Constant struct {
	Value octosql.Value
}

type CountingTrigger

type CountingTrigger struct {
	TriggerAfter uint
}

type Database

type Database interface {
	ListTables(ctx context.Context) ([]string, error)
	GetTable(ctx context.Context, name string) (DatasourceImplementation, Schema, error)
}

type Datasource

type Datasource struct {
	Name, Alias              string
	DatasourceImplementation DatasourceImplementation
	VariableMapping          map[string]string
	Predicates               []Expression
}

func (*Datasource) PushDownPredicates

func (node *Datasource) PushDownPredicates(newPredicates, pushedDownPredicates []Expression) (rejected []Expression, pushedDown []Expression, changed bool)

type DatasourceImplementation

type DatasourceImplementation interface {
	Materialize(ctx context.Context, env Environment, schema Schema, pushedDownPredicates []Expression) (execution.Node, error)
	PushDownPredicates(newPredicates, pushedDownPredicates []Expression) (rejected, pushedDown []Expression, changed bool)
}

type DatasourceRepository

type DatasourceRepository struct {
	// TODO: A może jednak ten bardziej dynamiczny interfejs? Że database.<table> i wtedy sie resolvuje
	// Bo inaczej będzie na start strasznie dużo rzeczy ładować niepotrzebnych dla wszystkich
	// skonfigurowanych baz danych.
	Databases       map[string]func() (Database, error)
	DefaultDatabase string
	FileHandlers    map[string]func(name string) (DatasourceImplementation, Schema, error)
}

func (*DatasourceRepository) GetDatasource

func (dr *DatasourceRepository) GetDatasource(ctx context.Context, name string) (DatasourceImplementation, Schema, error)

type Distinct

type Distinct struct {
	Source Node
}

type EndOfStreamTrigger

type EndOfStreamTrigger struct {
}

type Environment

type Environment struct {
	Aggregates      map[string]AggregateDetails
	Datasources     *DatasourceRepository
	Functions       map[string]FunctionDetails
	PhysicalConfig  map[string]interface{}
	VariableContext *VariableContext
}

TODO: There should be a seperate MaterializationContext.

func (Environment) WithRecordSchema

func (env Environment) WithRecordSchema(schema Schema) Environment

type Expression

type Expression struct {
	Type octosql.Type

	ExpressionType ExpressionType
	// Only one of the below may be non-null.
	Variable        *Variable
	Constant        *Constant
	FunctionCall    *FunctionCall
	And             *And
	Or              *Or
	QueryExpression *QueryExpression
	Coalesce        *Coalesce
	Tuple           *Tuple
	TypeAssertion   *TypeAssertion
	Cast            *Cast
}

func RenameVariablesExpr

func RenameVariablesExpr(oldToNew map[string]string, expr Expression) Expression

func (*Expression) Materialize

func (expr *Expression) Materialize(ctx context.Context, env Environment) (execution.Expression, error)

func (Expression) SplitByAnd

func (expr Expression) SplitByAnd() []Expression

func (Expression) VariablesUsed

func (expr Expression) VariablesUsed() []string

type ExpressionType

type ExpressionType int

const (
	ExpressionTypeVariable ExpressionType = iota
	ExpressionTypeConstant
	ExpressionTypeFunctionCall
	ExpressionTypeAnd
	ExpressionTypeOr
	ExpressionTypeQueryExpression
	ExpressionTypeCoalesce
	ExpressionTypeTuple
	ExpressionTypeTypeAssertion
	ExpressionTypeCast
)

func (ExpressionType) String

func (t ExpressionType) String() string

type Filter

type Filter struct {
	Source    Node
	Predicate Expression
}

type FunctionCall

type FunctionCall struct {
	Name               string
	Arguments          []Expression
	FunctionDescriptor FunctionDescriptor
}

type FunctionDescriptor

type FunctionDescriptor struct {
	ArgumentTypes []octosql.Type
	OutputType    octosql.Type
	TypeFn        func([]octosql.Type) (octosql.Type, bool) `json:"-"`
	Strict        bool
	Function      func([]octosql.Value) (octosql.Value, error) `json:"-"`
}

type FunctionDetails

type FunctionDetails struct {
	Description string
	Descriptors []FunctionDescriptor
}

type GroupBy

type GroupBy struct {
	Source               Node
	Aggregates           []Aggregate
	AggregateExpressions []Expression
	Key                  []Expression
	// Either index to Key, or -1, if none.
	KeyEventTimeIndex int
	Trigger           Trigger
}

type LookupJoin

type LookupJoin struct {
	Source, Joined Node
}

type Map

type Map struct {
	Source      Node
	Expressions []Expression
}

type MultiTrigger

type MultiTrigger struct {
	Triggers []Trigger
}

type Node

type Node struct {
	Schema Schema

	NodeType NodeType
	// Only one of the below may be non-null.
	Datasource          *Datasource
	Distinct            *Distinct
	Filter              *Filter
	GroupBy             *GroupBy
	LookupJoin          *LookupJoin
	StreamJoin          *StreamJoin
	Map                 *Map
	OrderBy             *OrderBy
	Requalifier         *Requalifier
	TableValuedFunction *TableValuedFunction
	Unnest              *Unnest
}

func RenameVariables

func RenameVariables(oldToNew map[string]string, node Node) Node

func (*Node) Materialize

func (node *Node) Materialize(ctx context.Context, env Environment) (execution.Node, error)

type NodeType

type NodeType int

const (
	NodeTypeDatasource NodeType = iota
	NodeTypeDistinct
	NodeTypeFilter
	NodeTypeGroupBy
	NodeTypeLookupJoin
	NodeTypeStreamJoin
	NodeTypeMap
	NodeTypeOrderBy
	NodeTypeRequalifier
	NodeTypeTableValuedFunction
	NodeTypeUnnest
)

func (NodeType) String

func (t NodeType) String() string

type Or

type Or struct {
	Arguments []Expression
}

type OrderBy

type OrderBy struct {
	Source               Node
	Key                  []Expression
	DirectionMultipliers []int
}

type QueryExpression

type QueryExpression struct {
	Source Node
}

type Requalifier

type Requalifier struct {
	Source    Node
	Qualifier string
}

type Schema

type Schema struct {
	Fields []SchemaField
	// TimeField is -1 if not present.
	TimeField int
}

func NewSchema

func NewSchema(fields []SchemaField, timeField int) Schema

type SchemaField

type SchemaField struct {
	Name string
	Type octosql.Type
}

type StreamJoin

type StreamJoin struct {
	Left, Right       Node
	LeftKey, RightKey []Expression
}

type TableValuedFunction

type TableValuedFunction struct {
	Name               string
	Arguments          map[string]TableValuedFunctionArgument
	FunctionDescriptor TableValuedFunctionDescriptor
}

type TableValuedFunctionArgument

type TableValuedFunctionArgument struct {
	TableValuedFunctionArgumentType TableValuedFunctionArgumentType
	// Only one of the below may be non-null.
	Expression *TableValuedFunctionArgumentExpression
	Table      *TableValuedFunctionArgumentTable
	Descriptor *TableValuedFunctionArgumentDescriptor
}

func (*TableValuedFunctionArgument) String

func (arg *TableValuedFunctionArgument) String() string

type TableValuedFunctionArgumentDescriptor

type TableValuedFunctionArgumentDescriptor struct {
	Descriptor string
}

type TableValuedFunctionArgumentExpression

type TableValuedFunctionArgumentExpression struct {
	Expression Expression
}

type TableValuedFunctionArgumentTable

type TableValuedFunctionArgumentTable struct {
	Table Node
}

type TableValuedFunctionArgumentType

type TableValuedFunctionArgumentType int

const (
	TableValuedFunctionArgumentTypeExpression TableValuedFunctionArgumentType = iota
	TableValuedFunctionArgumentTypeTable
	TableValuedFunctionArgumentTypeDescriptor
)

type TableValuedFunctionDescriptor

type TableValuedFunctionDescriptor struct {
	Materialize func(context.Context, Environment, map[string]TableValuedFunctionArgument) (execution.Node, error)
}

type Transformers

type Transformers struct {
	NodeTransformer       func(node Node) Node
	ExpressionTransformer func(expr Expression) Expression
}

func (*Transformers) TransformExpr

func (t *Transformers) TransformExpr(expr Expression) Expression

func (*Transformers) TransformNode

func (t *Transformers) TransformNode(node Node) Node

TODO: This should probably push down and make available the environment.

type Trigger

type Trigger struct {
	TriggerType TriggerType
	// Only one of the below may be non-null.
	CountingTrigger    *CountingTrigger
	EndOfStreamTrigger *EndOfStreamTrigger
	WatermarkTrigger   *WatermarkTrigger
	MultiTrigger       *MultiTrigger
}

func (*Trigger) Materialize

func (t *Trigger) Materialize(ctx context.Context, env Environment) func() execution.Trigger

type TriggerType

type TriggerType int

const (
	TriggerTypeCounting TriggerType = iota
	TriggerTypeEndOfStream
	TriggerTypeWatermark
	TriggerTypeMulti
)

func (TriggerType) String

func (t TriggerType) String() string

type Tuple

type Tuple struct {
	Arguments []Expression
}

type TypeAssertion

type TypeAssertion struct {
	Expression Expression
	TargetType octosql.Type
}

type Unnest

type Unnest struct {
	Source Node
	Field  string
}

type Variable

type Variable struct {
	Name     string
	IsLevel0 bool
}

type VariableContext

type VariableContext struct {
	Parent *VariableContext
	Fields []SchemaField
}

func (*VariableContext) WithRecordSchema

func (varCtx *VariableContext) WithRecordSchema(schema Schema) *VariableContext

type WatermarkTrigger

type WatermarkTrigger struct {
	TimeFieldIndex int
}