Define base dal pipeline implementation
This commit is contained in:
Tomaž Jerman
@@ -0,0 +1,350 @@
|
||||
package dal
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
type (
|
||||
// Pipeline defines a series of steps performed over the data
|
||||
//
|
||||
// General outline of how a pipeline is used:
|
||||
// 1. Populate the pipeline with steps to define the desired outcome
|
||||
// 2. Analyze the pipeline to collect statistics and assist optimizers
|
||||
// 3. Run structure optimization which may reorder/drop steps
|
||||
// 4. Run step optimization which may re-configure individual steps such as
|
||||
// push parts of a filter to lower level, request specific sort order, ...
|
||||
// 5. Use the pipeline as an iterator to pull the data from.
|
||||
Pipeline []PipelineStep
|
||||
|
||||
// PipelineStep defines an operation performed over the data in the pipeline
|
||||
PipelineStep interface {
|
||||
Identifier() string
|
||||
Sources() []string
|
||||
Initialize(context.Context, ...Iterator) (Iterator, error)
|
||||
|
||||
Analyze(ctx context.Context) error
|
||||
Analysis() stepAnalysis
|
||||
}
|
||||
|
||||
// Attribute mapping outlines specific attributes within a pipeline
|
||||
// @todo reconsider this interface; potentially remove it or split it up
|
||||
AttributeMapping interface {
|
||||
Identifier() string
|
||||
Expression() (expression string)
|
||||
Source() (ident string)
|
||||
Properties() MapProperties
|
||||
}
|
||||
|
||||
// MapProperties describe the attribute such as it's type and constraints
|
||||
MapProperties struct {
|
||||
IsPrimary bool
|
||||
Nullable bool
|
||||
Type Type
|
||||
}
|
||||
)
|
||||
|
||||
// LinkSteps links related steps into a tree structure
|
||||
//
|
||||
// @todo make it return a new slice and not mutate the original
|
||||
func (pp Pipeline) LinkSteps() (err error) {
|
||||
// map steps by identifiers
|
||||
steps := make(map[string]PipelineStep)
|
||||
for _, s := range pp {
|
||||
steps[s.Identifier()] = s
|
||||
}
|
||||
|
||||
// Link up...
|
||||
err = func() (err error) {
|
||||
for _, s := range pp {
|
||||
switch rs := s.(type) {
|
||||
case *Aggregate:
|
||||
rs.rel = steps[rs.RelSource]
|
||||
if rs.rel == nil {
|
||||
return fmt.Errorf("aggregate: missing source relation %s", rs.RelSource)
|
||||
}
|
||||
|
||||
case *Join:
|
||||
rs.relLeft = steps[rs.RelLeft]
|
||||
rs.relRight = steps[rs.RelRight]
|
||||
if rs.relLeft == nil {
|
||||
return fmt.Errorf("join: missing left relation %s", rs.relLeft)
|
||||
}
|
||||
if rs.relRight == nil {
|
||||
return fmt.Errorf("join: missing right relation %s", rs.relRight)
|
||||
}
|
||||
|
||||
case *Link:
|
||||
rs.relLeft = steps[rs.RelLeft]
|
||||
rs.relRight = steps[rs.RelRight]
|
||||
if rs.relLeft == nil {
|
||||
return fmt.Errorf("link: missing left relation %s", rs.relLeft)
|
||||
}
|
||||
if rs.relRight == nil {
|
||||
return fmt.Errorf("link: missing right relation %s", rs.relRight)
|
||||
}
|
||||
}
|
||||
}
|
||||
return
|
||||
}()
|
||||
if err != nil {
|
||||
return fmt.Errorf("unable to link steps: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Analyze runs analysis over each step in the pipeline
|
||||
//
|
||||
// Step analysis hints to the optimizers as to how expensive specific operations
|
||||
// are and the general dataset size involved.
|
||||
func (pp Pipeline) Analyze(ctx context.Context) (err error) {
|
||||
for _, p := range pp {
|
||||
err = p.Analyze(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Optimize runs all optimization and returns an optimized pipeline
|
||||
func (base Pipeline) Optimize(ctx context.Context) (optimized Pipeline, err error) {
|
||||
base, err = base.OptimizeStructure(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
return base.OptimizeSteps(ctx)
|
||||
}
|
||||
|
||||
// OptimizeStructure performs general pipeline structure optimizations such as
|
||||
// restructuring and clobbering steps onto the datasource layer
|
||||
func (base Pipeline) OptimizeStructure(ctx context.Context) (optimized Pipeline, err error) {
|
||||
optimized = base.Clone()
|
||||
|
||||
return optimized, optimized.walkSubtrees(optimized.root(), func(step PipelineStep, isRoot bool) (out PipelineStep, err error) {
|
||||
out = step
|
||||
for _, opt := range pipelineOptimizers {
|
||||
out, err = opt(out, isRoot)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return
|
||||
})
|
||||
}
|
||||
|
||||
// OptimizeSteps performs step specific optimizations such as pushing filters
|
||||
// on the lower levels, determining step-specific plans, ...
|
||||
func (base Pipeline) OptimizeSteps(ctx context.Context) (optimized Pipeline, err error) {
|
||||
optimized = base.Clone()
|
||||
return optimized, optimized.optimizeSteps(base.root(), internalFilter{})
|
||||
}
|
||||
|
||||
// Iterator constructs and returns an iterator based on the pipeline
|
||||
func (pp Pipeline) Iterator(ctx context.Context) (out Iterator, err error) {
|
||||
return pp.iterator(ctx, pp.root())
|
||||
}
|
||||
|
||||
// // // // // // // // // // // // // // // // // // // // // // // // //
|
||||
// Utilities
|
||||
|
||||
// iterator is the recursive counterpart to the .Iterator method
|
||||
func (p Pipeline) iterator(ctx context.Context, s PipelineStep) (it Iterator, err error) {
|
||||
switch s := s.(type) {
|
||||
case *Datasource:
|
||||
return s.Initialize(ctx)
|
||||
|
||||
case *Aggregate:
|
||||
it, err = p.iterator(ctx, s.rel)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
return s.Initialize(ctx, it)
|
||||
|
||||
case *Join:
|
||||
var left Iterator
|
||||
var right Iterator
|
||||
left, err = p.iterator(ctx, s.relLeft)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
right, err = p.iterator(ctx, s.relRight)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
return s.Initialize(ctx, left, right)
|
||||
|
||||
case *Link:
|
||||
var left Iterator
|
||||
var right Iterator
|
||||
left, err = p.iterator(ctx, s.relLeft)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
right, err = p.iterator(ctx, s.relRight)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
return s.Initialize(ctx, left, right)
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("unsupported step")
|
||||
}
|
||||
|
||||
func (pp Pipeline) root() PipelineStep {
|
||||
// @todo this is not ok; we need to analyze the tree, but I was lazy :seenoevil:
|
||||
return pp[len(pp)-1]
|
||||
}
|
||||
|
||||
// optimizeSteps is the recursive counterpart to the .OptimizeSteps method
|
||||
func (p Pipeline) optimizeSteps(node PipelineStep, inF internalFilter) (err error) {
|
||||
switch n := node.(type) {
|
||||
case *Datasource:
|
||||
inF, err = n.Optimize(inF)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if !inF.empty() {
|
||||
return fmt.Errorf("a datasource can not offload optimizations")
|
||||
}
|
||||
return
|
||||
|
||||
case *Aggregate:
|
||||
inF, err = n.Optimize(inF)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
return p.optimizeSteps(n.rel, inF)
|
||||
|
||||
case *Join:
|
||||
inF, err = n.Optimize(inF)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = p.optimizeSteps(n.relLeft, inF)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = p.optimizeSteps(n.relRight, inF)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
case *Link:
|
||||
inF, err = n.Optimize(inF)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = p.optimizeSteps(n.relLeft, inF)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = p.optimizeSteps(n.relRight, inF)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// walkSubtrees performs a DFS and invokes fn for every sub-tree node in the returning order
|
||||
func (p Pipeline) walkSubtrees(root PipelineStep, fn func(step PipelineStep, isRoot bool) (PipelineStep, error)) (err error) {
|
||||
err = p.walkSubtreesRec(root, true, fn)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// dfsRec reports only subtrees; in case no sub tree was there, do it
|
||||
switch root.(type) {
|
||||
case *Join, *Link:
|
||||
n, err := fn(root, true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.replace(root, n)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// walkSubtreesRec is the recursive counterpart to the .walkSubtrees method
|
||||
func (p Pipeline) walkSubtreesRec(root PipelineStep, isRoot bool, fn func(step PipelineStep, isRoot bool) (PipelineStep, error)) (err error) {
|
||||
var n PipelineStep
|
||||
|
||||
switch s := root.(type) {
|
||||
case *Datasource:
|
||||
// this one doesn't have anything under it
|
||||
return
|
||||
|
||||
case *Aggregate:
|
||||
return p.walkSubtreesRec(s.rel, false, fn)
|
||||
|
||||
case *Join:
|
||||
err = p.walkSubtreesRec(s.relLeft, false, fn)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = p.walkSubtreesRec(s.relRight, false, fn)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
n, err = fn(root, isRoot)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
p.replace(root, n)
|
||||
|
||||
case *Link:
|
||||
err = p.walkSubtreesRec(s.relLeft, false, fn)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = p.walkSubtreesRec(s.relRight, false, fn)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
n, err = fn(root, isRoot)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
p.replace(root, n)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (pp Pipeline) replace(o, n PipelineStep) {
|
||||
for i, p := range pp {
|
||||
if p == o {
|
||||
pp[i] = n
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p Pipeline) Clone() (out Pipeline) {
|
||||
out = make(Pipeline, 0, len(p))
|
||||
for _, s := range p {
|
||||
switch s := s.(type) {
|
||||
case *Aggregate:
|
||||
aux := *s
|
||||
out = append(out, &aux)
|
||||
|
||||
case *Join:
|
||||
aux := *s
|
||||
out = append(out, &aux)
|
||||
|
||||
case *Link:
|
||||
aux := *s
|
||||
out = append(out, &aux)
|
||||
|
||||
default:
|
||||
panic("unsupported step")
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
Reference in New Issue
Block a user