From 4da71e78ade78c470d3b33074cc341f4d12406cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Sun, 27 Jun 2021 13:01:32 +0200 Subject: [PATCH] Tweak relay model definitions and processing * tmp. remove paging * tweak join and group step definitions * better support multi frame requests --- pkg/report/frame.go | 53 ++++++----- pkg/report/model.go | 13 +-- pkg/report/step_group.go | 23 ++--- pkg/report/step_join.go | 88 ++++++++----------- store/rdbms/compose_record_datasource.go | 51 ++++++++--- system/service/report.go | 37 +++++++- system/types/report.go | 17 ++-- tests/reporter/joining_test.go | 2 +- .../integration/report_grouping_base.json | 4 +- .../integration/report_joining_base.json | 6 +- 10 files changed, 173 insertions(+), 121 deletions(-) diff --git a/pkg/report/frame.go b/pkg/report/frame.go index cb3717bd9..e99b0611e 100644 --- a/pkg/report/frame.go +++ b/pkg/report/frame.go @@ -15,11 +15,12 @@ type ( Name string `json:"name"` Source string `json:"source"` Ref string `json:"ref,omitempty"` - RefKey string `json:"refKey,omitempty"` + + RefValue string `json:"refValue,omitempty"` + RelColumn string `json:"relColumn,omitempty"` Columns FrameColumnSet `json:"columns"` Rows FrameRowSet `json:"rows"` - Error error `json:"error"` Paging *filter.Paging `json:"paging"` Sorting *filter.Sorting `json:"sorting"` @@ -44,14 +45,14 @@ type ( FrameDefinitionSet []*FrameDefinition FrameDefinition struct { - Name string - Source string - Ref string - Rows *RowDefinition - Columns FrameColumnSet + Name string `json:"name"` + Source string `json:"source"` + Ref string `json:"ref"` + Rows *RowDefinition `json:"rows"` + Columns FrameColumnSet `json:"columns"` - Paging *filter.Paging - Sorting filter.SortExprSet + Paging *filter.Paging `json:"paging"` + Sorting filter.SortExprSet `json:"sorting"` } ) @@ -143,22 +144,22 @@ func (b CellDefinition) OpToCmp() string { // Slice in place func (f *Frame) Slice(startIndex, size int) (a, b *Frame) { a = &Frame{ - Name: f.Name, - Source: f.Source, - Ref: f.Ref, - RefKey: f.RefKey, + Name: f.Name, + Source: f.Source, + Ref: f.Ref, + RefValue: f.RefValue, + RelColumn: f.RelColumn, Columns: f.Columns, - Error: f.Error, } b = &Frame{ - Name: f.Name, - Source: f.Source, - Ref: f.Ref, - RefKey: f.RefKey, + Name: f.Name, + Source: f.Source, + Ref: f.Ref, + RefValue: f.RefValue, + RelColumn: f.RelColumn, Columns: f.Columns, - Error: f.Error, } a.Rows = f.Rows[startIndex:size] @@ -213,10 +214,18 @@ func (f *Frame) Size() int { } func (f *Frame) FirstRow() FrameRow { + if f.Size() == 0 { + return nil + } + return f.Rows[0] } func (f *Frame) LastRow() FrameRow { + if f.Size() == 0 { + return nil + } + return f.Rows[f.Size()-1] } @@ -228,7 +237,7 @@ func (f *Frame) String() string { out := fmt.Sprintf("n: %10s; src: %10s\n", f.Name, f.Source) if f.Ref != "" { - out += fmt.Sprintf("ref: %10s; key: %10s\n", f.Ref, f.RefKey) + out += fmt.Sprintf("ref: %10s; col: %10s\n; key: %10s\n", f.Ref, f.RelColumn, f.RefValue) } for _, c := range f.Columns { @@ -348,6 +357,10 @@ func (r FrameRow) MarshalJSON() (out []byte, err error) { var s string for i, c := range r { + if c == nil { + continue + } + s, err = cast.ToStringE(c.Get()) if err != nil { return nil, err diff --git a/pkg/report/model.go b/pkg/report/model.go index 369338ccb..b66f94657 100644 --- a/pkg/report/model.go +++ b/pkg/report/model.go @@ -157,6 +157,8 @@ func (m *model) Load(ctx context.Context, dd ...*FrameDefinition) ([]*Frame, err i := 0 if def.Paging != nil && def.Paging.Limit > 0 { i = int(def.Paging.Limit) + } else { + i = -1 } ff, err := l(i + 1) @@ -165,15 +167,16 @@ func (m *model) Load(ctx context.Context, dd ...*FrameDefinition) ([]*Frame, err } dds := FrameDefinitionSet(dd) - for i, f := range ff { + for _, f := range ff { def = dds.FindBySourceRef(f.Source, f.Ref) if def == nil { return nil, fmt.Errorf("unable to find frame definition for frame: src-%s, ref-%s", f.Source, f.Ref) } - ff[i], err = m.calculatePaging(f, def.Paging, def.Sorting) - if err != nil { - return nil, err - } + + // ff[i], err = m.calculatePaging(f, def.Paging, def.Sorting) + // if err != nil { + // return nil, err + // } } return ff, err diff --git a/pkg/report/step_group.go b/pkg/report/step_group.go index fc757de3d..6ef232d42 100644 --- a/pkg/report/step_group.go +++ b/pkg/report/step_group.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "regexp" - "strings" ) type ( @@ -19,8 +18,8 @@ type ( } GroupDefinition struct { - Groups []*GroupKey `json:"groups"` - Columns []GroupColumn `json:"columns"` + Groups []*GroupColumn `json:"groups"` + Columns []*GroupColumn `json:"columns"` Rows *RowDefinition `json:"rows,omitempty"` } @@ -30,22 +29,17 @@ type ( GroupDefinition } - GroupKey struct { + GroupColumn struct { // Name defines the alias for the new column Name string `json:"name"` // Expr defines the expression to transform the column Expr string `json:"expr"` + // Aggregate defines the aggregation function to apply + Aggregate string `json:"aggregate"` // @todo imply from context Kind string `json:"kind"` } - - // Group columns define what columns we wish to produce and what operations - // we should perform over them. - // - // alias: operation: args; for example -- { "total": { "sum": "cost" } } - GroupColumn map[string]AggregateColumn - AggregateColumn map[string]string ) var ( @@ -106,13 +100,6 @@ func (d *stepGroup) Def() *StepDefinition { return &StepDefinition{Group: d.def} } -func (c AggregateColumn) GetOp() string { - for k := range c { - return strings.ToLower(k) - } - return "" -} - // // // // // @todo manual group step implementation for Datasources that don't provide it diff --git a/pkg/report/step_join.go b/pkg/report/step_join.go index 0e6401095..bce04f0ac 100644 --- a/pkg/report/step_join.go +++ b/pkg/report/step_join.go @@ -25,10 +25,12 @@ type ( } JoinStepDefinition struct { - Name string `json:"name"` - Local string `json:"local"` - Foreign string `json:"foreign"` - Rows *RowDefinition `json:"rows,omitempty"` + Name string `json:"name"` + LocalSource string `json:"localSource"` + LocalColumn string `json:"localColumn"` + ForeignSource string `json:"foreignSource"` + ForeignColumn string `json:"foreignColumn"` + Rows *RowDefinition `json:"rows,omitempty"` } ) @@ -42,7 +44,7 @@ func (j *stepJoin) Run(ctx context.Context, dd ...Datasource) (Datasource, error } if len(dd) < 2 { - return nil, fmt.Errorf("foreign join datasources not defined: %s", j.def.localDim()) + return nil, fmt.Errorf("foreign join datasources not defined: %s", j.def.LocalSource) } // @todo multiple joins @@ -59,13 +61,13 @@ func (j *stepJoin) Validate() error { case j.def.Name == "": return errors.New(pfx + "dimension name not defined") - case j.def.localDim() == "": + case j.def.LocalSource == "": return errors.New(pfx + "local dimension not defined") - case j.def.localColumn() == "": + case j.def.LocalColumn == "": return errors.New(pfx + "local column not defined") - case j.def.foreignDim() == "": + case j.def.ForeignSource == "": return errors.New(pfx + "foreign dimension not defined") - case j.def.foreignColumn() == "": + case j.def.ForeignColumn == "": return errors.New(pfx + "foreign column not defined") default: @@ -78,7 +80,7 @@ func (d *stepJoin) Name() string { } func (d *stepJoin) Source() []string { - return []string{d.def.localDim(), d.def.foreignDim()} + return []string{d.def.LocalSource, d.def.ForeignSource} } func (d *stepJoin) Def() *StepDefinition { @@ -98,16 +100,16 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade return func(cap int) ([]*Frame, error) { // determine local and foreign frame definitions - localDef := FrameDefinitionSet(dd).FindBySourceRef(d.Name(), d.def.localDim()) - foreignDef := FrameDefinitionSet(dd).FindBySourceRef(d.Name(), d.def.foreignDim()) + localDef := FrameDefinitionSet(dd).FindBySourceRef(d.Name(), d.def.LocalSource) + foreignDef := FrameDefinitionSet(dd).FindBySourceRef(d.Name(), d.def.ForeignSource) // basic pre-run validation // - definitions if localDef == nil { - return nil, fmt.Errorf("definition for local datasource not found: %s", d.def.localDim()) + return nil, fmt.Errorf("definition for local datasource not found: %s", d.def.LocalSource) } if foreignDef == nil { - return nil, fmt.Errorf("definition for foreign datasource not found: %s", d.def.foreignDim()) + return nil, fmt.Errorf("definition for foreign datasource not found: %s", d.def.ForeignSource) } if localDef.Paging == nil { localDef.Paging = &filter.Paging{} @@ -117,11 +119,11 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade } // - key columns - if localDef.Columns.Find(d.def.localColumn()) < 0 { - return nil, fmt.Errorf("local frame definition must include the key column: %s", d.def.localColumn()) + if len(localDef.Columns) > 0 && localDef.Columns.Find(d.def.LocalColumn) < 0 { + return nil, fmt.Errorf("local frame definition must include the key column: %s", d.def.LocalColumn) } - if foreignDef.Columns.Find(d.def.foreignColumn()) < 0 { - return nil, fmt.Errorf("foreign frame definition must include the key column: %s", d.def.foreignColumn()) + if len(foreignDef.Columns) > 0 && foreignDef.Columns.Find(d.def.ForeignColumn) < 0 { + return nil, fmt.Errorf("foreign frame definition must include the key column: %s", d.def.ForeignColumn) } // based on the passed sort, determine main/sub datasources @@ -174,7 +176,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade foreignDef.Paging.Limit = mainPageCap // - prepare loader, closer - mainLoader, mainCloser, err = prtDS.Partition(ctx, partitionSize, d.def.foreignColumn(), foreignDef) + mainLoader, mainCloser, err = prtDS.Partition(ctx, partitionSize, d.def.ForeignColumn, foreignDef) } else { mainPageCap = defaultPageSize if localDef.Paging != nil && localDef.Paging.Limit > 0 { @@ -199,7 +201,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade if useSubSort { // here we need to slice the partitioned datasource // @todo should this be layed off to the lowe level? - mainFrames, err = d.sliceFramesFurther(mainFrames, d.def.foreignColumn()) + mainFrames, err = d.sliceFramesFurther(mainFrames, d.def.ForeignColumn, d.def.LocalColumn) if err != nil { return nil, err } @@ -222,14 +224,14 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade keys := make([]string, 0, defaultPageSize) keySet := make(map[string]bool) if useSubSort { - mainKeyColIndex = mainFrames[0].Columns.Find(d.def.foreignColumn()) + mainKeyColIndex = mainFrames[0].Columns.Find(d.def.ForeignColumn) if mainKeyColIndex < 0 { - return nil, fmt.Errorf("key column on foreign datasource does not exist: %s", d.def.foreignColumn()) + return nil, fmt.Errorf("key column on foreign datasource does not exist: %s", d.def.ForeignColumn) } } else { - mainKeyColIndex = mainFrames[0].Columns.Find(d.def.localColumn()) + mainKeyColIndex = mainFrames[0].Columns.Find(d.def.LocalColumn) if mainKeyColIndex < 0 { - return nil, fmt.Errorf("key column on local datasource does not exist: %s", d.def.localColumn()) + return nil, fmt.Errorf("key column on local datasource does not exist: %s", d.def.LocalColumn) } } var k string @@ -269,7 +271,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade if useSubSort { // here we use the LOCAL datasource, because it's flipped // - prepare key pre-filter - localDef.Rows = d.keySliceToFilter(d.def.localColumn(), keys).MergeAnd(localDef.Rows) + localDef.Rows = d.keySliceToFilter(d.def.LocalColumn, keys).MergeAnd(localDef.Rows) // - go! loader, closer, err := d.base.Load(ctx, localDef) @@ -305,7 +307,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade r := mf.FirstRow() // get key value - k = mf.RefKey + k = mf.RefValue // get sort value sortBucketKey, err = d.cellsToString(r, sortKeyColIndexes) @@ -332,7 +334,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade prevSortBucketKey = sortBucketKey } - localColIndex := subFrames[0].Columns.Find(d.def.localColumn()) + localColIndex := subFrames[0].Columns.Find(d.def.LocalColumn) for i := range subFrames { subFrames[i], err = d.bucketSort(subFrames[i], buckets, localColIndex, localDef.Sorting...) if err != nil { @@ -357,10 +359,10 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade } // - prepare key pre-filter - foreignDef.Rows = d.keySliceToFilter(d.def.foreignColumn(), keys).MergeAnd(foreignDef.Rows) + foreignDef.Rows = d.keySliceToFilter(d.def.ForeignColumn, keys).MergeAnd(foreignDef.Rows) // - go! - loader, closer, err := prtDS.Partition(ctx, partitionSize, d.def.foreignColumn(), foreignDef) + loader, closer, err := prtDS.Partition(ctx, partitionSize, d.def.ForeignColumn, foreignDef) if closer != nil { defer closer() } @@ -368,7 +370,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loade return nil, err } subFrames, err = loader(0) - subFrames, err = d.sliceFramesFurther(subFrames, d.def.foreignColumn()) + subFrames, err = d.sliceFramesFurther(subFrames, d.def.ForeignColumn, d.def.LocalColumn) if err != nil { return nil, err } @@ -432,7 +434,7 @@ func (d *joinedDataset) keySliceToFilter(col string, keys []string) *RowDefiniti return cf } -func (d *joinedDataset) sliceFramesFurther(ff []*Frame, col string) (out []*Frame, err error) { +func (d *joinedDataset) sliceFramesFurther(ff []*Frame, selfCol, relCol string) (out []*Frame, err error) { outMap := make(map[string]int) cellToString := func(t expr.TypedValue) (string, error) { @@ -446,10 +448,11 @@ func (d *joinedDataset) sliceFramesFurther(ff []*Frame, col string) (out []*Fram i = len(out) outMap[k] = i out = append(out, &Frame{ - RefKey: k, - Columns: ff[0].Columns, - Paging: ff[0].Paging, - Sorting: ff[0].Sorting, + RefValue: k, + RelColumn: relCol, + Columns: ff[0].Columns, + Paging: ff[0].Paging, + Sorting: ff[0].Sorting, }) } @@ -459,7 +462,7 @@ func (d *joinedDataset) sliceFramesFurther(ff []*Frame, col string) (out []*Fram for _, f := range ff { // slice the output; one frame per key var k string - fColI := f.Columns.Find(col) + fColI := f.Columns.Find(selfCol) err = f.WalkRows(func(i int, r FrameRow) error { k, err = cellToString(r[fColI]) push(k, r) @@ -540,16 +543,3 @@ func (d *joinedDataset) cellsToString(row FrameRow, indexes []int) (out string, return } - -func (def *JoinStepDefinition) localDim() string { - return dimensionOf(def.Local) -} -func (def *JoinStepDefinition) localColumn() string { - return columnOf(def.Local) -} -func (def *JoinStepDefinition) foreignDim() string { - return dimensionOf(def.Foreign) -} -func (def *JoinStepDefinition) foreignColumn() string { - return columnOf(def.Foreign) -} diff --git a/store/rdbms/compose_record_datasource.go b/store/rdbms/compose_record_datasource.go index aed8f1bc5..51edfec10 100644 --- a/store/rdbms/compose_record_datasource.go +++ b/store/rdbms/compose_record_datasource.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "fmt" + "reflect" "strings" "github.com/Masterminds/squirrel" @@ -129,23 +130,34 @@ func (r *recordDatasource) Group(d report.GroupDefinition, name string) (bool, e GroupBy(e.String()) } + var e ql.ASTNode + var err error for _, c := range d.Columns { - for alias, op := range c { - o := op.GetOp() - e, err := parser.ParseExpression(fmt.Sprintf("%s(%s)", o, op[o])) + if c.Aggregate != "" { + e, err = parser.ParseExpression(fmt.Sprintf("%s(%s)", c.Aggregate, c.Expr)) + if err != nil { + return false, err + } + } else { + e, err = parser.ParseExpression(c.Expr) if err != nil { return false, err } - - // @todo imply based on context - c := report.MakeColumnOfKind("Number") - c.Name = alias - gCols = append(gCols, c) - - r.levelColumns[alias] = true - q = q. - Column(fmt.Sprintf("%s as `%s`", e.String(), alias)) } + + var col *report.FrameColumn + if c.Kind != "" { + col = report.MakeColumnOfKind(c.Kind) + } else { + // @todo imply based on context + col = report.MakeColumnOfKind("Number") + } + col.Name = c.Name + gCols = append(gCols, col) + r.levelColumns[c.Name] = true + + q = q. + Column(fmt.Sprintf("%s as `%s`", e.String(), c.Name)) } if d.Rows != nil { @@ -407,6 +419,10 @@ func (b *recordDatasource) Cast(row sqlx.ColScanner, out *report.Frame) error { } default: + if isNil(cv) { + continue + } + c, err := c.Caster(cv) if err != nil { return err @@ -498,3 +514,14 @@ func (ds *recordDatasource) rowFilterToString(conjunction string, cc report.Fram return strings.TrimSpace(base), nil } + +func isNil(i interface{}) bool { + if i == nil { + return true + } + switch reflect.TypeOf(i).Kind() { + case reflect.Ptr, reflect.Map, reflect.Array, reflect.Chan, reflect.Slice: + return reflect.ValueOf(i).IsNil() + } + return false +} diff --git a/system/service/report.go b/system/service/report.go index ba2daa111..ec23b5828 100644 --- a/system/service/report.go +++ b/system/service/report.go @@ -313,11 +313,40 @@ func (svc *report) RunFresh(ctx context.Context, src types.ReportDataSourceSet, return } - ff, err := model.Load(ctx, dd...) - if err != nil { - return err + auxdd := make([]*rep.FrameDefinition, 0, len(dd)) + for i, d := range dd { + // first one; nothing special needed + if i == 0 { + auxdd = append(auxdd, d) + continue + } + + // if the current source matches the prev. source, and they both define references, + // they fall into the same chunk. + if (d.Source == dd[i-1].Source) && (d.Ref != "" && dd[i-1].Ref != "") { + auxdd = append(auxdd, d) + continue + } + + // if the current one doesn't fall into the current chunk, process + // the chunk and reset it + ff, err := model.Load(ctx, auxdd...) + if err != nil { + return err + } + out = append(out, ff...) + + auxdd = make([]*rep.FrameDefinition, 0, len(dd)) + auxdd = append(auxdd, d) + } + + if len(auxdd) > 0 { + ff, err := model.Load(ctx, auxdd...) + if err != nil { + return err + } + out = append(out, ff...) } - out = append(out, ff...) // _ = svc.eventbus.WaitFor(ctx, event.ReportAfterUpdate(upd, report)) // return nil diff --git a/system/types/report.go b/system/types/report.go index ee1c99fe0..c82ebd208 100644 --- a/system/types/report.go +++ b/system/types/report.go @@ -48,14 +48,15 @@ type ( } ReportProjection struct { - Title string `json:"title"` - Description string `json:"description"` - Key string `json:"key"` - Kind string `json:"kind"` - Options map[string]interface{} `json:"options,omitempty"` - Elements []interface{} `json:"elements"` - XYWH [4]int `json:"xywh"` - Layout string `json:"layout"` + Title string `json:"title"` + Description string `json:"description"` + Key string `json:"key"` + Kind string `json:"kind"` + Options map[string]interface{} `json:"options,omitempty"` + Elements []interface{} `json:"elements"` + Sources report.StepDefinitionSet `json:"sources"` + XYWH [4]int `json:"xywh"` + Layout string `json:"layout"` } ReportProjectionSet []*ReportProjection diff --git a/tests/reporter/joining_test.go b/tests/reporter/joining_test.go index efa0d140e..9e58c5aa5 100644 --- a/tests/reporter/joining_test.go +++ b/tests/reporter/joining_test.go @@ -499,7 +499,7 @@ func indexJoinedResult(ff []*report.Frame) map[string]*report.Frame { out := make(map[string]*report.Frame) // the first one is the local ds for _, f := range ff[1:] { - out[f.RefKey] = f + out[f.RefValue] = f } return out diff --git a/tests/reporter/testdata/integration/report_grouping_base.json b/tests/reporter/testdata/integration/report_grouping_base.json index f1fe4958a..22d75c889 100644 --- a/tests/reporter/testdata/integration/report_grouping_base.json +++ b/tests/reporter/testdata/integration/report_grouping_base.json @@ -22,8 +22,8 @@ { "name": "by_name", "expr": "first_name", "kind": "String" } ], "columns": [ - { "count": { "count": "*" } }, - { "total": { "sum": "number_of_numbers" } } + { "name": "count", "aggregate": "count", "expr": "*" , "kind": "Number"}, + { "name": "total", "aggregate": "sum", "expr": "number_of_numbers", "kind": "Number" } ] }} ] diff --git a/tests/reporter/testdata/integration/report_joining_base.json b/tests/reporter/testdata/integration/report_joining_base.json index a242bbb08..e2671d0bc 100644 --- a/tests/reporter/testdata/integration/report_joining_base.json +++ b/tests/reporter/testdata/integration/report_joining_base.json @@ -25,8 +25,10 @@ { "join": { "name": "joined", - "local": "users.join_key", - "foreign": "jobs.usr" + "localSource": "users", + "localColumn": "join_key", + "foreignSource": "jobs", + "foreignColumn": "usr" }} ] }