DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Cleanup system/reporter - DAL/pipeline migration

Browse Source
This commit is contained in:
Tomaž Jerman
2022-09-01 16:55:21 +02:00
parent 21fe776cc6
commit 319a29fdc6
8 changed files with 521 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/boot_levels.go
-4
View File
@@ -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())
pkg/filter/filter.go
+27 -8
View File
@@ -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 }
pkg/reportutils/frame_builder.go → system/reporter/frame_builder.go
+32 -19
View File
@@ -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
pkg/reportutils/reportutils.go → system/reporter/reporter.go
+340 -182
View File
@@ -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
}
system/service/report.go
+19 -36
View File
@@ -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
}
system/types/report.go
+17 -5
View File
@@ -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 {
system/types/report_frame.go
+17 -3
View File
@@ -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
}
system/types/report_test.go
+69
View File
@@ -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: `<nil>`,
},
{
name: "ast node empty",
in: `{}`,
// this is what Stringer returns
out: `<nil>`,
},
}
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())
}
})
}
}