Add base support for offloading pipeline steps to the data source level

pkg/dal/def_aggregate.go

@@ -24,7 +24,7 @@ type (
 
 		rel      PipelineStep
 		plan     aggregatePlan
-		analysis stepAnalysis
+		analysis map[string]OpAnalysis
 	}
 
 	// aggregatePlan outlines how the optimizer determined the dataset should be
@@ -77,17 +77,19 @@ func (def *Aggregate) Attributes() [][]AttributeMapping {
 
 func (def *Aggregate) Analyze(ctx context.Context) (err error) {
 	// @todo proper analysis; for now we'll leave this as defaults
-	def.analysis = stepAnalysis{
-		scanCost:   costUnknown,
-		searchCost: costUnknown,
-		filterCost: costUnknown,
-		sortCost:   costUnknown,
-		outputSize: sizeUnknown,
+	def.analysis = map[string]OpAnalysis{
+		OpAnalysisIterate: {
+			ScanCost:   CostUnknown,
+			SearchCost: CostUnknown,
+			FilterCost: CostUnknown,
+			SortCost:   CostUnknown,
+			OutputSize: SizeUnknown,
+		},
 	}
 	return
 }
 
-func (def *Aggregate) Analysis() stepAnalysis {
+func (def *Aggregate) Analysis() map[string]OpAnalysis {
 	return def.analysis
 }
 
@@ -183,6 +185,8 @@ func (def *Aggregate) init(ctx context.Context, src Iterator) (exec *aggregate,
 	}
 	// - aggregates
 	for i, attr := range def.OutAttributes {
+		// @todo change when needed; currently, all aggregates are numbers
+		attr.Type = &TypeNumber{}
 		attr, err = prepAttr(attr)
 		if err != nil {
 			return

pkg/dal/def_datasource.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/cortezaproject/corteza-server/pkg/expr"
 	"github.com/cortezaproject/corteza-server/pkg/filter"
+	"github.com/cortezaproject/corteza-server/pkg/ql"
 )
 
 type (
@@ -18,13 +20,21 @@ type (
 
 		OutAttributes []AttributeMapping
 
-		analysis stepAnalysis
+		analysis   map[string]OpAnalysis
+		connection *ConnectionWrap
+		model      *Model
+
+		// clobbered lists all of the steps that are offloaded into the datasource.
+		// The list is provided in order; the first step is the first step which should execute
+		// in the report pipeline.
+		//
+		// @todo change to generic step; doing this because I can for now
+		clobbered []*Aggregate
 
 		// provided in the init step so we can omit some code in the exec step
+		// @todo consider removing this
 		auxIter Iterator
 	}
-
-	iterProvider func(ctx context.Context, mf ModelRef, f filter.Filter) (iter Iterator, model *Model, err error)
 )
 
 func (def *Datasource) Identifier() string {
@@ -36,22 +46,40 @@ func (def *Datasource) Sources() []string {
 }
 
 func (def *Datasource) Attributes() [][]AttributeMapping {
-	return [][]AttributeMapping{def.OutAttributes}
+	if len(def.clobbered) > 0 {
+		return def.offloadedAttributes()
+	}
+
+	return def.ownAttributes()
 }
 
 func (def *Datasource) Analyze(ctx context.Context) (err error) {
-	// @todo probe datasource; for now, RDBMS only so all is cheap
-	def.analysis = stepAnalysis{
-		scanCost:   costUnknown,
-		searchCost: costUnknown,
-		filterCost: costUnknown,
-		sortCost:   costUnknown,
-		outputSize: sizeUnknown,
+	a, err := def.connection.connection.Analyze(ctx, def.model)
+	if err != nil {
+		return
 	}
+
+	def.analysis = map[string]OpAnalysis{
+		OpAnalysisIterate: {
+			ScanCost:   CostUnknown,
+			SearchCost: CostUnknown,
+			FilterCost: CostUnknown,
+			SortCost:   CostUnknown,
+			OutputSize: SizeUnknown,
+		},
+	}
+
+	if _, ok := a[OpAnalysisAggregate]; ok {
+		def.analysis[OpAnalysisAggregate] = a[OpAnalysisAggregate]
+	}
+	if _, ok := a[OpAnalysisJoin]; ok {
+		def.analysis[OpAnalysisJoin] = a[OpAnalysisJoin]
+	}
+
 	return
 }
 
-func (def *Datasource) Analysis() stepAnalysis {
+func (def *Datasource) Analysis() map[string]OpAnalysis {
 	return def.analysis
 }
 
@@ -59,7 +87,14 @@ func (def *Datasource) Optimize(req internalFilter) (res internalFilter, err err
 	return internalFilter{}, fmt.Errorf("optimization not implemented")
 }
 
-func (def *Datasource) init(ctx context.Context, s iterProvider) (err error) {
+func (def *Datasource) init(ctx context.Context) (err error) {
+	if def.model == nil {
+		return fmt.Errorf("cannot initialize datasource: model not set")
+	}
+	if def.connection == nil {
+		return fmt.Errorf("cannot initialize datasource: connection not set")
+	}
+
 	if def.Filter != nil {
 		def.filter, err = toInternalFilter(def.Filter)
 		if err != nil {
@@ -67,17 +102,8 @@ func (def *Datasource) init(ctx context.Context, s iterProvider) (err error) {
 		}
 	}
 
-	// Do the first init to get the model.
-	// For now, this is a ~free operation, but it should change when we allow things
-	// like nesting reports, etc.
-	// @todo refactor datasource descriptors to avoid this
-	_, model, err := s(ctx, def.ModelRef, def.filter)
-	if err != nil {
-		return
-	}
-
 	if len(def.OutAttributes) == 0 {
-		def.OutAttributes = def.outAttrsFromModel(model)
+		def.OutAttributes = def.outAttrsFromModel(def.model)
 	}
 
 	pp := make([]string, 0, len(def.OutAttributes)/2+1)
@@ -94,18 +120,36 @@ func (def *Datasource) init(ctx context.Context, s iterProvider) (err error) {
 		return
 	}
 
-	// Get the iterator for actual use
-	def.auxIter, _, err = s(ctx, def.ModelRef, def.filter)
-	if err != nil {
-		return
-	}
-
+	// Firstly validate the base
 	err = def.validate()
 	if err != nil {
 		return
 	}
 
-	return nil
+	if len(def.clobbered) > 0 {
+		// @todo currently, we can only do one; change this when we tweak the DB
+		//       offloading code.
+		ag := def.clobbered[0]
+
+		// Invoke the aggregation's init to perform the validation and preparation logic
+		var wa *aggregate
+		wa, err = ag.init(ctx, nil)
+		if err != nil {
+			return
+		}
+
+		// Preprocess the filters to conform to connection API
+		f, having, err := def.getAggregationFilters(def.filter, wa.filter)
+		if err != nil {
+			return err
+		}
+
+		def.auxIter, err = def.connection.connection.Aggregate(ctx, def.model, f, wa.groupDefs, wa.aggregateDefs, having)
+		return err
+	}
+
+	def.auxIter, err = def.connection.connection.Search(ctx, def.model, def.filter)
+	return err
 }
 
 func (def *Datasource) exec(ctx context.Context) (out Iterator, err error) {
@@ -141,13 +185,178 @@ func (def *Datasource) outAttrsFromModel(model *Model) (attrs []AttributeMapping
 			Ident: attr.Ident,
 			Src:   attr.Ident,
 			Props: MapProperties{
-				IsPrimary: attr.PrimaryKey,
-				IsSystem:  attr.System,
-				Nullable:  attr.Type.IsNullable(),
-				Type:      attr.Type,
+				IsPrimary:    attr.PrimaryKey,
+				IsSystem:     attr.System,
+				IsFilterable: attr.Filterable,
+				IsSortable:   attr.Sortable,
+				Nullable:     attr.Type.IsNullable(),
+				Type:         attr.Type,
 			},
 		})
 	}
 
 	return
 }
+
+func (def *Datasource) shouldClobberAggregation() bool {
+	costs, ok := def.analysis[OpAnalysisAggregate]
+	if !ok {
+		return false
+	}
+
+	// @todo more to it; check and compare costs; for now we know that all rdbms
+	//       offloads will be faster.
+
+	_ = costs
+
+	return true
+}
+
+func (def *Datasource) ownAttributes() [][]AttributeMapping {
+	return [][]AttributeMapping{def.OutAttributes}
+}
+
+func (def *Datasource) offloadedAttributes() [][]AttributeMapping {
+	// The last offloaded step's attributes is what this step's iterator returns
+	return def.clobbered[len(def.clobbered)-1].Attributes()
+}
+
+func (def *Datasource) clobber(s PipelineStep) (ok bool) {
+	switch cs := s.(type) {
+	case *Aggregate:
+		if !def.shouldClobberAggregation() {
+			return false
+		}
+
+		def.clobbered = append(def.clobbered, cs)
+		return true
+	}
+
+	return
+}
+
+// getAggregationFilters transforms the base and the aggregation filters to conform
+// to the store/dal's model.Aggregate API
+//
+// - The full filter.Filter return parameter is applied to the base dataset
+// - The QL node return parameter is applied only to the aggregated dataset
+//
+// @todo should we change the store's API to accept 2x filter.Filter? I think
+//       that would allow the underlaying driver to decide how to handle them
+//       instead of relying on what SQL does.
+func (def *Datasource) getAggregationFilters(base, agg internalFilter) (filter internalFilter, having *ql.ASTNode, err error) {
+	var typedV expr.TypedValue
+	filter = base
+
+	// Move the aggregation's order and limit to the base filter because those
+	// are applied to the final output (per SQL)
+	//
+	// Leave base filtering parameters (filter, constraints, ...) as those are apploed
+	// to the dataset before aggregation (which is correct).
+	filter.orderBy = agg.orderBy
+	filter.limit = agg.limit
+
+	// Convert the rest of the filtering parameters defined on the aggregation's filter
+	// to the QL node.
+	// ALl of that is applied after the aggregation which is correct.
+	var nConstraints *ql.ASTNode
+	if len(agg.constraints) > 0 {
+		nConstraints = &ql.ASTNode{
+			Ref: "and",
+		}
+		for k, c := range agg.constraints {
+			arg := &ql.ASTNode{
+				Ref: "or",
+			}
+
+			for _, v := range c {
+				typedV, err = expr.Typify(v)
+				if err != nil {
+					return
+				}
+				arg.Args = append(arg.Args, &ql.ASTNode{
+					Symbol: k,
+					Value:  ql.WrapValue(typedV),
+				})
+			}
+
+			nConstraints.Args = append(nConstraints.Args, &ql.ASTNode{
+				Ref:  "group",
+				Args: ql.ASTNodeSet{arg},
+			})
+		}
+	}
+
+	var nStateConstraints *ql.ASTNode
+	if len(agg.stateConstraints) > 0 {
+		nStateConstraints = &ql.ASTNode{
+			Ref: "and",
+		}
+		for k, c := range agg.stateConstraints {
+			typedV, err = expr.Typify(c)
+			if err != nil {
+				return
+			}
+
+			nStateConstraints.Args = append(nStateConstraints.Args, &ql.ASTNode{
+				Symbol: k,
+				Value:  ql.WrapValue(typedV),
+			})
+		}
+	}
+
+	var nMetaConstraints *ql.ASTNode
+	if len(agg.stateConstraints) > 0 {
+		nMetaConstraints = &ql.ASTNode{
+			Ref: "and",
+		}
+		for k, c := range agg.stateConstraints {
+			typedV, err = expr.Typify(c)
+			if err != nil {
+				return
+			}
+
+			nMetaConstraints.Args = append(nMetaConstraints.Args, &ql.ASTNode{
+				Symbol: k,
+				Value:  ql.WrapValue(typedV),
+			})
+		}
+	}
+
+	nExpression := agg.expParsed
+
+	var nCursor *ql.ASTNode
+	if agg.cursor != nil {
+		nCursor, err = agg.cursor.ToAST(nil, nil)
+		if err != nil {
+			return
+		}
+	}
+
+	having = &ql.ASTNode{
+		Ref: "and",
+	}
+
+	if nConstraints != nil {
+		having.Args = append(having.Args, nConstraints)
+	}
+	if nStateConstraints != nil {
+		having.Args = append(having.Args, nStateConstraints)
+	}
+	if nMetaConstraints != nil {
+		having.Args = append(having.Args, nMetaConstraints)
+	}
+	if nExpression != nil {
+		having.Args = append(having.Args, nExpression)
+	}
+	if nCursor != nil {
+		having.Args = append(having.Args, nCursor)
+	}
+
+	// In case everything is empty, no need for the having part
+	if len(having.Ref) == 0 {
+		having = nil
+	}
+
+	return
+}

pkg/dal/def_join.go

@@ -30,7 +30,7 @@ type (
 		relLeft  PipelineStep
 		relRight PipelineStep
 		plan     joinPlan
-		analysis stepAnalysis
+		analysis map[string]OpAnalysis
 	}
 
 	// JoinPredicate determines the attributes the two datasets should get joined on
@@ -64,17 +64,21 @@ func (def *Join) Attributes() [][]AttributeMapping {
 }
 
 func (def *Join) Analyze(ctx context.Context) (err error) {
-	def.analysis = stepAnalysis{
-		scanCost:   costUnknown,
-		searchCost: costUnknown,
-		filterCost: costUnknown,
-		sortCost:   costUnknown,
-		outputSize: sizeUnknown,
+	// @todo proper analysis; for now we'll leave this as defaults
+	def.analysis = map[string]OpAnalysis{
+		OpAnalysisIterate: {
+			ScanCost:   CostUnknown,
+			SearchCost: CostUnknown,
+			FilterCost: CostUnknown,
+			SortCost:   CostUnknown,
+			OutputSize: SizeUnknown,
+		},
 	}
+
 	return
 }
 
-func (def *Join) Analysis() stepAnalysis {
+func (def *Join) Analysis() map[string]OpAnalysis {
 	return def.analysis
 }
 

pkg/dal/def_link.go

@@ -33,7 +33,7 @@ type (
 		relRight PipelineStep
 
 		plan     linkPlan
-		analysis stepAnalysis
+		analysis map[string]OpAnalysis
 	}
 
 	// LinkPredicate determines the attributes the two datasets should get joined on
@@ -75,17 +75,21 @@ func (def *Link) Attributes() [][]AttributeMapping {
 }
 
 func (def *Link) Analyze(ctx context.Context) (err error) {
-	def.analysis = stepAnalysis{
-		scanCost:   costUnknown,
-		searchCost: costUnknown,
-		filterCost: costUnknown,
-		sortCost:   costUnknown,
-		outputSize: sizeUnknown,
+	// @todo proper analysis; for now we'll leave this as defaults
+	def.analysis = map[string]OpAnalysis{
+		OpAnalysisIterate: {
+			ScanCost:   CostUnknown,
+			SearchCost: CostUnknown,
+			FilterCost: CostUnknown,
+			SortCost:   CostUnknown,
+			OutputSize: SizeUnknown,
+		},
 	}
+
 	return
 }
 
-func (def *Link) Analysis() stepAnalysis {
+func (def *Link) Analysis() map[string]OpAnalysis {
 	return def.analysis
 }
 

pkg/dal/driver.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/cortezaproject/corteza-server/pkg/expr"
 	"github.com/cortezaproject/corteza-server/pkg/filter"
+	"github.com/cortezaproject/corteza-server/pkg/ql"
 	"go.uber.org/zap"
 )
 
@@ -49,6 +50,12 @@ type (
 		// Search returns an iterator which can be used to access all if the bits
 		Search(context.Context, *Model, filter.Filter) (Iterator, error)
 
+		// Analyze returns the operation analysis the connection can perform for the model
+		Analyze(ctx context.Context, m *Model) (map[string]OpAnalysis, error)
+
+		// Aggregate returns the iterator with aggregated data from the base model
+		Aggregate(ctx context.Context, m *Model, f filter.Filter, groupBy []AggregateAttr, aggrExpr []AggregateAttr, having *ql.ASTNode) (i Iterator, _ error)
+
 		// Delete deletes the given value
 		Delete(ctx context.Context, m *Model, pkv ValueGetter) error
 

pkg/dal/optimization.go

@@ -1,43 +1,148 @@
 package dal
 
+import "fmt"
+
 type (
 	// opCost provides a general idea of expensive an operation is for a
 	// specific pipeline step.
 	//
 	// dsSize provides a general idea of how large an underlaying dataset is.
-	opCost       int
-	dsSize       int
-	stepAnalysis struct {
-		scanCost   opCost
-		searchCost opCost
-		filterCost opCost
-		sortCost   opCost
+	opCost int
+	dsSize int
 
-		outputSize dsSize
+	OpAnalysis struct {
+		ScanCost   opCost
+		SearchCost opCost
+		FilterCost opCost
+		SortCost   opCost
+
+		OutputSize dsSize
+	}
+
+	ppStepWrap struct {
+		step PipelineStep
+
+		parent *ppStepWrap
+		child  []*ppStepWrap
 	}
 )
 
 var (
-	pipelineOptimizers = []func(PipelineStep, bool) (PipelineStep, error){
-		// @todo add more optimizers
+	pipelineOptimizers = []func(Pipeline) (Pipeline, error){
+		pipelineClobberSteps,
 	}
 )
 
 const (
 	// operation computation indicators
-	costUnknown opCost = iota
-	costFree
-	costCheep
-	costAcceptable
-	costExpensive
-	costInfinite
+	CostUnknown opCost = iota
+	CostFree
+	CostCheep
+	CostAcceptable
+	CostExpensive
+	CostInfinite
 )
 
 const (
 	// dataset size indicators
-	sizeUnknown dsSize = iota
-	sizeTiny
-	sizeSmall
-	sizeMedium
-	sizeLarge
+	SizeUnknown dsSize = iota
+	SizeTiny
+	SizeSmall
+	SizeMedium
+	SizeLarge
 )
+
+const (
+	OpAnalysisIterate   string = "iterate"
+	OpAnalysisAggregate string = "aggregate"
+	OpAnalysisJoin      string = "join"
+)
+
+// wrapPpSteps wraps the pipeline steps in a more processing friendly format
+// and returns a slice of leave nodes
+//
+// @todo the pipeline representation might change which will make this obsolete
+func wrapPpSteps(pp Pipeline) (leaves []*ppStepWrap) {
+	ix := make(map[PipelineStep]*ppStepWrap)
+
+	for _, p := range pp {
+		ix[p] = &ppStepWrap{step: p}
+	}
+
+	for _, p := range pp {
+		switch c := p.(type) {
+		case *Aggregate:
+			ix[p].child = append(ix[p].child, ix[c.rel])
+			ix[c.rel].parent = ix[p]
+
+		case *Join:
+			ix[p].child = append(ix[p].child, ix[c.relLeft])
+			ix[c.relLeft].parent = ix[p]
+
+			ix[p].child = append(ix[p].child, ix[c.relRight])
+			ix[c.relRight].parent = ix[p]
+
+		case *Link:
+			ix[p].child = append(ix[p].child, ix[c.relLeft])
+			ix[c.relLeft].parent = ix[p]
+
+			ix[p].child = append(ix[p].child, ix[c.relRight])
+			ix[c.relRight].parent = ix[p]
+
+		case *Datasource:
+			continue
+
+		default:
+			panic(fmt.Errorf("impossible state: unknown pipeline step type %v", p))
+		}
+	}
+
+	for _, a := range ix {
+		if len(a.child) == 0 {
+			leaves = append(leaves, a)
+		}
+	}
+
+	return
+}
+
+// unwrapPpSteps unwraps the wrapped pipeline step into the classic Pipeline
+//
+// @todo the pipeline representation might change which will make this obsolete
+func unwrapPpSteps(n *ppStepWrap, seen map[*ppStepWrap]bool) (out Pipeline) {
+	for i, c := range n.child {
+		switch s := n.step.(type) {
+		case *Aggregate:
+			s.rel = c.step
+			s.RelSource = c.step.Identifier()
+		case *Join:
+			if i == 0 {
+				s.relLeft = c.step
+				s.RelLeft = c.step.Identifier()
+			} else {
+				s.relRight = c.step
+				s.RelRight = c.step.Identifier()
+			}
+		case *Link:
+			if i == 0 {
+				s.relLeft = c.step
+				s.RelLeft = c.step.Identifier()
+			} else {
+				s.relRight = c.step
+				s.RelRight = c.step.Identifier()
+			}
+		}
+	}
+
+	// @todo potentially cancancel sooner
+	if !seen[n] {
+		out = append(out, n.step)
+	}
+	seen[n] = true
+
+	if n.parent != nil {
+		out = append(unwrapPpSteps(n.parent, seen), out...)
+	}
+
+	return
+}

pkg/dal/optimization_clobber.go

@@ -0,0 +1,69 @@
+package dal
+
+// pipelineClobberSteps tries to reduce the amount of pipeline steps by offloading
+// higher operations to the lower levels
+//
+// As an example; the aggregation can be offloaded to the database for faster
+// execution.
+func pipelineClobberSteps(in Pipeline) (Pipeline, error) {
+	if len(in) <= 1 {
+		// Can't optimize further :upsidedownface:
+		return in, nil
+	}
+
+	// Outline
+	// - get a nicer pipeline representation
+	// 	 @todo make the pipeline nicer in the first place
+	// - traverse from the lief nodes up; try to clobber if the lief node allows it
+	//
+	// The clobbering for a branch ends when there is a node that can't be clobbered.
+	// The progression ends because application level nodes can't be offloaded to.
+	ll := wrapPpSteps(in)
+	for _, l := range ll {
+		for {
+			// When there is no parent, we can't progress further
+			if l.parent == nil {
+				break
+			}
+
+			// if step can't clobber, skip
+			cs, ok := l.step.(clobberableStep)
+			if !ok {
+				break
+			}
+
+			// if child fails to clobber parent, skip to the next child
+			// @note for now we can end the clobbering if any of the steps
+			//       can't be clobbered as all of the application defined steps
+			//       are focused on the single op. and can't do anything else.
+			if !cs.clobber(l.parent.step) {
+				break
+			}
+
+			// if clobbered successfully, update references
+			if l.parent != nil && l.parent.parent != nil {
+				// - update child ref of the parent's parent
+				for i, c := range l.parent.parent.child {
+					if c == l.parent {
+						l.parent.parent.child[i] = l
+					}
+				}
+			}
+
+			l.parent = l.parent.parent
+
+			// @todo for now, clobbering ends after one successfull instance; this is due
+			//       to the current DB implementation doesn't allow nested things.
+			break
+		}
+	}
+
+	// convert back to the pipeline representation; update deps in the process
+	out := make(Pipeline, 0, len(in))
+	seen := make(map[*ppStepWrap]bool, len(in))
+	for _, l := range ll {
+		out = append(out, unwrapPpSteps(l, seen)...)
+	}
+
+	return out, nil
+}

pkg/dal/optimization_clobber_test.go

@@ -0,0 +1,179 @@
+package dal
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func makeAnalysisDsAggregate() map[string]OpAnalysis {
+	return map[string]OpAnalysis{
+		OpAnalysisAggregate: {},
+	}
+}
+
+func TestClobberStep(t *testing.T) {
+	t.Run("no steps", func(t *testing.T) {
+		out, err := pipelineClobberSteps(nil)
+
+		require.NoError(t, err)
+		require.Nil(t, out)
+	})
+
+	t.Run("one step no optimize", func(t *testing.T) {
+		out, err := pipelineClobberSteps(Pipeline{&Datasource{}})
+
+		require.NoError(t, err)
+		require.Len(t, out, 1)
+	})
+
+	t.Run("agg ds", func(t *testing.T) {
+		ds := &Datasource{
+			Ident:    "ds_1",
+			analysis: makeAnalysisDsAggregate(),
+		}
+		agg := &Aggregate{
+			Ident:     "agg_1",
+			RelSource: "ds_1",
+			rel:       ds,
+		}
+
+		out, err := pipelineClobberSteps(Pipeline{agg, ds})
+
+		require.NoError(t, err)
+		require.Len(t, out, 1)
+		c := out[0].(*Datasource)
+		require.Len(t, c.clobbered, 1)
+	})
+
+	t.Run("agg agg ds", func(t *testing.T) {
+		// @note for now we're only offloading one aggregation
+		ds := &Datasource{
+			Ident:    "ds_1",
+			analysis: makeAnalysisDsAggregate(),
+		}
+		agg1 := &Aggregate{
+			Ident:     "agg_1",
+			RelSource: "ds_1",
+			rel:       ds,
+		}
+		agg2 := &Aggregate{
+			Ident:     "agg_2",
+			RelSource: "agg_1",
+			rel:       agg1,
+		}
+
+		out, err := pipelineClobberSteps(Pipeline{agg2, agg1, ds})
+
+		require.NoError(t, err)
+		require.Len(t, out, 2)
+		c := out[1].(*Datasource)
+		require.Len(t, c.clobbered, 1)
+	})
+
+	t.Run("join ds ds", func(t *testing.T) {
+		ds1 := &Datasource{
+			Ident:    "ds_1",
+			analysis: makeAnalysisDsAggregate(),
+		}
+		ds2 := &Datasource{
+			Ident:    "ds_2",
+			analysis: makeAnalysisDsAggregate(),
+		}
+
+		join := &Join{
+			Ident:    "join_1",
+			RelLeft:  "ds_1",
+			RelRight: "ds_2",
+
+			relLeft:  ds1,
+			relRight: ds2,
+		}
+
+		out, err := pipelineClobberSteps(Pipeline{join, ds1, ds2})
+
+		require.NoError(t, err)
+		require.Len(t, out, 3)
+	})
+
+	t.Run("join join ds ds ds", func(t *testing.T) {
+		ds1 := &Datasource{
+			Ident:    "ds_1",
+			analysis: makeAnalysisDsAggregate(),
+		}
+		ds2 := &Datasource{
+			Ident:    "ds_2",
+			analysis: makeAnalysisDsAggregate(),
+		}
+		ds3 := &Datasource{
+			Ident:    "ds_3",
+			analysis: makeAnalysisDsAggregate(),
+		}
+
+		join1 := &Join{
+			Ident:    "join_1",
+			RelLeft:  "ds_1",
+			RelRight: "ds_2",
+
+			relLeft:  ds1,
+			relRight: ds2,
+		}
+		join2 := &Join{
+			Ident:    "join_2",
+			RelLeft:  "join_1",
+			RelRight: "ds_3",
+
+			relLeft:  join1,
+			relRight: ds3,
+		}
+
+		out, err := pipelineClobberSteps(Pipeline{join2, join1, ds1, ds2, ds3})
+
+		require.NoError(t, err)
+		require.Len(t, out, 5)
+	})
+
+	t.Run("join join agg ds ds ds", func(t *testing.T) {
+		ds1 := &Datasource{
+			Ident:    "ds_1",
+			analysis: makeAnalysisDsAggregate(),
+		}
+		agg := &Aggregate{
+			Ident:     "agg_1",
+			RelSource: "ds_1",
+			rel:       ds1,
+		}
+		ds2 := &Datasource{
+			Ident:    "ds_2",
+			analysis: makeAnalysisDsAggregate(),
+		}
+		ds3 := &Datasource{
+			Ident:    "ds_3",
+			analysis: makeAnalysisDsAggregate(),
+		}
+
+		join1 := &Join{
+			Ident:    "join_1",
+			RelLeft:  "agg_1",
+			RelRight: "ds_2",
+
+			relLeft:  agg,
+			relRight: ds2,
+		}
+		join2 := &Join{
+			Ident:    "join_2",
+			RelLeft:  "join_1",
+			RelRight: "ds_3",
+
+			relLeft:  join1,
+			relRight: ds3,
+		}
+
+		out, err := pipelineClobberSteps(Pipeline{join2, join1, agg, ds1, ds2, ds3})
+
+		require.NoError(t, err)
+		require.Len(t, out, 5)
+
+		require.Len(t, ds1.clobbered, 1)
+	})
+}

pkg/dal/pipeline.go

@@ -24,7 +24,13 @@ type (
 		Attributes() [][]AttributeMapping
 
 		Analyze(ctx context.Context) error
-		Analysis() stepAnalysis
+		Analysis() map[string]OpAnalysis
+	}
+
+	// clobberableStep can have other steps offload their work to it.
+	// This is primarily used to offload the work to the database.
+	clobberableStep interface {
+		clobber(PipelineStep) bool
 	}
 
 	// Attribute mapping outlines specific attributes within a pipeline
@@ -38,11 +44,13 @@ type (
 
 	// MapProperties describe the attribute such as it's type and constraints
 	MapProperties struct {
-		Label     string
-		IsPrimary bool
-		IsSystem  bool
-		Nullable  bool
-		Type      Type
+		Label        string
+		IsPrimary    bool
+		IsSystem     bool
+		IsFilterable bool
+		IsSortable   bool
+		Nullable     bool
+		Type         Type
 	}
 )
 
@@ -96,56 +104,6 @@ func (pp Pipeline) LinkSteps() (err error) {
 	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{})
-}
-
-// // // // // // // // // // // // // // // // // // // // // // // // //
-// Utilities
-
 func (pp Pipeline) root() PipelineStep {
 	ix := make(map[string]PipelineStep)
 
@@ -168,19 +126,16 @@ func (pp Pipeline) root() PipelineStep {
 
 // Slice returns a new pipeline with all steps of the subtree with ident as root
 func (pp Pipeline) Slice(ident string) (out Pipeline) {
-	// Make a copy so we can assure the caller can go ham over the pipeline
-	ppc := pp.Clone()
-
 	// Find root
 	var r PipelineStep
-	for _, p := range ppc {
+	for _, p := range pp {
 		if p.Identifier() == ident {
 			r = p
 			break
 		}
 	}
 
-	return ppc.slice(r)
+	return pp.slice(r)
 }
 
 // slice is the recursive counterpart for the Slice method
@@ -203,157 +158,3 @@ func (pp Pipeline) slice(s PipelineStep) (out Pipeline) {
 
 	return
 }
-
-// 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)
-
-		case *Datasource:
-			aux := *s
-			out = append(out, &aux)
-
-		default:
-			panic("unsupported step")
-		}
-	}
-	return
-}

pkg/dal/service.go

@@ -429,21 +429,24 @@ func (svc *service) Search(ctx context.Context, mf ModelRef, operations Operatio
 	return cw.connection.Search(ctx, model, f)
 }
 
-// datasource provides a way for the pipeline steps to access DAL's model iterators
-func (svc *service) datasource(ctx context.Context, mf ModelRef, f filter.Filter) (iter Iterator, mod *Model, err error) {
-	model, cw, err := svc.storeOpPrep(ctx, mf, nil)
-	if err != nil {
-		err = fmt.Errorf("cannot search data entry: %w", err)
-		return
-	}
-
-	iter, err = cw.connection.Search(ctx, model, f)
-	return iter, model, err
-}
-
 // Run returns an iterator based on the provided Pipeline
 // @todo consider moving the Search method to utilize this also
 func (svc *service) Run(ctx context.Context, pp Pipeline) (iter Iterator, err error) {
+	pp, err = svc.pipelinePrerun(ctx, pp)
+	if err != nil {
+		return
+	}
+
+	err = svc.analyzePipeline(ctx, pp)
+	if err != nil {
+		return
+	}
+
+	pp, err = svc.optimizePipeline(ctx, pp)
+	if err != nil {
+		return
+	}
+
 	return svc.run(ctx, pp.root(), false)
 }
 
@@ -452,6 +455,12 @@ func (svc *service) Run(ctx context.Context, pp Pipeline) (iter Iterator, err er
 //
 // The method is primarily used by system reports to obtain some metadata
 func (svc *service) Dryrun(ctx context.Context, pp Pipeline) (err error) {
+	// @note we don't need to do any optimization or analisis here
+	pp, err = svc.pipelinePrerun(ctx, pp)
+	if err != nil {
+		return
+	}
+
 	_, err = svc.run(ctx, pp.root(), true)
 	return
 }
@@ -460,7 +469,7 @@ func (svc *service) Dryrun(ctx context.Context, pp Pipeline) (err error) {
 func (svc *service) run(ctx context.Context, s PipelineStep, dry bool) (it Iterator, err error) {
 	switch s := s.(type) {
 	case *Datasource:
-		err = s.init(ctx, svc.datasource)
+		err = s.init(ctx)
 		if err != nil {
 			return
 		}
@@ -920,6 +929,9 @@ func (svc *service) ReplaceModelAttribute(ctx context.Context, model *Model, old
 // @note refs are primarily used for DAL pipelines where steps can reference models
 //       by handles and slugs such as module and namespace.
 func (svc *service) FindModelByRefs(connectionID uint64, refs map[string]any) *Model {
+	if connectionID == 0 {
+		connectionID = svc.defConnID
+	}
 	return svc.models[connectionID].FindByRefs(refs)
 }
 
@@ -945,6 +957,10 @@ func (svc *service) FindModelByRef(ref ModelRef) *Model {
 		connectionID = svc.defConnID
 	}
 
+	if ref.Refs != nil {
+		return svc.FindModelByRefs(connectionID, ref.Refs)
+	}
+
 	if ref.ResourceID > 0 {
 		return svc.models[connectionID].FindByResourceID(ref.ResourceID)
 	}
@@ -1100,3 +1116,81 @@ func (svc *service) validateNewSensitivityLevels(levels *sensitivityLevelIndex)
 	}
 	return
 }
+
+// pipelinePrerun performs the common operations for both the Run and Dryrun
+func (svc *service) pipelinePrerun(ctx context.Context, pp Pipeline) (_ Pipeline, err error) {
+	err = pp.LinkSteps()
+	if err != nil {
+		return
+	}
+	err = svc.bindDatasourceConnections(ctx, pp)
+	if err != nil {
+		return
+	}
+
+	return pp, nil
+}
+
+// optimizePipeline runs optimization over the given Pipeline
+func (svc *service) optimizePipeline(ctx context.Context, pp Pipeline) (_ Pipeline, err error) {
+	pp, err = svc.optimizePipelineStructure(ctx, pp)
+	if err != nil {
+		return
+	}
+
+	// @todo add step-based optimization such as filter pushdown; omitting for now
+	//       since we don't have any of it in place yet
+
+	return pp, nil
+}
+
+// optimizePipelineStructure performs general pipeline structure optimizations
+// such as restructuring and clobbering steps onto the datasource layer
+func (svc *service) optimizePipelineStructure(ctx context.Context, pp Pipeline) (_ Pipeline, err error) {
+	for _, o := range pipelineOptimizers {
+		pp, err = o(pp)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return pp, nil
+}
+
+// bindDatasourceConnections special handles datasource pipeline steps and binds
+// a DAL connection to them
+func (svc *service) bindDatasourceConnections(ctx context.Context, pp Pipeline) error {
+	for _, p := range pp {
+		ds, ok := p.(*Datasource)
+		if !ok {
+			continue
+		}
+
+		ds.model = svc.FindModelByRef(ds.ModelRef)
+		if ds.model == nil {
+			return fmt.Errorf("model %v does not exist", ds.ModelRef)
+		}
+
+		ds.connection = svc.GetConnectionByID(ds.model.ConnectionID)
+		if ds.connection == nil {
+			return fmt.Errorf("connection %d does not exist", ds.model.ConnectionID)
+		}
+	}
+
+	return nil
+}
+
+// analyzePipeline 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 (svc *service) analyzePipeline(ctx context.Context, pp Pipeline) (err error) {
+	for _, p := range pp {
+		err = p.Analyze(ctx)
+		if err != nil {
+			return
+		}
+	}
+
+	return
+}

store/adapters/rdbms/dal/connection.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/cortezaproject/corteza-server/pkg/errors"
 	"github.com/cortezaproject/corteza-server/store/adapters/rdbms/ddl"
+	"github.com/cortezaproject/corteza-server/store/adapters/rdbms/ql"
 
 	"github.com/cortezaproject/corteza-server/pkg/dal"
 	"github.com/cortezaproject/corteza-server/pkg/filter"
@@ -106,6 +107,28 @@ func (c *connection) Search(ctx context.Context, m *dal.Model, f filter.Filter)
 	})
 }
 
+func (c *connection) Analyze(ctx context.Context, m *dal.Model) (a map[string]dal.OpAnalysis, err error) {
+	// @todo somehow (probably operations) bring in the info what can be done
+	//       for now, since we're quite rigid on the drivers, this will do.
+	a = map[string]dal.OpAnalysis{
+		dal.OpAnalysisAggregate: {
+			ScanCost:   dal.CostCheep,
+			SearchCost: dal.CostCheep,
+			FilterCost: dal.CostCheep,
+			SortCost:   dal.CostCheep,
+		},
+	}
+
+	return
+}
+
+func (c *connection) Aggregate(ctx context.Context, m *dal.Model, f filter.Filter, groupBy []dal.AggregateAttr, aggrExpr []dal.AggregateAttr, having *ql.ASTNode) (i dal.Iterator, _ error) {
+	return i, c.withModel(m, func(m *model) (err error) {
+		i, err = m.Aggregate(f, groupBy, aggrExpr, having)
+		return
+	})
+}
+
 func (c *connection) Delete(ctx context.Context, m *dal.Model, pkv dal.ValueGetter) (err error) {
 	return c.withModel(m, func(m *model) error {
 		return m.Delete(ctx, pkv)