From 6751d0ec8b357abf4e9bd3cc8e3b2675765a676d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Sun, 30 May 2021 15:14:25 +0200 Subject: [PATCH] Base pkg/report definition --- pkg/report/datasource.go | 60 ++++ pkg/report/datasource_filtering.go | 62 +++++ pkg/report/frame.go | 417 ++++++++++++++++++++++++++++ pkg/report/model.go | 430 +++++++++++++++++++++++++++++ pkg/report/step_group.go | 118 ++++++++ pkg/report/step_join.go | 198 +++++++++++++ pkg/report/step_load.go | 21 ++ pkg/report/util.go | 39 +++ 8 files changed, 1345 insertions(+) create mode 100644 pkg/report/datasource.go create mode 100644 pkg/report/datasource_filtering.go create mode 100644 pkg/report/frame.go create mode 100644 pkg/report/model.go create mode 100644 pkg/report/step_group.go create mode 100644 pkg/report/step_join.go create mode 100644 pkg/report/step_load.go create mode 100644 pkg/report/util.go diff --git a/pkg/report/datasource.go b/pkg/report/datasource.go new file mode 100644 index 000000000..659537dd5 --- /dev/null +++ b/pkg/report/datasource.go @@ -0,0 +1,60 @@ +package report + +import ( + "context" +) + +type ( + // DatasourceProvider provides access to system datasources, such as ComposeRecords + DatasourceProvider interface { + // Datasource initializes and returns the Datasource the reporter can use + Datasource(context.Context, *LoadStepDefinition) (Datasource, error) + } + + // Loader returns the next Frame from the Datasource + // @todo better memory reuse + Loader func(cap int) ([]*Frame, error) + // Closer closes the Datasource + Closer func() + + DatasourceSet []Datasource + Datasource interface { + Name() string + Load(context.Context, ...*FrameDefinition) (Loader, Closer, error) + } + + // GroupableDatasource is able to provide groupped data + GroupableDatasource interface { + Datasource + Group(GroupDefinition, string) (bool, error) + } + + // @todo TransformableDatasource +) + +// Merge merges the two DatasourceSets and overwrites any duplicates +func (dd DatasourceSet) Merge(mm DatasourceSet) DatasourceSet { +outer: + for _, m := range mm { + for i, d := range dd { + if d.Name() == m.Name() { + dd[i] = m + continue outer + } + } + dd = append(dd, m) + } + + return dd +} + +// Find searches for the Datasource by name +func (dd DatasourceSet) Find(name string) Datasource { + for _, d := range dd { + if d.Name() == name { + return d + } + } + + return nil +} diff --git a/pkg/report/datasource_filtering.go b/pkg/report/datasource_filtering.go new file mode 100644 index 000000000..1ecd69956 --- /dev/null +++ b/pkg/report/datasource_filtering.go @@ -0,0 +1,62 @@ +package report + +type ( + RowDefinition struct { + And []*RowDefinition `json:"and"` + Or []*RowDefinition `json:"or"` + Cells map[string]*CellDefinition `json:"cells"` + } + + CellDefinition struct { + Op string `json:"op"` + Value string `json:"value"` + } +) + +func (base *RowDefinition) MergeAnd(merge *RowDefinition) *RowDefinition { + // 1. merge the two + rr := &RowDefinition{ + And: make([]*RowDefinition, 0, 2), + } + if base != nil { + rr.And = append(rr.And, base) + } + if merge != nil { + rr.And = append(rr.And, merge) + } + + // 2. flatten the tree + // @todo do some more in-depth processing + if len(rr.And) == 1 { + return rr.And[0] + } + if len(rr.And)+len(rr.Cells)+len(rr.Or) == 0 { + return nil + } + + return rr +} + +func (base *RowDefinition) MergeOr(merge *RowDefinition) *RowDefinition { + // 1. merge the two + rr := &RowDefinition{ + Or: make([]*RowDefinition, 0, 2), + } + if base != nil { + rr.Or = append(rr.Or, base) + } + if merge != nil { + rr.Or = append(rr.Or, merge) + } + + // 2. flatten the tree + // @todo do some more in-depth processing + if len(rr.Or) == 1 { + return rr.Or[0] + } + if len(rr.And)+len(rr.Cells)+len(rr.Or) == 0 { + return nil + } + + return rr +} diff --git a/pkg/report/frame.go b/pkg/report/frame.go new file mode 100644 index 000000000..79cd0e767 --- /dev/null +++ b/pkg/report/frame.go @@ -0,0 +1,417 @@ +package report + +import ( + "encoding/json" + "fmt" + "sort" + + "github.com/cortezaproject/corteza-server/pkg/expr" + "github.com/cortezaproject/corteza-server/pkg/filter" + "github.com/spf13/cast" +) + +type ( + Frame struct { + Name string `json:"name"` + Source string `json:"source"` + Ref string `json:"ref"` + + Columns FrameColumnSet `json:"columns"` + Rows FrameRowSet `json:"rows"` + Error error `json:"error"` + + Paging *filter.Paging `json:"paging"` + Sorting *filter.Sorting `json:"sorting"` + + // params to help us perform things in place + startIndex int + size int + sliced bool + } + + FrameRowSet []FrameRow + FrameRow []expr.TypedValue + + frameCellCaster func(in interface{}) (expr.TypedValue, error) + FrameColumnSet []*FrameColumn + FrameColumn struct { + Name string `json:"name"` + Label string `json:"label"` + Kind string `json:"kind"` + Caster frameCellCaster `json:"-"` + } + + FrameDefinitionSet []*FrameDefinition + FrameDefinition struct { + Name string + Source string + Ref string + Rows *RowDefinition + Columns FrameColumnSet + + Paging *filter.Paging + Sorting filter.SortExprSet + } +) + +const ( + columnWildcard = "*" +) + +func MergeFrames(ff ...*Frame) (out *Frame) { + // @todo shape validation + for _, f := range ff { + if out == nil { + out = f + } else { + out.Rows = append(out.Rows, f.PullRows()...) + } + } + return out +} + +func MakeColumnOfKind(k string) *FrameColumn { + return &FrameColumn{ + Kind: k, + Caster: func(in interface{}) (expr.TypedValue, error) { + switch k { + case "Number": + return expr.NewFloat(in) + case "DateTime": + return expr.NewDateTime(in) + case "User", + "Record": + return expr.NewID(in) + case "Checkbox": + return expr.NewBoolean(in) + default: + return expr.NewString(in) + } + }, + } +} + +func KindOf(v expr.TypedValue) string { + // @todo ... + if v == nil { + return "String" + } + + switch v.Type() { + case "Integer", + "UnsignedInteger", + "Float": + return "Number" + case "DateTime": + return "DateTime" + case "ID": + return "Ref" + case "Boolean": + return "Checkbox" + default: + return "String" + } +} + +func (b *CellDefinition) UnmarshalJSON(data []byte) (err error) { + if b == nil { + *b = *(&CellDefinition{}) + } + + aux := make(map[string]string) + if err = json.Unmarshal(data, &aux); err != nil { + return err + } + + for op, val := range aux { + b.Value = val + b.Op = op + } + + return nil +} + +func (b CellDefinition) OpToCmp() string { + switch b.Op { + case "eq": + return "=" + case "ne": + return "!=" + case "lt": + return "<" + case "gt": + return ">" + case "le": + return "<=" + case "ge": + return ">=" + + default: + return "=" + } +} + +func (f *Frame) Sort(ss ...filter.SortExpr) error { + // @todo allow sorting sliced frames? + if f.sliced { + return fmt.Errorf("unable to sort a sliced frame") + } + + colIndex := make(map[string]int) + for _, s := range ss { + colIndex[s.Column] = f.Columns.Find(s.Column) + } + + // we use SliceStable for cases where the database applies some initial sorting + sort.SliceStable(f.Rows, func(i, j int) bool { + for _, s := range ss { + c, ok := f.Rows[i][colIndex[s.Column]].(expr.Comparable) + if !ok { + return true + } + + r, err := c.Compare(f.Rows[j][colIndex[s.Column]]) + if err != nil { + return false + } + + if r != 0 { + if s.Descending { + return r > 0 + } + return r < 0 + } + } + return false + }) + + return nil +} + +// Slice in place +func (f *Frame) Slice(startIndex, size int) (a, b *Frame) { + a = &Frame{ + Name: f.Name, + Source: f.Source, + Ref: f.Ref, + Columns: f.Columns, + Rows: f.Rows, + Error: f.Error, + + sliced: true, + } + b = &Frame{ + Name: f.Name, + Source: f.Source, + Ref: f.Ref, + Columns: f.Columns, + Rows: f.Rows, + Error: f.Error, + + sliced: true, + } + + a.startIndex = startIndex + // +1 to make it easier to work with indexes + a.size = size + 1 + + b.startIndex = size + 1 + b.size = f.Size() - startIndex + + return a, b +} + +func (f *Frame) WalkRows(cb func(i int, r FrameRow) error) (err error) { + limit := len(f.Rows) + if f.sliced { + limit = f.size + } + + for i := f.startIndex; i < limit; i++ { + if err = cb(i, f.Rows[i]); err != nil { + return err + } + } + + return nil +} + +func (f *Frame) WalkRowsR(cb func(i int, r FrameRow) error) (err error) { + i := len(f.Rows) - 1 + if f.sliced { + i = f.startIndex + f.size - 1 + } + + for i = i; i >= f.startIndex; i-- { + if err = cb(i, f.Rows[i]); err != nil { + return err + } + } + + return nil +} + +func (f *Frame) PeekRow(i int) FrameRow { + return f.Rows[f.startIndex+i] +} + +func (f *Frame) PeekRowSafe(i int) FrameRow { + ix := f.startIndex + i + if ix >= f.Size() { + return nil + } + + return f.Rows[ix] +} + +func (f *Frame) Size() int { + if f.sliced { + return f.size - f.startIndex + } + return len(f.Rows) +} + +func (f *Frame) FirstRow() FrameRow { + return f.Rows[f.startIndex] +} + +func (f *Frame) LastRow() FrameRow { + if f.sliced { + return f.Rows[f.startIndex+f.Size()-1] + } + return f.Rows[f.startIndex] +} + +func (f *Frame) PullRows() FrameRowSet { + if !f.sliced { + return f.Rows + } + + return f.Rows[f.startIndex : f.startIndex+f.size] +} + +// @todo nicer formatting and alignment +func (f *Frame) String() string { + if f == nil { + return "" + } + out := fmt.Sprintf("%s; %s; %s\n", f.Name, f.Source, f.Ref) + + for _, c := range f.Columns { + out += fmt.Sprintf("%s<%s>, ", c.Name, c.Kind) + } + out += "\n" + + f.WalkRows(func(i int, r FrameRow) error { + out += fmt.Sprintf("%d| ", i+1) + for _, c := range r { + if c == nil || c == nil { + out += ", " + } else { + v := cast.ToString(c.Get()) + out += fmt.Sprintf("%s, ", v) + } + } + out += "\n" + return nil + }) + + if f.Paging != nil { + out += "\n" + out += fmt.Sprintf("< %s; =%s; > %s", f.Paging.PrevPage.String(), f.Paging.PageCursor.String(), f.Paging.NextPage.String()) + } + + if f.Sorting != nil { + out += "\n" + out += f.Sorting.Sort.String() + } + + out += "\n" + out += fmt.Sprintf("ix %d; len %d", f.startIndex, f.Size()) + + return out +} + +func (cc FrameColumnSet) Find(name string) int { + for i, c := range cc { + if c.Name == name { + return i + } + } + + return -1 +} + +// Receivers to conform to rdbms field matcher +func (c *FrameColumn) IsBoolean() bool { + return c.Kind == "Bool" +} + +func (c *FrameColumn) IsNumeric() bool { + return c.Kind == "Number" +} + +func (c *FrameColumn) IsDateTime() bool { + return c.Kind == "DateTime" +} + +func (c *FrameColumn) IsRef() bool { + // @todo not quite right + return c.Kind == "Record" +} + +func (r FrameRow) ToVars(cc FrameColumnSet) (vv *expr.Vars, err error) { + vv, _ = expr.NewVars(nil) + + // The row + for i, c := range r { + if c == nil { + err = vv.AssignFieldValue(cc[i].Name, nil) + if err != nil { + return nil, err + } + } else { + err := vv.AssignFieldValue(cc[i].Name, c) + if err != nil { + return nil, err + } + } + } + return +} + +func (dd FrameDefinitionSet) Find(name string) *FrameDefinition { + for _, d := range dd { + if d.Name == name { + return d + } + } + + return nil +} + +func (dd FrameDefinitionSet) FindBySourceRef(source, ref string) *FrameDefinition { + for _, d := range dd { + if d.Source == source && d.Ref == ref { + return d + } + } + + return nil +} + +func (r FrameRow) MarshalJSON() (out []byte, err error) { + aux := make([]string, len(r)) + var s string + + for i, c := range r { + s, err = cast.ToStringE(c.Get()) + if err != nil { + return nil, err + } + + aux[i] = s + } + + return json.Marshal(aux) +} diff --git a/pkg/report/model.go b/pkg/report/model.go new file mode 100644 index 000000000..d6873b497 --- /dev/null +++ b/pkg/report/model.go @@ -0,0 +1,430 @@ +package report + +import ( + "context" + "errors" + "fmt" + + "github.com/cortezaproject/corteza-server/pkg/filter" + "github.com/spf13/cast" +) + +type ( + model struct { + steps []Step + datasources DatasourceSet + } + + M interface { + Add(...Step) M + Run(context.Context) error + Load(context.Context, ...*FrameDefinition) ([]*Frame, error) + } + + StepSet []Step + Step interface { + Name() string + Source() []string + Run(context.Context, ...Datasource) (Datasource, error) + Validate() error + Def() *StepDefinition + } + + StepDefinitionSet []*StepDefinition + StepDefinition struct { + Load *LoadStepDefinition `json:"load,omitempty"` + Join *JoinStepDefinition `json:"join,omitempty"` + Group *GroupStepDefinition `json:"group,omitempty"` + // @todo Transform + } + + modelGraphNode struct { + step Step + ds Datasource + + pp []*modelGraphNode + cc []*modelGraphNode + } +) + +func Model(ctx context.Context, sources map[string]DatasourceProvider, dd ...*StepDefinition) (M, error) { + steps := make([]Step, 0, len(dd)) + ss := make(DatasourceSet, 0, len(steps)*2) + + err := func() error { + for _, d := range dd { + switch { + case d.Load != nil: + if sources == nil { + return errors.New("no datasources defined") + } + + s, ok := sources[d.Load.Source] + if !ok { + return fmt.Errorf("unresolved data source: %s", d.Load.Source) + } + ds, err := s.Datasource(ctx, d.Load) + if err != nil { + return err + } + + ss = append(ss, ds) + + case d.Join != nil: + steps = append(steps, &stepJoin{def: d.Join}) + + case d.Group != nil: + steps = append(steps, &stepGroup{def: d.Group}) + + // @todo Transform + + default: + return errors.New("malformed step definition") + } + } + return nil + }() + + if err != nil { + return nil, fmt.Errorf("failed to create the model: %s", err.Error()) + } + + return &model{ + steps: steps, + datasources: ss, + }, nil +} + +func (m *model) Add(ss ...Step) M { + m.steps = append(m.steps, ss...) + return m +} + +func (m *model) Run(ctx context.Context) (err error) { + // initial validation + err = m.validateModel() + if err != nil { + return fmt.Errorf("failed to validate the model: %w", err) + } + + // nothing left to do + if len(m.steps) == 0 { + return nil + } + + // construct the step graph + gg, err := m.buildStepGraph(m.steps, m.datasources) + if err != nil { + return err + } + + m.datasources = nil + for _, n := range gg { + aux, err := m.reduceGraph(ctx, n) + if err != nil { + return err + } + m.datasources = append(m.datasources, aux) + } + + return nil +} + +func (m *model) Load(ctx context.Context, dd ...*FrameDefinition) ([]*Frame, error) { + var err error + + for _, d := range dd { + err = m.applyPaging(d, d.Paging, d.Sorting) + if err != nil { + return nil, err + } + } + + // @todo variable root def + def := dd[0] + + ds := m.datasources.Find(def.Source) + if ds == nil { + return nil, fmt.Errorf("unresolved source: %s", def.Source) + } + + l, c, err := ds.Load(ctx, dd...) + if err != nil { + return nil, err + } + defer c() + + i := 0 + if def.Paging != nil && def.Paging.Limit > 0 { + i = int(def.Paging.Limit) + } + + ff, err := l(i + 1) + if err != nil { + return nil, err + } + + dds := FrameDefinitionSet(dd) + for i, 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 + } + } + + return ff, err +} + +func (m *model) calculatePaging(f *Frame, p *filter.Paging, ss filter.SortExprSet) (*Frame, error) { + if p == nil { + p = &filter.Paging{} + } + + var ( + hasPrev = p.PageCursor != nil + hasNext = f.Size() > int(p.Limit) + out = &filter.Paging{} + ) + + out.Limit = p.Limit + + if hasNext { + f, _ = f.Slice(0, f.Size()-2) + out.NextPage = m.calculatePageCursor(f.LastRow(), f.Columns, ss) + } + + if hasPrev { + out.PrevPage = m.calculatePageCursor(f.FirstRow(), f.Columns, ss) + } + + f.Paging = out + f.Sorting = &filter.Sorting{ + Sort: ss, + } + + return f, nil +} + +func (m *model) calculatePageCursor(r FrameRow, cc FrameColumnSet, ss filter.SortExprSet) *filter.PagingCursor { + out := &filter.PagingCursor{LThen: ss.Reversed()} + + for _, s := range ss { + ci := cc.Find(s.Column) + out.Set(s.Column, r[ci].Get(), s.Descending) + } + + return out +} + +func (m *model) applyPaging(def *FrameDefinition, p *filter.Paging, ss filter.SortExprSet) (err error) { + if p == nil { + return nil + } + + ss, err = p.PageCursor.Sort(ss) + if err != nil { + return err + } + + // @todo somesort of a primary key to avoid edgecases + sort := ss.Clone() + if p.PageCursor != nil && p.PageCursor.ROrder { + sort.Reverse() + } + def.Sorting = sort + + // convert cursor to rows def + if p.PageCursor == nil { + return nil + } + + rd := &RowDefinition{ + Cells: make(map[string]*CellDefinition), + } + kk := p.PageCursor.Keys() + vv := p.PageCursor.Values() + for i, k := range kk { + v, err := cast.ToStringE(vv[i]) + if err != nil { + return err + } + + lt := p.PageCursor.Desc()[i] + if p.PageCursor.IsROrder() { + lt = !lt + } + op := "" + if lt { + op = "lt" + } else { + op = "gt" + } + + rd.Cells[k] = &CellDefinition{ + Op: op, + Value: fmt.Sprintf("'%s'", v), + } + } + def.Rows = rd.MergeAnd(def.Rows) + + return nil +} + +func (m *model) validateModel() error { + if len(m.steps)+len(m.datasources) == 0 { + return errors.New("no model steps defined") + } + + var err error + for _, s := range m.steps { + err = s.Validate() + if err != nil { + return err + } + } + + return nil +} + +func (m *model) buildStepGraph(ss StepSet, dd DatasourceSet) ([]*modelGraphNode, error) { + mp := make(map[string]*modelGraphNode) + + for _, s := range ss { + s := s + + // make sure that the step is in the graph + n, ok := mp[s.Name()] + if !ok { + n = &modelGraphNode{ + step: s, + } + mp[s.Name()] = n + } else { + n.step = s + } + + // make sure the child step is in there + for _, src := range s.Source() { + c, ok := mp[src] + if !ok { + c = &modelGraphNode{ + // will be added later + step: nil, + pp: []*modelGraphNode{n}, + + ds: dd.Find(src), + } + mp[src] = c + } + n.cc = append(n.cc, c) + } + } + + // return all of the root nodes + out := make([]*modelGraphNode, 0, len(ss)) + for _, n := range mp { + if len(n.pp) == 0 { + out = append(out, n) + } + } + + return out, nil +} + +func (m *model) reduceGraph(ctx context.Context, n *modelGraphNode) (out Datasource, err error) { + auxO := make([]Datasource, len(n.cc)) + if len(n.cc) > 0 { + for i, c := range n.cc { + out, err = m.reduceGraph(ctx, c) + if err != nil { + return nil, err + } + auxO[i] = out + } + } + + bail := func() (out Datasource, err error) { + if n.step == nil { + if n.ds != nil { + return n.ds, nil + } + + return out, nil + } + + aux, err := n.step.Run(ctx, auxO...) + if err != nil { + return nil, err + } + + return aux, nil + } + + if n.step == nil { + return bail() + } + + // check if this one can reduce the existing datasources + // + // for now, only "simple branches are supported" + var o Datasource + if len(auxO) > 1 { + return bail() + } else if len(auxO) > 0 { + // use the only available output + o = auxO[0] + } else { + // use own datasource (in case of leaves nodes) + o = n.ds + } + + if n.step.Def().Group != nil { + gds, ok := o.(GroupableDatasource) + if !ok { + return bail() + } + + ok, err = gds.Group(n.step.Def().Group.GroupDefinition, n.step.Name()) + if err != nil { + return nil, err + } else if !ok { + return bail() + } + + out = gds + // we've covered this step with the child step; ignore it + return out, nil + } + + return bail() +} + +// @todo cleanup the bellow two? + +func (sd *StepDefinition) source() string { + switch { + case sd.Load != nil: + return sd.Load.Source + case sd.Group != nil: + return sd.Group.Source + // @todo Transform + default: + return "" + } +} + +func (sd *StepDefinition) name() string { + switch { + case sd.Load != nil: + return sd.Load.Name + case sd.Group != nil: + return sd.Group.Name + // @todo Transform + default: + return "" + } +} diff --git a/pkg/report/step_group.go b/pkg/report/step_group.go new file mode 100644 index 000000000..fc757de3d --- /dev/null +++ b/pkg/report/step_group.go @@ -0,0 +1,118 @@ +package report + +import ( + "context" + "errors" + "fmt" + "regexp" + "strings" +) + +type ( + stepGroup struct { + def *GroupStepDefinition + } + + groupedDataset struct { + def *GroupStepDefinition + ds Datasource + } + + GroupDefinition struct { + Groups []*GroupKey `json:"groups"` + Columns []GroupColumn `json:"columns"` + Rows *RowDefinition `json:"rows,omitempty"` + } + + GroupStepDefinition struct { + Name string `json:"name"` + Source string `json:"source"` + GroupDefinition + } + + GroupKey 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"` + + // @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 ( + simpleExprMatcher = regexp.MustCompile("^\\*|\\w+$") +) + +const ( + stepGroupMaxFramers = 6 + stepGroupMaxFinalizers = 2 +) + +func (j *stepGroup) Run(ctx context.Context, dd ...Datasource) (Datasource, error) { + if len(dd) == 0 { + return nil, fmt.Errorf("unknown group dimension: %s", j.def.Source) + } + + return nil, nil + // @todo + // return &groupedDataset{ + // def: j.def, + // ds: dd[0], + // }, nil +} + +func (j *stepGroup) Validate() error { + pfx := "invalid group step: " + + // base things... + switch { + case j.def.Name == "": + return errors.New(pfx + "dimension name not defined") + + case j.def.Source == "": + return errors.New(pfx + "groupping dimension not defined") + case len(j.def.Groups) == 0: + return errors.New(pfx + "no group defined") + } + + // columns... + for i, g := range j.def.Groups { + if g.Name == "" { + return fmt.Errorf("%sgroup key alias missing for group: %d", pfx, i) + } + } + + return nil +} + +func (d *stepGroup) Name() string { + return d.def.Name +} + +func (d *stepGroup) Source() []string { + return []string{d.def.Source} +} + +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 new file mode 100644 index 000000000..495b15c58 --- /dev/null +++ b/pkg/report/step_join.go @@ -0,0 +1,198 @@ +package report + +import ( + "context" + "errors" + "fmt" + + "github.com/spf13/cast" +) + +type ( + stepJoin struct { + def *JoinStepDefinition + } + + joinedDataset struct { + def *JoinStepDefinition + + base Datasource + foreign Datasource + } + + JoinStepDefinition struct { + Name string `json:"name"` + Local string `json:"local"` + Foreign string `json:"foreign"` + Rows *RowDefinition `json:"rows,omitempty"` + } +) + +func (j *stepJoin) Run(ctx context.Context, dd ...Datasource) (Datasource, error) { + if len(dd) == 0 { + return nil, fmt.Errorf("unknown join datasources") + } + + if len(dd) < 2 { + return nil, fmt.Errorf("foreign join datasources not defined: %s", j.def.localDim()) + } + + // @todo multiple joins + return &joinedDataset{ + def: j.def, + base: dd[0], + foreign: dd[1], + }, nil +} + +func (j *stepJoin) Validate() error { + pfx := "invalid join step: " + switch { + case j.def.Name == "": + return errors.New(pfx + "dimension name not defined") + + case j.def.localDim() == "": + return errors.New(pfx + "local dimension not defined") + case j.def.localColumn() == "": + return errors.New(pfx + "local column not defined") + case j.def.foreignDim() == "": + return errors.New(pfx + "foreign dimension not defined") + case j.def.foreignColumn() == "": + return errors.New(pfx + "foreign column not defined") + + default: + return nil + } +} + +func (d *stepJoin) Name() string { + return d.def.Name +} + +func (d *stepJoin) Source() []string { + return []string{d.def.localDim(), d.def.foreignDim()} +} + +func (d *stepJoin) Def() *StepDefinition { + return &StepDefinition{Join: d.def} +} + +// // // // + +func (d *joinedDataset) Name() string { + return d.def.Name +} + +// @todo allow x-join sorting +// - determine the lead sort datasource (first sort expr. definition); use as base +// - sort the output based on the lead datasource +// +// @todo allow x-join filtering +// +// @todo improve join datasource loading +// use SQL partitioning & grouping to determine chunks that fall into the same group. +func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (Loader, Closer, error) { + // to hold closer references for all underlying datasources + closers := make([]Closer, 0, 10) + + return func(cap int) ([]*Frame, error) { + out := make([]*Frame, 0, 10) + + // fetch base frame + baseDef := FrameDefinitionSet(dd).FindBySourceRef(d.Name(), d.def.localDim()) + if baseDef == nil { + return nil, fmt.Errorf("could not find base definition: %s, %s", d.Name(), d.def.localDim()) + } + baseL, baseC, err := d.base.Load(ctx, baseDef) + if err != nil { + return nil, err + } + closers = append(closers, baseC) + baseFrame, err := baseL(cap) + if err != nil { + return nil, err + } + for i := range baseFrame { + baseFrame[i].Name = baseDef.Name + baseFrame[i].Source = baseDef.Source + baseFrame[i].Ref = baseDef.Ref + } + out = append(out, baseFrame...) + + // fetch foreign frames + // - foreign ref + foreignDef := FrameDefinitionSet(dd).FindBySourceRef(d.Name(), d.def.foreignDim()) + if foreignDef == nil { + return nil, fmt.Errorf("could not find foreign definition: %s", d.foreign.Name()) + } + + // - extract keys + kk := make([]string, 0, baseFrame[0].Size()) + kx := make(map[string]bool) + keyCol := baseFrame[0].Columns.Find(d.def.localColumn()) + if keyCol < 0 { + return nil, fmt.Errorf("could not find local key column: %s", d.def.localColumn()) + } + var ok bool + err = baseFrame[0].WalkRows(func(i int, r FrameRow) (err error) { + c := r[keyCol].Get() + k, err := cast.ToStringE(c) + if err != nil { + return err + } + if ok = kx[k]; !ok { + kk = append(kk, k) + kx[k] = true + } + return nil + }) + + // @todo partitioning + // @todo parallel + fdr := foreignDef.Rows + for _, k := range kk { + f := (&RowDefinition{ + Cells: map[string]*CellDefinition{ + d.def.foreignColumn(): {Op: "eq", Value: "'" + k + "'"}, + }, + }).MergeAnd(fdr) + + foreignDef.Rows = f + foreignL, foreignC, err := d.foreign.Load(ctx, foreignDef) + if err != nil { + return nil, err + } + closers = append(closers, foreignC) + foreignFrame, err := foreignL(cap) + if err != nil { + return nil, err + } + for i := range foreignFrame { + foreignFrame[i].Name = foreignDef.Name + foreignFrame[i].Source = foreignDef.Source + foreignFrame[i].Ref = foreignDef.Ref + } + + out = append(out, foreignFrame...) + } + + return out, nil + }, func() { + for _, c := range closers { + c() + } + }, nil +} + +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/pkg/report/step_load.go b/pkg/report/step_load.go new file mode 100644 index 000000000..423789202 --- /dev/null +++ b/pkg/report/step_load.go @@ -0,0 +1,21 @@ +package report + +type ( + stepLoad struct { + ds Datasource + def *LoadStepDefinition + } + + loadedDataset struct { + def *LoadStepDefinition + ds Datasource + } + + LoadStepDefinition struct { + Name string `json:"name"` + Source string `json:"source"` + Definition map[string]interface{} `json:"definition"` + Columns FrameColumnSet `json:"columns"` + Rows *RowDefinition `json:"rows,omitempty"` + } +) diff --git a/pkg/report/util.go b/pkg/report/util.go new file mode 100644 index 000000000..e787487fd --- /dev/null +++ b/pkg/report/util.go @@ -0,0 +1,39 @@ +package report + +import ( + "reflect" + "strings" +) + +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 +} + +func dimensionOf(k string) string { + pp := strings.Split(k, ".") + if len(pp) < 2 { + return "" + } + + return pp[0] +} + +func columnOf(k string) string { + if k == columnWildcard { + return k + } + + pp := strings.Split(k, ".") + if len(pp) < 2 { + return "" + } + + return strings.Join(pp[1:], ".") +}