diff --git a/pkg/dal/def_aggregate.go b/pkg/dal/def_aggregate.go index 6241d846d..12eb4f0d8 100644 --- a/pkg/dal/def_aggregate.go +++ b/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 diff --git a/pkg/dal/def_datasource.go b/pkg/dal/def_datasource.go index f30379920..e358a2565 100644 --- a/pkg/dal/def_datasource.go +++ b/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 +} diff --git a/pkg/dal/def_join.go b/pkg/dal/def_join.go index 81e186aa7..6d0fd6eab 100644 --- a/pkg/dal/def_join.go +++ b/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 } diff --git a/pkg/dal/def_link.go b/pkg/dal/def_link.go index b9a28e4a0..de858f70b 100644 --- a/pkg/dal/def_link.go +++ b/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 } diff --git a/pkg/dal/driver.go b/pkg/dal/driver.go index 3322905f7..7327c073b 100644 --- a/pkg/dal/driver.go +++ b/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 diff --git a/pkg/dal/optimization.go b/pkg/dal/optimization.go index 51ad41fce..7226c7ac7 100644 --- a/pkg/dal/optimization.go +++ b/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 +} diff --git a/pkg/dal/optimization_clobber.go b/pkg/dal/optimization_clobber.go new file mode 100644 index 000000000..b6865582c --- /dev/null +++ b/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 +} diff --git a/pkg/dal/optimization_clobber_test.go b/pkg/dal/optimization_clobber_test.go new file mode 100644 index 000000000..f27165b0a --- /dev/null +++ b/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) + }) +} diff --git a/pkg/dal/pipeline.go b/pkg/dal/pipeline.go index 77e62d7bb..c28b60c93 100644 --- a/pkg/dal/pipeline.go +++ b/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 -} diff --git a/pkg/dal/service.go b/pkg/dal/service.go index 895229f3e..4574f17b2 100644 --- a/pkg/dal/service.go +++ b/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 +} diff --git a/store/adapters/rdbms/dal/connection.go b/store/adapters/rdbms/dal/connection.go index 9afe5d777..effb6a469 100644 --- a/store/adapters/rdbms/dal/connection.go +++ b/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)