From f8595ca827a2e82466f0ea849753f38a7fe7ac1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Mon, 22 Aug 2022 12:22:36 +0200 Subject: [PATCH] Implement pipeline link (left) execution step --- pkg/dal/def_link.go | 41 ++ pkg/dal/exec_link_left.go | 549 ++++++++++++++++++ pkg/dal/exec_link_left_bench_test.go | 119 ++++ pkg/dal/exec_link_left_test.go | 819 +++++++++++++++++++++++++++ 4 files changed, 1528 insertions(+) create mode 100644 pkg/dal/exec_link_left.go create mode 100644 pkg/dal/exec_link_left_bench_test.go create mode 100644 pkg/dal/exec_link_left_test.go diff --git a/pkg/dal/def_link.go b/pkg/dal/def_link.go index 16b1c973b..2140dc037 100644 --- a/pkg/dal/def_link.go +++ b/pkg/dal/def_link.go @@ -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 +} diff --git a/pkg/dal/exec_link_left.go b/pkg/dal/exec_link_left.go new file mode 100644 index 000000000..babc87834 --- /dev/null +++ b/pkg/dal/exec_link_left.go @@ -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 + } +} diff --git a/pkg/dal/exec_link_left_bench_test.go b/pkg/dal/exec_link_left_bench_test.go new file mode 100644 index 000000000..bb3de15cf --- /dev/null +++ b/pkg/dal/exec_link_left_bench_test.go @@ -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) } diff --git a/pkg/dal/exec_link_left_test.go b/pkg/dal/exec_link_left_test.go new file mode 100644 index 000000000..3c261491b --- /dev/null +++ b/pkg/dal/exec_link_left_test.go @@ -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 +}