diff --git a/compose/service/record.go b/compose/service/record.go index 44a6f69d5..c087754d5 100644 --- a/compose/service/record.go +++ b/compose/service/record.go @@ -1769,22 +1769,21 @@ func loadRecord(ctx context.Context, s store.Storer, namespaceID, moduleID, reco func recordReportToDalPipeline(m *types.Module, metrics, dimensions, f string) (pp dal.Pipeline, err error) { // Map dimension to the aggregate group // @note we only ever used a single dimension so this is ok - dim := []dal.AttributeMapping{ - dal.SimpleAttr{ - Ident: "dimension_0", - Expr: dimensions, + dim := []dal.AggregateAttr{ + { + Identifier: "dimension_0", + RawExpr: dimensions, + Key: true, }, } // Map metrics to the aggregate attrs // - count is always present - mms := []dal.AttributeMapping{ - dal.SimpleAttr{ - Ident: "count", - Expr: "count(ID)", - Props: dal.MapProperties{ - Type: dal.TypeNumber{}, - }, + mms := []dal.AggregateAttr{ + { + Identifier: "count", + RawExpr: "count(ID)", + Type: dal.TypeNumber{}, }, } @@ -1797,12 +1796,10 @@ func recordReportToDalPipeline(m *types.Module, metrics, dimensions, f string) ( ident = strings.TrimSpace(pts[1]) } - mms = append(mms, dal.SimpleAttr{ - Ident: ident, - Expr: expr, - Props: dal.MapProperties{ - Type: dal.TypeNumber{}, - }, + mms = append(mms, dal.AggregateAttr{ + Identifier: ident, + RawExpr: expr, + Type: dal.TypeNumber{}, }) } diff --git a/pkg/dal/def_aggregate.go b/pkg/dal/def_aggregate.go index 3e173daf8..3a3532cac 100644 --- a/pkg/dal/def_aggregate.go +++ b/pkg/dal/def_aggregate.go @@ -17,8 +17,8 @@ type ( Filter filter.Filter filter internalFilter - Group []AttributeMapping - OutAttributes []AttributeMapping + Group []AggregateAttr + OutAttributes []AggregateAttr SourceAttributes []AttributeMapping @@ -41,11 +41,17 @@ type ( agg *aggregator } - // aggregateAttr is a simple wrapper to outline aggregated attribute definitions - aggregateAttr struct { - ident string - expr *ql.ASTNode - attrType Type + // AggregateAttr is a simple wrapper to outline aggregated attribute definitions + AggregateAttr struct { + Key bool + + // @todo change; temporary for compose service + RawExpr string + + Identifier string + Label string + Expression *ql.ASTNode + Type Type } ) @@ -58,7 +64,14 @@ func (def *Aggregate) Sources() []string { } func (def *Aggregate) Attributes() [][]AttributeMapping { - return [][]AttributeMapping{append(def.Group, def.OutAttributes...)} + aa := append(def.Group, def.OutAttributes...) + out := make([]AttributeMapping, 0, len(aa)) + + for _, a := range aa { + out = append(out, a.toSimpleAttr()) + } + + return [][]AttributeMapping{out} } func (def *Aggregate) Analyze(ctx context.Context) (err error) { @@ -126,27 +139,58 @@ func (def *Aggregate) init(ctx context.Context, src Iterator) (exec *aggregate, return ident, nil }) - // Convert & validate group definitions - // - groups - var gd aggregateAttr - outAttrs := make(map[string]bool, len(def.Group)+len(def.OutAttributes)) - for _, attr := range def.Group { - idtf := attr.Identifier() - gd, err = aggregateAttrFromExpr(pp, attr.Properties().Type, idtf, attr.Expression()) + prepAttr := func(attr AggregateAttr) (_ AggregateAttr, err error) { + if attr.RawExpr != "" { + // Parse (it already validates) raw expressions + attr.Expression, err = pp.Parse(attr.RawExpr) + if err != nil { + return + } + } else { + // Manually validate already parsed expressions + err = attr.Expression.Traverse(func(a *ql.ASTNode) (bool, *ql.ASTNode, error) { + if a.Symbol == "" { + return true, a, nil + } + if _, ok := srcAttrs[a.Symbol]; !ok { + return false, nil, fmt.Errorf("unknown attribute %s", a.Symbol) + } + + return true, a, nil + }) + } if err != nil { return } - exec.groupDefs = append(exec.groupDefs, gd) + return def.determineAttrType(attr, def.SourceAttributes) + } + + // Convert & validate group definitions + // - groups + outAttrs := make(map[string]bool, len(def.Group)+len(def.OutAttributes)) + for i, attr := range def.Group { + attr, err = prepAttr(attr) + if err != nil { + return + } + + def.Group[i] = attr + idtf := attr.Identifier + + exec.groupDefs = append(exec.groupDefs, attr) outAttrs[idtf] = true } // - aggregates - for _, attr := range def.OutAttributes { - idtf := attr.Identifier() - gd, err = aggregateAttrFromExpr(pp, attr.Properties().Type, idtf, attr.Expression()) + for i, attr := range def.OutAttributes { + attr, err = prepAttr(attr) if err != nil { return } - exec.aggregateDefs = append(exec.aggregateDefs, gd) + + def.OutAttributes[i] = attr + idtf := attr.Identifier + + exec.aggregateDefs = append(exec.aggregateDefs, attr) outAttrs[idtf] = true } @@ -180,15 +224,69 @@ func (def *Aggregate) init(ctx context.Context, src Iterator) (exec *aggregate, return } -func aggregateAttrFromExpr(pp *ql.Parser, t Type, ident, expr string) (out aggregateAttr, err error) { - n, err := pp.Parse(expr) - if err != nil { +// determineAttrType determines the type of the AggregateAttr based on it's definition +// and source attributes +func (def *Aggregate) determineAttrType(base AggregateAttr, ss []AttributeMapping) (out AggregateAttr, err error) { + out = base + if out.Type != nil { return } - return aggregateAttr{ - ident: ident, - expr: n, - attrType: t, - }, nil + var root *ql.ASTNode + var t Type + + // If we have a symbol, then we'll use it to determine the type. + // All current operations should return the same output type as the input one. + // + // In case of a function, use the output type of the root most function which has + // a known type. + // + // Note, some refs (group and add for example) may not know their types so we need + // to dig deeper. + base.Expression.Traverse(func(a *ql.ASTNode) (bool, *ql.ASTNode, error) { + if a.Symbol != "" { + root = a + return false, a, nil + } + + if a.Ref != "" { + tmp := refToGvalExp[a.Ref] + if tmp == nil || tmp.OutType == nil || tmp.OutTypeUnknown { + return true, a, nil + } + + if tmp.OutType != nil { + t = tmp.OutType + return false, a, nil + } + } + return true, a, nil + }) + + if root != nil { + for _, s := range ss { + if s.Identifier() == root.Symbol { + t = s.Properties().Type + break + } + } + } + + out.Type = t + return +} + +func (a AggregateAttr) toSimpleAttr() SimpleAttr { + return SimpleAttr{ + Ident: a.Identifier, + Props: MapProperties{ + Label: a.Label, + IsPrimary: a.Key, + Type: a.Type, + }, + + // @todo won't matter for now + Expr: "", + Src: "", + } } diff --git a/pkg/dal/exec_aggregate.go b/pkg/dal/exec_aggregate.go index 73feaf161..ce24419de 100644 --- a/pkg/dal/exec_aggregate.go +++ b/pkg/dal/exec_aggregate.go @@ -20,8 +20,8 @@ type ( scanRow *Row planned bool - groupDefs []aggregateAttr - aggregateDefs []aggregateAttr + groupDefs []AggregateAttr + aggregateDefs []AggregateAttr rowTester tester keyWalker keyWalker @@ -54,7 +54,7 @@ func (xs *aggregate) init(ctx context.Context) (err error) { // Initialize the key maker kk := make([]*ql.ASTNode, 0, len(xs.groupDefs)) for _, a := range xs.groupDefs { - kk = append(kk, a.expr) + kk = append(kk, a.Expression) } xs.keyWalker, err = aggregateGroupKeyWalker(kk...) @@ -335,7 +335,7 @@ func (s *aggregate) wrapGroup(ctx context.Context, key groupKey) (g *aggregateGr agg := Aggregator() for _, a := range s.aggregateDefs { - err = agg.AddAggregate(a.ident, a.expr) + err = agg.AddAggregate(a.Identifier, a.Expression) if err != nil { return } @@ -362,7 +362,7 @@ func (xs *aggregate) scanKey(g *aggregateGroup, dst *Row) (err error) { for i, attr := range xs.groupDefs { // @todo multi value support? // omitting err; internal row won't raise them - dst.SetValue(attr.ident, 0, g.key[i]) + dst.SetValue(attr.Identifier, 0, g.key[i]) } return nil @@ -379,7 +379,7 @@ func (s *aggregate) keep(ctx context.Context, r *Row) (bool, error) { func (s *aggregate) collectPrimaryAttributes() (out []string) { out = make([]string, 0, 2) for _, m := range s.def.Group { - out = append(out, m.Identifier()) + out = append(out, m.Identifier) } return @@ -424,9 +424,9 @@ func (s *aggregate) sortGroups() { }) } -func inKeys(kk []AttributeMapping, ident string) int { +func inKeys(kk []AggregateAttr, ident string) int { for i, k := range kk { - if k.Identifier() == ident { + if k.Identifier == ident { return i } } @@ -442,8 +442,8 @@ func (xs *aggregate) initScanRow() (out *Row) { // pre-populate with known attrs for _, attr := range append(xs.def.Group, xs.def.OutAttributes...) { - out.values[attr.Identifier()] = make([]any, 0, 2) - out.counters[attr.Identifier()] = 0 + out.values[attr.Identifier] = make([]any, 0, 2) + out.counters[attr.Identifier] = 0 } return diff --git a/pkg/dal/exec_aggregate_bench_test.go b/pkg/dal/exec_aggregate_bench_test.go index f640a3cf0..83b544f0a 100644 --- a/pkg/dal/exec_aggregate_bench_test.go +++ b/pkg/dal/exec_aggregate_bench_test.go @@ -32,8 +32,8 @@ func benchmarkExecAggregate(b *testing.B, n int) { for n := 0; n < b.N; n++ { def := Aggregate{ Ident: "agg", - Group: saToMapping(group...), - OutAttributes: saToMapping(outAttributes...), + Group: saToAggAttr(group...), + OutAttributes: saToAggAttr(outAttributes...), SourceAttributes: saToMapping(sourceAttributes...), } diff --git a/pkg/dal/exec_aggregate_test.go b/pkg/dal/exec_aggregate_test.go index a215c706b..c4c62c257 100644 --- a/pkg/dal/exec_aggregate_test.go +++ b/pkg/dal/exec_aggregate_test.go @@ -855,8 +855,8 @@ func TestStepAggregate(t *testing.T) { } sa.Ident = tc.name sa.SourceAttributes = saToMapping(tc.sourceAttributes...) - sa.Group = saToMapping(tc.group...) - sa.OutAttributes = saToMapping(tc.outAttributes...) + sa.Group = saToAggAttr(tc.group...) + sa.OutAttributes = saToAggAttr(tc.outAttributes...) sa.filter = tc.f aa, err := sa.iterator(ctx, b) @@ -891,8 +891,8 @@ func TestStepAggregateValidation(t *testing.T) { sa := &Aggregate{ Ident: "agg", SourceAttributes: saToMapping(basicAttrs...), - Group: saToMapping(groups...), - OutAttributes: saToMapping(attr...), + Group: saToAggAttr(groups...), + OutAttributes: saToAggAttr(attr...), } return sa.dryrun(ctx) @@ -902,8 +902,8 @@ func TestStepAggregateValidation(t *testing.T) { sa := &Aggregate{ Ident: "agg", SourceAttributes: saToMapping(basicAttrs...), - Group: saToMapping(groups...), - OutAttributes: saToMapping(attr...), + Group: saToAggAttr(groups...), + OutAttributes: saToAggAttr(attr...), Filter: f, } @@ -1003,8 +1003,8 @@ func TestStepAggregate_cursorCollect_forward(t *testing.T) { filter: internalFilter{ orderBy: c.ss, }, - Group: saToMapping(c.group...), - OutAttributes: saToMapping(c.outAttributes...), + Group: saToAggAttr(c.group...), + OutAttributes: saToAggAttr(c.outAttributes...), } out, err := (&aggregate{def: def}).ForwardCursor(c.in) @@ -1051,8 +1051,8 @@ func TestStepAggregate_cursorCollect_back(t *testing.T) { filter: internalFilter{ orderBy: c.ss, }, - Group: saToMapping(c.group...), - OutAttributes: saToMapping(c.outAttributes...), + Group: saToAggAttr(c.group...), + OutAttributes: saToAggAttr(c.outAttributes...), } out, err := (&aggregate{def: def}).BackCursor(c.in) @@ -1130,8 +1130,8 @@ func TestStepAggregate_more(t *testing.T) { } d := tc.def - d.Group = saToMapping(tc.group...) - d.OutAttributes = saToMapping(tc.outAttributes...) + d.Group = saToAggAttr(tc.group...) + d.OutAttributes = saToAggAttr(tc.outAttributes...) d.SourceAttributes = saToMapping(tc.sourceAttributes...) for _, k := range tc.group { d.filter.orderBy = append(d.filter.orderBy, &filter.SortExpr{Column: k.ident}) @@ -1343,8 +1343,8 @@ func TestStepAggregate_paging(t *testing.T) { prep := func(f internalFilter) { d = Aggregate{ Filter: f, - Group: saToMapping(tc.group...), - OutAttributes: saToMapping(tc.outAttributes...), + Group: saToAggAttr(tc.group...), + OutAttributes: saToAggAttr(tc.outAttributes...), SourceAttributes: saToMapping(tc.sourceAttributes...), } } diff --git a/pkg/dal/main_test.go b/pkg/dal/main_test.go index 92c51c914..5bfcd6635 100644 --- a/pkg/dal/main_test.go +++ b/pkg/dal/main_test.go @@ -26,6 +26,26 @@ func saToMapping(sa ...simpleAttribute) []AttributeMapping { return out } +func saToAggAttr(sa ...simpleAttribute) []AggregateAttr { + out := make([]AggregateAttr, 0, len(sa)) + for _, a := range sa { + aux := AggregateAttr{ + RawExpr: a.expr, + Identifier: a.ident, + Type: a.t, + } + if aux.RawExpr == "" { + aux.RawExpr = a.source + } + if aux.RawExpr == "" { + aux.RawExpr = a.ident + } + out = append(out, aux) + } + + return out +} + func (sa simpleAttribute) Properties() MapProperties { return MapProperties{ IsPrimary: sa.primary, diff --git a/pkg/dal/runner_gval.go b/pkg/dal/runner_gval.go index 2e9b03996..8fc48cdd9 100644 --- a/pkg/dal/runner_gval.go +++ b/pkg/dal/runner_gval.go @@ -19,6 +19,10 @@ type ( } exprHandlerGval struct { Handler func(...string) string + + // @todo temporarily added to support the current aggregate attribute type calculation + OutType Type + OutTypeUnknown bool } ) @@ -33,6 +37,7 @@ var ( Handler: func(args ...string) string { return fmt.Sprintf("(! %s)", args[0]) }, + OutType: TypeBoolean{}, }, // - bool @@ -40,16 +45,19 @@ var ( Handler: func(args ...string) string { return strings.Join(args, " && ") }, + OutType: TypeBoolean{}, }, "or": { Handler: func(args ...string) string { return strings.Join(args, " || ") }, + OutType: TypeBoolean{}, }, "xor": { Handler: func(args ...string) string { return fmt.Sprintf("(%s != %s)", args[0], args[1]) }, + OutType: TypeBoolean{}, }, // - comp. @@ -57,32 +65,38 @@ var ( Handler: func(args ...string) string { return fmt.Sprintf("%s == %s", args[0], args[1]) }, + OutType: TypeBoolean{}, }, "ne": { Handler: func(args ...string) string { return fmt.Sprintf("%s != %s", args[0], args[1]) }, + OutType: TypeBoolean{}, }, "lt": { Handler: func(args ...string) string { return fmt.Sprintf("%s < %s", args[0], args[1]) }, + OutType: TypeBoolean{}, }, "le": { Handler: func(args ...string) string { return fmt.Sprintf("%s <= %s", args[0], args[1]) }, + OutType: TypeBoolean{}, }, "gt": { Handler: func(args ...string) string { return fmt.Sprintf("%s > %s", args[0], args[1]) }, + OutType: TypeBoolean{}, }, "ge": { //Handler: makeGenericCompHandler(">="), Handler: func(args ...string) string { return fmt.Sprintf("%s >= %s", args[0], args[1]) }, + OutType: TypeBoolean{}, }, // - math @@ -90,21 +104,25 @@ var ( Handler: func(args ...string) string { return fmt.Sprintf("%s + %s", args[0], args[1]) }, + OutTypeUnknown: true, }, "sub": { Handler: func(args ...string) string { return fmt.Sprintf("%s - %s", args[0], args[1]) }, + OutTypeUnknown: true, }, "mult": { Handler: func(args ...string) string { return fmt.Sprintf("%s * %s", args[0], args[1]) }, + OutTypeUnknown: true, }, "div": { Handler: func(args ...string) string { return fmt.Sprintf("%s / %s", args[0], args[1]) }, + OutTypeUnknown: true, }, // - strings @@ -112,6 +130,7 @@ var ( Handler: func(args ...string) string { return fmt.Sprintf("concat(%s)", strings.Join(args, ", ")) }, + OutType: TypeText{}, }, // @todo implement; the commented versions are not good enough @@ -145,36 +164,43 @@ var ( Handler: func(args ...string) string { return "now()" }, + OutType: TypeTimestamp{}, }, "quarter": { Handler: func(args ...string) string { return fmt.Sprintf("quarter(%s)", args[0]) }, + OutType: TypeNumber{}, }, "year": { Handler: func(args ...string) string { return fmt.Sprintf("year(%s)", args[0]) }, + OutType: TypeNumber{}, }, "month": { Handler: func(args ...string) string { return fmt.Sprintf("month(%s)", args[0]) }, + OutType: TypeNumber{}, }, "date": { Handler: func(args ...string) string { return fmt.Sprintf("date(%s)", args[0]) }, + OutType: TypeDate{}, }, "day": { Handler: func(args ...string) string { return fmt.Sprintf("day(%s)", args[0]) }, + OutType: TypeNumber{}, }, "date_format": { Handler: func(args ...string) string { return fmt.Sprintf("strftime(%s, %s)", args[0], args[1]) }, + OutType: TypeText{}, }, // generic stuff @@ -182,16 +208,19 @@ var ( Handler: func(args ...string) string { return fmt.Sprintf("isNil(%s)", args[0]) }, + OutType: TypeBoolean{}, }, "nnull": { Handler: func(args ...string) string { return fmt.Sprintf("!isNil(%s)", args[0]) }, + OutType: TypeBoolean{}, }, "exists": { Handler: func(args ...string) string { return fmt.Sprintf("!isNil(%s)", args[0]) }, + OutType: TypeBoolean{}, }, // - typecast @@ -199,22 +228,26 @@ var ( Handler: func(args ...string) string { return fmt.Sprintf("float(%s)", args[0]) }, + OutType: TypeNumber{}, }, "int": { Handler: func(args ...string) string { return fmt.Sprintf("int(%s)", args[0]) }, + OutType: TypeNumber{}, }, "string": { Handler: func(args ...string) string { return fmt.Sprintf("string(%s)", args[0]) }, + OutType: TypeText{}, }, "group": { Handler: func(args ...string) string { return fmt.Sprintf("(%s)", args[0]) }, + OutTypeUnknown: true, }, } ) diff --git a/pkg/dal/service.go b/pkg/dal/service.go index 83f872a65..895229f3e 100644 --- a/pkg/dal/service.go +++ b/pkg/dal/service.go @@ -524,7 +524,14 @@ func collectAttributes(s PipelineStep) (out []AttributeMapping) { return s.OutAttributes case *Aggregate: - return append(s.Group, s.OutAttributes...) + out = make([]AttributeMapping, 0, 24) + for _, a := range s.Group { + out = append(out, a.toSimpleAttr()) + } + for _, a := range s.OutAttributes { + out = append(out, a.toSimpleAttr()) + } + return case *Join: return s.OutAttributes diff --git a/system/reporting/reporting.go b/system/reporting/reporting.go index f53a071d9..d483ed077 100644 --- a/system/reporting/reporting.go +++ b/system/reporting/reporting.go @@ -500,14 +500,33 @@ func convStepAggregate(step types.ReportStepAggregate, defs FrameDefinitionSet) return } + ggs := make([]dal.AggregateAttr, 0, len(step.Keys)) + for _, c := range step.Keys { + ggs = append(ggs, dal.AggregateAttr{ + Key: true, + Identifier: c.Name, + Label: c.Label, + Expression: c.Def.Node(), + }) + } + + vvs := make([]dal.AggregateAttr, 0, len(step.Columns)) + for _, c := range step.Columns { + vvs = append(vvs, dal.AggregateAttr{ + Identifier: c.Name, + Label: c.Label, + Expression: c.Def.Node(), + }) + } + // Make pipeline step out = &dal.Aggregate{ Ident: step.Name, RelSource: step.Source, Filter: f, - Group: step.Keys.DalMapping(), - OutAttributes: step.Columns.DalMapping(), + Group: ggs, + OutAttributes: vvs, } return } diff --git a/system/types/report.go b/system/types/report.go index 2e0471544..37555f9f6 100644 --- a/system/types/report.go +++ b/system/types/report.go @@ -6,7 +6,6 @@ import ( "encoding/json" "time" - "github.com/cortezaproject/corteza-server/pkg/dal" "github.com/cortezaproject/corteza-server/pkg/filter" "github.com/cortezaproject/corteza-server/pkg/ql" "github.com/cortezaproject/corteza-server/pkg/sql" @@ -147,42 +146,6 @@ type ( } ) -// // // // // // // // // // // // // // // // // // // // // // // // // - -// @todo redo/rething these col methods along with the rest of the attr -// interface rethinking. -func (c ReportAggregateColumn) Identifier() string { - return c.Name -} - -func (c ReportAggregateColumn) Expression() (expression string) { - if c.Def == nil || c.Def.ASTNode == nil { - return - } - return c.Def.String() -} - -func (c ReportAggregateColumn) Source() (ident string) { - return c.Name -} - -func (c ReportAggregateColumn) Properties() dal.MapProperties { - // @todo!!! - return dal.MapProperties{} -} - -func (cc ReportAggregateColumnSet) DalMapping() []dal.AttributeMapping { - out := make([]dal.AttributeMapping, 0, len(cc)) - - for _, c := range cc { - out = append(out, c) - } - - return out -} - -// // // // // // // // // // // // // // // // // // // // // // // // // - // ReportSteps returns a ReportStepSet collected from the ReportDataSourceSet func (ss ReportDataSourceSet) ReportSteps() ReportStepSet { out := make(ReportStepSet, 0, 124)