diff --git a/pkg/report/frame_buffer.go b/pkg/report/frame_buffer.go deleted file mode 100644 index 9e446c08b..000000000 --- a/pkg/report/frame_buffer.go +++ /dev/null @@ -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} -} diff --git a/pkg/report/frame_load_ctx.go b/pkg/report/frame_load_ctx.go new file mode 100644 index 000000000..a73fcb6f3 --- /dev/null +++ b/pkg/report/frame_load_ctx.go @@ -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, + }, + }, + }, + } +} diff --git a/pkg/report/step_join.go b/pkg/report/step_join.go index 43c4c4852..64247f214 100644 --- a/pkg/report/step_join.go +++ b/pkg/report/step_join.go @@ -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() } diff --git a/pkg/report/step_join_load_utils.go b/pkg/report/step_join_load_utils.go new file mode 100644 index 000000000..80c499fb2 --- /dev/null +++ b/pkg/report/step_join_load_utils.go @@ -0,0 +1 @@ +package report diff --git a/pkg/report/step_join_strategies.go b/pkg/report/step_join_strategies.go index 79da8b3ff..a5a24374d 100644 --- a/pkg/report/step_join_strategies.go +++ b/pkg/report/step_join_strategies.go @@ -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 } diff --git a/tests/reporter/3002_joining_base_nested_test.go b/tests/reporter/3002_joining_base_nested_test.go index e6c51b875..873bc25dd 100644 --- a/tests/reporter/3002_joining_base_nested_test.go +++ b/tests/reporter/3002_joining_base_nested_test.go @@ -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, label", 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, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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") } diff --git a/tests/reporter/3003_joining_base_nested_complex_test.go b/tests/reporter/3003_joining_base_nested_complex_test.go index a0451ef96..1d53255a4 100644 --- a/tests/reporter/3003_joining_base_nested_complex_test.go +++ b/tests/reporter/3003_joining_base_nested_complex_test.go @@ -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, label", 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, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_a, label", 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, fk_c, label", 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, fk_c, label", 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, fk_c, label", 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, fk_c, label", 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, fk_c, label", 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, fk_c, label", 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, fk_c, label", 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, fk_c, label", 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, fk_c, label", 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, fk_c, label", 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") } diff --git a/tests/reporter/3006_joining_paging_nested_test.go b/tests/reporter/3006_joining_paging_nested_test.go index 591816790..433fa50f8 100644 --- a/tests/reporter/3006_joining_paging_nested_test.go +++ b/tests/reporter/3006_joining_paging_nested_test.go @@ -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) }