Base pkg/report definition
This commit is contained in:
Tomaž Jerman
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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 "<NIL>"
|
||||
}
|
||||
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 += "<N/A>, "
|
||||
} 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)
|
||||
}
|
||||
@@ -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 ""
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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"`
|
||||
}
|
||||
)
|
||||
@@ -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:], ".")
|
||||
}
|
||||
Reference in New Issue
Block a user