Reporting join step cleanups/refactoring
This commit is contained in:
Tomaž Jerman
@@ -1,408 +0,0 @@
|
||||
package report
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"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 (
|
||||
// frameBuffer is a container for pulled frames that we may not use immediately
|
||||
frameBuffer struct {
|
||||
sourceName string
|
||||
|
||||
loader func(keyFilter *Filter, cap uint) ([]*Frame, error)
|
||||
closer Closer
|
||||
|
||||
localFrames []*Frame
|
||||
foreignFrames []*Frame
|
||||
|
||||
// more returns true if we need to pull more
|
||||
more func([]*Frame, []int) bool
|
||||
postFetch func([]*Frame) ([]*Frame, error)
|
||||
|
||||
// General state for easier work
|
||||
// All of these provided parameters must be validated beforehand; no validation
|
||||
// occurs in this struct
|
||||
metaInitialized bool
|
||||
sorting filter.SortExprSet
|
||||
sortColumns []int
|
||||
|
||||
keyCol string
|
||||
keyColIndex int
|
||||
}
|
||||
)
|
||||
|
||||
// load loads the next chunk into the buffer based on the provided loader
|
||||
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
|
||||
ff, err = bl.loader(keyFilter, cap)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if ff == nil {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
for _, f := range ff {
|
||||
f.Source = bl.sourceName
|
||||
}
|
||||
|
||||
// Bucket
|
||||
for _, f := range ff {
|
||||
if f.RefValue != "" {
|
||||
auxForeign = append(auxForeign, f)
|
||||
continue
|
||||
}
|
||||
auxLocal = append(auxLocal, f)
|
||||
}
|
||||
|
||||
// Post processing
|
||||
if bl.postFetch != nil {
|
||||
auxLocal, err = bl.postFetch(auxLocal)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
|
||||
// Push into the buffer
|
||||
if len(bl.localFrames) == 0 {
|
||||
bl.localFrames = auxLocal
|
||||
} else {
|
||||
// - when the source is partitioned we append frames to the buffer
|
||||
if len(auxLocal) > 1 {
|
||||
bl.localFrames = append(bl.localFrames, auxLocal...)
|
||||
} else {
|
||||
// - when the source is not partitioned, we append the fetched rows to the only buffer frame
|
||||
bl.localFrames[0].Rows = append(bl.localFrames[0].Rows, auxLocal[0].Rows...)
|
||||
}
|
||||
}
|
||||
|
||||
// foreign frames should not overwrite eachothers as they are all complete
|
||||
bl.foreignFrames = append(bl.foreignFrames, auxForeign...)
|
||||
|
||||
// Initialize meta parameters as they are dependant on the resulting frames
|
||||
if !bl.metaInitialized {
|
||||
bl.metaInitialized = true
|
||||
bl.sortColumns = make([]int, len(bl.sorting))
|
||||
for i, s := range bl.sorting {
|
||||
bl.sortColumns[i] = bl.localFrames[0].Columns.Find(s.Column)
|
||||
}
|
||||
|
||||
bl.keyColIndex = bl.localFrames[0].Columns.Find(bl.keyCol)
|
||||
}
|
||||
|
||||
// Do we need to fetch more?
|
||||
// When chunk size is 0, we are fetching all
|
||||
if cap == 0 || !bl.more(bl.localFrames, bl.sortColumns) {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// keys returns a slice of keys that should be used for filtering over the
|
||||
// sub loader.
|
||||
func (bl *frameBuffer) keys() (keys []string, err error) {
|
||||
keys = make([]string, 0, defaultPageSize)
|
||||
keySet := make(map[string]bool)
|
||||
var k string
|
||||
|
||||
for _, mf := range bl.localFrames {
|
||||
err = mf.WalkRows(func(i int, r FrameRow) error {
|
||||
k, err = cast.ToStringE(r[bl.keyColIndex].Get())
|
||||
if ok := keySet[k]; !ok {
|
||||
keys = append(keys, k)
|
||||
keySet[k] = true
|
||||
}
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// keyFilter prepares the filter that should be used when fetching related rows.
|
||||
//
|
||||
// @todo do some compression, ie "id > x && id < y"
|
||||
// this will return more stuff but it could be faster then the current thing
|
||||
func (bl *frameBuffer) keyFilter(keys []string) *Filter {
|
||||
aa := make(qlng.ASTNodeSet, len(keys))
|
||||
|
||||
for i, k := range keys {
|
||||
aa[i] = &qlng.ASTNode{
|
||||
Ref: "eq",
|
||||
Args: qlng.ASTNodeSet{
|
||||
&qlng.ASTNode{Symbol: bl.keyCol},
|
||||
&qlng.ASTNode{Value: qlng.MakeValueOf("String", k)},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
return &Filter{
|
||||
ASTNode: &qlng.ASTNode{
|
||||
Ref: "group",
|
||||
Args: qlng.ASTNodeSet{
|
||||
&qlng.ASTNode{
|
||||
Ref: "or",
|
||||
Args: aa,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (bl *frameBuffer) walkRowsLocal(fnc func(i int, r FrameRow) error) error {
|
||||
if len(bl.localFrames) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
if len(bl.localFrames) > 1 {
|
||||
for i, f := range bl.localFrames {
|
||||
fnc(i, f.FirstRow())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
return bl.localFrames[0].WalkRows(fnc)
|
||||
}
|
||||
|
||||
func (bl *frameBuffer) getByRefValue(ref expr.TypedValue) *Frame {
|
||||
s, err := expr.CastToString(ref.Get())
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
for _, f := range bl.localFrames {
|
||||
if f.RefValue == s {
|
||||
return f
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// cutLocal returns the cap of items from the buffers' localFrames and
|
||||
// buffers back the remaining items
|
||||
func (bl *frameBuffer) removeLocal(count int) {
|
||||
_, bl.localFrames, _ = bl.cut(bl.localFrames, count)
|
||||
return
|
||||
}
|
||||
|
||||
// cutForeign returns the cap of items from the buffers' foreignFrames and
|
||||
// buffers back the remaining items
|
||||
func (bl *frameBuffer) removeForeign(count int) {
|
||||
_, bl.foreignFrames, _ = bl.cut(bl.foreignFrames, count)
|
||||
return
|
||||
}
|
||||
|
||||
// cutLocal returns the cap of items from the buffers' localFrames and
|
||||
// buffers back the remaining items
|
||||
func (bl *frameBuffer) cutLocal(cap int) (out []*Frame, more bool) {
|
||||
out, bl.localFrames, more = bl.cut(bl.localFrames, cap)
|
||||
return
|
||||
}
|
||||
|
||||
// cutForeign returns the cap of items from the buffers' foreignFrames and
|
||||
// buffers back the remaining items
|
||||
func (bl *frameBuffer) cutForeign(cap int) (out []*Frame) {
|
||||
out, bl.foreignFrames, _ = bl.cut(bl.foreignFrames, cap)
|
||||
return
|
||||
}
|
||||
|
||||
func (bl *frameBuffer) sizeLocal() int {
|
||||
if len(bl.localFrames) == 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
if len(bl.localFrames) > 1 {
|
||||
return len(bl.localFrames)
|
||||
}
|
||||
|
||||
return len(bl.localFrames[0].Rows)
|
||||
}
|
||||
|
||||
func (bl *frameBuffer) sizeForeign() int {
|
||||
return len(bl.foreignFrames)
|
||||
}
|
||||
|
||||
// cut is a generic sub fnc for cutLocal and cutForeign
|
||||
func (bl *frameBuffer) cut(ff []*Frame, cap int) (out []*Frame, buffer []*Frame, more bool) {
|
||||
lfl := len(ff)
|
||||
|
||||
if lfl == 0 {
|
||||
return nil, nil, false
|
||||
}
|
||||
|
||||
if lfl == 1 {
|
||||
o := ff[0].CloneMeta()
|
||||
if ff[0].Size() <= cap {
|
||||
o.Rows = ff[0].Rows
|
||||
return []*Frame{o}, nil, false
|
||||
}
|
||||
|
||||
o.Rows = ff[0].Rows[0:cap]
|
||||
ff[0].Rows = ff[0].Rows[cap:]
|
||||
return []*Frame{o}, ff, true
|
||||
}
|
||||
|
||||
if lfl < cap {
|
||||
oo := ff
|
||||
return oo, nil, false
|
||||
}
|
||||
|
||||
return ff[0:cap], ff[cap:], true
|
||||
}
|
||||
|
||||
func (bl *frameBuffer) calculatePagingCursors(out []*Frame, sorting filter.SortExprSet, cursor *filter.PagingCursor, hasNext bool) []*Frame {
|
||||
if !hasNext {
|
||||
return out
|
||||
}
|
||||
|
||||
// index frames by ref
|
||||
index := make(map[string][]*Frame)
|
||||
for _, o := range out {
|
||||
index[o.Ref] = append(index[o.Ref], o)
|
||||
}
|
||||
|
||||
// @todo add support for prev cursors.
|
||||
// For now only next page is supported.
|
||||
// Keep a reference of previously seen cursors if you need this functionality.
|
||||
// if hasPrev {
|
||||
// out[0].Paging = &filter.Paging{}
|
||||
// out[0].Paging.PrevPage = bl.calculatePagingCursor(out[0].FirstRow(), out[0].Columns, index, true, sorting...)
|
||||
// out[0].Paging.PrevPage.ROrder = true
|
||||
// out[0].Paging.PrevPage.LThen = !sorting.Reversed()
|
||||
// }
|
||||
|
||||
if out[0].Paging == nil {
|
||||
out[0].Paging = &filter.Paging{}
|
||||
}
|
||||
out[0].Paging.NextPage = bl.calculatePagingCursor(out[0].LastRow(), out[0].Columns, index, false, sorting...)
|
||||
out[0].Paging.NextPage.LThen = sorting.Reversed()
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
func (bl *frameBuffer) calculatePagingCursor(r FrameRow, cols FrameColumnSet, index map[string][]*Frame, first bool, cc ...*filter.SortExpr) *filter.PagingCursor {
|
||||
// The check for existence should be performed way in advanced so we won't bother here.
|
||||
// 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
|
||||
|
||||
if strings.Contains(c.Column, ".") {
|
||||
pts := strings.Split(c.Column, ".")
|
||||
foreignFrames = index[pts[0]]
|
||||
var r FrameRow
|
||||
var f *Frame
|
||||
if first {
|
||||
f = foreignFrames[0]
|
||||
r = f.FirstRow()
|
||||
} else {
|
||||
f = foreignFrames[len(foreignFrames)-1]
|
||||
r = f.FirstRow()
|
||||
}
|
||||
|
||||
if r[f.Columns.Find(pts[1])] != nil {
|
||||
v = r[f.Columns.Find(pts[1])].Get()
|
||||
}
|
||||
cursor.Set(c.Column, v, c.Descending)
|
||||
} else {
|
||||
if r[cols.Find(c.Column)] != nil {
|
||||
v = r[cols.Find(c.Column)].Get()
|
||||
}
|
||||
cursor.Set(c.Column, v, c.Descending)
|
||||
}
|
||||
}
|
||||
|
||||
return cursor
|
||||
}
|
||||
|
||||
// prepareResponse takes the provided buffers, metadata and prepares the result of the step
|
||||
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
|
||||
if inverted {
|
||||
local = sub
|
||||
foreign = main
|
||||
} else {
|
||||
local = main
|
||||
foreign = sub
|
||||
}
|
||||
|
||||
more := false
|
||||
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
|
||||
}
|
||||
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{}
|
||||
}
|
||||
|
||||
if len(oo) == 0 {
|
||||
return prepareResponseEmpty(lfd, dscr), nil
|
||||
}
|
||||
|
||||
return oo, nil
|
||||
}
|
||||
|
||||
func prepareResponseEmpty(lfd *FrameDefinition, dd FrameDescriptionSet) (oo []*Frame) {
|
||||
dscr := dd.FilterByRef(lfd.Ref)[0]
|
||||
|
||||
f := &Frame{
|
||||
Name: lfd.Name,
|
||||
Ref: lfd.Ref,
|
||||
Source: lfd.Source,
|
||||
Columns: make(FrameColumnSet, 0, len(lfd.Columns)),
|
||||
}
|
||||
|
||||
if len(lfd.Columns) == 0 {
|
||||
f.Columns = dscr.Columns
|
||||
} else {
|
||||
for _, c := range lfd.Columns {
|
||||
f.Columns = append(f.Columns, dscr.Columns[dscr.Columns.Find(c.Name)])
|
||||
}
|
||||
}
|
||||
return []*Frame{f}
|
||||
}
|
||||
79
pkg/report/frame_load_ctx.go
Normal file
79
pkg/report/frame_load_ctx.go
Normal file
@@ -0,0 +1,79 @@
|
||||
package report
|
||||
|
||||
import (
|
||||
"github.com/cortezaproject/corteza-server/pkg/filter"
|
||||
"github.com/cortezaproject/corteza-server/pkg/qlng"
|
||||
"github.com/spf13/cast"
|
||||
)
|
||||
|
||||
type (
|
||||
// frameLoadCtx encapsulates some loading metadata to make it easier to work with
|
||||
frameLoadCtx struct {
|
||||
// initLoader initializes a fresh loader in case where it can't be evaluated beforehand
|
||||
initLoader func(int, *Filter) (Loader, Closer, error)
|
||||
|
||||
loader Loader
|
||||
closer Closer
|
||||
|
||||
// General state for easier work
|
||||
metaInitialized bool
|
||||
sorting filter.SortExprSet
|
||||
sortColumns []int
|
||||
|
||||
keyCol string
|
||||
keyColIndex int
|
||||
}
|
||||
)
|
||||
|
||||
// keys returns the unique key values based on the key column and the provided frames
|
||||
func (bl *frameLoadCtx) keys(ff []*Frame) (keys []string, err error) {
|
||||
keys = make([]string, 0, defaultPageSize)
|
||||
keySet := make(map[string]bool)
|
||||
var k string
|
||||
|
||||
for _, f := range ff {
|
||||
err = f.WalkRows(func(i int, r FrameRow) error {
|
||||
k, err = cast.ToStringE(r[bl.keyColIndex].Get())
|
||||
if ok := keySet[k]; !ok {
|
||||
keys = append(keys, k)
|
||||
keySet[k] = true
|
||||
}
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// keyFilter prepares the filter that should be used when fetching related rows.
|
||||
//
|
||||
// @todo do some compression, ie "id > x && id < y"
|
||||
// this will return more stuff but it could be faster then the current thing
|
||||
func (bl *frameLoadCtx) keyFilter(keys []string) *Filter {
|
||||
aa := make(qlng.ASTNodeSet, len(keys))
|
||||
|
||||
for i, k := range keys {
|
||||
aa[i] = &qlng.ASTNode{
|
||||
Ref: "eq",
|
||||
Args: qlng.ASTNodeSet{
|
||||
&qlng.ASTNode{Symbol: bl.keyCol},
|
||||
&qlng.ASTNode{Value: qlng.MakeValueOf("String", k)},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
return &Filter{
|
||||
ASTNode: &qlng.ASTNode{
|
||||
Ref: "group",
|
||||
Args: qlng.ASTNodeSet{
|
||||
&qlng.ASTNode{
|
||||
Ref: "or",
|
||||
Args: aa,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -25,6 +25,13 @@ type (
|
||||
partitioned bool
|
||||
partitionSize uint
|
||||
partitionCol string
|
||||
|
||||
// localFrames contains the pulled local frames
|
||||
localFrames []*Frame
|
||||
// foreignFrames contains the pulled foreign frames grouped by ds
|
||||
foreignFrames [][]*Frame
|
||||
// foreignSourceIndex maps the ds to the ds index of foreignFrames slice
|
||||
foreignSourceIndex map[string]int
|
||||
}
|
||||
|
||||
JoinStepDefinition struct {
|
||||
@@ -159,7 +166,7 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
|
||||
}
|
||||
|
||||
// - Validate the sort of the local frame for paging purposes
|
||||
err = d.validateSort(oLocalDef, dscr)
|
||||
err = d.validatePagingSort(oLocalDef, dscr)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -183,35 +190,36 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
|
||||
}
|
||||
}
|
||||
|
||||
// - Determine the join strategy to use.
|
||||
// For now the strategy is the combination of the following parameters, in the
|
||||
// future we can have a wrapper struct as well.
|
||||
mainLoader, subLoader, err := d.strategizeLoad(ctx, inverted, localDef, foreignDef)
|
||||
// - Do some preparation tasks before the loading occurs
|
||||
localLoader, foreignLoader, err := d.strategizeLoading(ctx, inverted, localDef, foreignDef)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Load and join data
|
||||
// Loading and joining data
|
||||
//
|
||||
// The loader function will iterate indefinitely until the requested frame
|
||||
// definition is satisfied.
|
||||
//
|
||||
// Outline:
|
||||
// . prepare additional (partial) filters based on the paging cursor
|
||||
// . pull data from the main source
|
||||
// . prepare an additional key-based filter for sub source
|
||||
// . pull data from the sub source
|
||||
// . additional processing
|
||||
// .. apply additional filtering based on page cursors
|
||||
// . prepare response
|
||||
// . load data from the two sources (local, foreign)
|
||||
// . remove rows that don't have any related data
|
||||
// . sort the pulled chunk
|
||||
// . apply additional processing
|
||||
// .. apply additional paging cursor based filtering
|
||||
// . update buffers
|
||||
//
|
||||
// When preparing the response, the paging cursor is calculated.
|
||||
isEmpty := false
|
||||
return func(cap int, processed bool) (oo []*Frame, err error) {
|
||||
var keys []string
|
||||
if isEmpty {
|
||||
return
|
||||
}
|
||||
|
||||
// We need to go 1 over to see if we can calculate paging cursors
|
||||
loadCap := cap
|
||||
if processed {
|
||||
cap++
|
||||
loadCap++
|
||||
}
|
||||
|
||||
// The modified flag will help us determine if we need another iteration or not.
|
||||
@@ -220,70 +228,72 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
|
||||
// Underlying loaders must be able to provide all of the requested data, so if
|
||||
// we don't do any modifications here, the iteration should not repeat.
|
||||
modified := false
|
||||
// m is defined here so we don't re initialize it every time later on
|
||||
m := false
|
||||
|
||||
pagingSatisfied := oLocalDef.Paging.PageCursor == nil
|
||||
more := false
|
||||
var buffLocal []*Frame
|
||||
var buffForeign []*Frame
|
||||
for {
|
||||
modified = false
|
||||
|
||||
// . Pull data from the main source
|
||||
more, err = mainLoader.load(uint(cap), nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if !more {
|
||||
isEmpty = true
|
||||
break
|
||||
}
|
||||
|
||||
// . Prepare an additional key-based filter for sub sources
|
||||
keys, err = mainLoader.keys()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
keyFilter := subLoader.keyFilter(keys)
|
||||
|
||||
// . Pull data from the sub source
|
||||
more, err = subLoader.load(uint(cap), keyFilter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !more {
|
||||
break
|
||||
}
|
||||
|
||||
// Cut out any local rows that do not have any foreign rows
|
||||
modified = d.validateRows(mainLoader, subLoader, inverted)
|
||||
if modified {
|
||||
continue
|
||||
}
|
||||
|
||||
// . Sort the collected data
|
||||
// . load data from the two sources (local, foreign)
|
||||
//
|
||||
// @todo Most of the sorting has already been done by the datasource,
|
||||
// so we can get away with a lot of partial processing.
|
||||
// if d.shouldSort(oLocalDef.Sort) {
|
||||
// }
|
||||
err = d.strategizeSort(mainLoader, subLoader, inverted, oLocalDef, d.def.LocalColumn)
|
||||
// The loading order differs a bit based on the provided sort.
|
||||
if !inverted {
|
||||
buffLocal, buffForeign, err = d.localBasedLoad(ctx, loadCap, localLoader, foreignLoader, localDef, foreignDef)
|
||||
} else {
|
||||
buffLocal, buffForeign, err = d.foreignBasedLoad(ctx, loadCap, localLoader, foreignLoader, localDef, foreignDef)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return
|
||||
}
|
||||
|
||||
// . Additional processing
|
||||
// .. if both pulled buffersare empty, there is nothing left to do
|
||||
if d.sizeOfBuffer(buffLocal, d.partitioned)+d.sizeOfBuffer(buffForeign, true) == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
// . remove rows that don't have any related data
|
||||
buffLocal, buffForeign, modified = d.removeOrphanRows(buffLocal, buffForeign)
|
||||
|
||||
// . sort the pulled chunk
|
||||
//
|
||||
// The way we are fetching data, most of the sorting is already done.
|
||||
// We need to assure the sort based on both local and foreign data sources.
|
||||
buffLocal, buffForeign = d.sortFrameBuffers(localLoader, foreignLoader, buffLocal, buffForeign, localDef.Sort, inverted)
|
||||
|
||||
// . apply additional processing
|
||||
//
|
||||
// Any additional filtering and goes here.
|
||||
// Transformations and other bits should reside in the buffers when loading data.
|
||||
// Any processing should only occur based on the provided buffer to simplify
|
||||
// the entire algorithm.
|
||||
|
||||
// .. Additional filters based on page cursors
|
||||
// .. apply additional paging cursor based filtering
|
||||
//
|
||||
// As we're combining data from N sources, the paging cursors are also
|
||||
// based on the combination of sources and we can't do the "classical"
|
||||
// "cut the data and call it a day" as we do with records.
|
||||
//
|
||||
// The "primary" DS (the one that defines the initial order) may strictly
|
||||
// cut off unneeded rows on the DS level, but the rest need to be calculated
|
||||
// manually here.
|
||||
//
|
||||
// This evaluation must only occur while the paging filter is not yet
|
||||
// satisfied.
|
||||
if !pagingSatisfied {
|
||||
m, pagingSatisfied = d.pagingFilter(mainLoader, subLoader, pp)
|
||||
buffLocal, buffForeign, m, pagingSatisfied, err = d.applyPagingFilter(buffLocal, buffForeign, pp)
|
||||
modified = modified || m
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// . Prepare response
|
||||
// . Update buffers
|
||||
//
|
||||
// As we're fetching and providing sorted data, the buffer
|
||||
// merging algorithm may be a simple concatenation.
|
||||
d.updateBuffers(buffLocal, buffForeign)
|
||||
|
||||
// If there were no modifications to what we pulled, we can safely
|
||||
// assume that we can produce a response.
|
||||
if !modified {
|
||||
@@ -291,13 +301,17 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
|
||||
}
|
||||
}
|
||||
|
||||
return prepareResponse(mainLoader, subLoader, inverted, processed, oLocalDef, d.def.LocalColumn, dscr)
|
||||
}, func() {
|
||||
if mainLoader.closer != nil {
|
||||
mainLoader.closer()
|
||||
if d.bufferSize() == 0 {
|
||||
return d.prepareResponseGeneric(oLocalDef, dscr), nil
|
||||
}
|
||||
if subLoader.closer != nil {
|
||||
subLoader.closer()
|
||||
|
||||
return d.prepareResponse(oLocalDef, oForeignDef, dscr, cap, processed)
|
||||
}, func() {
|
||||
if localLoader.closer != nil {
|
||||
localLoader.closer()
|
||||
}
|
||||
if foreignLoader.closer != nil {
|
||||
foreignLoader.closer()
|
||||
}
|
||||
}, nil
|
||||
}
|
||||
@@ -393,16 +407,27 @@ func (d *joinedDataset) sliceFrames(ff []*Frame, selfCol, relCol string) (out []
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (d *joinedDataset) validateSort(def *FrameDefinition, dd FrameDescriptionSet) (err error) {
|
||||
// validatePagingSort assures that we can apply paging to the provided sort
|
||||
//
|
||||
// The final datasource that is defined in the given sort must be sorted by
|
||||
// a primary key or a unique value.
|
||||
func (d *joinedDataset) validatePagingSort(def *FrameDefinition, dd FrameDescriptionSet) (err error) {
|
||||
// The first one is always local so this is ok
|
||||
localDscr := dd[0]
|
||||
var dscr *FrameDescription
|
||||
|
||||
// - determine the last ds of the given sort definition
|
||||
sortDS := ""
|
||||
auxSS := make(filter.SortExprSet, 0, len(def.Sort))
|
||||
|
||||
for i := len(def.Sort) - 2; i >= 0; i-- {
|
||||
aa := strings.Split(def.Sort[i].Column, ".")
|
||||
bb := strings.Split(def.Sort[i+1].Column, ".")
|
||||
aa := strings.Split(def.Sort[i+1].Column, ".")
|
||||
bb := strings.Split(def.Sort[i].Column, ".")
|
||||
|
||||
if len(aa) != len(bb) || (len(aa) > 1 && aa[0] != aa[1]) {
|
||||
auxSS = append(auxSS, def.Sort[i+1])
|
||||
if len(aa) > 1 {
|
||||
sortDS = aa[0]
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
@@ -412,16 +437,7 @@ func (d *joinedDataset) validateSort(def *FrameDefinition, dd FrameDescriptionSe
|
||||
}
|
||||
}
|
||||
|
||||
// @todo temporarily disabled
|
||||
if sortDS != "" {
|
||||
return fmt.Errorf("[temporary] initial sort can not by by a foreign column: %s", sortDS)
|
||||
}
|
||||
|
||||
// The first one is always local so this is ok
|
||||
localDscr := dd[0]
|
||||
var dscr *FrameDescription
|
||||
|
||||
// When local, ref is omitted
|
||||
// when local, ref is omitted
|
||||
if sortDS == "" {
|
||||
// Do this to avoid extra work afterwords
|
||||
auxSS = def.Sort
|
||||
@@ -432,7 +448,7 @@ func (d *joinedDataset) validateSort(def *FrameDefinition, dd FrameDescriptionSe
|
||||
dscr = dd.FilterByRef(sortDS)[0]
|
||||
}
|
||||
|
||||
// Check if we're sorting by a unique value
|
||||
// - check if we're sorting by a unique value
|
||||
def.Sort = func() filter.SortExprSet {
|
||||
unique := ""
|
||||
for _, c := range dscr.Columns {
|
||||
@@ -451,11 +467,14 @@ func (d *joinedDataset) validateSort(def *FrameDefinition, dd FrameDescriptionSe
|
||||
} else {
|
||||
return append(def.Sort, &filter.SortExpr{Column: fmt.Sprintf("%s.%s", sortDS, unique), Descending: auxSS.LastDescending()})
|
||||
}
|
||||
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// prepareSorting parses the provided local sort definition and applies it to
|
||||
// the local and foreign definition
|
||||
//
|
||||
// This allows the store to provide the ordered data, simplifying our lives.
|
||||
func (d *joinedDataset) prepareSorting(local, foreign *FrameDefinition) (inverted bool, err error) {
|
||||
var (
|
||||
localSS filter.SortExprSet
|
||||
@@ -490,37 +509,435 @@ func (d *joinedDataset) prepareSorting(local, foreign *FrameDefinition) (inverte
|
||||
return
|
||||
}
|
||||
|
||||
func (d *joinedDataset) validateRows(local, foreign *frameBuffer, inverted bool) (modified bool) {
|
||||
keyCol := foreign.localFrames[0].RelColumn
|
||||
cols := local.localFrames[0].Columns
|
||||
keyColIx := cols.Find(keyCol)
|
||||
// localBasedLoad primary loads from local source then foreign one
|
||||
func (d *joinedDataset) localBasedLoad(ctx context.Context, cap int, local, foreign *frameLoadCtx, localDef, foreignDef *FrameDefinition) (buffLocal, buffForeign []*Frame, err error) {
|
||||
// Fetch data from local ds
|
||||
var localPost func(ff []*Frame) ([]*Frame, error)
|
||||
if d.partitioned {
|
||||
localPost = func(ff []*Frame) ([]*Frame, error) {
|
||||
var err error
|
||||
|
||||
// - index foreign frames into buckets; here the foreign sort must be respected
|
||||
buckets := make(map[string]int)
|
||||
for i, mf := range foreign.localFrames {
|
||||
buckets[mf.RefValue] = i
|
||||
ff, err = d.sliceFrames(ff, d.def.LocalColumn, d.def.ForeignColumn)
|
||||
for _, f := range ff {
|
||||
f.RefValue = ""
|
||||
f.RelColumn = ""
|
||||
}
|
||||
return ff, err
|
||||
}
|
||||
}
|
||||
|
||||
var k string
|
||||
aux := make([]FrameRow, 0, 100)
|
||||
local.walkRowsLocal(func(i int, r FrameRow) error {
|
||||
v := r[keyColIx]
|
||||
if v == nil {
|
||||
k = ""
|
||||
} else {
|
||||
k = cast.ToString(v.Get())
|
||||
buffLocal, err = d.frameLoader(local.loader, local, cap, d.partitioned, localPost)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if len(buffLocal) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// Init local frame loader metadata
|
||||
if !local.metaInitialized {
|
||||
local.metaInitialized = true
|
||||
local.sortColumns = make([]int, len(local.sorting))
|
||||
for i, s := range local.sorting {
|
||||
local.sortColumns[i] = buffLocal[0].Columns.Find(s.Column)
|
||||
}
|
||||
|
||||
if _, ok := buckets[k]; ok {
|
||||
aux = append(aux, r)
|
||||
local.keyColIndex = buffLocal[0].Columns.Find(local.keyCol)
|
||||
}
|
||||
|
||||
// Prepare filtering for foreign DS
|
||||
keys, err := local.keys(buffLocal)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if len(keys) == 0 {
|
||||
return
|
||||
}
|
||||
keyFilter := foreign.keyFilter(keys)
|
||||
|
||||
// Fetch data from foreign ds
|
||||
foreignPost := func(ff []*Frame) ([]*Frame, error) {
|
||||
ff, err = d.sliceFrames(ff, d.def.ForeignColumn, d.def.LocalColumn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
for i := range ff {
|
||||
if ff[i].Name == "" {
|
||||
ff[i].Name = foreignDef.Name
|
||||
}
|
||||
if ff[i].Source == "" {
|
||||
ff[i].Source = foreignDef.Source
|
||||
}
|
||||
if ff[i].Ref == "" {
|
||||
ff[i].Ref = foreignDef.Ref
|
||||
}
|
||||
}
|
||||
|
||||
modified = len(local.localFrames[0].Rows) != len(aux)
|
||||
local.localFrames[0].Rows = aux
|
||||
return ff, nil
|
||||
}
|
||||
|
||||
fLdr, fClsr, err := foreign.initLoader(cap, keyFilter)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer fClsr()
|
||||
buffForeign, err = d.frameLoader(fLdr, foreign, 0, true, foreignPost)
|
||||
|
||||
if len(buffForeign) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// Init foreign frame loader metadata
|
||||
if !foreign.metaInitialized {
|
||||
foreign.metaInitialized = true
|
||||
foreign.sortColumns = make([]int, len(foreign.sorting))
|
||||
for i, s := range foreign.sorting {
|
||||
foreign.sortColumns[i] = buffLocal[0].Columns.Find(s.Column)
|
||||
}
|
||||
|
||||
foreign.keyColIndex = buffLocal[0].Columns.Find(foreign.keyCol)
|
||||
}
|
||||
|
||||
return
|
||||
|
||||
}
|
||||
|
||||
// foreignBasedLoad primary loads from foreign source and then the local one
|
||||
func (d *joinedDataset) foreignBasedLoad(ctx context.Context, cap int, local, foreign *frameLoadCtx, localDef, foreignDef *FrameDefinition) (buffLocal, buffForeign []*Frame, err error) {
|
||||
// Fetch data from foreign ds
|
||||
foreignPost := func(ff []*Frame) ([]*Frame, error) {
|
||||
ff, err = d.sliceFrames(ff, d.def.ForeignColumn, d.def.LocalColumn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for i := range ff {
|
||||
if ff[i].Name == "" {
|
||||
ff[i].Name = foreignDef.Name
|
||||
}
|
||||
if ff[i].Source == "" {
|
||||
ff[i].Source = foreignDef.Source
|
||||
}
|
||||
if ff[i].Ref == "" {
|
||||
ff[i].Ref = foreignDef.Ref
|
||||
}
|
||||
}
|
||||
|
||||
return ff, nil
|
||||
}
|
||||
|
||||
buffForeign, err = d.frameLoader(foreign.loader, foreign, 0, true, foreignPost)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Init foreign frame loader metadata
|
||||
if !foreign.metaInitialized {
|
||||
foreign.metaInitialized = true
|
||||
foreign.sortColumns = make([]int, len(foreign.sorting))
|
||||
for i, s := range foreign.sorting {
|
||||
foreign.sortColumns[i] = buffForeign[0].Columns.Find(s.Column)
|
||||
}
|
||||
|
||||
foreign.keyColIndex = buffForeign[0].Columns.Find(foreign.keyCol)
|
||||
}
|
||||
|
||||
// Prepare filtering for foreign DS
|
||||
keys, err := foreign.keys(buffForeign)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if len(keys) == 0 {
|
||||
return
|
||||
}
|
||||
keyFilter := local.keyFilter(keys)
|
||||
|
||||
// Fetch data from local ds
|
||||
var localPost func(ff []*Frame) ([]*Frame, error)
|
||||
if d.partitioned {
|
||||
localPost = func(ff []*Frame) ([]*Frame, error) {
|
||||
var err error
|
||||
|
||||
ff, err = d.sliceFrames(ff, d.def.LocalColumn, d.def.ForeignColumn)
|
||||
for _, f := range ff {
|
||||
f.RefValue = ""
|
||||
f.RelColumn = ""
|
||||
}
|
||||
return ff, err
|
||||
}
|
||||
}
|
||||
|
||||
lLdr, lClsr, err := local.initLoader(cap, keyFilter)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer lClsr()
|
||||
|
||||
buffLocal, err = d.frameLoader(lLdr, local, 0, true, localPost)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Init local frame loader metadata
|
||||
if !local.metaInitialized {
|
||||
local.metaInitialized = true
|
||||
local.sortColumns = make([]int, len(local.sorting))
|
||||
for i, s := range local.sorting {
|
||||
local.sortColumns[i] = buffLocal[0].Columns.Find(s.Column)
|
||||
}
|
||||
|
||||
local.keyColIndex = buffLocal[0].Columns.Find(local.keyCol)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// updateBuffers updates existing frame buffers with newly provided ones
|
||||
func (d *joinedDataset) updateBuffers(local, foreign []*Frame) {
|
||||
// - local
|
||||
if len(d.localFrames) == 0 {
|
||||
d.localFrames = local
|
||||
} else {
|
||||
if d.partitioned {
|
||||
d.localFrames = append(d.localFrames, local...)
|
||||
} else {
|
||||
d.localFrames[0].Rows = append(d.localFrames[0].Rows, local[0].Rows...)
|
||||
}
|
||||
}
|
||||
|
||||
// - foreign
|
||||
if d.foreignSourceIndex == nil {
|
||||
d.foreignSourceIndex = make(map[string]int)
|
||||
}
|
||||
if ix, ok := d.foreignSourceIndex[foreign[0].Ref]; !ok {
|
||||
d.foreignSourceIndex[foreign[0].Ref] = len(d.foreignFrames)
|
||||
d.foreignFrames = append(d.foreignFrames, foreign)
|
||||
} else {
|
||||
d.foreignFrames[ix] = append(d.foreignFrames[ix], foreign...)
|
||||
}
|
||||
}
|
||||
|
||||
// removeOrphanRows removes all local rows with no related foreign frame
|
||||
func (d *joinedDataset) removeOrphanRows(local, foreign []*Frame) ([]*Frame, []*Frame, bool) {
|
||||
// - index foreign frames based on ref value so that we have an easier time validating
|
||||
fIndex := make(map[string]bool)
|
||||
for _, ff := range foreign {
|
||||
fIndex[ff.RefValue] = true
|
||||
}
|
||||
|
||||
// - go over local frame rows and see which ones need to be removed
|
||||
var k string
|
||||
var modified bool
|
||||
|
||||
cols := local[0].Columns
|
||||
keyColIx := cols.Find(foreign[0].RelColumn)
|
||||
for _, lf := range local {
|
||||
aux := make([]FrameRow, 0, len(lf.Rows))
|
||||
|
||||
lf.WalkRows(func(i int, r FrameRow) error {
|
||||
if r[keyColIx] == nil {
|
||||
return nil
|
||||
} else {
|
||||
k = cast.ToString(r[keyColIx].Get())
|
||||
if k == "" {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
if fIndex[k] {
|
||||
aux = append(aux, r)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
modified = modified || len(aux) != len(lf.Rows)
|
||||
lf.Rows = aux
|
||||
}
|
||||
|
||||
return local, foreign, modified
|
||||
}
|
||||
|
||||
func (d *joinedDataset) prepareResponse(localDef, foreignDef *FrameDefinition, dd FrameDescriptionSet, cap int, applyPaging bool) (oo []*Frame, err error) {
|
||||
// - Use the requested size of the two buffers
|
||||
// -- first come local frames
|
||||
oo = append(oo, d.cutLocalFrameBuffer(cap)...)
|
||||
localDelimiter := len(oo)
|
||||
//
|
||||
// -- followed by foreign
|
||||
aux := d.cutForeignFrameBuffer(cap)
|
||||
for _, a := range aux {
|
||||
a.RelSource = oo[0].Ref
|
||||
}
|
||||
oo = append(oo, aux...)
|
||||
|
||||
// - Default for empty response
|
||||
if len(oo) == 0 {
|
||||
return d.prepareResponseGeneric(localDef, dd), nil
|
||||
}
|
||||
|
||||
// - Optionally calculate paging
|
||||
if applyPaging {
|
||||
// when the buffer is empty, the isn't anything else to pull -- no next page
|
||||
if d.bufferSize() == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// Paging can only be applied to non-partitioned local frames
|
||||
// @todo generalize?
|
||||
if localDelimiter != 1 {
|
||||
return
|
||||
}
|
||||
|
||||
// @todo calculate prev page cursor; for now only next page is provided
|
||||
|
||||
if oo[0].Paging == nil {
|
||||
oo[0].Paging = &filter.Paging{}
|
||||
}
|
||||
oo[0].Paging.NextPage = d.calculatePagingCursor(oo[0], oo[1:], true, localDef.Sort)
|
||||
oo[0].Paging.NextPage.LThen = localDef.Sort.Reversed()
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// prepareResponseGeneric constructs an empty response in case where no data would be provided
|
||||
func (d *joinedDataset) prepareResponseGeneric(lfd *FrameDefinition, dd FrameDescriptionSet) (oo []*Frame) {
|
||||
dscr := dd.FilterByRef(lfd.Ref)[0]
|
||||
|
||||
f := &Frame{
|
||||
Name: lfd.Name,
|
||||
Ref: lfd.Ref,
|
||||
Source: lfd.Source,
|
||||
Columns: make(FrameColumnSet, 0, len(lfd.Columns)),
|
||||
}
|
||||
|
||||
if len(lfd.Columns) == 0 {
|
||||
f.Columns = dscr.Columns
|
||||
} else {
|
||||
for _, c := range lfd.Columns {
|
||||
f.Columns = append(f.Columns, dscr.Columns[dscr.Columns.Find(c.Name)])
|
||||
}
|
||||
}
|
||||
return []*Frame{f}
|
||||
}
|
||||
|
||||
// calculatePagingCursor calculates the paging cursor for the given frame
|
||||
//
|
||||
// The check for existence should be performed way in advanced so we won't bother here.
|
||||
// A unique value is also assured at way before.
|
||||
func (d *joinedDataset) calculatePagingCursor(local *Frame, foreign []*Frame, desc bool, ss filter.SortExprSet) *filter.PagingCursor {
|
||||
cursor := &filter.PagingCursor{LThen: ss.Reversed()}
|
||||
var foreignFrames []*Frame
|
||||
var v interface{}
|
||||
var localRow FrameRow
|
||||
|
||||
if desc {
|
||||
localRow = local.LastRow()
|
||||
} else {
|
||||
localRow = local.FirstRow()
|
||||
}
|
||||
|
||||
// Index foreign frames by ref for nicer lookups
|
||||
foreignIndex := make(map[string][]*Frame)
|
||||
for _, f := range foreign {
|
||||
foreignIndex[f.Ref] = append(foreignIndex[f.Ref], f)
|
||||
}
|
||||
|
||||
// Go over sorting and construct the cursor
|
||||
for _, s := range ss {
|
||||
foreignFrames = nil
|
||||
|
||||
if strings.Contains(s.Column, ".") {
|
||||
pts := strings.Split(s.Column, ".")
|
||||
foreignFrames = foreignIndex[pts[0]]
|
||||
var r FrameRow
|
||||
var f *Frame
|
||||
if desc {
|
||||
f = foreignFrames[len(foreignFrames)-1]
|
||||
r = f.FirstRow()
|
||||
} else {
|
||||
f = foreignFrames[0]
|
||||
r = f.FirstRow()
|
||||
}
|
||||
|
||||
if r[f.Columns.Find(pts[1])] != nil {
|
||||
v = r[f.Columns.Find(pts[1])].Get()
|
||||
}
|
||||
cursor.Set(s.Column, v, s.Descending)
|
||||
} else {
|
||||
if localRow[local.Columns.Find(s.Column)] != nil {
|
||||
v = localRow[local.Columns.Find(s.Column)].Get()
|
||||
}
|
||||
cursor.Set(s.Column, v, s.Descending)
|
||||
}
|
||||
}
|
||||
|
||||
return cursor
|
||||
}
|
||||
|
||||
func (d *joinedDataset) cutLocalFrameBuffer(cap int) (oo []*Frame) {
|
||||
// When partitioned, extract frames
|
||||
if d.partitioned {
|
||||
// We can use everything
|
||||
if len(d.localFrames) <= cap {
|
||||
oo = d.localFrames
|
||||
d.localFrames = nil
|
||||
return
|
||||
}
|
||||
|
||||
oo = d.localFrames[0:cap]
|
||||
d.localFrames = d.localFrames[cap:]
|
||||
return
|
||||
}
|
||||
|
||||
// When not partitioned, extract rows
|
||||
aux := d.localFrames[0].CloneMeta()
|
||||
if d.localFrames[0].Size() <= cap {
|
||||
aux.Rows = d.localFrames[0].Rows
|
||||
d.localFrames = nil
|
||||
return []*Frame{aux}
|
||||
}
|
||||
|
||||
aux.Rows = d.localFrames[0].Rows[0:cap]
|
||||
d.localFrames[0].Rows = d.localFrames[0].Rows[cap:]
|
||||
return []*Frame{aux}
|
||||
}
|
||||
|
||||
func (d *joinedDataset) cutForeignFrameBuffer(cap int) (oo []*Frame) {
|
||||
for i, dsFrames := range d.foreignFrames {
|
||||
// We can use everything
|
||||
if len(dsFrames) <= cap {
|
||||
oo = append(oo, dsFrames...)
|
||||
d.foreignFrames[i] = nil
|
||||
continue
|
||||
}
|
||||
|
||||
oo = append(oo, dsFrames[0:cap]...)
|
||||
d.foreignFrames[i] = dsFrames[cap:]
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (d *joinedDataset) bufferSize() int {
|
||||
if len(d.localFrames) == 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
if d.partitioned {
|
||||
return len(d.localFrames)
|
||||
}
|
||||
|
||||
return d.localFrames[0].Size()
|
||||
}
|
||||
|
||||
func (d *joinedDataset) sizeOfBuffer(b []*Frame, partitioned bool) int {
|
||||
if partitioned {
|
||||
return len(b)
|
||||
}
|
||||
|
||||
if len(b) == 0 {
|
||||
return 0
|
||||
}
|
||||
return b[0].Size()
|
||||
}
|
||||
|
||||
1
pkg/report/step_join_load_utils.go
Normal file
1
pkg/report/step_join_load_utils.go
Normal file
@@ -0,0 +1 @@
|
||||
package report
|
||||
@@ -22,379 +22,395 @@ type (
|
||||
}
|
||||
)
|
||||
|
||||
// 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) {
|
||||
// strategizeLoading initializes and prepares a set of load contexts based on the provided state
|
||||
func (d *joinedDataset) strategizeLoading(ctx context.Context, inverted bool, local, foreign *FrameDefinition) (localLoader *frameLoadCtx, foreignLoader *frameLoadCtx, err error) {
|
||||
if !inverted {
|
||||
return d.stratLocalMain(ctx, local, foreign)
|
||||
return d.stratLoadPrimary(ctx, inverted, local, foreign)
|
||||
}
|
||||
|
||||
return d.stratForeignMain(ctx, local, foreign)
|
||||
return d.stratLoadInverted(ctx, inverted, local, foreign)
|
||||
}
|
||||
|
||||
// stratLocalMain uses the local datasource for the main loader
|
||||
// stratLoadPrimary is the generic strategy where the local DS dictates the output
|
||||
//
|
||||
// The strategy is used always, except for when the initial sort is controlled
|
||||
// by the foreign datasource.
|
||||
//
|
||||
// In this strategy, the local datasource controlls the initial sort and the
|
||||
// filter that should be applied when pulling data from the foreign source.
|
||||
func (d *joinedDataset) stratLocalMain(ctx context.Context, local, foreign *FrameDefinition) (ml *frameBuffer, sl *frameBuffer, err error) {
|
||||
var ok bool
|
||||
|
||||
// Prepare the main loader from the local source
|
||||
// Local source can be partitioned to support nested joining.
|
||||
if d.partitioned {
|
||||
ok, err = (d.local.(PartitionableDatasource)).Partition(d.partitionSize, d.partitionCol)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if !ok {
|
||||
err = fmt.Errorf("local datasource is not partitionable: %s", d.local.Name())
|
||||
return
|
||||
}
|
||||
}
|
||||
// The strategy should be always be used, except when the foreign DS defines the initial sort.
|
||||
func (d *joinedDataset) stratLoadPrimary(ctx context.Context, inverted bool, local, foreign *FrameDefinition) (localLoader *frameLoadCtx, foreignLoader *frameLoadCtx, err error) {
|
||||
// @todo partitioned local when needed
|
||||
|
||||
// - local
|
||||
ldr, clsr, err := d.local.Load(ctx, local)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
localLoader = &frameLoadCtx{
|
||||
loader: ldr,
|
||||
closer: clsr,
|
||||
|
||||
ml = &frameBuffer{
|
||||
sourceName: d.Name(),
|
||||
sorting: local.Sort,
|
||||
keyCol: d.def.LocalColumn,
|
||||
keyColIndex: -1,
|
||||
|
||||
loader: func(_ *Filter, cap uint) ([]*Frame, error) {
|
||||
return ldr(int(cap), false)
|
||||
},
|
||||
closer: clsr,
|
||||
sorting: local.Sort,
|
||||
|
||||
// Overfetch frames when the last two entries define the same sort.
|
||||
// This is required to support paging.
|
||||
more: func(ff []*Frame, sc []int) bool {
|
||||
// No sorting, we don't care
|
||||
if len(local.Sort) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
// With sorting and using primary/unique columns, we don't care
|
||||
for _, s := range local.Sort {
|
||||
c := local.Columns[local.Columns.Find(s.Column)]
|
||||
if c.Primary || c.Unique {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// This is the last frame we can pull out, we don't care
|
||||
if uint(ff[len(ff)-1].Size()) < local.Paging.Limit {
|
||||
return false
|
||||
}
|
||||
|
||||
// With sorting and regular'ol columns, we care only if the over-fetched row
|
||||
// is the same as the last requested row.
|
||||
//
|
||||
// If we have multiple frames that means that it is partitioned; check whole frames
|
||||
if len(ff) > 1 {
|
||||
tmpl := len(ff)
|
||||
return ff[tmpl-1].FirstRow().Compare(ff[tmpl-2].FirstRow(), sc...) == 0
|
||||
}
|
||||
|
||||
// Else it is a regular'ol regular'ol frame; check rows
|
||||
return ff[0].LastRow().Compare(ff[0].LastLastRow(), sc...) == 0
|
||||
},
|
||||
}
|
||||
|
||||
// Partitioned sources need to be sliced
|
||||
if d.partitioned {
|
||||
ml.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
|
||||
}
|
||||
}
|
||||
|
||||
// - foreign
|
||||
//
|
||||
//
|
||||
|
||||
// Prepare the sub loader from the foreign source
|
||||
subDS, ok := d.foreign.(PartitionableDatasource)
|
||||
// -- only allow partitionable datasources
|
||||
pForeignDS, ok := d.foreign.(PartitionableDatasource)
|
||||
if !ok {
|
||||
// @todo allow alternatives also
|
||||
err = fmt.Errorf("foreign datasource is not partitionable")
|
||||
return
|
||||
}
|
||||
|
||||
//
|
||||
// Clone foreign filter so that we don't corrupt the initial one
|
||||
ffilter := foreign.Filter.Clone()
|
||||
sl = &frameBuffer{
|
||||
sourceName: d.Name(),
|
||||
// meta...
|
||||
keyCol: d.def.ForeignColumn,
|
||||
keyColIndex: -1,
|
||||
//
|
||||
foreignLoader = &frameLoadCtx{
|
||||
initLoader: func(cap int, f *Filter) (Loader, Closer, error) {
|
||||
foreign.Filter = merger(ffilter.Clone(), f, "and")
|
||||
|
||||
loader: func(keyFilter *Filter, cap uint) ([]*Frame, error) {
|
||||
foreign.Filter = merger(ffilter.Clone(), keyFilter, "and")
|
||||
|
||||
ok, err := subDS.Partition(cap, d.def.ForeignColumn)
|
||||
// - partition
|
||||
ok, err := pForeignDS.Partition(uint(cap), d.def.ForeignColumn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, nil, err
|
||||
}
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("foreign datasource is not partitionable: %s", d.foreign.Name())
|
||||
return nil, nil, fmt.Errorf("foreign datasource is not partitionable: %s", d.foreign.Name())
|
||||
}
|
||||
loader, closer, err := subDS.Load(ctx, foreign)
|
||||
if closer != nil {
|
||||
defer closer()
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return loader(0, false)
|
||||
},
|
||||
sorting: foreign.Sort,
|
||||
|
||||
more: func(_ []*Frame, _ []int) bool {
|
||||
return false
|
||||
return pForeignDS.Load(ctx, foreign)
|
||||
},
|
||||
|
||||
postFetch: func(ff []*Frame) ([]*Frame, error) {
|
||||
ff, err = d.sliceFrames(ff, d.def.ForeignColumn, d.def.LocalColumn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for i := range ff {
|
||||
if ff[i].Name == "" {
|
||||
ff[i].Name = foreign.Name
|
||||
}
|
||||
if ff[i].Source == "" {
|
||||
ff[i].Source = foreign.Source
|
||||
}
|
||||
if ff[i].Ref == "" {
|
||||
ff[i].Ref = foreign.Ref
|
||||
}
|
||||
}
|
||||
|
||||
return ff, nil
|
||||
},
|
||||
sorting: foreign.Sort,
|
||||
keyCol: d.def.ForeignColumn,
|
||||
keyColIndex: -1,
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (d *joinedDataset) stratForeignMain(ctx context.Context, local, foreign *FrameDefinition) (ml *frameBuffer, sl *frameBuffer, err error) {
|
||||
// @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")
|
||||
// stratLoadInverted makes the foreign DS dictate the output
|
||||
//
|
||||
// This strategy should only be used for cases where the foreign DS defines the initial sort.
|
||||
// When we're using inverted sort, the foreign DS should define the initial sort.
|
||||
func (d *joinedDataset) stratLoadInverted(ctx context.Context, inverted bool, local, foreign *FrameDefinition) (localLoader *frameLoadCtx, foreignLoader *frameLoadCtx, err error) {
|
||||
// - foreign
|
||||
//
|
||||
// -- only allow partitionable datasources
|
||||
pForeignDS, ok := d.foreign.(PartitionableDatasource)
|
||||
if !ok {
|
||||
// @todo allow alternatives also
|
||||
err = fmt.Errorf("foreign datasource is not partitionable")
|
||||
return
|
||||
}
|
||||
//
|
||||
// - partition
|
||||
ok, err = pForeignDS.Partition(uint(foreign.Paging.Limit), d.def.ForeignColumn)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if !ok {
|
||||
return nil, nil, fmt.Errorf("foreign datasource is not partitionable: %s", d.foreign.Name())
|
||||
}
|
||||
//
|
||||
// - init laoder
|
||||
ldr, clsr, err := d.foreign.Load(ctx, foreign)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
foreignLoader = &frameLoadCtx{
|
||||
loader: ldr,
|
||||
closer: clsr,
|
||||
|
||||
sorting: foreign.Sort,
|
||||
keyCol: d.def.ForeignColumn,
|
||||
keyColIndex: -1,
|
||||
}
|
||||
|
||||
// - local
|
||||
//
|
||||
// Clone foreign filter so that we don't corrupt the initial one
|
||||
// @todo partitioned local when needed
|
||||
lfilter := local.Filter.Clone()
|
||||
localLoader = &frameLoadCtx{
|
||||
initLoader: func(cap int, f *Filter) (Loader, Closer, error) {
|
||||
local.Filter = merger(lfilter.Clone(), f, "and")
|
||||
|
||||
return d.local.Load(ctx, local)
|
||||
},
|
||||
|
||||
sorting: local.Sort,
|
||||
keyCol: d.def.LocalColumn,
|
||||
keyColIndex: -1,
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// frameLoader is a utility function to load the appropriate data from the given source
|
||||
//
|
||||
// Outline:
|
||||
// . load a chunk of data
|
||||
// . post processing
|
||||
// . overfetch rows when required
|
||||
func (d *joinedDataset) frameLoader(loader Loader, loadCtx *frameLoadCtx, cap int, partitioned bool, post func(ff []*Frame) ([]*Frame, error)) (out []*Frame, err error) {
|
||||
var ff []*Frame
|
||||
out = make([]*Frame, 0, 32)
|
||||
|
||||
for {
|
||||
// . load a chunk of data
|
||||
ff, err = loader(cap, false)
|
||||
if err != nil || ff == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// . post processing
|
||||
// .. update metadata
|
||||
for _, f := range ff {
|
||||
f.Source = d.Name()
|
||||
}
|
||||
//
|
||||
// .. additional provided post processing
|
||||
if post != nil {
|
||||
ff, err = post(ff)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
//
|
||||
// .. update the output buffer
|
||||
if len(out) == 0 {
|
||||
out = append(out, ff...)
|
||||
} else {
|
||||
if !partitioned {
|
||||
// When not partitioned, we always pull one frame
|
||||
out[0].Rows = append(out[0].Rows, ff[0].Rows...)
|
||||
} else {
|
||||
// - Append to existing buffer when partitioned.
|
||||
// Pulled chunks can be considered complete so we don't have to worry about that
|
||||
out = append(out, ff...)
|
||||
}
|
||||
}
|
||||
|
||||
// . overfetch rows when required
|
||||
if cap == 0 || !d.overfetch(ff, partitioned, ff[0].Columns, loadCtx.sorting) {
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// overfetch checks if we need to overfetch data based on the provided sort
|
||||
//
|
||||
// We're overfetching when the last rows of the given buffer define the same values
|
||||
// for the sort columns.
|
||||
// If we don't overfetch, we may experience issues when applying paging.
|
||||
func (d *joinedDataset) overfetch(ff []*Frame, partitioned bool, cc FrameColumnSet, ss filter.SortExprSet) bool {
|
||||
// No sorting, we don't care
|
||||
if len(ss) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
// When sorting over primary/unique columns, we don't care
|
||||
for _, s := range ss {
|
||||
c := cc[cc.Find(s.Column)]
|
||||
if c.Primary || c.Unique {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// When the last two things have the same sort column values, we need to overfetch
|
||||
var (
|
||||
a, b FrameRow
|
||||
)
|
||||
|
||||
if !partitioned {
|
||||
if ff[0].Size() < 2 {
|
||||
return true
|
||||
}
|
||||
|
||||
a = ff[0].LastLastRow()
|
||||
b = ff[0].LastRow()
|
||||
} else {
|
||||
if len(ff) < 2 {
|
||||
return true
|
||||
}
|
||||
|
||||
a = ff[len(ff)-2].FirstRow()
|
||||
b = ff[len(ff)-1].FirstRow()
|
||||
}
|
||||
|
||||
sci := make([]int, 0, len(ss))
|
||||
for _, s := range ss {
|
||||
sci = append(sci, cc.Find(s.Column))
|
||||
}
|
||||
|
||||
return a.Compare(b, sci...) == 0
|
||||
}
|
||||
|
||||
// // // // // // // // // // // // // // // // // // // // // // // // //
|
||||
// Sorting
|
||||
|
||||
// shouldSort determines if we need to perform additional sorting based on the
|
||||
// provided frame definition sorting.
|
||||
// sortFrameBuffers sorts the provided local, foreign frame buffers
|
||||
//
|
||||
// * If no sort was requested; consider it sorted.
|
||||
// * If all sort expressions use only the local source; consider it sorted.
|
||||
// * If at least one sort expression references a foreign source; consider it NOT sorted.
|
||||
func (d *joinedDataset) shouldSort(ss filter.SortExprSet) bool {
|
||||
if len(ss) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
// Outline:
|
||||
// . bucket foreign frames based on their ref value
|
||||
// . sort entire local frames when more then one is proviced (covers partitioned local ds)
|
||||
// . sort rows for each local frame
|
||||
// . assure correct foreign frame order to match local frame rows
|
||||
func (d *joinedDataset) sortFrameBuffers(localLoader, foreignLoader *frameLoadCtx, localBuffer, foreignBuffer []*Frame, ss filter.SortExprSet, inverted bool) ([]*Frame, []*Frame) {
|
||||
// - extract sort expressions for local
|
||||
//
|
||||
// Foreign frames are chunked and those are already sorted as they should be.
|
||||
localSS := make(filter.SortExprSet, 0, len(ss))
|
||||
for _, s := range ss {
|
||||
if strings.Contains(s.Column, ".") {
|
||||
return true
|
||||
if !strings.Contains(s.Column, ".") {
|
||||
localSS = append(localSS, s)
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func (d *joinedDataset) strategizeSort(main, sub *frameBuffer, inverted bool, lfd *FrameDefinition, keyColumn string) (err error) {
|
||||
var local, foreign *frameBuffer
|
||||
|
||||
// Determine which one was local/foreign
|
||||
if inverted {
|
||||
local = sub
|
||||
foreign = main
|
||||
} else {
|
||||
local = main
|
||||
foreign = sub
|
||||
}
|
||||
|
||||
// - index foreign frames into buckets; here the foreign sort must be respected
|
||||
// . bucket foreign frames based on their ref value
|
||||
//
|
||||
// This will speed up later processing
|
||||
buckets := make(map[string]int)
|
||||
for i, mf := range foreign.localFrames {
|
||||
for i, mf := range foreignBuffer {
|
||||
buckets[mf.RefValue] = i
|
||||
}
|
||||
|
||||
localSort := make(filter.SortExprSet, 0, len(lfd.Sort))
|
||||
for _, s := range lfd.Sort {
|
||||
if !strings.Contains(s.Column, ".") {
|
||||
localSort = append(localSort, s)
|
||||
// . sort entire local frames when more then one is proviced (covers partitioned local ds)
|
||||
sort.SliceStable(localBuffer, func(i, j int) bool {
|
||||
frameI := localBuffer[i]
|
||||
frameDelimiterI := frameI.FirstRow()
|
||||
frameJ := localBuffer[j]
|
||||
frameDelimiterJ := frameJ.FirstRow()
|
||||
|
||||
// what bucket the frame corresponds to
|
||||
bucketI := buckets[cast.ToString(frameDelimiterI[localLoader.keyColIndex].Get())]
|
||||
bucketJ := buckets[cast.ToString(frameDelimiterJ[localLoader.keyColIndex].Get())]
|
||||
|
||||
// when inverted, use foreign frames to determine initial sort
|
||||
if inverted {
|
||||
if bucketI < bucketJ {
|
||||
return !ss.Reversed()
|
||||
}
|
||||
}
|
||||
|
||||
// go through the sort definitions and sort based on that
|
||||
for si, s := range localSS {
|
||||
ci, ok := frameDelimiterI[localLoader.sortColumns[si]].(expr.Comparable)
|
||||
if !ok {
|
||||
return !s.Descending
|
||||
}
|
||||
|
||||
r, err := ci.Compare(frameDelimiterJ[localLoader.sortColumns[si]])
|
||||
if err != nil {
|
||||
return s.Descending
|
||||
}
|
||||
|
||||
if r != 0 {
|
||||
if s.Descending {
|
||||
return r > 0
|
||||
}
|
||||
return r < 0
|
||||
}
|
||||
}
|
||||
|
||||
return bucketI < bucketJ
|
||||
})
|
||||
|
||||
// . sort rows for each local frame
|
||||
for _, l := range localBuffer {
|
||||
sort.SliceStable(l.Rows, func(i, j int) bool {
|
||||
rowI := l.Rows[i]
|
||||
rowJ := l.Rows[j]
|
||||
|
||||
// what bucket the frame corresponds to
|
||||
bucketI := buckets[cast.ToString(rowI[localLoader.keyColIndex].Get())]
|
||||
bucketJ := buckets[cast.ToString(rowJ[localLoader.keyColIndex].Get())]
|
||||
|
||||
// when inverted, use foreign frames to determine initial sort
|
||||
if inverted {
|
||||
if bucketI < bucketJ {
|
||||
return !ss.Reversed()
|
||||
}
|
||||
}
|
||||
|
||||
// go through the sort definitions and sort based on that
|
||||
for si, s := range localSS {
|
||||
ci, ok := rowI[localLoader.sortColumns[si]].(expr.Comparable)
|
||||
if !ok {
|
||||
return !s.Descending
|
||||
}
|
||||
|
||||
r, err := ci.Compare(rowJ[localLoader.sortColumns[si]])
|
||||
if err != nil {
|
||||
return s.Descending
|
||||
}
|
||||
|
||||
if r != 0 {
|
||||
if s.Descending {
|
||||
return r > 0
|
||||
}
|
||||
return r < 0
|
||||
}
|
||||
}
|
||||
|
||||
return bucketI < bucketJ
|
||||
})
|
||||
}
|
||||
|
||||
if len(local.localFrames) == 1 {
|
||||
err = d.sortFrameRows(local, localSort, buckets, inverted)
|
||||
} else {
|
||||
err = d.sortBufferFrames(local, localSort, buckets, inverted)
|
||||
// . assure correct foreign frame order to match local frame rows
|
||||
//
|
||||
// Each foreign frame is on the same index as the corresponding local row.
|
||||
// This simplifies later algorithms and removes the need for additional
|
||||
// mapping structures.
|
||||
for _, l := range localBuffer {
|
||||
l.WalkRows(func(i int, r FrameRow) error {
|
||||
buckets[cast.ToString(r[localLoader.keyColIndex].Get())] = i
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// - assure bucket's order
|
||||
// @todo can we do this in one go alongside the original sort?
|
||||
aux := make([]*Frame, len(foreign.localFrames))
|
||||
for _, f := range foreign.localFrames {
|
||||
//
|
||||
// Update foreign frame order based on reordered buckets
|
||||
aux := make([]*Frame, len(foreignBuffer))
|
||||
for _, f := range foreignBuffer {
|
||||
aux[buckets[f.RefValue]] = f
|
||||
}
|
||||
foreign.localFrames = aux
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (d *joinedDataset) sortFrameRows(local *frameBuffer, ss filter.SortExprSet, buckets map[string]int, inverted bool) error {
|
||||
f := local.localFrames[0]
|
||||
|
||||
// - sort based on bucket index
|
||||
var ri FrameRow
|
||||
var rj FrameRow
|
||||
|
||||
var bi int
|
||||
var bj int
|
||||
|
||||
var ci expr.Comparable
|
||||
var ok bool
|
||||
|
||||
sort.SliceStable(f.Rows, func(i, j int) bool {
|
||||
ri = f.Rows[i]
|
||||
rj = f.Rows[j]
|
||||
bi = buckets[cast.ToString(ri[local.keyColIndex].Get())]
|
||||
bj = buckets[cast.ToString(rj[local.keyColIndex].Get())]
|
||||
|
||||
if inverted {
|
||||
// if bi is before bj, the row bellongs to a complitely different bucket
|
||||
// so we shouldn't do anything extra
|
||||
if bi < bj {
|
||||
return !ss.Reversed()
|
||||
}
|
||||
}
|
||||
|
||||
// go through the sort definitions and sort based on that
|
||||
for si, s := range ss {
|
||||
ci, ok = ri[local.sortColumns[si]].(expr.Comparable)
|
||||
if !ok {
|
||||
return !s.Descending
|
||||
}
|
||||
|
||||
r, err := ci.Compare(rj[local.sortColumns[si]])
|
||||
if err != nil {
|
||||
return s.Descending
|
||||
}
|
||||
|
||||
if r != 0 {
|
||||
if s.Descending {
|
||||
return r > 0
|
||||
}
|
||||
return r < 0
|
||||
}
|
||||
}
|
||||
|
||||
return bi < bj
|
||||
})
|
||||
|
||||
f.WalkRows(func(i int, r FrameRow) error {
|
||||
buckets[cast.ToString(r[local.keyColIndex].Get())] = i
|
||||
return nil
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *joinedDataset) sortBufferFrames(local *frameBuffer, ss filter.SortExprSet, buckets map[string]int, inverted bool) error {
|
||||
f := local.localFrames[0]
|
||||
|
||||
var ri *Frame
|
||||
var rj *Frame
|
||||
|
||||
var bi int
|
||||
var bj int
|
||||
|
||||
var ci expr.Comparable
|
||||
var ok bool
|
||||
|
||||
sort.SliceStable(local.localFrames, func(i, j int) bool {
|
||||
ri = local.localFrames[i]
|
||||
rj = local.localFrames[j]
|
||||
bi = buckets[cast.ToString(ri.Rows[0][local.keyColIndex].Get())]
|
||||
bj = buckets[cast.ToString(rj.Rows[0][local.keyColIndex].Get())]
|
||||
|
||||
if inverted {
|
||||
// if bi is before bj, the row bellongs to a complitely different bucket
|
||||
// so we shouldn't do anything extra
|
||||
if bi < bj {
|
||||
return !ss.Reversed()
|
||||
}
|
||||
}
|
||||
|
||||
// go through the sort definitions and sort based on that
|
||||
for si, s := range ss {
|
||||
ci, ok = ri.Rows[0][local.sortColumns[si]].(expr.Comparable)
|
||||
if !ok {
|
||||
return !s.Descending
|
||||
}
|
||||
|
||||
r, err := ci.Compare(rj.Rows[0][local.sortColumns[si]])
|
||||
if err != nil {
|
||||
return s.Descending
|
||||
}
|
||||
|
||||
if r != 0 {
|
||||
if s.Descending {
|
||||
return r > 0
|
||||
}
|
||||
return r < 0
|
||||
}
|
||||
}
|
||||
|
||||
return bi < bj
|
||||
})
|
||||
|
||||
f.WalkRows(func(i int, r FrameRow) error {
|
||||
buckets[cast.ToString(r[local.keyColIndex].Get())] = i
|
||||
return nil
|
||||
})
|
||||
|
||||
return nil
|
||||
return localBuffer, aux
|
||||
}
|
||||
|
||||
// // // // // // // // // // // // // // // // // // // // // // // // //
|
||||
// Paging
|
||||
|
||||
// strategizePaging breaks down the given paging cursor into partial conditions and
|
||||
// applies it to the frame definitions.
|
||||
func (d *joinedDataset) strategizePaging(local, foreign *FrameDefinition, inverted bool) (pp []partialPagingCnd, err error) {
|
||||
pp, err = d.calculatePagingFilters(local, inverted)
|
||||
// break down the cursor
|
||||
pp, err = d.calcualteCursorConditions(local, inverted)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
local.Filter = merger(&Filter{pp[0].filterCut}, local.Filter, "and")
|
||||
local.Paging.PageCursor = nil
|
||||
|
||||
// apply the "cut" filter to the appropriate DS; foreign when inverted,
|
||||
// local otherwise.
|
||||
switch pp[0].ref {
|
||||
case local.Ref, "":
|
||||
local.Filter = merger(&Filter{pp[0].filterCut}, local.Filter, "and")
|
||||
case foreign.Ref:
|
||||
foreign.Filter = merger(&Filter{pp[0].filterCut}, foreign.Filter, "and")
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// calculatePagingFilters produces additional filtering that should be done
|
||||
// on the datasource level and/or in the join logic.
|
||||
// calcualteCursorConditions processes the given paging cursor and returns a set
|
||||
// of partial conditions
|
||||
//
|
||||
// The core logic is extracted from store/rdbms/builders/cursor.go
|
||||
func (d *joinedDataset) calculatePagingFilters(local *FrameDefinition, inverted bool) (partials []partialPagingCnd, err error) {
|
||||
//
|
||||
// Partial cursor conditions are to be used for additional DS filtering and
|
||||
// additional post filtering when processing the data.
|
||||
func (d *joinedDataset) calcualteCursorConditions(local *FrameDefinition, inverted bool) (partials []partialPagingCnd, err error) {
|
||||
if len(local.Paging.PageCursor.Keys()) == 0 {
|
||||
return
|
||||
}
|
||||
@@ -662,60 +678,99 @@ out:
|
||||
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
|
||||
// applyPagingFilter performs additional filtering based on the provided paging partials
|
||||
func (d *joinedDataset) applyPagingFilter(buffLocal, buffForeign []*Frame, pp []partialPagingCnd) ([]*Frame, []*Frame, bool, bool, error) {
|
||||
var (
|
||||
err error
|
||||
modified, satisfied bool
|
||||
cutIndex int
|
||||
)
|
||||
|
||||
// When there is only one partial, it was already handled by the DS.
|
||||
// Sorting assures a unique column when working with cursors.
|
||||
if len(pp) <= 1 {
|
||||
// Already satisfied by the DS
|
||||
return false, true
|
||||
return buffLocal, buffForeign, false, true, nil
|
||||
}
|
||||
|
||||
inclCondition := pp[0]
|
||||
fCondition := pp[1]
|
||||
colIndex := make(map[string]FrameColumnSet)
|
||||
colIndex[""] = buffLocal[0].Columns
|
||||
colIndex[buffLocal[0].Ref] = buffLocal[0].Columns
|
||||
colIndex[buffForeign[0].Ref] = buffForeign[0].Columns
|
||||
|
||||
main.walkRowsLocal(func(i int, r FrameRow) error {
|
||||
if done {
|
||||
return nil
|
||||
}
|
||||
rowIndex := 0
|
||||
nextRows := func() (map[string]FrameRow, bool) {
|
||||
out := make(map[string]FrameRow)
|
||||
more := true
|
||||
|
||||
// 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
|
||||
}
|
||||
// local
|
||||
if d.partitioned {
|
||||
// local can just be ""
|
||||
out[""] = buffLocal[rowIndex].FirstRow()
|
||||
out[buffLocal[0].Ref] = buffLocal[rowIndex].FirstRow()
|
||||
} else {
|
||||
done = true
|
||||
return nil
|
||||
// local can just be ""
|
||||
out[""] = buffLocal[0].PeekRow(rowIndex)
|
||||
out[buffLocal[0].Ref] = buffLocal[0].PeekRow(rowIndex)
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
// foreign
|
||||
out[buffForeign[0].Ref] = buffForeign[rowIndex].FirstRow()
|
||||
|
||||
if cutSize > 0 {
|
||||
main.removeLocal(cutSize)
|
||||
main.removeForeign(cutSize)
|
||||
sub.removeLocal(cutSize)
|
||||
sub.removeForeign(cutSize)
|
||||
rowIndex++
|
||||
more = rowIndex < len(buffForeign)
|
||||
|
||||
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 out, more
|
||||
}
|
||||
|
||||
return false, true
|
||||
outer:
|
||||
for {
|
||||
// peek next rows and break if no more
|
||||
rr, more := nextRows()
|
||||
|
||||
// check paging partials; break if cursor passes, continue if it does not
|
||||
pass := true
|
||||
for i, p := range pp {
|
||||
n := p.filterInclude
|
||||
if i > 0 {
|
||||
n = p.filterCut
|
||||
}
|
||||
tmp := d.eval(n, rr[p.ref], colIndex[p.ref])
|
||||
// the first partial is less strict so it only matches the rows that
|
||||
// can potentially be included.
|
||||
if !tmp && i == 0 {
|
||||
break outer
|
||||
}
|
||||
|
||||
pass = pass && tmp
|
||||
if !pass {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// if all conditions passed, this is the delimiter for what is included
|
||||
if pass {
|
||||
break
|
||||
}
|
||||
|
||||
cutIndex++
|
||||
if !more {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// cut based on cut index; satisfied when we don't remove everything due to paging
|
||||
modified = cutIndex > 0
|
||||
satisfied = cutIndex < len(buffForeign)-1
|
||||
//
|
||||
// - local
|
||||
if d.partitioned {
|
||||
buffLocal = buffLocal[cutIndex:]
|
||||
} else {
|
||||
buffLocal[0].Rows = buffLocal[0].Rows[cutIndex:]
|
||||
}
|
||||
//
|
||||
// - foreign
|
||||
buffForeign = buffForeign[cutIndex:]
|
||||
|
||||
return buffLocal, buffForeign, modified, satisfied, err
|
||||
}
|
||||
|
||||
@@ -2,163 +2,161 @@ package reporter
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/report"
|
||||
)
|
||||
|
||||
func Test3002_joining_base_nested(t *testing.T) {
|
||||
|
||||
t.Skip("@todo temporarily disabled")
|
||||
|
||||
var (
|
||||
ctx, h, s = setup(t)
|
||||
m, _, dd = loadScenario(ctx, s, t, h)
|
||||
ff = loadNoErr(ctx, h, m, dd...)
|
||||
local, foreign *report.Frame
|
||||
)
|
||||
// var (
|
||||
// ctx, h, s = setup(t)
|
||||
// m, _, dd = loadScenario(ctx, s, t, h)
|
||||
// ff = loadNoErr(ctx, h, m, dd...)
|
||||
// local, foreign *report.Frame
|
||||
// )
|
||||
|
||||
// The joining here looks like this:
|
||||
//
|
||||
// (nested)
|
||||
// (nested_aux) (cc)
|
||||
// (aa) (bb)
|
||||
// // The joining here looks like this:
|
||||
// //
|
||||
// // (nested)
|
||||
// // (nested_aux) (cc)
|
||||
// // (aa) (bb)
|
||||
|
||||
h.a.Len(ff, 11)
|
||||
// h.a.Len(ff, 11)
|
||||
|
||||
ix := indexJoinedResult(ff)
|
||||
_ = ix
|
||||
// ix := indexJoinedResult(ff)
|
||||
// _ = ix
|
||||
|
||||
// // joined -- 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", local.Source)
|
||||
h.a.Equal("aa", local.Ref)
|
||||
checkRows(h, local,
|
||||
"aa_01, aa :: 01",
|
||||
"aa_02, aa :: 02",
|
||||
"aa_03, aa :: 03",
|
||||
"aa_04, aa :: 04",
|
||||
"aa_05, aa :: 05")
|
||||
// // local
|
||||
// local = ff[0]
|
||||
// h.a.Equal("pk<String>, label<String>", local.Columns.OmitSys().String())
|
||||
// h.a.Equal("joined", local.Source)
|
||||
// h.a.Equal("aa", local.Ref)
|
||||
// checkRows(h, local,
|
||||
// "aa_01, aa :: 01",
|
||||
// "aa_02, aa :: 02",
|
||||
// "aa_03, aa :: 03",
|
||||
// "aa_04, aa :: 04",
|
||||
// "aa_05, aa :: 05")
|
||||
|
||||
// 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)
|
||||
h.a.Equal("bb", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_01", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"bb_01, aa_01, bb :: 01",
|
||||
"bb_02, aa_01, bb :: 02",
|
||||
"bb_03, aa_01, bb :: 03")
|
||||
// // 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)
|
||||
// h.a.Equal("bb", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_01", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "bb_01, aa_01, bb :: 01",
|
||||
// "bb_02, aa_01, bb :: 02",
|
||||
// "bb_03, aa_01, bb :: 03")
|
||||
|
||||
// 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)
|
||||
h.a.Equal("bb", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_02", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"bb_04, aa_02, bb :: 04",
|
||||
"bb_05, aa_02, bb :: 05")
|
||||
// // 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)
|
||||
// h.a.Equal("bb", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_02", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "bb_04, aa_02, bb :: 04",
|
||||
// "bb_05, aa_02, bb :: 05")
|
||||
|
||||
// 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)
|
||||
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")
|
||||
// // 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)
|
||||
// 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")
|
||||
|
||||
// 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)
|
||||
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")
|
||||
// // 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)
|
||||
// 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")
|
||||
|
||||
// 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)
|
||||
h.a.Equal("bb", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_05", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"bb_08, aa_05, bb :: 08",
|
||||
"bb_09, aa_05, bb :: 09",
|
||||
"bb_10, aa_05, bb :: 10")
|
||||
// // 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)
|
||||
// h.a.Equal("bb", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_05", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "bb_08, aa_05, bb :: 08",
|
||||
// "bb_09, aa_05, bb :: 09",
|
||||
// "bb_10, aa_05, bb :: 10")
|
||||
|
||||
// The other foreign
|
||||
// // The other foreign
|
||||
|
||||
// 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("cc", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_01", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"cc_01, aa_01, cc :: 01",
|
||||
"cc_02, aa_01, cc :: 02")
|
||||
// // 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("cc", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_01", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "cc_01, aa_01, cc :: 01",
|
||||
// "cc_02, aa_01, cc :: 02")
|
||||
|
||||
// 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("cc", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_02", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"cc_03, aa_02, cc :: 03",
|
||||
"cc_04, aa_02, cc :: 04")
|
||||
// // 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("cc", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_02", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "cc_03, aa_02, cc :: 03",
|
||||
// "cc_04, aa_02, cc :: 04")
|
||||
|
||||
// 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("cc", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_03", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"cc_05, aa_03, cc :: 05",
|
||||
"cc_06, aa_03, cc :: 06")
|
||||
// // 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("cc", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_03", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "cc_05, aa_03, cc :: 05",
|
||||
// "cc_06, aa_03, cc :: 06")
|
||||
|
||||
// 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("cc", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_04", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"cc_07, aa_04, cc :: 07")
|
||||
// // 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("cc", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_04", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "cc_07, aa_04, cc :: 07")
|
||||
|
||||
// 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("cc", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_05", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"cc_08, aa_05, cc :: 08")
|
||||
// // 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("cc", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_05", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "cc_08, aa_05, cc :: 08")
|
||||
}
|
||||
|
||||
@@ -2,192 +2,190 @@ package reporter
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/report"
|
||||
)
|
||||
|
||||
func Test3003_joining_base_nested_complex(t *testing.T) {
|
||||
|
||||
t.Skip("@todo temporarily disabled")
|
||||
|
||||
var (
|
||||
ctx, h, s = setup(t)
|
||||
m, _, dd = loadScenario(ctx, s, t, h)
|
||||
ff = loadNoErr(ctx, h, m, dd...)
|
||||
local, foreign *report.Frame
|
||||
)
|
||||
// var (
|
||||
// ctx, h, s = setup(t)
|
||||
// m, _, dd = loadScenario(ctx, s, t, h)
|
||||
// ff = loadNoErr(ctx, h, m, dd...)
|
||||
// local, foreign *report.Frame
|
||||
// )
|
||||
|
||||
// The joining here looks like this:
|
||||
//
|
||||
// (nested)
|
||||
// (nested_lft) (nested_rgh)
|
||||
// (aa) (bb) (cc) (dd)
|
||||
// // The joining here looks like this:
|
||||
// //
|
||||
// // (nested)
|
||||
// // (nested_lft) (nested_rgh)
|
||||
// // (aa) (bb) (cc) (dd)
|
||||
|
||||
h.a.Len(ff, 19)
|
||||
// h.a.Len(ff, 19)
|
||||
|
||||
ix := indexJoinedResult(ff)
|
||||
_ = ix
|
||||
// ix := indexJoinedResult(ff)
|
||||
// _ = ix
|
||||
|
||||
// // left join
|
||||
// // // left join
|
||||
|
||||
local = ff[0]
|
||||
h.a.Equal("pk<String>, label<String>", local.Columns.OmitSys().String())
|
||||
h.a.Equal("joined", local.Source)
|
||||
h.a.Equal("aa", local.Ref)
|
||||
checkRows(h, local,
|
||||
"aa_01, aa :: 01",
|
||||
"aa_02, aa :: 02",
|
||||
"aa_03, aa :: 03",
|
||||
"aa_04, aa :: 04",
|
||||
"aa_05, aa :: 05")
|
||||
// local = ff[0]
|
||||
// h.a.Equal("pk<String>, label<String>", local.Columns.OmitSys().String())
|
||||
// h.a.Equal("joined", local.Source)
|
||||
// h.a.Equal("aa", local.Ref)
|
||||
// checkRows(h, local,
|
||||
// "aa_01, aa :: 01",
|
||||
// "aa_02, aa :: 02",
|
||||
// "aa_03, aa :: 03",
|
||||
// "aa_04, aa :: 04",
|
||||
// "aa_05, aa :: 05")
|
||||
|
||||
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)
|
||||
h.a.Equal("bb", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_01", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"bb_01, aa_01, bb :: 01",
|
||||
"bb_02, aa_01, bb :: 02",
|
||||
"bb_03, aa_01, bb :: 03")
|
||||
// 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)
|
||||
// h.a.Equal("bb", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_01", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "bb_01, aa_01, bb :: 01",
|
||||
// "bb_02, aa_01, bb :: 02",
|
||||
// "bb_03, aa_01, bb :: 03")
|
||||
|
||||
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)
|
||||
h.a.Equal("bb", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_02", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"bb_04, aa_02, bb :: 04",
|
||||
"bb_05, aa_02, bb :: 05")
|
||||
// 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)
|
||||
// h.a.Equal("bb", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_02", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "bb_04, aa_02, bb :: 04",
|
||||
// "bb_05, aa_02, bb :: 05")
|
||||
|
||||
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)
|
||||
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["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)
|
||||
// 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["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)
|
||||
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["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)
|
||||
// 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["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)
|
||||
h.a.Equal("bb", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
h.a.Equal("aa_05", foreign.RefValue)
|
||||
checkRows(h, foreign,
|
||||
"bb_08, aa_05, bb :: 08",
|
||||
"bb_09, aa_05, bb :: 09",
|
||||
"bb_10, aa_05, bb :: 10")
|
||||
// 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)
|
||||
// h.a.Equal("bb", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// h.a.Equal("aa_05", foreign.RefValue)
|
||||
// checkRows(h, foreign,
|
||||
// "bb_08, aa_05, bb :: 08",
|
||||
// "bb_09, aa_05, bb :: 09",
|
||||
// "bb_10, aa_05, bb :: 10")
|
||||
|
||||
// // right join
|
||||
// // // right join
|
||||
|
||||
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("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["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("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["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("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["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("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["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("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["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("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["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("cc", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
checkRows(h, foreign,
|
||||
"cc_07, aa_04, cc :: 07")
|
||||
// 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("cc", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// checkRows(h, foreign,
|
||||
// "cc_07, aa_04, cc :: 07")
|
||||
|
||||
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("cc", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
checkRows(h, foreign,
|
||||
"cc_08, aa_05, cc :: 08")
|
||||
// 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("cc", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// checkRows(h, foreign,
|
||||
// "cc_08, aa_05, cc :: 08")
|
||||
|
||||
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", 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["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", 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["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", 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["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", 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["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", 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["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", 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["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", 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["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", 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["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", foreign.Source)
|
||||
h.a.Equal("dd", foreign.Ref)
|
||||
h.a.Equal("pk", foreign.RelColumn)
|
||||
checkRows(h, foreign,
|
||||
"dd_05, cc_05, dd :: 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", foreign.Source)
|
||||
// h.a.Equal("dd", foreign.Ref)
|
||||
// h.a.Equal("pk", foreign.RelColumn)
|
||||
// checkRows(h, foreign,
|
||||
// "dd_05, cc_05, dd :: 05")
|
||||
}
|
||||
|
||||
@@ -2,94 +2,92 @@ package reporter
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/report"
|
||||
)
|
||||
|
||||
func Test3006_joining_paging_nested(t *testing.T) {
|
||||
|
||||
t.Skip("@todo temporarily disabled")
|
||||
|
||||
var (
|
||||
ctx, h, s = setup(t)
|
||||
m, _, dd = loadScenario(ctx, s, t, h)
|
||||
ff []*report.Frame
|
||||
def = dd[0]
|
||||
local, foreign *report.Frame
|
||||
)
|
||||
// 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
|
||||
// // // // 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")
|
||||
// // 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_01"]
|
||||
// h.a.NotNil(foreign)
|
||||
|
||||
foreign = ix["bb/aa/aa_02"]
|
||||
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_01"]
|
||||
// h.a.NotNil(foreign)
|
||||
|
||||
foreign = ix["cc/aa/aa_02"]
|
||||
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
|
||||
// // // // 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")
|
||||
// // 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_03"]
|
||||
// h.a.NotNil(foreign)
|
||||
|
||||
foreign = ix["bb/aa/aa_04"]
|
||||
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_03"]
|
||||
// h.a.NotNil(foreign)
|
||||
|
||||
foreign = ix["cc/aa/aa_04"]
|
||||
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
|
||||
// // // // 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")
|
||||
// // 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["bb/aa/aa_05"]
|
||||
// h.a.NotNil(foreign)
|
||||
|
||||
foreign = ix["cc/aa/aa_05"]
|
||||
h.a.NotNil(foreign)
|
||||
// foreign = ix["cc/aa/aa_05"]
|
||||
// h.a.NotNil(foreign)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user