DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Data handling and processing improvements

Browse Source
This commit is contained in:
Tomaž Jerman
2021-09-02 10:43:58 +02:00
parent b73a76263c
commit 00c47ed5c9
19 changed files with 839 additions and 760 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/report/frame.go
+6 -3
View File
@@ -290,13 +290,15 @@ func (f *Frame) String() string {
}
func (f *Frame) CollectCursorValues(r FrameRow, cc ...*filter.SortExpr) *filter.PagingCursor {
// @todo pk and unique things; how should we do it?
cursor := &filter.PagingCursor{LThen: filter.SortExprSet(cc).Reversed()}
for _, c := range cc {
var v interface{}
if r[f.Columns.Find(c.Column)] != nil {
v = r[f.Columns.Find(c.Column)].Get()
}
// the check for existence should be performed way in advanced so we won't bother here
cursor.Set(c.Column, r[f.Columns.Find(c.Column)].Get(), c.Descending)
cursor.Set(c.Column, v, c.Descending)
}
return cursor
@@ -309,6 +311,7 @@ func (f *Frame) CloneMeta() *Frame {
Ref: f.Ref,
RefValue: f.RefValue,
RelColumn: f.RelColumn,
RelSource: f.RelSource,
Columns: f.Columns.Clone(),
Paging: f.Paging.Clone(),
Sort: f.Sort.Clone(),
pkg/report/frame_buffer.go
+47 -26
View File
@@ -14,9 +14,8 @@ type (
frameBuffer struct {
sourceName string
chunkSize uint
loader func(keyFilter *Filter, cap uint) ([]*Frame, error)
closer Closer
loader func(keyFilter *Filter, cap uint) ([]*Frame, error)
closer Closer
localFrames []*Frame
foreignFrames []*Frame
@@ -38,18 +37,14 @@ type (
)
// load loads the next chunk into the buffer based on the provided loader
func (bl *frameBuffer) load(keyFilter *Filter, paged bool) (more bool, err error) {
func (bl *frameBuffer) load(cap uint, keyFilter *Filter) (more bool, err error) {
var ff []*Frame
auxLocal := make([]*Frame, 0, 32)
auxForeign := make([]*Frame, 0, 32)
for {
// Load
if paged {
ff, err = bl.loader(keyFilter, bl.chunkSize+1)
} else {
ff, err = bl.loader(keyFilter, bl.chunkSize)
}
ff, err = bl.loader(keyFilter, cap)
if err != nil {
return false, err
}
@@ -107,7 +102,7 @@ func (bl *frameBuffer) load(keyFilter *Filter, paged bool) (more bool, err error
// Do we need to fetch more?
// When chunk size is 0, we are fetching all
if bl.chunkSize == 0 || !bl.more(bl.localFrames, bl.sortColumns) {
if cap == 0 || !bl.more(bl.localFrames, bl.sortColumns) {
return true, nil
}
}
@@ -303,6 +298,7 @@ func (bl *frameBuffer) calculatePagingCursor(r FrameRow, cols FrameColumnSet, in
// A unique value is also assured at way before.
cursor := &filter.PagingCursor{LThen: filter.SortExprSet(cc).Reversed()}
var foreignFrames []*Frame
var v interface{}
for _, c := range cc {
foreignFrames = nil
@@ -320,9 +316,15 @@ func (bl *frameBuffer) calculatePagingCursor(r FrameRow, cols FrameColumnSet, in
r = f.FirstRow()
}
cursor.Set(c.Column, r[f.Columns.Find(pts[1])].Get(), c.Descending)
if r[f.Columns.Find(pts[1])] != nil {
v = r[f.Columns.Find(pts[1])].Get()
}
cursor.Set(c.Column, v, c.Descending)
} else {
cursor.Set(c.Column, r[cols.Find(c.Column)].Get(), c.Descending)
if r[cols.Find(c.Column)] != nil {
v = r[cols.Find(c.Column)].Get()
}
cursor.Set(c.Column, v, c.Descending)
}
}
@@ -330,7 +332,7 @@ func (bl *frameBuffer) calculatePagingCursor(r FrameRow, cols FrameColumnSet, in
}
// prepareResponse takes the provided buffers, metadata and prepares the result of the step
func prepareResponse(main, sub *frameBuffer, inverted bool, lfd *FrameDefinition, keyColumn string, dscr FrameDescriptionSet) (oo []*Frame, err error) {
func prepareResponse(main, sub *frameBuffer, inverted, processed bool, lfd *FrameDefinition, keyColumn string, dscr FrameDescriptionSet) (oo []*Frame, err error) {
var local, foreign *frameBuffer
// Determine which one was local/foreign
@@ -342,22 +344,41 @@ func prepareResponse(main, sub *frameBuffer, inverted bool, lfd *FrameDefinition
foreign = sub
}
// cut
// - things from local take priority
more := false
oo, more = local.cutLocal(int(lfd.Paging.Limit))
oo = append(oo, local.cutForeign(int(lfd.Paging.Limit))...)
if processed {
// cut
// - things from local take priority
oo, more = local.cutLocal(int(lfd.Paging.Limit))
oo = append(oo, local.cutForeign(int(lfd.Paging.Limit))...)
// - followed by things in foreign
aux, _ := foreign.cutLocal(int(lfd.Paging.Limit))
for _, a := range aux {
a.RelSource = oo[0].Ref
// - followed by things in foreign
aux, _ := foreign.cutLocal(int(lfd.Paging.Limit))
for _, a := range aux {
a.RelSource = oo[0].Ref
}
oo = append(oo, aux...)
oo = append(oo, foreign.cutForeign(int(lfd.Paging.Limit))...)
// paging
oo = local.calculatePagingCursors(oo, lfd.Sort, lfd.Paging.PageCursor, more)
} else {
// whole
// - things from local take priority
oo = local.localFrames
local.localFrames = []*Frame{}
oo = append(oo, local.foreignFrames...)
local.foreignFrames = []*Frame{}
// - followed by things in foreign
aux := foreign.localFrames
for _, a := range aux {
a.RelSource = oo[0].Ref
}
oo = append(oo, aux...)
foreign.localFrames = []*Frame{}
oo = append(oo, foreign.foreignFrames...)
foreign.foreignFrames = []*Frame{}
}
oo = append(oo, aux...)
oo = append(oo, foreign.cutForeign(int(lfd.Paging.Limit))...)
// paging
oo = local.calculatePagingCursors(oo, lfd.Sort, lfd.Paging.PageCursor, more)
if len(oo) == 0 {
return prepareResponseEmpty(lfd, dscr), nil
pkg/report/step_join.go
+72 -330
View File
@@ -8,7 +8,6 @@ import (
"github.com/cortezaproject/corteza-server/pkg/expr"
"github.com/cortezaproject/corteza-server/pkg/filter"
"github.com/cortezaproject/corteza-server/pkg/qlng"
"github.com/spf13/cast"
)
@@ -102,24 +101,20 @@ func (d *joinedDataset) Describe() FrameDescriptionSet {
local := d.local.Describe()
for _, l := range local {
out = append(out,
&FrameDescription{
Source: d.Name(),
Ref: l.Source,
Columns: l.Columns,
},
)
l.Source = d.Name()
if l.Ref == "" {
l.Ref = l.Source
}
out = append(out, l)
}
foreign := d.foreign.Describe()
for _, f := range foreign {
out = append(out,
&FrameDescription{
Source: d.Name(),
Ref: f.Source,
Columns: f.Columns,
},
)
f.Source = d.Name()
if f.Ref == "" {
f.Ref = f.Source
}
out = append(out, f)
}
return out
@@ -145,17 +140,18 @@ func (d *joinedDataset) Partition(partitionSize uint, partitionCol string) (bool
}
func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loader, c Closer, err error) {
dscr := d.Describe()
// Preparation
// - Assure sort columns for paging purposes
// - Assure local/foreign definitions
// Keep a cloned original version so we don't overwrite the initial definition.
oLocalDef, oForeignDef, err := d.prepareDefinitions(FrameDefinitionSet(dd))
oLocalDef, oForeignDef, err := d.prepareDefinitions(FrameDefinitionSet(dd), dscr)
if err != nil {
return
}
// - Validate the sort of the local frame for paging purposes
dscr := d.Describe()
err = d.validateSort(oLocalDef, dscr)
if err != nil {
return
@@ -172,12 +168,12 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
}
// - Preprocess additional paging filtering
var cndMain, apxx, cndSub *qlng.ASTNode
var pp []partialPagingCnd
if oLocalDef.Paging.PageCursor != nil {
cndMain, apxx, cndSub, err = d.calculatePagingFilters(localDef, inverted)
localDef.Filter = merger(&Filter{cndMain}, localDef.Filter, "and")
localDef.Paging.PageCursor = nil
pp, err = d.strategizePaging(localDef, foreignDef, inverted)
if err != nil {
return
}
}
// - Determine the join strategy to use.
@@ -202,11 +198,14 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
// .. apply additional filtering based on page cursors
// . prepare response
isEmpty := false
return func(cap int, paged bool) (oo []*Frame, err error) {
return func(cap int, processed bool) (oo []*Frame, err error) {
var keys []string
if isEmpty {
return
}
if processed {
cap++
}
// The modified flag will help us determine if we need another iteration or not.
// The flag is only set to true if we do any additional modifications in here.
@@ -223,7 +222,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
modified = false
// . Pull data from the main source
more, err = mainLoader.load(nil, paged)
more, err = mainLoader.load(uint(cap), nil)
if err != nil {
return
}
@@ -240,7 +239,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
keyFilter := subLoader.keyFilter(keys)
// . Pull data from the sub source
more, err = subLoader.load(keyFilter, paged)
more, err = subLoader.load(uint(cap), keyFilter)
if err != nil {
return nil, err
}
@@ -266,7 +265,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
// .. Additional filters based on page cursors
if !pagingSatisfied {
m, pagingSatisfied = d.pagingFilter(mainLoader, subLoader, apxx, cndSub)
m, pagingSatisfied = d.pagingFilter(mainLoader, subLoader, pp)
modified = modified || m
}
@@ -279,13 +278,18 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
}
}
return prepareResponse(mainLoader, subLoader, inverted, oLocalDef, d.def.LocalColumn, dscr)
return prepareResponse(mainLoader, subLoader, inverted, processed, oLocalDef, d.def.LocalColumn, dscr)
}, func() {
return
if mainLoader.closer != nil {
mainLoader.closer()
}
if subLoader.closer != nil {
subLoader.closer()
}
}, nil
}
func (d *joinedDataset) prepareDefinitions(dd FrameDefinitionSet) (localDef *FrameDefinition, foreignDef *FrameDefinition, err error) {
func (d *joinedDataset) prepareDefinitions(dd FrameDefinitionSet, dscr FrameDescriptionSet) (localDef *FrameDefinition, foreignDef *FrameDefinition, err error) {
if len(dd) == 0 {
err = errors.New("joining requires at least one frame definition")
return
@@ -301,6 +305,7 @@ func (d *joinedDataset) prepareDefinitions(dd FrameDefinitionSet) (localDef *Fra
Ref: d.def.LocalSource,
Paging: dd[0].Paging,
Sort: dd[0].Sort,
Filter: dd[0].Filter,
}
}
@@ -316,6 +321,18 @@ func (d *joinedDataset) prepareDefinitions(dd FrameDefinitionSet) (localDef *Fra
}
}
if len(localDef.Columns) == 0 {
dscr = d.local.Describe()
sc := dscr.FilterBySource(localDef.Ref)[0]
localDef.Columns = sc.Columns
}
if len(foreignDef.Columns) == 0 {
dscr = d.foreign.Describe()
sc := dscr.FilterBySource(foreignDef.Ref)[0]
foreignDef.Columns = sc.Columns
}
return
}
@@ -363,313 +380,41 @@ func (d *joinedDataset) sliceFrames(ff []*Frame, selfCol, relCol string) (out []
return out, nil
}
// pagingFilter applies additional filtering based on the given page cursor
func (d *joinedDataset) pagingFilter(main, sub *frameBuffer, cndMain, cndSub *qlng.ASTNode) (modified, satisfied bool) {
cutSize := 0
done := false
if cndMain == nil {
return false, true
}
main.walkRowsLocal(func(i int, r FrameRow) error {
if done {
return nil
}
// Firstly we evaluate if the local row falls in the "danger zone"
// (if the row was right on the edge of where the paging cursor filter applied)
if d.eval(cndMain, r, main.localFrames[0].Columns) {
// If we are in the "danger zone", we check what foreign frames don't pass
// the cursor filter.
//
// If the foreign frame does not pass it, we should remove it along with the local row.
if cndSub != nil && !d.eval(cndSub, sub.getByRefValue(r[main.keyColIndex]).FirstRow(), sub.localFrames[0].Columns) {
cutSize++
} else {
done = true
return nil
}
} else {
done = true
return nil
}
return nil
})
if cutSize > 0 {
main.removeLocal(cutSize)
sub.removeLocal(cutSize)
if main.sizeLocal() <= cutSize {
// We removed all of the local buffer so the paging is not yet satisfied
return true, false
}
// We removed the portion of the local buffer, so the paging is satisfied
return true, true
}
return false, true
}
// calculatePagingFilters produces additional filtering that should be done
// on the datasource level and/or in the join logic.
//
// The core logic is extracted from store/rdbms/builders/cursor.go
func (d *joinedDataset) calculatePagingFilters(local *FrameDefinition, inverted bool) (localCondition, localAppendix, foreignCondition *qlng.ASTNode, err error) {
if len(local.Paging.PageCursor.Keys()) == 0 {
return
}
var (
cur = local.Paging.PageCursor
// baseCndAppx is the initial AST for finding rows that match the sort column
// It's basically the second part of the wrap condition (if the value equals)
//
// The correlated string version is: (%s OR ((%s IS NULL AND %s) OR %s = %s))
baseCndAppx = func(field string, checkNull bool, value interface{}) *qlng.ASTNode {
return &qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "and",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "is",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Symbol: field,
},
&qlng.ASTNode{
Ref: "null",
},
},
}, &qlng.ASTNode{
Value: qlng.MakeValueOf("Boolean", checkNull),
},
},
},
&qlng.ASTNode{
Ref: "eq",
Args: qlng.ASTNodeSet{{
Symbol: field,
}, {
// @todo type
Value: qlng.MakeValueOf("String", value),
}},
},
},
}
}
// baseCnd is the initial AST for filtering over the given sort column
//
// The correlated string version is: ((%s IS %s AND %s) OR (%s %s %s))
baseCnd = func(field string, nullVal *qlng.ASTNode, checkNull bool, compOp string, value interface{}, appendix bool) *qlng.ASTNode {
pp := strings.Split(field, ".")
field = pp[len(pp)-1]
out := &qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{&qlng.ASTNode{
Ref: "and",
Args: qlng.ASTNodeSet{&qlng.ASTNode{
Ref: "is",
Args: qlng.ASTNodeSet{{
Symbol: field,
}, nullVal},
}, &qlng.ASTNode{
Value: qlng.MakeValueOf("Boolean", checkNull),
}},
}, &qlng.ASTNode{
Ref: compOp,
Args: qlng.ASTNodeSet{{
Symbol: field,
}, {
// @todo type
Value: qlng.MakeValueOf("String", value),
}},
},
},
}
if appendix {
localAppendix = baseCndAppx(field, checkNull, value)
return &qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{
out,
localAppendix,
},
}
}
return out
}
// wrapCnd is the conjunction between two paging cursor columns
//
// The correlated string version is: (%s OR (((%s IS NULL AND %s) OR %s = %s) AND %s))
wrapCnd = func(base *qlng.ASTNode, field string, value interface{}, checkNull bool, condition *qlng.ASTNode) *qlng.ASTNode {
pp := strings.Split(field, ".")
field = pp[len(pp)-1]
return &qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{
base,
&qlng.ASTNode{
Ref: "and",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "and",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "is",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Symbol: field,
},
&qlng.ASTNode{
Ref: "null",
},
},
}, &qlng.ASTNode{
Value: qlng.MakeValueOf("Boolean", checkNull),
},
},
},
&qlng.ASTNode{
Ref: "eq",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Symbol: field,
},
&qlng.ASTNode{
// @todo type
Value: qlng.MakeValueOf("String", value),
},
},
},
},
},
condition,
},
},
},
}
}
)
var (
cc = cur.Keys()
vv = cur.Values()
ltOp = map[bool]string{
true: "lt",
false: "gt",
}
notOp = map[bool]*qlng.ASTNode{
true: {Ref: "nnull"},
false: {Ref: "null"},
}
isNull = func(i int, neg bool) bool {
if (isNil(vv[i]) && !neg) || (!isNil(vv[i]) && neg) {
return true
}
return false
}
)
// Determine the point at which we switch local sorts and foreign sorts
sourceDelimiter := len(cc) - 1
for j := range cc {
if j > 0 {
if strings.Contains(cc[j-1], ".") != strings.Contains(cc[j], ".") {
sourceDelimiter = j
break
}
}
}
// Some temporary variables to avoid initialization
var tmp []string
var field string
calculateAST := func(cc []string, vv []interface{}, dsc []bool, cut bool) (cnd *qlng.ASTNode) {
// going from the last key/column to the 1st one
for i := len(cc) - 1; i >= 0; i-- {
// We need to cut off the values that are before the cursor (when ascending)
// and vice-versa for descending.
lt := dsc[i]
if cut && cur.IsROrder() {
lt = !lt
}
op := ltOp[lt]
tmp = strings.Split(cc[i], ".")
field = tmp[len(tmp)-1]
base := baseCnd(field, notOp[!lt], isNull(i, lt), op, vv[i], cut && i == len(cc)-1)
if cnd == nil {
cnd = base
} else {
cnd = wrapCnd(base, field, vv[i], isNull(i, false), cnd)
}
}
return
}
// when there is no delimiter we can fully filter the ds
if sourceDelimiter == len(cc)-1 {
localCondition = calculateAST(cc, vv, cur.Desc(), false)
} else {
localCondition = calculateAST(cc[0:sourceDelimiter], vv[0:sourceDelimiter], cur.Desc()[0:sourceDelimiter], true)
foreignCondition = calculateAST(cc[sourceDelimiter:], vv[sourceDelimiter:], cur.Desc()[sourceDelimiter:], false)
}
return
}
func (d *joinedDataset) validateSort(def *FrameDefinition, dd FrameDescriptionSet) (err error) {
sortDS := ""
auxSS := make(filter.SortExprSet, 0, len(def.Sort))
// Get the last sorting delimiter
for i := len(def.Sort) - 1; i >= 0; i-- {
s := def.Sort[i]
for i := len(def.Sort) - 2; i >= 0; i-- {
aa := strings.Split(def.Sort[i].Column, ".")
bb := strings.Split(def.Sort[i+1].Column, ".")
spts := strings.Split(s.Column, ".")
if len(spts) == 1 && sortDS != "" {
if len(aa) != len(bb) || (len(aa) > 1 && aa[0] != aa[1]) {
auxSS = append(auxSS, def.Sort[i+1])
break
}
if len(spts) > 1 {
if sortDS == "" {
sortDS = spts[0]
} else if sortDS != spts[0] {
break
}
auxSS = append(auxSS, def.Sort[i+1])
if len(aa) > 1 {
sortDS = aa[0]
}
auxSS = append(auxSS, s)
}
// The first one is always local so this is ok
localDscr := dd[0]
var dscr *FrameDescription
// When local, ref is omitted
if sortDS == "" {
// Do this to avoid extra work afterwords
auxSS = def.Sort
dscr = localDscr
sortDS = dscr.Ref
} else {
dscr = dd.FilterByRef(sortDS)[0]
}
// Check if we're sorting by a unique value
if sortDS == "" {
sortDS = def.Ref
}
dscr := dd.FilterByRef(sortDS)[0]
def.Sort = func() filter.SortExprSet {
unique := ""
for _, c := range dscr.Columns {
@@ -682,7 +427,8 @@ func (d *joinedDataset) validateSort(def *FrameDefinition, dd FrameDescriptionSe
}
}
}
if sortDS == def.Ref {
if sortDS == localDscr.Ref {
return append(def.Sort, &filter.SortExpr{Column: unique, Descending: auxSS.LastDescending()})
} else {
return append(def.Sort, &filter.SortExpr{Column: fmt.Sprintf("%s.%s", sortDS, unique), Descending: auxSS.LastDescending()})
@@ -720,10 +466,6 @@ func (d *joinedDataset) prepareSorting(local, foreign *FrameDefinition) (inverte
}
}
if foreignDS != "" && foreignDS != foreign.Ref {
return false, fmt.Errorf("foreign datasource in sort expression not found: %s", foreignDS)
}
local.Sort = localSS
foreign.Sort = append(foreignSS, foreign.Sort...)
pkg/report/step_join_strategies.go
+360 -122
View File
@@ -8,9 +8,20 @@ import (
"github.com/cortezaproject/corteza-server/pkg/expr"
"github.com/cortezaproject/corteza-server/pkg/filter"
"github.com/cortezaproject/corteza-server/pkg/qlng"
"github.com/spf13/cast"
)
type (
// partialPagingCnd is a wrapper struck for parts of the processed paging cursor
partialPagingCnd struct {
filterCut *qlng.ASTNode
filterInclude *qlng.ASTNode
// when ref is "" that means local source
ref string
}
)
// strategizeLoad uses the given context to determine what join strategy we should
// use to achieve the correct result as optimally as possible
func (d *joinedDataset) strategizeLoad(ctx context.Context, inverted bool, local, foreign *FrameDefinition) (ml *frameBuffer, sl *frameBuffer, err error) {
@@ -57,9 +68,8 @@ func (d *joinedDataset) stratLocalMain(ctx context.Context, local, foreign *Fram
loader: func(_ *Filter, cap uint) ([]*Frame, error) {
return ldr(int(cap), false)
},
closer: clsr,
chunkSize: local.Paging.Limit,
sorting: local.Sort,
closer: clsr,
sorting: local.Sort,
// Overfetch frames when the last two entries define the same sort.
// This is required to support paging.
@@ -126,7 +136,6 @@ func (d *joinedDataset) stratLocalMain(ctx context.Context, local, foreign *Fram
keyCol: d.def.ForeignColumn,
keyColIndex: -1,
chunkSize: foreign.Paging.Limit,
loader: func(keyFilter *Filter, cap uint) ([]*Frame, error) {
foreign.Filter = merger(ffilter.Clone(), keyFilter, "and")
@@ -178,124 +187,8 @@ func (d *joinedDataset) stratLocalMain(ctx context.Context, local, foreign *Fram
}
func (d *joinedDataset) stratForeignMain(ctx context.Context, local, foreign *FrameDefinition) (ml *frameBuffer, sl *frameBuffer, err error) {
// - main
mainDS, ok := d.foreign.(PartitionableDatasource)
if !ok {
// @todo allow alternatives also
err = fmt.Errorf("foreign datasource is not partitionable")
return
}
partitionSize := foreign.Paging.Limit
ok, err = mainDS.Partition(partitionSize, d.def.ForeignColumn)
if err != nil {
return
}
if !ok {
err = fmt.Errorf("foreign datasource is not partitionable")
return
}
ldr, clsr, err := mainDS.Load(ctx, foreign)
ml = &frameBuffer{
// meta...
keyCol: d.def.ForeignColumn,
keyColIndex: -1,
loader: func(_ *Filter, cap uint) ([]*Frame, error) {
return ldr(int(cap), false)
},
closer: clsr,
// This is the maximum if all of the pages have full partitions
chunkSize: local.Paging.Limit * partitionSize,
sorting: foreign.Sort,
// In case where there is no unique column present in the sort, we need
// to overfetch rows until we reach a row that has a different sort.
more: func(ff []*Frame, sc []int) bool {
f := ff[0]
// No sorting, we don't care
if len(foreign.Sort) == 0 {
return false
}
// This is the last frame we can pull out, we don't care
if uint(f.Size()) < foreign.Paging.Limit {
return false
}
// With sorting and using primary/unique columns, we don't care
for _, s := range foreign.Sort {
c := foreign.Columns[foreign.Columns.Find(s.Column)]
if c.Primary || c.Unique {
return false
}
}
// Only the last frame of the buffer is passed in here, so when it is too small
if f.Size() <= 1 {
return true
}
// With sorting and regular'ol columns, we care only if the over-fetched row
// is the same as the last requested row.
return f.LastRow().Compare(f.LastLastRow(), sc...) == 0
},
postFetch: func(f []*Frame) ([]*Frame, error) {
return d.sliceFrames(f, d.def.ForeignColumn, d.def.LocalColumn)
},
}
// - sub
if d.partitioned {
ok, err = (d.local.(PartitionableDatasource)).Partition(d.partitionSize, d.partitionCol)
if err != nil {
return
}
if !ok {
err = fmt.Errorf("foreign datasource is not partitionable")
return
}
}
lfilter := local.Filter.Clone()
sl = &frameBuffer{
// meta...
keyCol: d.def.LocalColumn,
keyColIndex: -1,
loader: func(keyFilter *Filter, _ uint) ([]*Frame, error) {
// @todo not ok for multiple pulls!
local.Filter = merger(lfilter.Clone(), keyFilter, "and")
loader, closer, err := d.local.Load(ctx, local)
if closer != nil {
defer closer()
}
if err != nil {
return nil, err
}
return loader(0, false)
},
sorting: local.Sort,
more: func(ff []*Frame, sc []int) bool {
return false
},
}
if d.partitioned {
sl.postFetch = func(f []*Frame) ([]*Frame, error) {
ff, err := d.sliceFrames(f, d.def.LocalColumn, d.def.ForeignColumn)
for _, f := range ff {
f.RefValue = ""
f.RelColumn = ""
}
return ff, err
}
}
return
// @todo this will be added at the very and as it's an inverse of the above strategy
return nil, nil, fmt.Errorf("unable to sort by a joined column")
}
// // // // // // // // // // // // // // // // // // // // // // // // //
@@ -481,3 +374,348 @@ func (d *joinedDataset) sortBufferFrames(local *frameBuffer, ss filter.SortExprS
return nil
}
// // // // // // // // // // // // // // // // // // // // // // // // //
// Paging
func (d *joinedDataset) strategizePaging(local, foreign *FrameDefinition, inverted bool) (pp []partialPagingCnd, err error) {
pp, err = d.calculatePagingFilters(local, inverted)
if err != nil {
return
}
local.Filter = merger(&Filter{pp[0].filterCut}, local.Filter, "and")
local.Paging.PageCursor = nil
return
}
// calculatePagingFilters produces additional filtering that should be done
// on the datasource level and/or in the join logic.
//
// The core logic is extracted from store/rdbms/builders/cursor.go
func (d *joinedDataset) calculatePagingFilters(local *FrameDefinition, inverted bool) (partials []partialPagingCnd, err error) {
if len(local.Paging.PageCursor.Keys()) == 0 {
return
}
var localAppendix *qlng.ASTNode
var (
cur = local.Paging.PageCursor
// baseCndAppx is the initial AST for finding rows that match the sort column
// It's basically the second part of the wrap condition (if the value equals)
//
// The correlated string version is: (%s OR ((%s IS NULL AND %s) OR %s = %s))
baseCndAppx = func(field string, checkNull bool, value interface{}) *qlng.ASTNode {
return &qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "and",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "is",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Symbol: field,
},
&qlng.ASTNode{
Ref: "null",
},
},
}, &qlng.ASTNode{
Value: qlng.MakeValueOf("Boolean", checkNull),
},
},
},
&qlng.ASTNode{
Ref: "eq",
Args: qlng.ASTNodeSet{{
Symbol: field,
}, {
// @todo type
Value: qlng.MakeValueOf("String", value),
}},
},
},
}
}
// baseCnd is the initial AST for filtering over the given sort column
//
// The correlated string version is: ((%s IS %s AND %s) OR (%s %s %s))
baseCnd = func(field string, nullVal *qlng.ASTNode, checkNull bool, compOp string, value interface{}, appendix bool) *qlng.ASTNode {
pp := strings.Split(field, ".")
field = pp[len(pp)-1]
out := &qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{&qlng.ASTNode{
Ref: "and",
Args: qlng.ASTNodeSet{&qlng.ASTNode{
Ref: "is",
Args: qlng.ASTNodeSet{{
Symbol: field,
}, nullVal},
}, &qlng.ASTNode{
Value: qlng.MakeValueOf("Boolean", checkNull),
}},
}, &qlng.ASTNode{
Ref: compOp,
Args: qlng.ASTNodeSet{{
Symbol: field,
}, {
// @todo type
Value: qlng.MakeValueOf("String", value),
}},
},
},
}
if appendix {
localAppendix = baseCndAppx(field, checkNull, value)
return &qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{
out,
localAppendix,
},
}
}
return out
}
// wrapCnd is the conjunction between two paging cursor columns
//
// The correlated string version is: (%s OR (((%s IS NULL AND %s) OR %s = %s) AND %s))
wrapCnd = func(base *qlng.ASTNode, field string, value interface{}, checkNull bool, condition *qlng.ASTNode) *qlng.ASTNode {
pp := strings.Split(field, ".")
field = pp[len(pp)-1]
return &qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{
base,
&qlng.ASTNode{
Ref: "and",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "or",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "and",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Ref: "is",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Symbol: field,
},
&qlng.ASTNode{
Ref: "null",
},
},
}, &qlng.ASTNode{
Value: qlng.MakeValueOf("Boolean", checkNull),
},
},
},
&qlng.ASTNode{
Ref: "eq",
Args: qlng.ASTNodeSet{
&qlng.ASTNode{
Symbol: field,
},
&qlng.ASTNode{
// @todo type
Value: qlng.MakeValueOf("String", value),
},
},
},
},
},
condition,
},
},
},
}
}
)
var (
cc = cur.Keys()
vv = cur.Values()
ltOp = map[bool]string{
true: "lt",
false: "gt",
}
notOp = map[bool]*qlng.ASTNode{
true: {Ref: "nnull"},
false: {Ref: "null"},
}
isNull = func(i int, neg bool) bool {
if (isNil(vv[i]) && !neg) || (!isNil(vv[i]) && neg) {
return true
}
return false
}
)
// Some temporary variables to avoid initialization
var tmp []string
var field string
calculateAST := func(cc []string, vv []interface{}, dsc []bool, cut bool) (cnd *qlng.ASTNode) {
// going from the last key/column to the 1st one
for i := len(cc) - 1; i >= 0; i-- {
// We need to cut off the values that are before the cursor (when ascending)
// and vice-versa for descending.
lt := dsc[i]
if cut && cur.IsROrder() {
lt = !lt
}
op := ltOp[lt]
tmp = strings.Split(cc[i], ".")
field = tmp[len(tmp)-1]
base := baseCnd(field, notOp[!lt], isNull(i, lt), op, vv[i], cut && i == len(cc)-1)
if cnd == nil {
cnd = base
} else {
cnd = wrapCnd(base, field, vv[i], isNull(i, false), cnd)
}
}
return
}
// Edge case where only 1 source is used
ref := ""
for j := range cc {
if j > 0 {
aa := strings.Split(cc[j-1], ".")
bb := strings.Split(cc[j], ".")
if len(aa) > 1 {
ref = aa[0]
}
if len(aa) != len(bb) || aa[0] != bb[0] {
goto out
}
}
}
partials = append(partials, partialPagingCnd{
filterCut: calculateAST(cc, vv, cur.Desc(), false),
ref: ref,
})
return
out:
// Create a partial filter for each bit of the cursor
startIndex := 0
for j := range cc {
if j == startIndex {
continue
}
aa := strings.Split(cc[startIndex], ".")
bb := strings.Split(cc[j], ".")
if len(aa) != len(bb) || aa[0] != bb[0] {
aux := partialPagingCnd{
filterCut: calculateAST(cc[startIndex:j], vv[startIndex:j], cur.Desc()[startIndex:j], startIndex == 0),
}
if len(aa) > 1 {
aux.ref = aa[0]
}
if startIndex == 0 {
aux.filterInclude = localAppendix
}
partials = append(partials, aux)
startIndex = j
}
if j == len(cc)-1 {
aux := partialPagingCnd{
filterCut: calculateAST(cc[startIndex:], vv[startIndex:], cur.Desc()[startIndex:], startIndex == 0),
}
if len(aa) > 1 {
aux.ref = aa[0]
}
if startIndex == 0 {
aux.filterInclude = localAppendix
}
partials = append(partials, aux)
}
}
return
}
// pagingFilter applies additional filtering based on the given page cursor
func (d *joinedDataset) pagingFilter(main, sub *frameBuffer, pp []partialPagingCnd) (modified, satisfied bool) {
cutSize := 0
done := false
if len(pp) <= 1 {
// Already satisfied by the DS
return false, true
}
inclCondition := pp[0]
fCondition := pp[1]
main.walkRowsLocal(func(i int, r FrameRow) error {
if done {
return nil
}
// Firstly we evaluate if the local row falls in the "danger zone"
// (if the row was right on the edge of where the paging cursor filter applied)
if d.eval(inclCondition.filterInclude, r, main.localFrames[0].Columns) {
// If we are in the "danger zone", we check what foreign frames don't pass
// the cursor filter.
//
// If the foreign frame does not pass it, we should remove it along with the local row.
if fCondition.filterCut != nil && !d.eval(fCondition.filterCut, sub.getByRefValue(r[main.keyColIndex]).FirstRow(), sub.localFrames[0].Columns) {
cutSize++
} else {
done = true
return nil
}
} else {
done = true
return nil
}
return nil
})
if cutSize > 0 {
main.removeLocal(cutSize)
main.removeForeign(cutSize)
sub.removeLocal(cutSize)
sub.removeForeign(cutSize)
if main.sizeLocal() <= cutSize {
// We removed all of the local buffer so the paging is not yet satisfied
return true, false
}
// We removed the portion of the local buffer, so the paging is satisfied
return true, true
}
return false, true
}
store/rdbms/compose_record_datasource.go
+33 -4
View File
@@ -85,6 +85,7 @@ func (r *recordDatasource) Describe() report.FrameDescriptionSet {
return report.FrameDescriptionSet{
&report.FrameDescription{
Source: r.Name(),
Ref: r.Name(),
Columns: r.cols,
},
}
@@ -314,6 +315,12 @@ func (r *recordDatasource) load(ctx context.Context, def *report.FrameDefinition
}
}
// Make sure results are always sorted at least by primary keys
var canPage bool
if canPage, err = r.validateSort(def); err != nil {
return
}
// Cloned sorting instructions for the actual sorting
// Original must be kept for cursor creation
sort = def.Sort.Clone()
@@ -380,7 +387,7 @@ func (r *recordDatasource) load(ctx context.Context, def *report.FrameDefinition
}
i = 0
if processed {
return r.calculatePaging(out, def.Sort, uint(cap), def.Paging.PageCursor), nil
return r.calculatePaging(out, def.Sort, uint(cap), def.Paging.PageCursor, canPage), nil
}
return out, nil
}
@@ -388,7 +395,7 @@ func (r *recordDatasource) load(ctx context.Context, def *report.FrameDefinition
if i > 0 {
if processed {
return r.calculatePaging([]*report.Frame{f}, def.Sort, uint(cap), def.Paging.PageCursor), nil
return r.calculatePaging([]*report.Frame{f}, def.Sort, uint(cap), def.Paging.PageCursor, canPage), nil
} else {
return []*report.Frame{f}, nil
}
@@ -484,7 +491,7 @@ func (r *recordDatasource) baseQuery(f *report.Filter) (sqb squirrel.SelectBuild
return sqb, nil
}
func (b *recordDatasource) calculatePaging(out []*report.Frame, sorting filter.SortExprSet, limit uint, cursor *filter.PagingCursor) []*report.Frame {
func (b *recordDatasource) calculatePaging(out []*report.Frame, sorting filter.SortExprSet, limit uint, cursor *filter.PagingCursor, canPage bool) []*report.Frame {
for _, o := range out {
var (
hasPrev = cursor != nil
@@ -508,7 +515,7 @@ func (b *recordDatasource) calculatePaging(out []*report.Frame, sorting filter.S
hasPrev, hasNext = hasNext, hasPrev
}
if ignoreLimit {
if ignoreLimit || !canPage {
return out
}
@@ -619,3 +626,25 @@ func (r *recordDatasource) sortExpr(sorting filter.SortExprSet) ([]string, error
return ss, nil
}
func (r *recordDatasource) validateSort(def *report.FrameDefinition) (canPage bool, err error) {
unique := ""
def.Sort = func() filter.SortExprSet {
for _, c := range r.cols {
if c.Primary || c.Unique {
if unique == "" {
unique = c.Name
}
if def.Sort.Get(c.Name) != nil {
unique = c.Name
return def.Sort
}
}
}
if unique != "" {
return append(def.Sort, &filter.SortExpr{Column: unique, Descending: def.Sort.LastDescending()})
}
return def.Sort
}()
return unique != "", nil
}
tests/reporter/3001_joining_base_test.go
+5 -5
View File
@@ -32,7 +32,7 @@ func Test3001_joining_base(t *testing.T) {
"aa_05, aa :: 05")
// aa_01
foreign = ix["aa_01"]
foreign = ix["bb/aa/aa_01"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
@@ -45,7 +45,7 @@ func Test3001_joining_base(t *testing.T) {
"bb_03, aa_01, bb :: 03")
// aa_02
foreign = ix["aa_02"]
foreign = ix["bb/aa/aa_02"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
@@ -57,7 +57,7 @@ func Test3001_joining_base(t *testing.T) {
"bb_05, aa_02, bb :: 05")
// aa_03
foreign = ix["aa_03"]
foreign = ix["bb/aa/aa_03"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
@@ -68,7 +68,7 @@ func Test3001_joining_base(t *testing.T) {
"bb_06, aa_03, bb :: 06")
// aa_04
foreign = ix["aa_04"]
foreign = ix["bb/aa/aa_04"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
@@ -79,7 +79,7 @@ func Test3001_joining_base(t *testing.T) {
"bb_07, aa_04, bb :: 07")
// aa_05
foreign = ix["aa_05"]
foreign = ix["bb/aa/aa_05"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
tests/reporter/3002_joining_base_nested_test.go
+17 -17
View File
@@ -25,12 +25,12 @@ func Test3002_joining_base_nested(t *testing.T) {
ix := indexJoinedResult(ff)
_ = ix
// // joined_aux -- the initial join
// // joined -- the initial join
// local
local = ff[0]
h.a.Equal("pk<String>, label<String>", local.Columns.OmitSys().String())
h.a.Equal("joined_aux", local.Source)
h.a.Equal("joined", local.Source)
h.a.Equal("aa", local.Ref)
checkRows(h, local,
"aa_01, aa :: 01",
@@ -40,10 +40,10 @@ func Test3002_joining_base_nested(t *testing.T) {
"aa_05, aa :: 05")
// aa_01
foreign = ix["joined_aux/aa_01"]
foreign = ix["bb/aa/aa_01"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_aux", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_01", foreign.RefValue)
@@ -53,10 +53,10 @@ func Test3002_joining_base_nested(t *testing.T) {
"bb_03, aa_01, bb :: 03")
// aa_02
foreign = ix["joined_aux/aa_02"]
foreign = ix["bb/aa/aa_02"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_aux", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_02", foreign.RefValue)
@@ -65,10 +65,10 @@ func Test3002_joining_base_nested(t *testing.T) {
"bb_05, aa_02, bb :: 05")
// aa_03
foreign = ix["joined_aux/aa_03"]
foreign = ix["bb/aa/aa_03"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_aux", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_03", foreign.RefValue)
@@ -76,10 +76,10 @@ func Test3002_joining_base_nested(t *testing.T) {
"bb_06, aa_03, bb :: 06")
// aa_04
foreign = ix["joined_aux/aa_04"]
foreign = ix["bb/aa/aa_04"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_aux", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_04", foreign.RefValue)
@@ -87,10 +87,10 @@ func Test3002_joining_base_nested(t *testing.T) {
"bb_07, aa_04, bb :: 07")
// aa_05
foreign = ix["joined_aux/aa_05"]
foreign = ix["bb/aa/aa_05"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_aux", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_05", foreign.RefValue)
@@ -102,7 +102,7 @@ func Test3002_joining_base_nested(t *testing.T) {
// The other foreign
// aa_01
foreign = ix["joined/aa_01"]
foreign = ix["cc/aa/aa_01"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
@@ -114,7 +114,7 @@ func Test3002_joining_base_nested(t *testing.T) {
"cc_02, aa_01, cc :: 02")
// aa_02
foreign = ix["joined/aa_02"]
foreign = ix["cc/aa/aa_02"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
@@ -126,7 +126,7 @@ func Test3002_joining_base_nested(t *testing.T) {
"cc_04, aa_02, cc :: 04")
// aa_03
foreign = ix["joined/aa_03"]
foreign = ix["cc/aa/aa_03"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
@@ -138,7 +138,7 @@ func Test3002_joining_base_nested(t *testing.T) {
"cc_06, aa_03, cc :: 06")
// aa_04
foreign = ix["joined/aa_04"]
foreign = ix["cc/aa/aa_04"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
@@ -149,7 +149,7 @@ func Test3002_joining_base_nested(t *testing.T) {
"cc_07, aa_04, cc :: 07")
// aa_05
foreign = ix["joined/aa_05"]
foreign = ix["cc/aa/aa_05"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
tests/reporter/3003_joining_base_nested_complex_test.go
+34 -31
View File
@@ -7,6 +7,9 @@ import (
)
func Test3003_joining_base_nested_complex(t *testing.T) {
t.Skip("@todo")
var (
ctx, h, s = setup(t)
m, _, dd = loadScenario(ctx, s, t, h)
@@ -29,7 +32,7 @@ func Test3003_joining_base_nested_complex(t *testing.T) {
local = ff[0]
h.a.Equal("pk<String>, label<String>", local.Columns.OmitSys().String())
h.a.Equal("joined_lft", local.Source)
h.a.Equal("joined", local.Source)
h.a.Equal("aa", local.Ref)
checkRows(h, local,
"aa_01, aa :: 01",
@@ -38,10 +41,10 @@ func Test3003_joining_base_nested_complex(t *testing.T) {
"aa_04, aa :: 04",
"aa_05, aa :: 05")
foreign = ix["joined_lft/aa_01"]
foreign = ix["bb/aa/aa_01"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_lft", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_01", foreign.RefValue)
@@ -50,10 +53,10 @@ func Test3003_joining_base_nested_complex(t *testing.T) {
"bb_02, aa_01, bb :: 02",
"bb_03, aa_01, bb :: 03")
foreign = ix["joined_lft/aa_02"]
foreign = ix["bb/aa/aa_02"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_lft", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_02", foreign.RefValue)
@@ -61,30 +64,30 @@ func Test3003_joining_base_nested_complex(t *testing.T) {
"bb_04, aa_02, bb :: 04",
"bb_05, aa_02, bb :: 05")
foreign = ix["joined_lft/aa_03"]
foreign = ix["bb/aa/aa_03"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_lft", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_03", foreign.RefValue)
checkRows(h, foreign,
"bb_06, aa_03, bb :: 06")
foreign = ix["joined_lft/aa_04"]
foreign = ix["bb/aa/aa_04"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_lft", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_04", foreign.RefValue)
checkRows(h, foreign,
"bb_07, aa_04, bb :: 07")
foreign = ix["joined_lft/aa_05"]
foreign = ix["bb/aa/aa_05"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_lft", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("bb", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
h.a.Equal("aa_05", foreign.RefValue)
@@ -95,94 +98,94 @@ func Test3003_joining_base_nested_complex(t *testing.T) {
// // right join
foreign = ix["joined/aa_01"]
foreign = ix["cc/aa/aa_01"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
h.a.Equal("joined_rgh", foreign.Ref)
h.a.Equal("cc", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
"cc_01, aa_01, cc :: 01",
"cc_02, aa_01, cc :: 02")
foreign = ix["joined/aa_02"]
foreign = ix["cc/aa/aa_02"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
h.a.Equal("joined_rgh", foreign.Ref)
h.a.Equal("cc", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
"cc_03, aa_02, cc :: 03",
"cc_04, aa_02, cc :: 04")
foreign = ix["joined/aa_03"]
foreign = ix["cc/aa/aa_03"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
h.a.Equal("joined_rgh", foreign.Ref)
h.a.Equal("cc", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
"cc_05, aa_03, cc :: 05",
"cc_06, aa_03, cc :: 06")
foreign = ix["joined/aa_04"]
foreign = ix["cc/aa/aa_04"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
h.a.Equal("joined_rgh", foreign.Ref)
h.a.Equal("cc", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
"cc_07, aa_04, cc :: 07")
foreign = ix["joined/aa_05"]
foreign = ix["cc/aa/aa_05"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_a<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined", foreign.Source)
h.a.Equal("joined_rgh", foreign.Ref)
h.a.Equal("cc", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
"cc_08, aa_05, cc :: 08")
foreign = ix["joined_rgh/cc_01"]
foreign = ix["dd/cc/cc_01"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_c<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_rgh", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("dd", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
"dd_01, cc_01, dd :: 01")
foreign = ix["joined_rgh/cc_02"]
foreign = ix["dd/cc/cc_02"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_c<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_rgh", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("dd", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
"dd_02, cc_02, dd :: 02")
foreign = ix["joined_rgh/cc_03"]
foreign = ix["dd/cc/cc_03"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_c<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_rgh", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("dd", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
"dd_03, cc_03, dd :: 03")
foreign = ix["joined_rgh/cc_04"]
foreign = ix["dd/cc/cc_04"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_c<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_rgh", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("dd", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
"dd_04, cc_04, dd :: 04")
foreign = ix["joined_rgh/cc_05"]
foreign = ix["dd/cc/cc_05"]
h.a.NotNil(foreign)
h.a.Equal("pk<String>, fk_c<String>, label<String>", foreign.Columns.OmitSys().String())
h.a.Equal("joined_rgh", foreign.Source)
h.a.Equal("joined", foreign.Source)
h.a.Equal("dd", foreign.Ref)
h.a.Equal("pk", foreign.RelColumn)
checkRows(h, foreign,
tests/reporter/3004_filtering_test.go
+2 -2
View File
@@ -27,7 +27,7 @@ func Test3004_filtering(t *testing.T) {
", Maria_Königsmann, Maria, Königsmann")
// Engel_Kiefer
foreign = ix["Engel_Kiefer"]
foreign = ix["jobs/users/Engel_Kiefer"]
h.a.NotNil(foreign)
h.a.Equal(4, foreign.Size())
h.a.Equal("id<Record>, usr<String>, name<String>, type<Select>, cost<Number>, time_spent<Number>", foreign.Columns.String())
@@ -38,7 +38,7 @@ func Test3004_filtering(t *testing.T) {
", Engel_Kiefer, u12 j9, a, 71, 90")
// Engel_Loritz
foreign = ix["Maria_Königsmann"]
foreign = ix["jobs/users/Maria_Königsmann"]
h.a.NotNil(foreign)
h.a.Equal(2, foreign.Size())
h.a.Equal("id<Record>, usr<String>, name<String>, type<Select>, cost<Number>, time_spent<Number>", foreign.Columns.String())
tests/reporter/3004_joining_paging_test.go → tests/reporter/3005_joining_paging_test.go
+6 -6
View File
@@ -6,7 +6,7 @@ import (
"github.com/cortezaproject/corteza-server/pkg/report"
)
func Test3004_joining_paging(t *testing.T) {
func Test3005_joining_paging(t *testing.T) {
var (
ctx, h, s = setup(t)
m, _, dd = loadScenario(ctx, s, t, h)
@@ -30,10 +30,10 @@ func Test3004_joining_paging(t *testing.T) {
", aa_05, aa :: 05",
", aa_04, aa :: 04")
foreign = ix["aa_05"]
foreign = ix["bb/aa/aa_05"]
h.a.NotNil(foreign)
foreign = ix["aa_04"]
foreign = ix["bb/aa/aa_04"]
h.a.NotNil(foreign)
// // // PAGE 2
@@ -52,10 +52,10 @@ func Test3004_joining_paging(t *testing.T) {
", aa_03, aa :: 03",
", aa_02, aa :: 02")
foreign = ix["aa_03"]
foreign = ix["bb/aa/aa_03"]
h.a.NotNil(foreign)
foreign = ix["aa_02"]
foreign = ix["bb/aa/aa_02"]
h.a.NotNil(foreign)
// // // PAGE 1
@@ -72,6 +72,6 @@ func Test3004_joining_paging(t *testing.T) {
checkRows(h, local,
", aa_01, aa :: 01")
foreign = ix["aa_01"]
foreign = ix["bb/aa/aa_01"]
h.a.NotNil(foreign)
}
tests/reporter/3005_sorting_test.go
-111
View File
@@ -1,111 +0,0 @@
package reporter
import (
"testing"
"github.com/cortezaproject/corteza-server/pkg/report"
)
func Test3005_sorting(t *testing.T) {
var (
ctx, h, s = setup(t)
m, _, dd = loadScenario(ctx, s, t, h)
ff []*report.Frame
local, foreign *report.Frame
)
ff = loadNoErr(ctx, h, m, dd...)
h.a.Len(ff, 7)
local = ff[0]
ix := indexJoinedResult(ff)
_ = ix
// local
h.a.Equal(12, local.Size())
h.a.Equal("id<Record>, join_key<String>, first_name<String>, last_name<String>", local.Columns.String())
h.a.Equal("first_name, last_name DESC", local.Sort.String())
checkRows(h, local,
", Engel, Loritz",
", Engel, Kiefer",
", Engel, Kempf",
", Manu, Specht",
", Maria, Spannagel",
", Maria, Königsmann",
", Maria, Krüger",
", Sascha, Jans",
", Sigi, Goldschmidt",
", Ulli, Haupt",
", Ulli, Förstner",
", Ulli, Böhler")
// Maria_Königsmann
foreign = ix["Maria_Königsmann"]
h.a.NotNil(foreign)
h.a.Equal("id<Record>, usr<String>, name<String>, type<Select>, cost<Number>, time_spent<Number>", foreign.Columns.String())
h.a.Equal("type", foreign.Sort.String())
checkRows(h, foreign,
", Maria_Königsmann, u1 j1, a, 10, 2",
", Maria_Königsmann, u1 j5, a, 4, 4",
", Maria_Königsmann, u1 j2, b, 20, 5",
", Maria_Königsmann, u1 j6, b, 25, 25",
", Maria_Königsmann, u1 j7, b, 9, 91",
", Maria_Königsmann, u1 j4, c, 3, 0",
", Maria_Königsmann, u1 j3, d, 11, 1")
// Engel_Loritz
foreign = ix["Engel_Loritz"]
h.a.NotNil(foreign)
h.a.Equal("id<Record>, usr<String>, name<String>, type<Select>, cost<Number>, time_spent<Number>", foreign.Columns.String())
h.a.Equal("type", foreign.Sort.String())
checkRows(h, foreign,
", Engel_Loritz, u3 j1, a, 10, 1",
", Engel_Loritz, u3 j2, a, 0, 0",
", Engel_Loritz, u3 j3, a, 19, 99")
// Sigi_Goldschmidt
foreign = ix["Sigi_Goldschmidt"]
h.a.NotNil(foreign)
h.a.Equal("id<Record>, usr<String>, name<String>, type<Select>, cost<Number>, time_spent<Number>", foreign.Columns.String())
h.a.Equal("type", foreign.Sort.String())
checkRows(h, foreign,
", Sigi_Goldschmidt, u7 j2, a, 10, 21",
", Sigi_Goldschmidt, u7 j3, b, 10, 99",
", Sigi_Goldschmidt, u7 j1, d, 10, 29")
// Engel_Kiefer
foreign = ix["Engel_Kiefer"]
h.a.NotNil(foreign)
h.a.Equal("id<Record>, usr<String>, name<String>, type<Select>, cost<Number>, time_spent<Number>", foreign.Columns.String())
h.a.Equal("type", foreign.Sort.String())
checkRows(h, foreign,
", Engel_Kiefer, u12 j1, a, 42, 69",
", Engel_Kiefer, u12 j4, a, 35, 26",
", Engel_Kiefer, u12 j5, a, 34, 29",
", Engel_Kiefer, u12 j9, a, 71, 90",
", Engel_Kiefer, u12 j2, b, 65, 99",
", Engel_Kiefer, u12 j8, b, 74, 39",
", Engel_Kiefer, u12 j3, c, 38, 71",
", Engel_Kiefer, u12 j10, c, 79, 25",
", Engel_Kiefer, u12 j11, c, 19, 66",
", Engel_Kiefer, u12 j6, d, 92, 16",
", Engel_Kiefer, u12 j7, d, 14, 71")
// Manu_Specht
foreign = ix["Manu_Specht"]
h.a.NotNil(foreign)
h.a.Equal("id<Record>, usr<String>, name<String>, type<Select>, cost<Number>, time_spent<Number>", foreign.Columns.String())
h.a.Equal("type", foreign.Sort.String())
checkRows(h, foreign,
", Manu_Specht, u10 j3, b, 53, 12",
", Manu_Specht, u10 j4, b, 60, 22",
", Manu_Specht, u10 j2, c, 83, 70",
", Manu_Specht, u10 j1, d, 45, 56")
// Ulli_Böhler
foreign = ix["Ulli_Böhler"]
h.a.NotNil(foreign)
h.a.Equal("id<Record>, usr<String>, name<String>, type<Select>, cost<Number>, time_spent<Number>", foreign.Columns.String())
h.a.Equal("type", foreign.Sort.String())
checkRows(h, foreign,
", Ulli_Böhler, u5 j1, a, 1, 2")
}
tests/reporter/3006_joining_paging_nested_test.go
+92
View File
@@ -0,0 +1,92 @@
package reporter
import (
"testing"
"github.com/cortezaproject/corteza-server/pkg/report"
)
func Test3006_joining_paging_nested(t *testing.T) {
var (
ctx, h, s = setup(t)
m, _, dd = loadScenario(ctx, s, t, h)
ff []*report.Frame
def = dd[0]
local, foreign *report.Frame
)
// // // PAGE 1
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 5)
local = ff[0]
ix := indexJoinedResult(ff)
_ = ix
// local
h.a.Equal(2, local.Size())
h.a.NotNil(local.Paging)
h.a.NotNil(local.Paging.NextPage)
checkRows(h, local,
", aa_01, aa :: 01",
", aa_02, aa :: 02")
foreign = ix["bb/aa/aa_01"]
h.a.NotNil(foreign)
foreign = ix["bb/aa/aa_02"]
h.a.NotNil(foreign)
foreign = ix["cc/aa/aa_01"]
h.a.NotNil(foreign)
foreign = ix["cc/aa/aa_02"]
h.a.NotNil(foreign)
// // // PAGE 2
def.Paging.PageCursor = local.Paging.NextPage
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 5)
local = ff[0]
ix = indexJoinedResult(ff)
_ = ix
// local
h.a.Equal(2, local.Size())
h.a.NotNil(local.Paging)
h.a.NotNil(local.Paging.NextPage)
checkRows(h, local,
", aa_03, aa :: 03",
", aa_04, aa :: 04")
foreign = ix["bb/aa/aa_03"]
h.a.NotNil(foreign)
foreign = ix["bb/aa/aa_04"]
h.a.NotNil(foreign)
foreign = ix["cc/aa/aa_03"]
h.a.NotNil(foreign)
foreign = ix["cc/aa/aa_04"]
h.a.NotNil(foreign)
// // // PAGE 3
def.Paging.PageCursor = local.Paging.NextPage
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 3)
local = ff[0]
ix = indexJoinedResult(ff)
_ = ix
// local
h.a.Equal(1, local.Size())
h.a.Nil(local.Paging)
checkRows(h, local,
", aa_05, aa :: 05")
foreign = ix["bb/aa/aa_05"]
h.a.NotNil(foreign)
foreign = ix["cc/aa/aa_05"]
h.a.NotNil(foreign)
}
tests/reporter/grouping_base_test.go
+1 -49
View File
@@ -77,53 +77,5 @@ func Test2003_grouping_sorting(t *testing.T) {
}
func Test2004_grouping_paging(t *testing.T) {
var (
ctx, h, s = setup(t)
m, _, dd = loadScenario(ctx, s, t, h)
ff []*report.Frame
f *report.Frame
def = dd[0]
)
// ^ going up ^
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 1)
f = ff[0]
h.a.NotNil(f.Paging)
h.a.NotNil(f.Paging.NextPage)
h.a.Nil(f.Paging.PrevPage)
h.a.Equal(4, f.Size())
checkRows(h, f,
"Engel, 3, 179",
"Manu, 1, 61",
"Maria, 3, 183",
"Sascha, 1, 38")
def.Paging.PageCursor = f.Paging.NextPage
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 1)
f = ff[0]
h.a.NotNil(f.Paging)
h.a.Nil(f.Paging.NextPage)
h.a.NotNil(f.Paging.PrevPage)
h.a.Equal(2, f.Size())
checkRows(h, f,
"Sigi, 1, 67",
"Ulli, 3, 122")
// v going down v
def.Paging.PageCursor = f.Paging.PrevPage
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 1)
f = ff[0]
h.a.NotNil(f.Paging)
h.a.NotNil(f.Paging.NextPage)
h.a.Nil(f.Paging.PrevPage)
h.a.Equal(4, f.Size())
checkRows(h, f,
"Engel, 3, 179",
"Manu, 1, 61",
"Maria, 3, 183",
"Sascha, 1, 38")
t.Skip("@todo how can we support paging for groupped data? We need to assure something unique")
}
tests/reporter/loading_base_test.go
+83
View File
@@ -161,3 +161,86 @@ func Test1004_loading_paging(t *testing.T) {
", Manu_Specht, Manu",
", Maria_Krüger, Maria")
}
func Test1005_loading_paging_no_sort(t *testing.T) {
var (
ctx, h, s = setup(t)
m, _, dd = loadScenario(ctx, s, t, h)
ff []*report.Frame
f *report.Frame
def = dd[0]
)
// ^ going up ^
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 1)
f = ff[0]
h.a.NotNil(f.Paging)
h.a.NotNil(f.Paging.NextPage)
h.a.Nil(f.Paging.PrevPage)
h.a.Equal(5, f.Size())
checkRows(h, f,
", Maria_Königsmann, Maria",
", Ulli_Haupt, Ulli",
", Engel_Loritz, Engel",
", Sascha_Jans, Sascha",
", Ulli_Böhler, Ulli")
def.Paging.PageCursor = f.Paging.NextPage
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 1)
f = ff[0]
h.a.NotNil(f.Paging)
h.a.NotNil(f.Paging.NextPage)
h.a.NotNil(f.Paging.PrevPage)
h.a.Equal(5, f.Size())
checkRows(h, f,
", Maria_Spannagel, Maria",
", Sigi_Goldschmidt, Sigi",
", Engel_Kempf, Engel",
", Maria_Krüger, Maria",
", Manu_Specht, Manu")
def.Paging.PageCursor = f.Paging.NextPage
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 1)
f = ff[0]
h.a.NotNil(f.Paging)
h.a.Nil(f.Paging.NextPage)
h.a.NotNil(f.Paging.PrevPage)
h.a.Equal(2, f.Size())
checkRows(h, f,
", Ulli_Förstner, Ulli",
", Engel_Kiefer, Engel")
// v going down v
def.Paging.PageCursor = f.Paging.PrevPage
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 1)
f = ff[0]
h.a.NotNil(f.Paging)
h.a.NotNil(f.Paging.NextPage)
h.a.NotNil(f.Paging.PrevPage)
h.a.Equal(5, f.Size())
checkRows(h, f,
", Maria_Spannagel, Maria",
", Sigi_Goldschmidt, Sigi",
", Engel_Kempf, Engel",
", Maria_Krüger, Maria",
", Manu_Specht, Manu")
def.Paging.PageCursor = f.Paging.PrevPage
ff = loadNoErr(ctx, h, m, def)
h.a.Len(ff, 1)
f = ff[0]
h.a.NotNil(f.Paging)
h.a.NotNil(f.Paging.NextPage)
h.a.Nil(f.Paging.PrevPage)
h.a.Equal(5, f.Size())
checkRows(h, f,
", Maria_Königsmann, Maria",
", Ulli_Haupt, Ulli",
", Engel_Loritz, Engel",
", Sascha_Jans, Sascha",
", Ulli_Böhler, Ulli")
}
tests/reporter/main_test.go
+1 -1
View File
@@ -354,7 +354,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.RefValue] = f
out[fmt.Sprintf("%s/%s/%s", f.Ref, f.RelSource, f.RefValue)] = f
}
return out
tests/reporter/testdata/S1005_loading_paging_no_sort/report.json
Vendored
+24
View File
@@ -0,0 +1,24 @@
{
"handle": "testing_report",
"sources": [{
"step": { "load": {
"name": "users",
"source": "composeRecords",
"definition": {
"module": "user",
"namespace": "ns"
}
}}}],
"frames": [{
"name": "result",
"source": "users",
"columns": [
{ "name": "id", "label": "id" },
{ "name": "join_key", "label": "join_key" },
{ "name": "first_name", "label": "first_name" }
],
"paging": {
"limit": 5
}
}]
}
tests/reporter/testdata/S3004_joining_paging/report.json → tests/reporter/testdata/S3005_joining_paging/report.json
Vendored
View File
tests/reporter/testdata/S3005_sorting/report.json
Vendored
-53
View File
@@ -1,53 +0,0 @@
{
"handle": "testing_report",
"sources": [
{ "step": { "load": {
"name": "users",
"source": "composeRecords",
"definition": {
"module": "user",
"namespace": "ns"
}
}}},
{ "step": { "load": {
"name": "jobs",
"source": "composeRecords",
"definition": {
"module": "job",
"namespace": "ns"
}
}}},
{ "step": { "join": {
"name": "joined",
"localSource": "users",
"localColumn": "join_key",
"foreignSource": "jobs",
"foreignColumn": "usr"
}}}
],
"frames": [{
"name": "result",
"source": "joined",
"ref": "users",
"columns": [
{ "name": "id", "label": "id" },
{ "name": "join_key", "label": "join_key" },
{ "name": "first_name", "label": "first_name" },
{ "name": "last_name", "label": "last_name" }
],
"sort": "first_name, last_name DESC"
}, {
"name": "result",
"source": "joined",
"ref": "jobs",
"columns": [
{ "name": "id", "label": "id" },
{ "name": "usr", "label": "usr" },
{ "name": "name", "label": "name" },
{ "name": "type", "label": "type" },
{ "name": "cost", "label": "cost" },
{ "name": "time_spent", "label": "time_spent" }
],
"sort": "type"
}]
}
tests/reporter/testdata/S3006_joining_paging_nested/report.json
Vendored
+56
View File
@@ -0,0 +1,56 @@
{
"handle": "testing_report",
"sources": [
{ "step": { "load": {
"name": "aa",
"source": "composeRecords",
"definition": {
"module": "aa",
"namespace": "ns"
}
}}},
{ "step": { "load": {
"name": "bb",
"source": "composeRecords",
"definition": {
"module": "bb",
"namespace": "ns"
}
}}},
{ "step": { "load": {
"name": "cc",
"source": "composeRecords",
"definition": {
"module": "cc",
"namespace": "ns"
}
}}},
{ "step": { "join": {
"name": "joined_aux",
"localSource": "aa",
"localColumn": "pk",
"foreignSource": "bb",
"foreignColumn": "fk_a"
}}},
{ "step": { "join": {
"name": "joined",
"localSource": "joined_aux",
"localColumn": "pk",
"foreignSource": "cc",
"foreignColumn": "fk_a"
}}}
],
"frames": [{
"name": "result",
"source": "joined",
"sort": "label, cc.label DESC",
"paging": {
"limit": 2
}
}]
}