Documentation

Index

Constants

const AnyKind = "*** any procedure kind ***"

const DefaultYieldName = "_result"

DefaultYieldName is the name of a result that doesn't have any name assigned.


const NextPlanNodeIDKey = "NextPlanNodeID"

Variables

var DefaultTriggerSpec = AfterWatermarkTriggerSpec{}

var EmptyBounds = &Bounds{
	Start: values.Time(0),
	Stop:  values.Time(0),
}

EmptyBounds is a time range containing only a single point


Functions

func ClearRegisteredRules

func ClearRegisteredRules()

func ComputeBounds

func ComputeBounds(node Node) error

ComputeBounds computes the time bounds for a plan node from the bounds of its predecessors.

func Formatted

func Formatted(p *Spec, opts ...FormatOption) fmt.Formatter

Formatted accepts a plan.Spec and options, and returns a Formatter that can be used with the standard fmt package, e.g.,

fmt.Println(Formatted(plan, WithDetails())

func HasSideEffect

func HasSideEffect(spec ProcedureSpec) bool

func RegisterLogicalRules

func RegisterLogicalRules(rules ...Rule)

RegisterLogicalRules registers the rule created by createFn with the logical plan.

func RegisterPhysicalRules

func RegisterPhysicalRules(rules ...Rule)

RegisterPhysicalRules registers the rule created by createFn with the physical plan.

func RegisterProcedureSpec

func RegisterProcedureSpec(k ProcedureKind, c CreateProcedureSpec, qks ...flux.OperationKind)

RegisterProcedureSpec registers a new procedure with the specified kind. The call panics if the kind is not unique.

func RegisterProcedureSpecWithSideEffect

func RegisterProcedureSpecWithSideEffect(k ProcedureKind, c CreateProcedureSpec, qks ...flux.OperationKind)

RegisterProcedureSpecWithSideEffect registers a new procedure that produces side effects

func ReplaceNode

func ReplaceNode(oldNode, newNode Node)

ReplaceNode accepts two nodes and attaches all the predecessors of the old node to the new node.

S1   S2        S1   S2
  \ /
oldNode   =>   newNode
  / \            / \
P1   P2        P1   P2

As is convention, newNode will not have any successors attached. The planner will take care of this.

func SetTriggerSpec

func SetTriggerSpec(node Node) error

func WalkPredecessors

func WalkPredecessors(roots []Node, visitFn func(node Node) error) error

WalkPredecessor visits every node in the plan rooted at `roots` in topological order, following predecessor links. If a cycle is detected, no node is visited and an error is returned.

func WalkSuccessors

func WalkSuccessors(roots []Node, visitFn func(node Node) error) error

WalkSuccessors visits every node in the plan rooted at `roots` in topological order, following successor links. If a cycle is detected, no node is visited and an error is returned.

Types

type Administration

type Administration interface {
	Now() time.Time
}

type AfterAtLeastCountTriggerSpec

type AfterAtLeastCountTriggerSpec struct {
	Count int
}

func (AfterAtLeastCountTriggerSpec) Kind

func (AfterAtLeastCountTriggerSpec) Kind() TriggerKind

type AfterProcessingTimeTriggerSpec

type AfterProcessingTimeTriggerSpec struct {
	Duration flux.Duration
}

func (AfterProcessingTimeTriggerSpec) Kind

func (AfterProcessingTimeTriggerSpec) Kind() TriggerKind

type AfterWatermarkTriggerSpec

type AfterWatermarkTriggerSpec struct {
	AllowedLateness flux.Duration
}

func (AfterWatermarkTriggerSpec) Kind

func (AfterWatermarkTriggerSpec) Kind() TriggerKind

type AnyPattern

type AnyPattern struct{}

AnyPattern describes (and matches) any plan node

func (AnyPattern) Match

func (AnyPattern) Match(node Node) bool

func (AnyPattern) Roots

func (AnyPattern) Roots() []ProcedureKind

type Bounds

type Bounds struct {
	Start values.Time
	Stop  values.Time
}

Bounds is a range of time

func (*Bounds) Contains

func (b *Bounds) Contains(t values.Time) bool

Contains reports whether a given time is contained within the time range

func (*Bounds) Intersect

func (b *Bounds) Intersect(o *Bounds) *Bounds

Intersect returns the intersection of two bounds. It returns empty bounds if one of the input bounds are empty.

func (*Bounds) IsEmpty

func (b *Bounds) IsEmpty() bool

IsEmpty reports whether the given bounds contain at most a single point

func (*Bounds) Overlaps

func (b *Bounds) Overlaps(o *Bounds) bool

Overlaps reports whether two given bounds have overlapping time ranges

func (*Bounds) Shift

func (b *Bounds) Shift(d values.Duration) *Bounds

Shift moves the start and stop values of a time range by a specified duration

func (*Bounds) Union

func (b *Bounds) Union(o *Bounds) *Bounds

Union returns the smallest bounds which contain both input bounds. It returns empty bounds if one of the input bounds are empty.

type BoundsAwareProcedureSpec

type BoundsAwareProcedureSpec interface {
	TimeBounds(predecessorBounds *Bounds) *Bounds
}

BoundsAwareProcedureSpec is any procedure that modifies the time bounds of its data.

type Cost

type Cost struct {
	Disk int64
	CPU  int64
	GPU  int64
	MEM  int64
	NET  int64
}

Cost stores various dimensions of the cost of a query plan

func Add

func Add(a Cost, b Cost) Cost

Add two cost structures together

type CreateProcedureSpec

type CreateProcedureSpec func(flux.OperationSpec, Administration) (ProcedureSpec, error)

CreateProcedureSpec creates a ProcedureSpec from an OperationSpec and Administration

type DefaultCost

type DefaultCost struct {
}

func (DefaultCost) Cost

func (c DefaultCost) Cost(inStats []Statistics) (Cost, Statistics)

type Detailer

type Detailer interface {
	PlanDetails() string
}

Detailer provides an optional interface that ProcedureSpecs can implement. Implementors of this interface will have their details appear in the formatted output for a plan if the WithDetails() option is set.

type FormatOption

type FormatOption func(*formatter)

func WithDetails

func WithDetails() FormatOption

WithDetails returns a FormatOption that can be used to provide extra details in a formatted plan.

type GeneratedYieldProcedureSpec

type GeneratedYieldProcedureSpec struct {
	DefaultCost
	Name string
}

GeneratedYieldProcedureSpec provides a special planner-generated yield for queries that don't have explicit calls to yield().

func (*GeneratedYieldProcedureSpec) Copy

func (y *GeneratedYieldProcedureSpec) Copy() ProcedureSpec

func (*GeneratedYieldProcedureSpec) Kind

func (y *GeneratedYieldProcedureSpec) Kind() ProcedureKind

func (*GeneratedYieldProcedureSpec) YieldName

func (y *GeneratedYieldProcedureSpec) YieldName() string

type LogicalNode

type LogicalNode struct {
	Spec   ProcedureSpec
	Source []interpreter.StackEntry
	// contains filtered or unexported fields
}

LogicalNode consists of the input and output edges and a procedure spec that describes what the node does.

func CreateLogicalNode

func CreateLogicalNode(id NodeID, spec ProcedureSpec) *LogicalNode

CreateLogicalNode creates a single logical plan node from a procedure spec. The newly created logical node has no incoming or outgoing edges.

func (*LogicalNode) AddPredecessors

func (e *LogicalNode) AddPredecessors(nodes ...Node)

func (*LogicalNode) AddSuccessors

func (e *LogicalNode) AddSuccessors(nodes ...Node)

func (*LogicalNode) Bounds

func (b *LogicalNode) Bounds() *Bounds

func (*LogicalNode) CallStack

func (lpn *LogicalNode) CallStack() []interpreter.StackEntry

CallStack returns the call stack that created this LogicalNode.

func (*LogicalNode) ClearPredecessors

func (e *LogicalNode) ClearPredecessors()

func (*LogicalNode) ClearSuccessors

func (e *LogicalNode) ClearSuccessors()

func (*LogicalNode) ID

func (lpn *LogicalNode) ID() NodeID

ID returns a human-readable identifier unique to this plan.

func (*LogicalNode) Kind

func (lpn *LogicalNode) Kind() ProcedureKind

Kind returns the kind of procedure performed by this plan node.

func (*LogicalNode) Predecessors

func (e *LogicalNode) Predecessors() []Node

func (*LogicalNode) ProcedureSpec

func (lpn *LogicalNode) ProcedureSpec() ProcedureSpec

ProcedureSpec returns the procedure spec for this plan node.

func (*LogicalNode) ReplaceSpec

func (lpn *LogicalNode) ReplaceSpec(newSpec ProcedureSpec) error

func (*LogicalNode) SetBounds

func (b *LogicalNode) SetBounds(bounds *Bounds)

func (*LogicalNode) ShallowCopy

func (lpn *LogicalNode) ShallowCopy() Node

func (*LogicalNode) Successors

func (e *LogicalNode) Successors() []Node

type LogicalOption

type LogicalOption interface {
	// contains filtered or unexported methods
}

LogicalOption is an option to configure the behavior of the logical plan.

func AddLogicalRules

func AddLogicalRules(rules ...Rule) LogicalOption

func DisableIntegrityChecks

func DisableIntegrityChecks() LogicalOption

DisableIntegrityChecks disables integrity checks in the logical planner.

func OnlyLogicalRules

func OnlyLogicalRules(rules ...Rule) LogicalOption

OnlyLogicalRules produces a logical plan option that forces only a set of particular rules to be applied.

func RemoveLogicalRules

func RemoveLogicalRules(rules ...string) LogicalOption

type LogicalPlanner

type LogicalPlanner interface {
	CreateInitialPlan(spec *flux.Spec) (*Spec, error)
	Plan(context.Context, *Spec) (*Spec, error)
}

LogicalPlanner translates a flux.Spec into a plan.Spec and applies any registered logical rules to the plan.

Logical planning should transform the plan in ways that are independent of actual physical algorithms used to implement operations, and independent of the actual data being processed.

func NewLogicalPlanner

func NewLogicalPlanner(options ...LogicalOption) LogicalPlanner

NewLogicalPlanner returns a new logical plan with the given options. The plan will be configured to apply any logical rules that have been registered.

type NarrowTransformationTriggerSpec

type NarrowTransformationTriggerSpec struct{}

func (NarrowTransformationTriggerSpec) Kind

func (NarrowTransformationTriggerSpec) Kind() TriggerKind

type Node

type Node interface {
	// ID returns an identifier for this plan node.
	ID() NodeID

	// Bounds returns the time bounds for this plan node.
	Bounds() *Bounds

	// Predecessors returns plan nodes executed immediately before this node.
	Predecessors() []Node

	// Successors returns plan nodes executed immediately after this node.
	Successors() []Node

	// ProcedureSpec returns the specification of the procedure represented by this node.
	ProcedureSpec() ProcedureSpec

	// ReplaceSpec replaces the procedure spec of this node with another.
	ReplaceSpec(ProcedureSpec) error

	// Kind returns the type of procedure represented by this node.
	Kind() ProcedureKind

	// CallStack returns the list of StackEntry values that created this
	// Node. A Node may have no associated call stack. This happens
	// when a Node is constructed from a planner rule and not from a
	// source location.
	CallStack() []interpreter.StackEntry

	// Helper methods for manipulating a plan
	// These methods are used during planning
	SetBounds(bounds *Bounds)
	AddSuccessors(...Node)
	AddPredecessors(...Node)
	ClearSuccessors()
	ClearPredecessors()

	ShallowCopy() Node
}

Node defines the common interface for interacting with logical and physical plan nodes.

func MergeToLogicalNode

func MergeToLogicalNode(top, bottom Node, procSpec ProcedureSpec) (Node, error)

MergeToLogicalNode merges top and bottom plan nodes into a new plan node, with the given procedure spec.

V1     V2       V1            V2       <-- successors
  \   /
   top             mergedNode
    |      ==>         |
  bottom               W
    |
    W

The returned node will have its predecessors set to the predecessors of "bottom", however, it's successors will not be set---it will be the responsibility of the plan to attach the merged node to its successors.

func MergeToPhysicalNode

func MergeToPhysicalNode(top, bottom Node, procSpec PhysicalProcedureSpec) (Node, error)

func SwapPlanNodes

func SwapPlanNodes(top, bottom Node) (Node, error)

SwapPlanNodes swaps two plan nodes and returns an equivalent sub-plan with the nodes swapped.

V1   V2        V1   V2
  \ /
   A              B
   |     ==>      |
   B          copy of A
   |              |
   W              W

Note that successors of the original top node will not be updated, and the returned plan node will have no successors. It will be the responsibility of the plan to attach the swapped nodes to successors.

type NodeID

type NodeID string

type OrFinallyTriggerSpec

type OrFinallyTriggerSpec struct {
	Main    TriggerSpec
	Finally TriggerSpec
}

func (OrFinallyTriggerSpec) Kind

func (OrFinallyTriggerSpec) Kind() TriggerKind

type Pattern

type Pattern interface {
	Roots() []ProcedureKind
	Match(Node) bool
}

Pattern represents an operator tree pattern It can match itself against a query plan

func Any

func Any() Pattern

Any returns a pattern that matches anything.

func OneOf

func OneOf(kinds []ProcedureKind, predecessors ...Pattern) Pattern

OneOf matches any plan node from a given set of ProcedureKind and whose predecessors match the given predecessor patterns. This is identical to Pat, except for matching any pattern root from a set of ProcedureKinds.

func Pat

func Pat(kind ProcedureKind, predecessors ...Pattern) Pattern

Pat returns a pattern that can match a plan node with the given ProcedureKind and whose predecessors match the given predecessor patterns.

For example, to construct a pattern that matches a join followed by a sum:

 sum
  |
 |X|    <=>  join(A, B) |> sum()  <=>  Pat(SumKind, Pat(JoinKind, Any(), Any()))
/   \

A B

func PhysPat

func PhysPat(kind ProcedureKind, predecessors ...Pattern) Pattern

PhysPat returns a pattern that matches a physical plan node with the given ProcedureKind and whose predecessors match the given predecessor patterns.

type PhysicalAttributes

type PhysicalAttributes struct {
}

PhysicalAttributes encapsulates sny physical attributes of the result produced by a physical plan node, such as collation, etc.

type PhysicalOneKindPattern

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

PhysicalOneKindPattern matches a physical operator pattern

func (PhysicalOneKindPattern) Match

func (p PhysicalOneKindPattern) Match(node Node) bool

func (PhysicalOneKindPattern) Roots

func (p PhysicalOneKindPattern) Roots() []ProcedureKind

type PhysicalOption

type PhysicalOption interface {
	// contains filtered or unexported methods
}

PhysicalOption is an option to configure the behavior of the physical plan.

func DisableValidation

func DisableValidation() PhysicalOption

DisableValidation disables validation in the physical planner.

func OnlyPhysicalRules

func OnlyPhysicalRules(rules ...Rule) PhysicalOption

OnlyPhysicalRules produces a physical plan option that forces only a particular set of rules to be applied.

func RemovePhysicalRules

func RemovePhysicalRules(rules ...string) PhysicalOption

func WithDefaultMemoryLimit

func WithDefaultMemoryLimit(memBytes int64) PhysicalOption

WithDefaultMemoryLimit sets the default memory limit for plans generated by the plan. If the query spec explicitly sets a memory limit, that limit is used instead of the default.

type PhysicalPlanNode

type PhysicalPlanNode struct {
	Spec   PhysicalProcedureSpec
	Source []interpreter.StackEntry

	// The trigger spec defines how and when a transformation
	// sends its tables to downstream operators
	TriggerSpec TriggerSpec

	// The attributes required from inputs to this node
	RequiredAttrs []PhysicalAttributes

	// The attributes provided to consumers of this node's output
	OutputAttrs PhysicalAttributes
	// contains filtered or unexported fields
}

PhysicalPlanNode represents a physical operation in a plan.

func CreatePhysicalNode

func CreatePhysicalNode(id NodeID, spec PhysicalProcedureSpec) *PhysicalPlanNode

CreatePhysicalNode creates a single physical plan node from a procedure spec. The newly created physical node has no incoming or outgoing edges.

func CreateUniquePhysicalNode

func CreateUniquePhysicalNode(ctx context.Context, prefix string, spec PhysicalProcedureSpec) *PhysicalPlanNode

func (*PhysicalPlanNode) AddPredecessors

func (e *PhysicalPlanNode) AddPredecessors(nodes ...Node)

func (*PhysicalPlanNode) AddSuccessors

func (e *PhysicalPlanNode) AddSuccessors(nodes ...Node)

func (*PhysicalPlanNode) Bounds

func (b *PhysicalPlanNode) Bounds() *Bounds

func (*PhysicalPlanNode) CallStack

func (ppn *PhysicalPlanNode) CallStack() []interpreter.StackEntry

func (*PhysicalPlanNode) ClearPredecessors

func (e *PhysicalPlanNode) ClearPredecessors()

func (*PhysicalPlanNode) ClearSuccessors

func (e *PhysicalPlanNode) ClearSuccessors()

func (*PhysicalPlanNode) Cost

func (ppn *PhysicalPlanNode) Cost(inStats []Statistics) (cost Cost, outStats Statistics)

Cost provides the self-cost (i.e., does not include the cost of its predecessors) for this plan node. Caller must provide statistics of predecessors to this node.

func (*PhysicalPlanNode) ID

func (ppn *PhysicalPlanNode) ID() NodeID

ID returns a human-readable id for this plan node.

func (*PhysicalPlanNode) Kind

func (ppn *PhysicalPlanNode) Kind() ProcedureKind

Kind returns the procedure kind for this plan node.

func (*PhysicalPlanNode) Predecessors

func (e *PhysicalPlanNode) Predecessors() []Node

func (*PhysicalPlanNode) ProcedureSpec

func (ppn *PhysicalPlanNode) ProcedureSpec() ProcedureSpec

ProcedureSpec returns the procedure spec for this plan node.

func (*PhysicalPlanNode) ReplaceSpec

func (ppn *PhysicalPlanNode) ReplaceSpec(newSpec ProcedureSpec) error

func (*PhysicalPlanNode) SetBounds

func (b *PhysicalPlanNode) SetBounds(bounds *Bounds)

func (*PhysicalPlanNode) ShallowCopy

func (ppn *PhysicalPlanNode) ShallowCopy() Node

func (*PhysicalPlanNode) Successors

func (e *PhysicalPlanNode) Successors() []Node

type PhysicalPlanner

type PhysicalPlanner interface {
	Plan(ctx context.Context, lplan *Spec) (*Spec, error)
}

PhysicalPlanner performs transforms a logical plan to a physical plan, by applying any registered physical rules.

func NewPhysicalPlanner

func NewPhysicalPlanner(options ...PhysicalOption) PhysicalPlanner

NewPhysicalPlanner creates a new physical plan with the specified options. The new plan will be configured to apply any physical rules that have been registered.

type PhysicalProcedureSpec

type PhysicalProcedureSpec interface {
	Kind() ProcedureKind
	Copy() ProcedureSpec
	Cost(inStats []Statistics) (cost Cost, outStats Statistics)
}

PhysicalProcedureSpec is similar to its logical counterpart but must provide a method to determine cost.

type Planner

type Planner interface {
	Plan(context.Context, *flux.Spec) (*Spec, error)
}

type PlannerBuilder

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

PlannerBuilder provides clients with an easy way to create planners.

func (*PlannerBuilder) AddLogicalOptions

func (pb *PlannerBuilder) AddLogicalOptions(lopt ...LogicalOption)

AddLogicalOptions lets callers specify attributes of the logical planner that will be part of the created planner.

func (*PlannerBuilder) AddPhysicalOptions

func (pb *PlannerBuilder) AddPhysicalOptions(popt ...PhysicalOption)

AddPhysicalOptions lets callers specify attributes of the physical planner that will be part of the created planner.

func (PlannerBuilder) Build

func (pb PlannerBuilder) Build() Planner

Build builds a planner with specified attributes.

type PostPhysicalValidator

type PostPhysicalValidator interface {
	PostPhysicalValidate(id NodeID) error
}

PostPhysicalValidator provides an interface that can be implemented by PhysicalProcedureSpecs for any validation checks to be performed post-physical planning.

type ProcedureKind

type ProcedureKind string

ProcedureKind denotes the kind of operation

type ProcedureSpec

type ProcedureSpec interface {
	Kind() ProcedureKind
	Copy() ProcedureSpec
}

ProcedureSpec specifies a query operation

type RepeatedTriggerSpec

type RepeatedTriggerSpec struct {
	Trigger TriggerSpec
}

func (RepeatedTriggerSpec) Kind

func (RepeatedTriggerSpec) Kind() TriggerKind

type Rule

type Rule interface {
	// Name of this rule (must be unique).
	Name() string

	// Pattern for this rule to match against.
	Pattern() Pattern

	// Rewrite an operation into an equivalent one.
	// The returned node is the new root of the sub tree.
	// The boolean return value should be true if anything changed during the rewrite.
	Rewrite(context.Context, Node) (Node, bool, error)
}

Rule is transformation rule for a query operation

type Spec

type Spec struct {
	Roots     map[Node]struct{}
	Resources flux.ResourceManagement
	Now       time.Time
}

Spec holds the result nodes of a query plan with associated metadata

func NewPlanSpec

func NewPlanSpec() *Spec

NewPlanSpec initializes a new query plan

func (*Spec) BottomUpWalk

func (plan *Spec) BottomUpWalk(f func(Node) error) error

BottomUpWalk will execute f for each plan node in the Spec, starting from the sources, and only visiting a node after all its predecessors have been visited.

func (*Spec) CheckIntegrity

func (plan *Spec) CheckIntegrity() error

CheckIntegrity checks the integrity of the plan, i.e.:

- node A is predecessor of B iff B is successor of A;
- there is no cycle.

This check only detects this problem (N2 is predecessor of R, but not viceversa):

N1 <----> R
          |
N2 <-------

And this one (R is successor of N2, but not viceversa):

N1 <-------
|         |--> R
N2 --------

But not this one, because N2 is not reachable from R (root):

N1 <-------
          |--> R
N2 --------

func (*Spec) Replace

func (plan *Spec) Replace(root, with Node)

Replace replaces one of the root nodes of the query plan

func (*Spec) TopDownWalk

func (plan *Spec) TopDownWalk(f func(node Node) error) error

TopDownWalk will execute f for each plan node in the Spec. It always visits a node before visiting its predecessors.

func (*Spec) TopologicalWalk

func (plan *Spec) TopologicalWalk(visitFn func(node Node) error) error

TopologicalWalk visits every node in the plan in topological order. If a cycle is detected, no node is visited and an error is returned.

type Statistics

type Statistics struct {
	Cardinality      int64
	GroupCardinality int64
}

type TriggerAwareProcedureSpec

type TriggerAwareProcedureSpec interface {
	TriggerSpec() TriggerSpec
}

type TriggerKind

type TriggerKind int
const (
	NarrowTransformation TriggerKind = iota
	AfterWatermark
	Repeated
	AfterProcessingTime
	AfterAtLeastCount
	OrFinally
)

type TriggerSpec

type TriggerSpec interface {
	Kind() TriggerKind
}

type UnionKindPattern

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

UnionKindPattern matches any one of a set of procedures that have a specified predecessor pattern.

For example, UnionKindPattern( { Proc1Kind, Proc2Kind }, { Pat1, Pat2 } ) will match either Proc1Kind { Pat1, Pat2 } or Proc2Kind { Pat1, Pat2 }

         [ ProcedureKind ]
         /       |  ...  \
pattern1     pattern2  ... patternK

func (UnionKindPattern) Match

func (okp UnionKindPattern) Match(node Node) bool

func (UnionKindPattern) Roots

func (okp UnionKindPattern) Roots() []ProcedureKind

type WindowSpec

type WindowSpec struct {
	Every  flux.Duration
	Period flux.Duration
	Offset flux.Duration
}

type YieldProcedureSpec

type YieldProcedureSpec interface {
	YieldName() string
}

YieldProcedureSpec is a special procedure that has the side effect of returning a result to the client.

Directories

Path Synopsis
plantest Package plantest contains utilities for testing each query planning phase