Implement pipeline link (left) execution step
This commit is contained in:
Tomaž Jerman
@@ -48,6 +48,10 @@ type (
|
||||
// as the data is provided in the correct order.
|
||||
partialScan bool
|
||||
}
|
||||
|
||||
rowLink struct {
|
||||
a, b ValueGetter
|
||||
}
|
||||
)
|
||||
|
||||
func (def *Link) Identifier() string {
|
||||
@@ -128,3 +132,40 @@ func (def *Link) validate(ii []Iterator) (err error) {
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (r *rowLink) SelectGVal(ctx context.Context, k string) (interface{}, error) {
|
||||
return r.GetValue(k, 0)
|
||||
}
|
||||
|
||||
func (r *rowLink) GetValue(name string, pos uint) (v any, err error) {
|
||||
a := r.a.CountValues()
|
||||
if cc, ok := a[name]; ok {
|
||||
if pos >= cc {
|
||||
return nil, nil
|
||||
}
|
||||
return r.a.GetValue(name, pos)
|
||||
}
|
||||
|
||||
b := r.b.CountValues()
|
||||
if cc, ok := b[name]; ok {
|
||||
if pos >= cc {
|
||||
return nil, nil
|
||||
}
|
||||
return r.b.GetValue(name, pos)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (r *rowLink) CountValues() (out map[string]uint) {
|
||||
out = make(map[string]uint)
|
||||
|
||||
for k, c := range r.a.CountValues() {
|
||||
out[k] = c
|
||||
}
|
||||
for k, c := range r.b.CountValues() {
|
||||
out[k] = c
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
@@ -0,0 +1,549 @@
|
||||
package dal
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"sort"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/filter"
|
||||
"github.com/spf13/cast"
|
||||
)
|
||||
|
||||
type (
|
||||
linkLeft struct {
|
||||
def Link
|
||||
filter internalFilter
|
||||
|
||||
leftSource Iterator
|
||||
rightSource Iterator
|
||||
err error
|
||||
scanRow *row
|
||||
planned bool
|
||||
filtered bool
|
||||
|
||||
// Index the attributes for easier lookups later on
|
||||
outLeftAttrIndex map[string]int
|
||||
outRightAttrIndex map[string]int
|
||||
leftAttrIndex map[string]int
|
||||
rightAttrIndex map[string]int
|
||||
|
||||
linkRightAttr AttributeMapping
|
||||
linkLeftAttr AttributeMapping
|
||||
rightSortAttrs []string
|
||||
|
||||
rowTester tester
|
||||
|
||||
// Buffer to keep track of pulled left rows
|
||||
leftRows []*row
|
||||
relIndex *relIndex
|
||||
keepLeft bool
|
||||
leftIndex int
|
||||
rightIndex int
|
||||
|
||||
// Some helper fields for temporary data
|
||||
leftRow *row
|
||||
rightRow *row
|
||||
relScanBuffer *relIndexBuffer
|
||||
}
|
||||
)
|
||||
|
||||
func (xs *linkLeft) init(ctx context.Context) (err error) {
|
||||
// @note the index is keeping track of the right source attributes so we can
|
||||
// simplify the sorting logic.
|
||||
xs.relIndex = newRelIndex(xs.rightSortAttrs...)
|
||||
xs.indexAttributes()
|
||||
|
||||
// basic sort breadown
|
||||
for _, o := range xs.filter.OrderBy() {
|
||||
if _, ok := xs.rightAttrIndex[o.Column]; ok {
|
||||
xs.rightSortAttrs = append(xs.rightSortAttrs, o.Column)
|
||||
}
|
||||
}
|
||||
|
||||
xs.linkLeftAttr = xs.def.LeftAttributes[xs.leftAttrIndex[xs.def.On.Left]]
|
||||
xs.linkRightAttr = xs.def.RightAttributes[xs.rightAttrIndex[xs.def.On.Right]]
|
||||
|
||||
xs.rowTester, err = prepareGenericRowTester(xs.filter)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
return xs.applyPlan(ctx)
|
||||
}
|
||||
|
||||
func (xs *linkLeft) Next(ctx context.Context) (more bool) {
|
||||
xs.err = xs.applyPlan(ctx)
|
||||
if xs.err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
more, xs.err = xs.next(ctx)
|
||||
return
|
||||
}
|
||||
|
||||
func (xs *linkLeft) More(limit uint, v ValueGetter) (err error) {
|
||||
xs.filter.cursor, err = xs.ForwardCursor(v)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Redo row tester
|
||||
xs.rowTester, err = prepareGenericRowTester(xs.filter)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Redo the state
|
||||
// @todo adjust based on aggregation plan; reuse buffered, etc.
|
||||
xs.relIndex = newRelIndex(xs.rightSortAttrs...)
|
||||
xs.leftRows = make([]*row, 0, 128)
|
||||
xs.scanRow = nil
|
||||
xs.planned = false
|
||||
xs.keepLeft = false
|
||||
xs.leftIndex = 0
|
||||
xs.rightIndex = 0
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (xs *linkLeft) Err() error { return xs.err }
|
||||
|
||||
func (xs *linkLeft) Scan(s ValueSetter) (err error) {
|
||||
for k, cc := range xs.scanRow.CountValues() {
|
||||
for i := uint(0); i < cc; i++ {
|
||||
// @note internal row won't raise errors so we can safely omit them
|
||||
v, _ := xs.scanRow.GetValue(k, i)
|
||||
err = s.SetValue(k, i, v)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (xs *linkLeft) Close() (err error) {
|
||||
if xs == nil {
|
||||
return
|
||||
}
|
||||
|
||||
cc := []io.Closer{
|
||||
xs.leftSource,
|
||||
xs.rightSource,
|
||||
}
|
||||
|
||||
for _, c := range cc {
|
||||
if c != nil {
|
||||
err = c.Close()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (xs *linkLeft) BackCursor(v ValueGetter) (pc *filter.PagingCursor, err error) {
|
||||
g := &rowLink{
|
||||
a: xs.leftRow,
|
||||
b: v,
|
||||
}
|
||||
pc, err = filter.PagingCursorFrom(xs.filter.OrderBy(), g, xs.collectPrimaryAttributes()...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pc.ROrder = true
|
||||
pc.LThen = xs.filter.OrderBy().Reversed()
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (xs *linkLeft) ForwardCursor(v ValueGetter) (pc *filter.PagingCursor, err error) {
|
||||
g := &rowLink{
|
||||
a: xs.leftRow,
|
||||
b: v,
|
||||
}
|
||||
pc, err = filter.PagingCursorFrom(xs.filter.OrderBy(), g, xs.collectPrimaryAttributes()...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// // // // // // // // // // // // // // // // // // // // // // // // //
|
||||
// Utility methods
|
||||
|
||||
// next prepares the next scan row based on the defined link plan
|
||||
func (xs *linkLeft) next(ctx context.Context) (more bool, err error) {
|
||||
more, err = xs.pullNext(ctx)
|
||||
if !more || err != nil {
|
||||
return more, err
|
||||
}
|
||||
|
||||
return xs.nextBuffered()
|
||||
}
|
||||
|
||||
// pullNext pulls additional data so we can produce more
|
||||
//
|
||||
// This step may be omitted based on the join plan.
|
||||
func (xs *linkLeft) pullNext(ctx context.Context) (more bool, err error) {
|
||||
// Pull next chunk from source if not entirely buffered
|
||||
if xs.def.plan.partialScan {
|
||||
return false, fmt.Errorf("partialScan join plan strategy not implemented")
|
||||
}
|
||||
|
||||
// Check if buffer has more
|
||||
// We need to check for gt because the right bits may still be relevant
|
||||
if xs.leftIndex > len(xs.leftRows) {
|
||||
return false, nil
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// nextBuffered prepares the next scan row from the buffers
|
||||
func (xs *linkLeft) nextBuffered() (more bool, err error) {
|
||||
// keepLeft indicates if we should keep the left row and move onto the next
|
||||
// in the right buffers.
|
||||
//
|
||||
// If we're keeping it, take the next row from the other side, else take the
|
||||
// next left one and reset right counters.
|
||||
|
||||
var (
|
||||
ok bool
|
||||
)
|
||||
|
||||
for {
|
||||
if !xs.keepLeft {
|
||||
// Go for the next left row
|
||||
if xs.leftIndex >= len(xs.leftRows) {
|
||||
return
|
||||
}
|
||||
xs.leftRow = xs.leftRows[xs.leftIndex]
|
||||
xs.leftIndex++
|
||||
|
||||
// Go for the next right buffer
|
||||
xs.relScanBuffer, ok, err = xs.getRelatedBuffer(xs.leftRow)
|
||||
if !ok || err != nil {
|
||||
return
|
||||
}
|
||||
// if len(rel) > 1 {
|
||||
// // @todo implement this; not entirely sure how it should be so I'll block it for now
|
||||
// // @todo move this check futher up
|
||||
// return false, fmt.Errorf("multi-value link predicates not supported")
|
||||
// }
|
||||
// xs.relScanBuffer = rel[0]
|
||||
xs.rightIndex = 0
|
||||
|
||||
xs.scanRow = xs.leftRow
|
||||
xs.keepLeft = true
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// Related buffer done
|
||||
if xs.rightIndex >= len(xs.relScanBuffer.rows) {
|
||||
xs.keepLeft = false
|
||||
continue
|
||||
}
|
||||
|
||||
xs.rightRow = xs.relScanBuffer.rows[xs.rightIndex]
|
||||
xs.rightIndex++
|
||||
xs.scanRow = xs.rightRow
|
||||
break
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// getRelatedBuffer returns all of the right rows corresponding to the given left row
|
||||
func (xs *linkLeft) getRelatedBuffer(l *row) (out *relIndexBuffer, ok bool, err error) {
|
||||
attrIdent := xs.linkLeftAttr.Identifier()
|
||||
attrType := xs.linkLeftAttr.Properties().Type
|
||||
|
||||
// @todo mv link predicate attrs
|
||||
v, _ := l.GetValue(attrIdent, 0)
|
||||
|
||||
switch attrType.(type) {
|
||||
case TypeNumber:
|
||||
out, ok = xs.relIndex.GetInt(cast.ToInt64(v))
|
||||
return
|
||||
|
||||
case TypeText:
|
||||
out, ok = xs.relIndex.GetString(cast.ToString(v))
|
||||
return
|
||||
|
||||
case TypeID,
|
||||
TypeRef:
|
||||
out, ok = xs.relIndex.GetID(cast.ToUint64(v))
|
||||
return
|
||||
|
||||
default:
|
||||
// @note this should be validated way before
|
||||
err = fmt.Errorf("cannot use type %s ad join predicate", attrType.Type())
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// applyPlan runs plan specific logic to prepare the state
|
||||
func (xs *linkLeft) applyPlan(ctx context.Context) (err error) {
|
||||
if xs.planned || xs.err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
xs.planned = true
|
||||
switch {
|
||||
case !xs.def.plan.partialScan:
|
||||
return xs.pullEntireSource(ctx)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// pullEntireSource pulls both sources into memory and indexes them for later use
|
||||
func (xs *linkLeft) pullEntireSource(ctx context.Context) (err error) {
|
||||
// This bit does the filtering so just mark it of as such here
|
||||
xs.filtered = true
|
||||
|
||||
// First the right source
|
||||
err = xs.pullEntireRightSource(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Next the left source
|
||||
err = xs.pullEntireLeftSource(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Sort if needed
|
||||
return xs.sortLeftRows()
|
||||
}
|
||||
|
||||
// pullEntireRightSource pulls and indexes all of the right bits
|
||||
func (xs *linkLeft) pullEntireRightSource(ctx context.Context) (err error) {
|
||||
// Drain the source
|
||||
for xs.rightSource.Next(ctx) {
|
||||
r := &row{
|
||||
counters: make(map[string]uint),
|
||||
values: make(valueSet),
|
||||
}
|
||||
|
||||
err = xs.rightSource.Scan(r)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = xs.indexRightRow(r)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return xs.rightSource.Err()
|
||||
}
|
||||
|
||||
// pullEntireLeftSource pulls and indexes all of the left bits
|
||||
func (xs *linkLeft) pullEntireLeftSource(ctx context.Context) (err error) {
|
||||
var (
|
||||
rel *relIndexBuffer
|
||||
ok bool
|
||||
)
|
||||
|
||||
// Drain the source
|
||||
for xs.leftSource.Next(ctx) {
|
||||
l := &row{
|
||||
counters: make(map[string]uint),
|
||||
values: make(valueSet),
|
||||
}
|
||||
|
||||
err = xs.leftSource.Scan(l)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
rel, ok, err = xs.getRelatedBuffer(l)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
if !xs.keep(ctx, l, rel) {
|
||||
continue
|
||||
}
|
||||
|
||||
xs.leftRows = append(xs.leftRows, l)
|
||||
}
|
||||
return xs.leftSource.Err()
|
||||
}
|
||||
|
||||
// indexRightRow pushes the provided row onto the rel index
|
||||
// @todo consider moving most of this logic to the relIndex struct.
|
||||
func (xs *linkLeft) indexRightRow(r *row) (err error) {
|
||||
attrIdent := xs.linkRightAttr.Identifier()
|
||||
attrType := xs.linkRightAttr.Properties().Type
|
||||
|
||||
// @todo mv link predicate attrs; should be prevented higher up for now
|
||||
v, err := r.GetValue(attrIdent, 0)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// @todo not so sure about this switch; see above coment about moving this out
|
||||
switch attrType.(type) {
|
||||
case TypeNumber:
|
||||
xs.relIndex.AddInt(cast.ToInt64(v), r)
|
||||
return
|
||||
|
||||
case TypeText:
|
||||
xs.relIndex.AddString(cast.ToString(v), r)
|
||||
return
|
||||
|
||||
case TypeID,
|
||||
TypeRef:
|
||||
xs.relIndex.AddID(cast.ToUint64(v), r)
|
||||
return
|
||||
|
||||
default:
|
||||
// @note this should be validated way before
|
||||
return fmt.Errorf("cannot use type %s as link predicate", attrType.Type())
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// sortLeftRows sorts the left rows into the correct order
|
||||
//
|
||||
// Algorithm outline:
|
||||
// Compare the two left rows based on the defined sort order.
|
||||
// If the two left rows match up, or we're primarily sorting using right rows,
|
||||
// check the min/max stats values produced by the index buffer.
|
||||
//
|
||||
// If some min/max of a chunk is greater/lesser then the other, no row in the other
|
||||
// can appear before/after the prior.
|
||||
func (xs *linkLeft) sortLeftRows() (err error) {
|
||||
var (
|
||||
leftRelBufferA *relIndexBuffer
|
||||
leftRelBufferB *relIndexBuffer
|
||||
|
||||
a, b any
|
||||
)
|
||||
|
||||
// Use stable sort just so we don't needlesly messup the initial order if we decide
|
||||
// to preserve sorts produced by further steps.
|
||||
sort.SliceStable(xs.leftRows, func(i, j int) bool {
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// Prepare the data
|
||||
leftRowA := xs.leftRows[i]
|
||||
leftRowB := xs.leftRows[j]
|
||||
|
||||
leftRelBufferA, _, err = xs.getRelatedBuffer(leftRowA)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
leftRelBufferB, _, err = xs.getRelatedBuffer(leftRowB)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
for _, s := range xs.filter.OrderBy() {
|
||||
if _, ok := xs.leftAttrIndex[s.Column]; ok {
|
||||
// This bit here orders based on the left attributes
|
||||
less, skip := evalCmpResult(compareGetters(leftRowA, leftRowB, leftRowA.counters, leftRowB.counters, s.Column), s)
|
||||
if !skip {
|
||||
return less
|
||||
}
|
||||
} else {
|
||||
// This bit here orders based on the right attributes
|
||||
// Check the stats of the buffer; make sure to adjust based on direction
|
||||
|
||||
// Use chunk's values
|
||||
if !s.Descending {
|
||||
a = leftRelBufferA.min[s.Column]
|
||||
b = leftRelBufferB.min[s.Column]
|
||||
} else {
|
||||
a = leftRelBufferA.max[s.Column]
|
||||
b = leftRelBufferB.max[s.Column]
|
||||
}
|
||||
|
||||
less, skip := evalCmpResult(compareValues(a, b), s)
|
||||
if !skip {
|
||||
return less
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
})
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// keep checks if the row should be kept or discarded
|
||||
//
|
||||
// Link's keep is a bit more complicated and it looks at the related buffer as well.
|
||||
func (xs *linkLeft) keep(ctx context.Context, left *row, buffer *relIndexBuffer) (keep bool) {
|
||||
// If no buffer, we won't keep -- left inner join like behavior
|
||||
if buffer == nil {
|
||||
return false
|
||||
}
|
||||
// No tester include all ok rows
|
||||
if xs.rowTester == nil {
|
||||
return true
|
||||
}
|
||||
|
||||
ch := &rowLink{a: left}
|
||||
for _, ch.b = range buffer.rows {
|
||||
if !xs.rowTester.Test(ctx, ch) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (xs *linkLeft) collectPrimaryAttributes() (out []string) {
|
||||
out = make([]string, 0, 2)
|
||||
for _, m := range xs.def.OutLeftAttributes {
|
||||
if m.Properties().IsPrimary {
|
||||
out = append(out, m.Identifier())
|
||||
}
|
||||
}
|
||||
|
||||
for _, m := range xs.def.OutRightAttributes {
|
||||
if m.Properties().IsPrimary {
|
||||
out = append(out, m.Identifier())
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (xs *linkLeft) indexAttributes() {
|
||||
xs.outLeftAttrIndex = make(map[string]int)
|
||||
for i, a := range xs.def.LeftAttributes {
|
||||
xs.outLeftAttrIndex[a.Identifier()] = i
|
||||
}
|
||||
xs.outRightAttrIndex = make(map[string]int)
|
||||
for i, a := range xs.def.RightAttributes {
|
||||
xs.outRightAttrIndex[a.Identifier()] = i
|
||||
}
|
||||
|
||||
xs.leftAttrIndex = make(map[string]int)
|
||||
for i, a := range xs.def.LeftAttributes {
|
||||
xs.leftAttrIndex[a.Identifier()] = i
|
||||
}
|
||||
|
||||
xs.rightAttrIndex = make(map[string]int)
|
||||
for i, a := range xs.def.RightAttributes {
|
||||
xs.rightAttrIndex[a.Identifier()] = i
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
package dal
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math/rand"
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/filter"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func benchmarkExecLink_left(b *testing.B, n int) {
|
||||
ctx := context.Background()
|
||||
lattrs := []simpleAttribute{
|
||||
{ident: "l_k"},
|
||||
{ident: "l_v1"},
|
||||
{ident: "l_v2"},
|
||||
}
|
||||
fattrs := []simpleAttribute{
|
||||
{ident: "f_k"},
|
||||
{ident: "f_ref"},
|
||||
{ident: "f_v1"},
|
||||
{ident: "f_v2"},
|
||||
}
|
||||
|
||||
la := []simpleAttribute{
|
||||
{ident: "l_k", t: TypeID{}},
|
||||
{ident: "l_v1"},
|
||||
{ident: "l_v2"},
|
||||
}
|
||||
fa := []simpleAttribute{
|
||||
{ident: "f_k", t: TypeID{}},
|
||||
{ident: "f_ref", t: TypeID{}},
|
||||
{ident: "f_v1"},
|
||||
{ident: "f_v2"},
|
||||
}
|
||||
|
||||
// Inmem buffer for example
|
||||
l := InMemoryBuffer()
|
||||
f := InMemoryBuffer()
|
||||
for i := 0; i < n; i++ {
|
||||
require.NoError(b, l.Add(ctx, simpleRow{"l_k": i + 1, "l_v1": "a", "l_v2": rand.Intn(200)}))
|
||||
require.NoError(b, f.Add(ctx, simpleRow{"f_k": i + 1, "f_ref": i + 1, "f_v1": "a", "f_v2": rand.Intn(200)}))
|
||||
}
|
||||
|
||||
b.ResetTimer()
|
||||
|
||||
for n := 0; n < b.N; n++ {
|
||||
def := Link{
|
||||
Ident: "link",
|
||||
OutLeftAttributes: saToMapping(lattrs...),
|
||||
OutRightAttributes: saToMapping(fattrs...),
|
||||
LeftAttributes: saToMapping(la...),
|
||||
RightAttributes: saToMapping(fa...),
|
||||
Filter: internalFilter{orderBy: filter.SortExprSet{{Column: "f_k"}}},
|
||||
On: LinkPredicate{Left: "l_k", Right: "f_ref"},
|
||||
}
|
||||
|
||||
def.Initialize(ctx, l, f)
|
||||
|
||||
l.Seek(ctx, 0)
|
||||
f.Seek(ctx, 0)
|
||||
}
|
||||
}
|
||||
|
||||
// goos: linux
|
||||
// goarch: amd64
|
||||
// pkg: github.com/cortezaproject/corteza-server/pkg/dal
|
||||
// cpu: Intel(R) Core(TM) i7-8750H CPU @ 2.20GHz
|
||||
// BenchmarkExecLink_left_200-12 1874 633602 ns/op
|
||||
// BenchmarkExecLink_left_400-12 910 1369591 ns/op
|
||||
// BenchmarkExecLink_left_600-12 590 2162512 ns/op
|
||||
// BenchmarkExecLink_left_800-12 451 2725109 ns/op
|
||||
// BenchmarkExecLink_left_1000-12 358 3568684 ns/op
|
||||
// BenchmarkExecLink_left_1200-12 260 4059573 ns/op
|
||||
// BenchmarkExecLink_left_1400-12 250 4742408 ns/op
|
||||
// BenchmarkExecLink_left_1600-12 219 5383900 ns/op
|
||||
// BenchmarkExecLink_left_1800-12 181 6880839 ns/op
|
||||
// BenchmarkExecLink_left_2000-12 165 6821432 ns/op
|
||||
// BenchmarkExecLink_left_2200-12 134 8386884 ns/op
|
||||
// BenchmarkExecLink_left_2400-12 142 8471661 ns/op
|
||||
// BenchmarkExecLink_left_2600-12 135 8702328 ns/op
|
||||
// BenchmarkExecLink_left_2800-12 126 9375744 ns/op
|
||||
// BenchmarkExecLink_left_3000-12 114 10156339 ns/op
|
||||
// BenchmarkExecLink_left_3200-12 96 10682149 ns/op
|
||||
// BenchmarkExecLink_left_3400-12 104 11361907 ns/op
|
||||
// BenchmarkExecLink_left_3600-12 80 12616645 ns/op
|
||||
// BenchmarkExecLink_left_3800-12 73 13761230 ns/op
|
||||
// BenchmarkExecLink_left_4000-12 81 13795284 ns/op
|
||||
// BenchmarkExecLink_left_4200-12 69 15898381 ns/op
|
||||
// BenchmarkExecLink_left_4400-12 72 15162550 ns/op
|
||||
// BenchmarkExecLink_left_4600-12 74 15650527 ns/op
|
||||
// BenchmarkExecLink_left_4800-12 70 19573447 ns/op
|
||||
// BenchmarkExecLink_left_5000-12 69 17214960 ns/op
|
||||
func BenchmarkExecLink_left_200(b *testing.B) { benchmarkExecLink_left(b, 200) }
|
||||
func BenchmarkExecLink_left_400(b *testing.B) { benchmarkExecLink_left(b, 400) }
|
||||
func BenchmarkExecLink_left_600(b *testing.B) { benchmarkExecLink_left(b, 600) }
|
||||
func BenchmarkExecLink_left_800(b *testing.B) { benchmarkExecLink_left(b, 800) }
|
||||
func BenchmarkExecLink_left_1000(b *testing.B) { benchmarkExecLink_left(b, 1000) }
|
||||
func BenchmarkExecLink_left_1200(b *testing.B) { benchmarkExecLink_left(b, 1200) }
|
||||
func BenchmarkExecLink_left_1400(b *testing.B) { benchmarkExecLink_left(b, 1400) }
|
||||
func BenchmarkExecLink_left_1600(b *testing.B) { benchmarkExecLink_left(b, 1600) }
|
||||
func BenchmarkExecLink_left_1800(b *testing.B) { benchmarkExecLink_left(b, 1800) }
|
||||
func BenchmarkExecLink_left_2000(b *testing.B) { benchmarkExecLink_left(b, 2000) }
|
||||
func BenchmarkExecLink_left_2200(b *testing.B) { benchmarkExecLink_left(b, 2200) }
|
||||
func BenchmarkExecLink_left_2400(b *testing.B) { benchmarkExecLink_left(b, 2400) }
|
||||
func BenchmarkExecLink_left_2600(b *testing.B) { benchmarkExecLink_left(b, 2600) }
|
||||
func BenchmarkExecLink_left_2800(b *testing.B) { benchmarkExecLink_left(b, 2800) }
|
||||
func BenchmarkExecLink_left_3000(b *testing.B) { benchmarkExecLink_left(b, 3000) }
|
||||
func BenchmarkExecLink_left_3200(b *testing.B) { benchmarkExecLink_left(b, 3200) }
|
||||
func BenchmarkExecLink_left_3400(b *testing.B) { benchmarkExecLink_left(b, 3400) }
|
||||
func BenchmarkExecLink_left_3600(b *testing.B) { benchmarkExecLink_left(b, 3600) }
|
||||
func BenchmarkExecLink_left_3800(b *testing.B) { benchmarkExecLink_left(b, 3800) }
|
||||
func BenchmarkExecLink_left_4000(b *testing.B) { benchmarkExecLink_left(b, 4000) }
|
||||
func BenchmarkExecLink_left_4200(b *testing.B) { benchmarkExecLink_left(b, 4200) }
|
||||
func BenchmarkExecLink_left_4400(b *testing.B) { benchmarkExecLink_left(b, 4400) }
|
||||
func BenchmarkExecLink_left_4600(b *testing.B) { benchmarkExecLink_left(b, 4600) }
|
||||
func BenchmarkExecLink_left_4800(b *testing.B) { benchmarkExecLink_left(b, 4800) }
|
||||
func BenchmarkExecLink_left_5000(b *testing.B) { benchmarkExecLink_left(b, 5000) }
|
||||
@@ -0,0 +1,819 @@
|
||||
package dal
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/filter"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestStepLinkleft(t *testing.T) {
|
||||
crs1 := &filter.PagingCursor{}
|
||||
crs1.Set("l_pk", 1, false)
|
||||
crs1.Set("l_val", "l1 v1", false)
|
||||
crs1.Set("f_pk", 2, false)
|
||||
crs1.Set("f_fk", 1, false)
|
||||
crs1.Set("f_val", "f2 v1", false)
|
||||
|
||||
basicLeftAttrs := []simpleAttribute{
|
||||
{ident: "l_pk", t: TypeID{}},
|
||||
{ident: "l_val", t: TypeText{}},
|
||||
}
|
||||
basicRightAttrs := []simpleAttribute{
|
||||
{ident: "f_pk", t: TypeID{}},
|
||||
{ident: "f_fk", t: TypeRef{}},
|
||||
{ident: "f_val", t: TypeText{}},
|
||||
}
|
||||
basicOutLeftAttrs := []simpleAttribute{
|
||||
{ident: "l_pk", t: TypeID{}},
|
||||
{ident: "l_val", t: TypeText{}},
|
||||
}
|
||||
basicOutRightAttrs := []simpleAttribute{
|
||||
{ident: "f_pk", t: TypeID{}},
|
||||
{ident: "f_fk", t: TypeRef{}},
|
||||
{ident: "f_val", t: TypeText{}},
|
||||
}
|
||||
|
||||
tcc := []struct {
|
||||
name string
|
||||
|
||||
leftAttributes []simpleAttribute
|
||||
rightAttributes []simpleAttribute
|
||||
leftOutAttributes []simpleAttribute
|
||||
rightOutAttributes []simpleAttribute
|
||||
linkPred LinkPredicate
|
||||
|
||||
lIn []simpleRow
|
||||
fIn []simpleRow
|
||||
out []simpleRow
|
||||
|
||||
f internalFilter
|
||||
}{
|
||||
// Basic behavior
|
||||
{
|
||||
name: "basic link",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
out: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "basic link multiple right",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 1, "f_val": "f2 v1"},
|
||||
{"f_pk": 3, "f_fk": 1, "f_val": "f3 v1"},
|
||||
{"f_pk": 4, "f_fk": 2, "f_val": "f4 v1"},
|
||||
},
|
||||
out: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 1, "f_val": "f2 v1"},
|
||||
{"f_pk": 3, "f_fk": 1, "f_val": "f3 v1"},
|
||||
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
{"f_pk": 4, "f_fk": 2, "f_val": "f4 v1"},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "basic link omit missing rows",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 9999, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "basic link no rows joined",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 123, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 9999, "f_val": "f2 v1"},
|
||||
},
|
||||
out: []simpleRow{},
|
||||
},
|
||||
{
|
||||
name: "basic link empty right",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{},
|
||||
out: []simpleRow{},
|
||||
},
|
||||
{
|
||||
name: "basic link empty left",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
lIn: []simpleRow{},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 123, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 9999, "f_val": "f2 v1"},
|
||||
},
|
||||
out: []simpleRow{},
|
||||
},
|
||||
{
|
||||
name: "empty input",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
lIn: []simpleRow{},
|
||||
fIn: []simpleRow{},
|
||||
out: []simpleRow{},
|
||||
},
|
||||
|
||||
// Filtering
|
||||
{
|
||||
name: "filtering constraints single attr",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_const": "c1", "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_const": "c2", "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{
|
||||
{"l_pk": 1, "l_const": "c1", "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
},
|
||||
|
||||
f: internalFilter{
|
||||
constraints: map[string][]any{
|
||||
"l_const": {"c1"},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "filtering constraints right single attr",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_const": "c1", "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_const": "c2", "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{
|
||||
{"l_pk": 1, "l_const": "c1", "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
},
|
||||
|
||||
f: internalFilter{
|
||||
constraints: map[string][]any{
|
||||
"f_val": {"f1 v1"},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "filtering constraints both single attr",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_const": "c1", "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_const": "c2", "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{},
|
||||
|
||||
f: internalFilter{
|
||||
constraints: map[string][]any{
|
||||
"l_const": {"c2"},
|
||||
"f_val": {"f1 v1"},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "filtering constraints multiple attrs",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_const_a": "cac1", "l_const_b": "cbc1", "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_const_a": "cac1", "l_const_b": "cbc2", "l_val": "l2 v1"},
|
||||
},
|
||||
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{
|
||||
{"l_pk": 1, "l_const_a": "cac1", "l_const_b": "cbc1", "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
},
|
||||
|
||||
f: internalFilter{
|
||||
constraints: map[string][]any{"l_const_a": {"cac1"}, "l_const_b": {"cbc1"}},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "filtering constraints single attr multiple options",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_const_a": "cac1", "l_const_b": "cbc1", "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_const_a": "cac1", "l_const_b": "cbc2", "l_val": "l2 v1"},
|
||||
{"l_pk": 3, "l_const_a": "cac2", "l_const_b": "cbc3", "l_val": "l3 v1"},
|
||||
},
|
||||
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{
|
||||
{"l_pk": 1, "l_const_a": "cac1", "l_const_b": "cbc1", "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"l_pk": 2, "l_const_a": "cac1", "l_const_b": "cbc2", "l_val": "l2 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
f: internalFilter{
|
||||
constraints: map[string][]any{"l_const_b": {"cbc1", "cbc2"}},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "filtering expressions constant true",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
f: internalFilter{
|
||||
expression: "true",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "filtering expressions constant false",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{},
|
||||
|
||||
f: internalFilter{
|
||||
expression: "false",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "filtering expressions simple",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
f: internalFilter{
|
||||
expression: "l_val == 'l2 v1'",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "filtering expressions right simple",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"},
|
||||
},
|
||||
|
||||
out: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
},
|
||||
|
||||
f: internalFilter{
|
||||
expression: "f_val == 'f1 v1'",
|
||||
},
|
||||
},
|
||||
|
||||
// Paging
|
||||
{
|
||||
name: "paging cut off first entry",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
f: internalFilter{
|
||||
cursor: crs1,
|
||||
},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 1, "f_val": "f2 v1"},
|
||||
{"f_pk": 3, "f_fk": 1, "f_val": "f3 v1"},
|
||||
{"f_pk": 4, "f_fk": 2, "f_val": "f4 v1"},
|
||||
},
|
||||
out: []simpleRow{
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
{"f_pk": 4, "f_fk": 2, "f_val": "f4 v1"},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "paging cut off first entry with constant true",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
f: internalFilter{
|
||||
cursor: crs1,
|
||||
expression: "true",
|
||||
},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 1, "f_val": "f2 v1"},
|
||||
{"f_pk": 3, "f_fk": 1, "f_val": "f3 v1"},
|
||||
{"f_pk": 4, "f_fk": 2, "f_val": "f4 v1"},
|
||||
},
|
||||
out: []simpleRow{
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
{"f_pk": 4, "f_fk": 2, "f_val": "f4 v1"},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "paging cut off first entry with constant false",
|
||||
leftAttributes: basicLeftAttrs,
|
||||
rightAttributes: basicRightAttrs,
|
||||
leftOutAttributes: basicOutLeftAttrs,
|
||||
rightOutAttributes: basicOutRightAttrs,
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
f: internalFilter{
|
||||
cursor: crs1,
|
||||
expression: "false",
|
||||
},
|
||||
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 1, "f_val": "f2 v1"},
|
||||
{"f_pk": 3, "f_fk": 1, "f_val": "f3 v1"},
|
||||
{"f_pk": 4, "f_fk": 2, "f_val": "f4 v1"},
|
||||
},
|
||||
out: []simpleRow{},
|
||||
},
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
for _, tc := range tcc {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
l := InMemoryBuffer()
|
||||
for _, r := range tc.lIn {
|
||||
require.NoError(t, l.Add(ctx, r))
|
||||
}
|
||||
|
||||
f := InMemoryBuffer()
|
||||
for _, r := range tc.fIn {
|
||||
require.NoError(t, f.Add(ctx, r))
|
||||
}
|
||||
|
||||
def := Link{
|
||||
Ident: "foo",
|
||||
On: tc.linkPred,
|
||||
LeftAttributes: saToMapping(tc.leftAttributes...),
|
||||
RightAttributes: saToMapping(tc.rightAttributes...),
|
||||
OutLeftAttributes: saToMapping(tc.leftOutAttributes...),
|
||||
OutRightAttributes: saToMapping(tc.rightOutAttributes...),
|
||||
Filter: tc.f,
|
||||
}
|
||||
|
||||
xs, err := def.Initialize(ctx, l, f)
|
||||
require.NoError(t, err)
|
||||
|
||||
i := 0
|
||||
for xs.Next(ctx) {
|
||||
require.NoError(t, xs.Err())
|
||||
out := simpleRow{}
|
||||
require.NoError(t, xs.Err())
|
||||
require.NoError(t, xs.Scan(out))
|
||||
|
||||
require.Equal(t, tc.out[i], out)
|
||||
|
||||
i++
|
||||
}
|
||||
require.Equal(t, len(tc.out), i)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStepLinkleft_cursorCollect_forward(t *testing.T) {
|
||||
tcc := []struct {
|
||||
name string
|
||||
ss filter.SortExprSet
|
||||
in simpleRow
|
||||
state simpleRow
|
||||
leftAttrs []simpleAttribute
|
||||
rightAttrs []simpleAttribute
|
||||
outleftAttrs []simpleAttribute
|
||||
outrightAttrs []simpleAttribute
|
||||
out func() *filter.PagingCursor
|
||||
err bool
|
||||
}{
|
||||
{
|
||||
name: "one",
|
||||
in: simpleRow{"f_pk1": 25, "f1": "v25"},
|
||||
state: simpleRow{"l_pk1": 1, "f1": "v1"},
|
||||
leftAttrs: []simpleAttribute{{
|
||||
ident: "l_pk1",
|
||||
primary: true,
|
||||
}},
|
||||
rightAttrs: []simpleAttribute{{
|
||||
ident: "f_pk1",
|
||||
primary: true,
|
||||
}},
|
||||
outleftAttrs: []simpleAttribute{{
|
||||
ident: "l_pk1",
|
||||
primary: true,
|
||||
}},
|
||||
outrightAttrs: []simpleAttribute{{
|
||||
ident: "f_pk1",
|
||||
primary: true,
|
||||
}},
|
||||
out: func() *filter.PagingCursor {
|
||||
pc := &filter.PagingCursor{}
|
||||
pc.Set("l_pk1", 1, false)
|
||||
pc.Set("f_pk1", 25, false)
|
||||
return pc
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, c := range tcc {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
xs := &linkLeft{
|
||||
def: Link{
|
||||
Filter: internalFilter{
|
||||
orderBy: c.ss,
|
||||
},
|
||||
LeftAttributes: saToMapping(c.leftAttrs...),
|
||||
RightAttributes: saToMapping(c.rightAttrs...),
|
||||
OutLeftAttributes: saToMapping(c.outleftAttrs...),
|
||||
OutRightAttributes: saToMapping(c.outrightAttrs...),
|
||||
},
|
||||
leftRow: simpleToRow(c.state),
|
||||
}
|
||||
|
||||
out, err := xs.ForwardCursor(c.in)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Equal(t, c.out(), out)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStepLinkleft_cursorCollect_back(t *testing.T) {
|
||||
tcc := []struct {
|
||||
name string
|
||||
ss filter.SortExprSet
|
||||
in simpleRow
|
||||
state simpleRow
|
||||
leftAttrs []simpleAttribute
|
||||
rightAttrs []simpleAttribute
|
||||
outleftAttrs []simpleAttribute
|
||||
outrightAttrs []simpleAttribute
|
||||
out func() *filter.PagingCursor
|
||||
err bool
|
||||
}{
|
||||
{
|
||||
name: "one",
|
||||
in: simpleRow{"f_pk1": 25, "f1": "v25"},
|
||||
state: simpleRow{"l_pk1": 1, "f1": "v1"},
|
||||
leftAttrs: []simpleAttribute{{
|
||||
ident: "l_pk1",
|
||||
primary: true,
|
||||
}},
|
||||
rightAttrs: []simpleAttribute{{
|
||||
ident: "f_pk1",
|
||||
primary: true,
|
||||
}},
|
||||
outleftAttrs: []simpleAttribute{{
|
||||
ident: "l_pk1",
|
||||
primary: true,
|
||||
}},
|
||||
outrightAttrs: []simpleAttribute{{
|
||||
ident: "f_pk1",
|
||||
primary: true,
|
||||
}},
|
||||
out: func() *filter.PagingCursor {
|
||||
pc := &filter.PagingCursor{}
|
||||
pc.Set("l_pk1", 1, false)
|
||||
pc.Set("f_pk1", 25, false)
|
||||
pc.ROrder = true
|
||||
return pc
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, c := range tcc {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
jj := &linkLeft{
|
||||
def: Link{
|
||||
Filter: internalFilter{
|
||||
orderBy: c.ss,
|
||||
},
|
||||
LeftAttributes: saToMapping(c.leftAttrs...),
|
||||
RightAttributes: saToMapping(c.rightAttrs...),
|
||||
OutLeftAttributes: saToMapping(c.outleftAttrs...),
|
||||
OutRightAttributes: saToMapping(c.outrightAttrs...),
|
||||
},
|
||||
leftRow: simpleToRow(c.state),
|
||||
}
|
||||
|
||||
out, err := jj.BackCursor(c.in)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Equal(t, c.out(), out)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStepLinkleft_more(t *testing.T) {
|
||||
tcc := []struct {
|
||||
name string
|
||||
linkPred LinkPredicate
|
||||
leftAttrs []simpleAttribute
|
||||
rightAttrs []simpleAttribute
|
||||
outleftAttrs []simpleAttribute
|
||||
outrightAttrs []simpleAttribute
|
||||
|
||||
lIn []simpleRow
|
||||
fIn []simpleRow
|
||||
out1 []simpleRow
|
||||
out2 []simpleRow
|
||||
|
||||
f internalFilter
|
||||
}{
|
||||
{
|
||||
name: "one",
|
||||
leftAttrs: []simpleAttribute{{
|
||||
ident: "l_pk",
|
||||
primary: true,
|
||||
t: TypeID{},
|
||||
}},
|
||||
rightAttrs: []simpleAttribute{{
|
||||
ident: "f_pk",
|
||||
primary: true,
|
||||
t: TypeID{},
|
||||
}, {
|
||||
ident: "f_fk",
|
||||
primary: false,
|
||||
t: TypeID{},
|
||||
}},
|
||||
outleftAttrs: []simpleAttribute{{
|
||||
ident: "l_pk",
|
||||
primary: true,
|
||||
t: TypeID{},
|
||||
}},
|
||||
outrightAttrs: []simpleAttribute{{
|
||||
ident: "f_pk",
|
||||
primary: true,
|
||||
t: TypeID{},
|
||||
}, {
|
||||
ident: "f_fk",
|
||||
primary: false,
|
||||
t: TypeID{},
|
||||
}},
|
||||
linkPred: LinkPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
lIn: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
},
|
||||
fIn: []simpleRow{
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
{"f_pk": 2, "f_fk": 1, "f_val": "f2 v1"},
|
||||
{"f_pk": 3, "f_fk": 1, "f_val": "f3 v1"},
|
||||
{"f_pk": 4, "f_fk": 2, "f_val": "f4 v1"},
|
||||
},
|
||||
|
||||
out1: []simpleRow{
|
||||
{"l_pk": 1, "l_val": "l1 v1"},
|
||||
{"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"},
|
||||
},
|
||||
out2: []simpleRow{
|
||||
{"l_pk": 2, "l_val": "l2 v1"},
|
||||
{"f_pk": 4, "f_fk": 2, "f_val": "f4 v1"},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
for _, tc := range tcc {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
l := InMemoryBuffer()
|
||||
for _, r := range tc.lIn {
|
||||
require.NoError(t, l.Add(ctx, r))
|
||||
}
|
||||
|
||||
f := InMemoryBuffer()
|
||||
for _, r := range tc.fIn {
|
||||
require.NoError(t, f.Add(ctx, r))
|
||||
}
|
||||
|
||||
def := Link{
|
||||
Ident: "foo",
|
||||
On: tc.linkPred,
|
||||
LeftAttributes: saToMapping(tc.leftAttrs...),
|
||||
RightAttributes: saToMapping(tc.rightAttrs...),
|
||||
OutLeftAttributes: saToMapping(tc.outleftAttrs...),
|
||||
OutRightAttributes: saToMapping(tc.outrightAttrs...),
|
||||
Filter: tc.f,
|
||||
}
|
||||
|
||||
xs, err := def.Initialize(ctx, l, f)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.True(t, xs.Next(ctx))
|
||||
out := simpleRow{}
|
||||
require.NoError(t, xs.Err())
|
||||
require.NoError(t, xs.Scan(out))
|
||||
require.Equal(t, tc.out1[0], out)
|
||||
|
||||
require.True(t, xs.Next(ctx))
|
||||
out = simpleRow{}
|
||||
require.NoError(t, xs.Err())
|
||||
require.NoError(t, xs.Scan(out))
|
||||
require.Equal(t, tc.out1[1], out)
|
||||
|
||||
l.Seek(ctx, 0)
|
||||
f.Seek(ctx, 0)
|
||||
require.NoError(t, xs.More(0, out))
|
||||
|
||||
i := 0
|
||||
for xs.Next(ctx) {
|
||||
out := simpleRow{}
|
||||
require.NoError(t, xs.Err())
|
||||
require.NoError(t, xs.Scan(out))
|
||||
|
||||
require.Equal(t, tc.out2[i], out)
|
||||
|
||||
i++
|
||||
}
|
||||
require.Equal(t, len(tc.out2), i)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func simpleToRow(in simpleRow) (out *row) {
|
||||
out = &row{}
|
||||
for k, v := range in {
|
||||
out.SetValue(k, 0, v)
|
||||
}
|
||||
return
|
||||
}
|
||||
Reference in New Issue
Block a user