DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Tweak relay model definitions and processing

Browse Source 
* tmp. remove paging
* tweak join and group step definitions
* better support multi frame requests
This commit is contained in:
Tomaž Jerman
2021-08-16 09:16:07 +02:00
parent 0ee6ceb55b
commit 4da71e78ad
10 changed files with 173 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/report/frame.go
+33 -20
View File
@@ -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
pkg/report/model.go
+8 -5
View File
@@ -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
pkg/report/step_group.go
+5 -18
View File
@@ -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
pkg/report/step_join.go
+39 -49
View File
@@ -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)
}
store/rdbms/compose_record_datasource.go
+39 -12
View File
@@ -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
}
system/service/report.go
+33 -4
View File
@@ -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
system/types/report.go
+9 -8
View File
@@ -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
tests/reporter/joining_test.go
+1 -1
View File
@@ -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
tests/reporter/testdata/integration/report_grouping_base.json
Vendored
+2 -2
View File
@@ -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" }
]
}}
]
tests/reporter/testdata/integration/report_joining_base.json
Vendored
+4 -2
View File
@@ -25,8 +25,10 @@
{ "join": {
"name": "joined",
"local": "users.join_key",
"foreign": "jobs.usr"
"localSource": "users",
"localColumn": "join_key",
"foreignSource": "jobs",
"foreignColumn": "usr"
}}
]
}