From 319a29fdc624eda42a6a0bdba740389656bfcb75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Thu, 1 Sep 2022 14:46:35 +0200 Subject: [PATCH] Cleanup system/reporter - DAL/pipeline migration --- app/boot_levels.go | 4 - pkg/filter/filter.go | 35 +- .../reporter}/frame_builder.go | 51 +- .../reporter/reporter.go | 522 ++++++++++++------ system/service/report.go | 55 +- system/types/report.go | 22 +- system/types/report_frame.go | 20 +- system/types/report_test.go | 69 +++ 8 files changed, 521 insertions(+), 257 deletions(-) rename {pkg/reportutils => system/reporter}/frame_builder.go (72%) rename pkg/reportutils/reportutils.go => system/reporter/reporter.go (50%) create mode 100644 system/types/report_test.go diff --git a/app/boot_levels.go b/app/boot_levels.go index 63e6dce39..163a3562a 100644 --- a/app/boot_levels.go +++ b/app/boot_levels.go @@ -445,10 +445,6 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) { } } - // Register reporters - // @todo additional datasource providers; generate? - sysService.DefaultReport.RegisterReporter("composeRecords", cmpService.DefaultRecord) - // Initializing seeder _ = seeder.Seeder(ctx, app.Store, dal.Service(), seeder.Faker()) diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go index ada88a388..5ce5e9413 100644 --- a/pkg/filter/filter.go +++ b/pkg/filter/filter.go @@ -1,16 +1,19 @@ package filter +import "github.com/cortezaproject/corteza-server/pkg/ql" + type ( filterOpt func(*filter) filter struct { - constaints map[string][]any - stateConditions map[string]State - metaConditions map[string]any - expression string - orderBy SortExprSet - limit uint - cursor *PagingCursor + constaints map[string][]any + stateConditions map[string]State + metaConditions map[string]any + expression string + expressionParsed *ql.ASTNode + orderBy SortExprSet + limit uint + cursor *PagingCursor } Filter interface { @@ -39,7 +42,7 @@ type ( } ) -func Generic(oo ...filterOpt) Filter { +func Generic(oo ...filterOpt) *filter { f := &filter{} for _, o := range oo { @@ -99,6 +102,13 @@ func WithExpression(e string) filterOpt { } } +// WithExpressionParsed sets parsed expression to filter +func WithExpressionParsed(e *ql.ASTNode) filterOpt { + return func(f *filter) { + f.expressionParsed = e + } +} + // WithOrderBy sets order by expression func WithOrderBy(o SortExprSet) filterOpt { return func(f *filter) { @@ -120,10 +130,19 @@ func WithCursor(p *PagingCursor) filterOpt { } } +func (f *filter) With(oo ...filterOpt) *filter { + for _, o := range oo { + o(f) + } + + return f +} + func (f *filter) Constraints() map[string][]any { return f.constaints } func (f *filter) StateConstraints() map[string]State { return f.stateConditions } func (f *filter) MetaConstraints() map[string]any { return f.metaConditions } func (f *filter) Expression() string { return f.expression } +func (f *filter) ExpressionParsed() string { return f.expression } func (f *filter) OrderBy() SortExprSet { return f.orderBy } func (f *filter) Limit() uint { return f.limit } func (f *filter) Cursor() *PagingCursor { return f.cursor } diff --git a/pkg/reportutils/frame_builder.go b/system/reporter/frame_builder.go similarity index 72% rename from pkg/reportutils/frame_builder.go rename to system/reporter/frame_builder.go index 37b143bb9..96ee9d54b 100644 --- a/pkg/reportutils/frame_builder.go +++ b/system/reporter/frame_builder.go @@ -1,4 +1,4 @@ -package reportutils +package reporter import ( "fmt" @@ -10,20 +10,33 @@ import ( ) type ( - // reportFrameBuilder is a helper struct for building report frames from - // dal iterators + // reportFrameBuilder assist in frame construction from iterators + // + // Primarily simplifies mapping the correct iterator attributes to correct + // frame row columns and having them encoded properly. reportFrameBuilder struct { - def *types.ReportFrameDefinition - frame *types.ReportFrame + def *types.ReportFrameDefinition + frame *types.ReportFrame + attrMapping map[string]int + attrMvDel map[string]string } ) +// newReportFrameBuilder initializes a new report frame builder func newReportFrameBuilder(def *types.ReportFrameDefinition) *reportFrameBuilder { - // Index requested columns for easier lookup + // Index cols for easier lookups attrMap := make(map[string]int) + mvDelMap := make(map[string]string) for i, c := range def.Columns { attrMap[c.Name] = i + + if c.Multivalue { + mvDelMap[c.Name] = c.MultivalueDelimiter + if mvDelMap[c.Name] == "" { + mvDelMap[c.Name] = "\n" + } + } } out := &reportFrameBuilder{ @@ -31,13 +44,13 @@ func newReportFrameBuilder(def *types.ReportFrameDefinition) *reportFrameBuilder attrMapping: attrMap, } + // Init output frame out.freshFrame() - return out } -// withRefs includes additional metadata required by the link step -func (b *reportFrameBuilder) withRefs(col string) { +// linked includes additional metadata required by the link step +func (b *reportFrameBuilder) linked(col string) { b.frame.RelColumn = col } @@ -67,16 +80,6 @@ func (b *reportFrameBuilder) addRow(r *dal.Row) { } } -func (b *reportFrameBuilder) stringifyVal(col string, val any) string { - // @todo nicer formatting and such? V1 didn't do much different - return fmt.Sprintf("%v", val) -} - -func (b *reportFrameBuilder) joinMultiVal(col string, vals []string) string { - // @todo add delimiter (extend attrs) - return strings.Join(vals, "\n") -} - // done returns the constructed frame and prepares a new frame with the same // metadata as the original one func (b *reportFrameBuilder) done() *types.ReportFrame { @@ -86,7 +89,17 @@ func (b *reportFrameBuilder) done() *types.ReportFrame { return out } +func (b *reportFrameBuilder) stringifyVal(col string, val any) string { + // @todo nicer formatting and such? V1 didn't do much different + return fmt.Sprintf("%v", val) +} + +func (b *reportFrameBuilder) joinMultiVal(col string, vals []string) string { + return strings.Join(vals, b.attrMvDel[col]) +} + func (b *reportFrameBuilder) freshFrame() { + // reuse the old frame metadata and clears out the rows if b.frame != nil { aux := *b.frame b.frame = &aux diff --git a/pkg/reportutils/reportutils.go b/system/reporter/reporter.go similarity index 50% rename from pkg/reportutils/reportutils.go rename to system/reporter/reporter.go index c42bd315c..f25ec2c56 100644 --- a/pkg/reportutils/reportutils.go +++ b/system/reporter/reporter.go @@ -1,4 +1,4 @@ -package reportutils +package reporter import ( "context" @@ -11,90 +11,114 @@ import ( ) type ( - ReportWorkload struct { - Pipeline dal.Pipeline - FrameDefs types.ReportFrameDefinitionSet + run struct { + Pipeline dal.Pipeline + Defs types.ReportFrameDefinitionSet } - PipelineRunner interface { + modelFinder interface { FindModel(dal.ModelRef) *dal.Model - Run(context.Context, dal.Pipeline) (dal.Iterator, error) + } + + dryRunner interface { + modelFinder Dryrun(context.Context, dal.Pipeline) error } ) -// Workloads creates a set of workloads for the given pipeline and frame definitions -func Workloads(pr PipelineRunner, steps types.ReportStepSet, defs types.ReportFrameDefinitionSet) (out []ReportWorkload, err error) { - // Construct a pipeline from the steps; we'll slice it later - base, err := Pipeline(pr, steps) - if err != nil { - return - } - - // Prepare workloads based on the provided definitions +// Runs create a set of runs based on step and frame definitions +func Runs(pr modelFinder, steps types.ReportStepSet, defs types.ReportFrameDefinitionSet) (out []run, err error) { + // Prepare runs based on the provided definitions // // - If consecutive definitions point to the same source with the same name // consider them to fall under the same workload (the link step) // - else, one def per workload auxDefs := make(types.ReportFrameDefinitionSet, 0) - var auxOut ReportWorkload + var aux run for i, def := range defs { if i == 0 { auxDefs = append(auxDefs, def) continue } - // This is for the link step + // Definitions fall together if def.Name == defs[i-1].Name && def.Source == defs[i-1].Source { auxDefs = append(auxDefs, def) continue } - // This is for the rest - auxOut, err = makeWorkload(base, auxDefs) + // Make run for the previous definition (exclude current!!) + aux, err = makeRun(pr, steps, auxDefs) if err != nil { return } - out = append(out, auxOut) + out = append(out, aux) + + // Prepare next definition batch including the current one auxDefs = make(types.ReportFrameDefinitionSet, 0) auxDefs = append(auxDefs, def) } // Handle the ones (potentially) not covered by the above loop if len(auxDefs) > 0 { - auxOut, err = makeWorkload(base, auxDefs) + aux, err = makeRun(pr, steps, auxDefs) if err != nil { return } - out = append(out, auxOut) + out = append(out, aux) } return } -// Frames returns a set of frames for the given workload & iterator combo -func Frames(ctx context.Context, iter dal.Iterator, workload ReportWorkload) (ff []*types.ReportFrame, err error) { - // Preprocessing on the workload's frame definitions; assure all columns/metdata are there - // to avoid nonesense later down the line - updateDefAttrs(workload) +// Frames returns a set of ReportFrame for the given workload & iterator combo +func Frames(ctx context.Context, iter dal.Iterator, r run) (ff []*types.ReportFrame, err error) { + // Preprocessing on the workload's frame definitions; assure all + // columns/metdata are there to avoid nonesense later down the line + updateDefinitionColumns(r) // @todo perhaps need to change; for now only this scenario introduces multiple // frame defs per workload - if len(workload.FrameDefs) > 1 { - return framifyLinkIter(ctx, iter, workload) + if len(r.Defs) > 1 { + return stepLinkFrames(ctx, iter, r) } - return framifyIter(ctx, iter, workload) + return stepFrames(ctx, iter, r) } -// framifyLinkIter is a handler dedicated for the link step due to it's unique output -func framifyLinkIter(ctx context.Context, iter dal.Iterator, workload ReportWorkload) (ff []*types.ReportFrame, err error) { - defs := workload.FrameDefs +// Describe returns a set of frame descriptions based on the given pipeline +func Describe(ctx context.Context, rr dryRunner, ss types.ReportStepSet, sources []string) (out types.FrameDescriptionSet, err error) { + // Make a run for the whole thing + pp, err := makePipeline(rr, ss, nil) + if err != nil { + return + } + var aux types.FrameDescriptionSet + + for _, src := range sources { + // Use the requested source as root + sub := pp.Slice(src) + s := sub[0] + + // Describe + aux, err = describePipeline(s, src) + if err != nil { + return + } + out = append(out, aux...) + } + + return +} + +// stepLinkFrames is dedicated for the link step due to it's unique output +func stepLinkFrames(ctx context.Context, iter dal.Iterator, r run) (ff []*types.ReportFrame, err error) { + defs := r.Defs // @note this will only be called for the link step so it can freely panic if violated - r := workload.Pipeline[0].(*dal.Link) + defLink := r.Pipeline[0].(*dal.Link) // Unpack frame definitions for the link - defLeft, defRight := unpackLinkDefs(defs, workload.Pipeline) + defLeft, defRight := unpackLinkDefs(defs, r.Pipeline) // Init vars to keep track of the progress // @note true is left, false is right @@ -103,7 +127,7 @@ func framifyLinkIter(ctx context.Context, iter dal.Iterator, workload ReportWork builders := make(map[bool]*reportFrameBuilder) builders[true] = newReportFrameBuilder(defLeft) builders[false] = newReportFrameBuilder(defRight) - builders[false].withRefs(r.On.Right) + builders[false].linked(defLink.On.Right) limits := make(map[bool]uint) if defLeft.Paging != nil { @@ -200,9 +224,9 @@ func unpackLinkDefs(defs types.ReportFrameDefinitionSet, pp dal.Pipeline) (left, return find(defs, l.RelLeft), find(defs, l.RelRight) } -// framifyIter is a generic iter to frame handler -func framifyIter(ctx context.Context, iter dal.Iterator, workload ReportWorkload) (ff []*types.ReportFrame, err error) { - defs := workload.FrameDefs +// stepFrames is a generic iter to frame converter +func stepFrames(ctx context.Context, iter dal.Iterator, r run) (ff []*types.ReportFrame, err error) { + defs := r.Defs // @note only the link step takes multiple defs and that one is not covered // by this function @@ -254,64 +278,6 @@ func framifyIter(ctx context.Context, iter dal.Iterator, workload ReportWorkload return append(ff, builder.done()), nil } -// Pipeline creates a pipeline from the given steps -func Pipeline(pr PipelineRunner, steps types.ReportStepSet) (out dal.Pipeline, err error) { - out = make(dal.Pipeline, 0, len(steps)) - - for _, step := range steps { - switch { - case step.Load != nil: - aux, err := makeStepLoad(pr, *step.Load) - if err != nil { - return nil, err - } - out = append(out, aux) - - case step.Join != nil: - aux, err := makeStepJoin(*step.Join) - if err != nil { - return nil, err - } - out = append(out, aux) - - case step.Link != nil: - aux, err := makeStepLink(*step.Link) - if err != nil { - return nil, err - } - out = append(out, aux) - - case step.Aggregate != nil: - aux, err := makeStepAggregate(*step.Aggregate) - if err != nil { - return nil, err - } - out = append(out, aux) - } - } - - return out, out.LinkSteps() -} - -// DescribePipeline returns a set of frame descriptions based on the given pipeline -func DescribePipeline(pp dal.Pipeline, sources []string) (out types.FrameDescriptionSet, err error) { - var aux types.FrameDescriptionSet - - for _, src := range sources { - sub := pp.Slice(src) - s := sub[0] - - aux, err = describePipeline(s, src) - if err != nil { - return - } - - out = append(out, aux...) - } - - return -} - func describePipeline(s dal.PipelineStep, src string) (out types.FrameDescriptionSet, err error) { aa := s.Attributes() @@ -331,26 +297,120 @@ func describePipeline(s dal.PipelineStep, src string) (out types.FrameDescriptio return } -func makeWorkload(pp dal.Pipeline, defs types.ReportFrameDefinitionSet) (out ReportWorkload, err error) { - // We'll trust all of the defs point to the same source; this should be validated - // way sooner - def := defs[0] - - out.FrameDefs = defs - out.Pipeline = pp.Slice(def.Source) +func makeRun(pr modelFinder, ss types.ReportStepSet, defs types.ReportFrameDefinitionSet) (out run, err error) { + var pp dal.Pipeline + pp, err = makePipeline(pr, ss, defs) + if err != nil { + return + } + out.Defs = defs + out.Pipeline = pp.Slice(defs[0].Source) return } -// @todo address with col/attr rework/rethink +func makePipeline(mf modelFinder, ss types.ReportStepSet, defs types.ReportFrameDefinitionSet) (pp dal.Pipeline, err error) { + for _, step := range ss { + switch { + case step.Load != nil: + aux, err := convStepLoad(mf, *step.Load, defs.FilterBySource(step.Load.Name)) + if err != nil { + return nil, err + } + pp = append(pp, aux) + + case step.Aggregate != nil: + aux, err := convStepAggregate(*step.Aggregate, defs.FilterBySource(step.Aggregate.Name)) + if err != nil { + return nil, err + } + pp = append(pp, aux) + + case step.Join != nil: + aux, err := convStepJoin(*step.Join, defs.FilterBySource(step.Join.Name)) + if err != nil { + return nil, err + } + pp = append(pp, aux) + + case step.Link != nil: + aux, err := convStepLink(*step.Link, defs.FilterBySource(step.Link.Name)) + if err != nil { + return nil, err + } + pp = append(pp, aux) + } + } + + return pp, pp.LinkSteps() +} + +// mappingToFrameCols converts pipeline AttributeMapping to ReportFrameColumnSet func mappingToFrameCols(mm []dal.AttributeMapping) types.ReportFrameColumnSet { out := make(types.ReportFrameColumnSet, 0, len(mm)) for _, m := range mm { - out = append(out, types.ReportFrameColumn{ - Name: m.Identifier(), - Kind: "String", - }) + out = append(out, mappingToFrameCol(m)) + } + + return out +} + +// @note current implementation a bit _rushed_ since I'll probably rethink +// how the pipeline handles attributes -- will revisit then. +func mappingToFrameCol(m dal.AttributeMapping) types.ReportFrameColumn { + p := m.Properties() + + const ( + // Coppied around to reduce imports + emailLength = 254 + urlLength = 2048 + + attachmentResType = "corteza::system:attachment" + userResType = "corteza::system:user" + moduleResType = "corteza::compose:module" + ) + + out := types.ReportFrameColumn{ + Name: m.Identifier(), + // @todo use another method/push into meta? + Label: m.Identifier(), + Kind: "String", + + Primary: p.IsPrimary, + } + + switch t := p.Type.(type) { + case *dal.TypeBoolean: + out.Kind = "Boolean" + case *dal.TypeDate, *dal.TypeTime, *dal.TypeTimestamp: + out.Kind = "DateTime" + + case *dal.TypeNumber: + out.Kind = "Number" + + case *dal.TypeEnum: + out.Kind = "Select" + + case *dal.TypeText: + // @note temporary solution; we should push some meta along with it + if t.Length == emailLength { + out.Kind = "Email" + } else if t.Length == urlLength { + out.Kind = "URL" + } else { + out.Kind = "String" + } + + case *dal.TypeRef: + switch t.RefModel.ResourceType { + case moduleResType: + out.Kind = "Record" + case userResType: + out.Kind = "User" + case attachmentResType: + out.Kind = "File" + } } return out @@ -358,22 +418,121 @@ func mappingToFrameCols(mm []dal.AttributeMapping) types.ReportFrameColumnSet { // Report step -> DAL step conversion -func makeStepJoin(step types.ReportStepJoin) (out *dal.Join, err error) { +// convStepLoad converts ReportStepLoad to dal.Datasource +func convStepLoad(pr modelFinder, step types.ReportStepLoad, defs types.ReportFrameDefinitionSet) (out *dal.Datasource, err error) { + // Validation + if len(defs) > 1 { + err = fmt.Errorf("cannot convert load step: expecting at most one definition, got %d", len(defs)) + return + } + + // Get additional filtering + var extf filter.Filter + if len(defs) == 1 { + extf = filterFromDef(defs[0]) + } + + // Prepare model ref + mfr, err := makeModelRef(step) + if err != nil { + return + } + + f, err := dal.FilterFromExpr(step.Filter.Node()). + MergeFilters(extf) + if err != nil { + return + } + + // Make pipeline step + return &dal.Datasource{ + Ident: step.Name, + Filter: f, + ModelRef: mfr, + OutAttributes: filteredModelAttributes(pr, step, mfr), + }, nil +} + +// convStepAggregate converts ReportStepAggregate to dal.Aggregate +func convStepAggregate(step types.ReportStepAggregate, defs types.ReportFrameDefinitionSet) (out *dal.Aggregate, err error) { + // Validation + if len(defs) > 1 { + err = fmt.Errorf("cannot convert aggregate step: expecting at most one definition, got %d", len(defs)) + return + } + + // Get additional filtering + var extf filter.Filter + if len(defs) == 1 { + extf = filterFromDef(defs[0]) + } + + f, err := dal.FilterFromExpr(step.Filter.Node()). + MergeFilters(extf) + if err != nil { + return + } + + // Make pipeline step + out = &dal.Aggregate{ + Ident: step.Name, + RelSource: step.Source, + Filter: f, + + Group: step.Keys.DalMapping(), + OutAttributes: step.Columns.DalMapping(), + } + return +} + +// convStepJoin converts ReportStepJoin to dal.Join +func convStepJoin(step types.ReportStepJoin, defs types.ReportFrameDefinitionSet) (out *dal.Join, err error) { + // Validation + if len(defs) > 1 { + err = fmt.Errorf("cannot convert join step: expecting at most one definition, got %d", len(defs)) + return + } + + // Get additional filtering + var extf filter.Filter + if len(defs) == 1 { + extf = filterFromDef(defs[0]) + } + + f, err := dal.FilterFromExpr(step.Filter.Node()). + MergeFilters(extf) + if err != nil { + return + } + + // Make pipeline step out = &dal.Join{ Ident: step.Name, RelLeft: step.LocalSource, RelRight: step.ForeignSource, + Filter: f, + On: dal.JoinPredicate{ Left: step.LocalColumn, Right: step.ForeignColumn, }, - Filter: dal.FilterForExpr(step.Filter.Node()), } return } -func makeStepLink(step types.ReportStepLink) (out *dal.Link, err error) { +// convStepLink converts ReportStepLink to dal.Link +func convStepLink(step types.ReportStepLink, defs types.ReportFrameDefinitionSet) (out *dal.Link, err error) { + // Validation + if len(defs) > 2 { + err = fmt.Errorf("cannot convert join step: expecting at most two definitions, got %d", len(defs)) + return + } + + // @todo additional filtering; will need to split the dal.Link filter into + // left and right for more control and clarity + + // Make pipeline step out = &dal.Link{ Ident: step.Name, RelLeft: step.LocalSource, @@ -383,48 +542,83 @@ func makeStepLink(step types.ReportStepLink) (out *dal.Link, err error) { Left: step.LocalColumn, Right: step.ForeignColumn, }, - Filter: dal.FilterForExpr(step.Filter.Node()), + Filter: dal.FilterFromExpr(step.Filter.Node()), } return } -func makeStepAggregate(step types.ReportStepAggregate) (out *dal.Aggregate, err error) { - out = &dal.Aggregate{ - Ident: step.Name, - RelSource: step.Source, - Filter: dal.FilterForExpr(step.Filter.Node()), +// updateDefinitionColumns assures run's frame column completeness +func updateDefinitionColumns(r run) { + ppAttrs := r.Pipeline[0].Attributes() + for i, def := range r.Defs { + if len(def.Columns) > 0 { + continue + } - Group: step.Keys.DalMapping(), - OutAttributes: step.Columns.DalMapping(), + def.Columns = mappingToFrameCols(ppAttrs[i]) } - return } -func makeStepLoad(pr PipelineRunner, step types.ReportStepLoad) (out *dal.Datasource, err error) { - mfr, c, err := getModelRef(step) - if err != nil { +// makeModelRef returns the model ref based on the step load definition +// @todo should be expanded when we support models that are not compose modules +func makeModelRef(step types.ReportStepLoad) (out dal.ModelRef, err error) { + var ( + connectionID uint64 + moduleID, namespaceID uint64 + module, namespace string + + aux any + ok bool + ) + + if aux, ok = step.Definition["moduleID"]; ok { + moduleID = cast.ToUint64(aux) + } else if aux, ok = step.Definition["module"]; ok { + module = cast.ToString(aux) + } else { + err = fmt.Errorf("step definition is missing moduleID or module") return } - return &dal.Datasource{ - Ident: step.Name, - Filter: dal.FilterForExpr(step.Filter.Node()). - WithConstraints(c), - ModelRef: mfr, - OutAttributes: filteredModelAttributes(pr, step, mfr), - }, nil + if aux, ok = step.Definition["namespaceID"]; ok { + namespaceID = cast.ToUint64(aux) + } else if aux, ok = step.Definition["module"]; ok { + namespace = cast.ToString(aux) + } else { + err = fmt.Errorf("step definition is missing namespaceID or namespace") + return + } + // Connection is optional, default is primary connection + if aux, ok = step.Definition["connectionID"]; ok { + connectionID = cast.ToUint64(aux) + } + + out.ConnectionID = connectionID + out.Refs = make(map[string]any) + + // Use only one of the two identifier variations with priority to ID + if moduleID > 0 { + out.Refs["moduleID"] = moduleID + } else { + out.Refs["module"] = module + } + + if namespaceID > 0 { + out.Refs["namespaceID"] = namespaceID + } else { + out.Refs["namespace"] = namespace + } + + return } -// // // // // // // // // // // // // // // // // // // // // // // // // -// Utilities... - // filteredModelAttributes returns the requested attributes based on the step // definition or all attributes if none are specified // // The function collects the attributes from the DAL model to omit the attribute // construction step. -func filteredModelAttributes(pr PipelineRunner, step types.ReportStepLoad, mfr dal.ModelRef) (out []dal.AttributeMapping) { +func filteredModelAttributes(pr modelFinder, step types.ReportStepLoad, mfr dal.ModelRef) (out []dal.AttributeMapping) { out = make([]dal.AttributeMapping, 0, 100) // All of the attributes @@ -443,7 +637,6 @@ func filteredModelAttributes(pr PipelineRunner, step types.ReportStepLoad, mfr d for _, col := range step.Columns { reqIndex[col.Name] = true } - for _, attr := range fullAttrs { if reqIndex[attr.Ident] { out = append(out, attrToMapping(attr)...) @@ -453,7 +646,7 @@ func filteredModelAttributes(pr PipelineRunner, step types.ReportStepLoad, mfr d return } -func getModelAttrs(pr PipelineRunner, mfr dal.ModelRef) (attrs dal.AttributeSet, err error) { +func getModelAttrs(pr modelFinder, mfr dal.ModelRef) (attrs dal.AttributeSet, err error) { m := pr.FindModel(mfr) if m == nil { return nil, fmt.Errorf("model not found: %v", mfr) @@ -477,53 +670,18 @@ func attrToMapping(aa ...*dal.Attribute) (out []dal.AttributeMapping) { return } -// @todo support for ns/mod by handle -func getModelRef(step types.ReportStepLoad) (out dal.ModelRef, constraints map[string][]any, err error) { - var ( - moduleID, connectionID, namespaceID uint64 - aux any - ok bool +func filterFromDef(def *types.ReportFrameDefinition) (out filter.Filter) { + aux := filter.Generic( + filter.WithExpressionParsed(def.Filter.Node()), + filter.WithOrderBy(def.Sort), ) - constraints = make(map[string][]any) - - if aux, ok = step.Definition["moduleID"]; ok { - moduleID = cast.ToUint64(aux) - } else if aux, ok = step.Definition["module"]; ok { - moduleID = cast.ToUint64(aux) - } else { - err = fmt.Errorf("step definition is missing moduleID") - return + if def.Paging != nil { + aux = aux.With( + filter.WithCursor(def.Paging.PageCursor), + filter.WithLimit(def.Paging.Limit), + ) } - if aux, ok = step.Definition["namespaceID"]; ok { - namespaceID = cast.ToUint64(aux) - } else if aux, ok = step.Definition["module"]; ok { - namespaceID = cast.ToUint64(aux) - } else { - err = fmt.Errorf("step definition is missing namespaceID") - return - } - - if aux, ok = step.Definition["connectionID"]; ok { - connectionID = cast.ToUint64(aux) - } else if aux, ok = step.Definition["connection"]; ok { - connectionID = cast.ToUint64(aux) - } - - constraints["moduleID"] = []any{moduleID} - constraints["namespaceID"] = []any{namespaceID} - - return dal.ModelRef{ConnectionID: connectionID, ResourceID: moduleID}, constraints, nil -} - -func updateDefAttrs(workload ReportWorkload) { - ppAttrs := workload.Pipeline[0].Attributes() - for i, def := range workload.FrameDefs { - if len(def.Columns) > 0 { - continue - } - - def.Columns = mappingToFrameCols(ppAttrs[i]) - } + return aux } diff --git a/system/service/report.go b/system/service/report.go index e672eccf0..7c57626e2 100644 --- a/system/service/report.go +++ b/system/service/report.go @@ -6,7 +6,7 @@ import ( "github.com/cortezaproject/corteza-server/pkg/dal" "github.com/cortezaproject/corteza-server/pkg/errors" - "github.com/cortezaproject/corteza-server/pkg/reportutils" + "github.com/cortezaproject/corteza-server/system/reporter" "github.com/cortezaproject/corteza-server/pkg/actionlog" "github.com/cortezaproject/corteza-server/pkg/filter" @@ -26,7 +26,7 @@ type ( users UserService - pipelineRunner reportutils.PipelineRunner + pipelineRunner pipelineRunner } reportAccessController interface { @@ -37,10 +37,12 @@ type ( CanDeleteReport(context.Context, *types.Report) bool CanRunReport(context.Context, *types.Report) bool } -) -var ( - reporters = make(map[string]any) + pipelineRunner interface { + Run(context.Context, dal.Pipeline) (dal.Iterator, error) + Dryrun(context.Context, dal.Pipeline) error + FindModel(dal.ModelRef) *dal.Model + } ) // Report is a default report service initializer @@ -57,10 +59,6 @@ func Report(s store.Storer, ac reportAccessController, al actionlog.Recorder, eb } } -func (svc *report) RegisterReporter(key string, r any) { - reporters[key] = r -} - func (svc *report) LookupByID(ctx context.Context, ID uint64) (report *types.Report, err error) { var ( aaProps = &reportActionProps{report: &types.Report{ID: ID}} @@ -294,7 +292,7 @@ func (svc *report) Undelete(ctx context.Context, ID uint64) (err error) { return svc.recordAction(ctx, aaProps, ReportActionUndelete, err) } -// actionlog? +// @todo actionlog? func (svc *report) Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out types.FrameDescriptionSet, err error) { out = make(types.FrameDescriptionSet, 0, len(sources)*2) @@ -303,26 +301,10 @@ func (svc *report) Describe(ctx context.Context, src types.ReportDataSourceSet, return ReportErrNotAllowedToCreate() } - ss := src.ModelSteps() + ss := src.ReportSteps() ss = append(ss, st...) - pp, err := reportutils.Pipeline(svc.pipelineRunner, ss) - if err != nil { - return - } - - err = pp.LinkSteps() - if err != nil { - return - } - - // Dryrun the pipeline to go over all init steps - err = svc.pipelineRunner.Dryrun(ctx, pp) - if err != nil { - return - } - - out, err = reportutils.DescribePipeline(pp, sources) + out, err = reporter.Describe(ctx, svc.pipelineRunner, ss, sources) return err }() @@ -351,23 +333,24 @@ func (svc *report) Run(ctx context.Context, reportID uint64, dd types.ReportFram } // Get all of the steps - ss := r.Sources.ModelSteps() - ss = append(ss, r.Blocks.ModelSteps()...) + ss := r.Sources.ReportSteps() + ss = append(ss, r.Blocks.ReportSteps()...) - // Prepare required pipelines - workloads, err := reportutils.Workloads(svc.pipelineRunner, ss, dd) + // Prepare a set of runs for the provided definitions + runs, err := reporter.Runs(svc.pipelineRunner, ss, dd) if err != nil { return } - // Run the pipelines and produce report frames - for _, workload := range workloads { - iter, err = svc.pipelineRunner.Run(ctx, workload.Pipeline) + // Run the reports and produce the frames + // @todo this can be ran in paralel + for _, run := range runs { + iter, err = svc.pipelineRunner.Run(ctx, run.Pipeline) if err != nil { return } - ff, err = reportutils.Frames(ctx, iter, workload) + ff, err = reporter.Frames(ctx, iter, run) if err != nil { return } diff --git a/system/types/report.go b/system/types/report.go index da163a830..c0ae4242f 100644 --- a/system/types/report.go +++ b/system/types/report.go @@ -139,8 +139,8 @@ type ( filter.Paging } - // qlExprWrap is a wrapper for ql.ASTNode to implement custom JSON unmarshal - // required by reporter. + // qlExprWrap is a wrapper for ql.ASTNode to implement custom JSON + // unmarshal required by reporter. // @todo consider moving this to the ql package qlExprWrap struct { *ql.ASTNode @@ -148,6 +148,8 @@ type ( } ) +// // // // // // // // // // // // // // // // // // // // // // // // // + // @todo redo/rething these col methods along with the rest of the attr // interface rethinking. func (c ReportAggregateColumn) Identifier() string { @@ -155,7 +157,9 @@ func (c ReportAggregateColumn) Identifier() string { } func (c ReportAggregateColumn) Expression() (expression string) { - // @todo!!! + if c.Def == nil || c.Def.ASTNode == nil { + return + } return c.Def.String() } @@ -180,7 +184,8 @@ func (cc ReportAggregateColumnSet) DalMapping() []dal.AttributeMapping { // // // // // // // // // // // // // // // // // // // // // // // // // -func (ss ReportDataSourceSet) ModelSteps() ReportStepSet { +// ReportSteps returns a ReportStepSet collected from the ReportDataSourceSet +func (ss ReportDataSourceSet) ReportSteps() ReportStepSet { out := make(ReportStepSet, 0, 124) for _, s := range ss { @@ -190,7 +195,8 @@ func (ss ReportDataSourceSet) ModelSteps() ReportStepSet { return out } -func (pp ReportBlockSet) ModelSteps() ReportStepSet { +// ReportSteps returns a ReportStepSet collected from the ReportBlockSet +func (pp ReportBlockSet) ReportSteps() ReportStepSet { out := make(ReportStepSet, 0, 124) for _, p := range pp { @@ -202,6 +208,7 @@ func (pp ReportBlockSet) ModelSteps() ReportStepSet { // Initial ReportBlock struct definition omitted string casting for the BlockID (sorry) // so we need to handle that edge case when reading from DB. +// @todo consider dropping this in the next/one of the following releases func (b *ReportBlock) UnmarshalJSON(data []byte) (err error) { type internalReportBlock ReportBlock i := struct { @@ -239,6 +246,7 @@ func (vv ReportDataSourceSet) Value() (driver.Value, error) { return json.Marsha func (vv *ReportScenarioSet) Scan(src any) error { return sql.ParseJSON(src, vv) } func (vv ReportScenarioSet) Value() (driver.Value, error) { return json.Marshal(vv) } +// Node is a helper for accessing the wrapped QL node to omit nil checks func (f *qlExprWrap) Node() *ql.ASTNode { if f == nil { return nil @@ -246,6 +254,10 @@ func (f *qlExprWrap) Node() *ql.ASTNode { return f.ASTNode } +// UnmarshalJSON parses the wrap into a proper QL node and an optional error +// +// The function can work over JSON strings (where FE provides a QL node) or +// raw expression strings (where FE sends over the easeier stringified expression). func (f *qlExprWrap) UnmarshalJSON(data []byte) (err error) { var aux interface{} if err = json.Unmarshal(data, &aux); err != nil { diff --git a/system/types/report_frame.go b/system/types/report_frame.go index 95b3fc004..6e292e593 100644 --- a/system/types/report_frame.go +++ b/system/types/report_frame.go @@ -5,7 +5,6 @@ import ( "strings" "github.com/cortezaproject/corteza-server/pkg/filter" - "github.com/cortezaproject/corteza-server/pkg/ql" ) type ( @@ -26,7 +25,7 @@ type ( Paging *filter.Paging `json:"paging"` Sort filter.SortExprSet `json:"sort"` - Filter *ql.ASTNode `json:"filter"` + Filter *qlExprWrap `json:"filter"` } FrameDescriptionSet []ReportFrameDescription @@ -46,6 +45,9 @@ type ( Primary bool `json:"primary"` Unique bool `json:"unique"` System bool `json:"system"` + + Multivalue bool `json:"multivalue"` + MultivalueDelimiter string `json:"multivalueDelimiter"` } ReportFrameDefinitionSet []*ReportFrameDefinition @@ -55,7 +57,7 @@ type ( Ref string `json:"ref"` Columns ReportFrameColumnSet `json:"columns"` - Filter *ql.ASTNode `json:"filter"` + Filter *qlExprWrap `json:"filter"` Paging *filter.Paging `json:"paging"` Sort filter.SortExprSet `json:"sort"` } @@ -123,3 +125,15 @@ func (cc ReportFrameColumnSet) OmitSys() ReportFrameColumnSet { func (r ReportFrameRow) String() string { return strings.Join(r, ", ") } + +// FilterBySource returns a set of definitions for the requested identifier +func (dd ReportFrameDefinitionSet) FilterBySource(ident string) ReportFrameDefinitionSet { + out := make(ReportFrameDefinitionSet, 0, len(dd)) + for _, d := range dd { + if d.Source == ident { + out = append(out, d) + } + } + + return out +} diff --git a/system/types/report_test.go b/system/types/report_test.go new file mode 100644 index 000000000..f0c2c1774 --- /dev/null +++ b/system/types/report_test.go @@ -0,0 +1,69 @@ +package types + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestQLWrapParsing(t *testing.T) { + tcc := []struct { + name string + in string + out string + err bool + }{ + { + name: "ast node valid", + in: `{"ref": "or", "args": [{"value": {"@type": "String","@value": "Maria"}}, {"value": {"@type": "String","@value": "Maria"}}]}`, + out: `or("Maria", "Maria")`, + }, + { + name: "ast raw expr valid", + in: `{"raw": "a || b"}`, + out: `or(a, b)`, + }, + { + name: "raw expr valid", + in: `"c || d"`, + out: `or(c, d)`, + }, + { + name: "ast node fnc valid", + in: `{"ref": "year", "args": [{"ref": "now"}]}`, + out: `year(now())`, + }, + { + name: "expr fnc valid", + in: `"year(now())"`, + out: `year(now())`, + }, + { + name: "raw expr empty", + in: `""`, + // this is what Stringer returns + out: ``, + }, + { + name: "ast node empty", + in: `{}`, + // this is what Stringer returns + out: ``, + }, + } + + for _, tc := range tcc { + t.Run(tc.name, func(t *testing.T) { + w := &qlExprWrap{} + + err := json.Unmarshal([]byte(tc.in), &w) + if tc.err { + require.Error(t, err) + } else { + require.NoError(t, err) + require.Equal(t, tc.out, w.ASTNode.String()) + } + }) + } +}