From b3cdbd060f87561b77790a8a62c84fe88cba6c20 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Wed, 21 Sep 2022 13:23:44 +0200 Subject: [PATCH] Add paging tests for pipeline aggregate and join --- pkg/dal/exec_aggregate.go | 7 + pkg/dal/exec_aggregate_test.go | 238 ++++++++++++++++++++++++++++- pkg/dal/exec_join_left.go | 4 + pkg/dal/exec_join_left_test.go | 270 +++++++++++++++++++++++++++++++++ 4 files changed, 518 insertions(+), 1 deletion(-) diff --git a/pkg/dal/exec_aggregate.go b/pkg/dal/exec_aggregate.go index e72ec432b..167a21038 100644 --- a/pkg/dal/exec_aggregate.go +++ b/pkg/dal/exec_aggregate.go @@ -33,6 +33,8 @@ type ( groupIndex *btree.Generic[*aggregateGroup] groups []*aggregateGroup i int + + ctr int } keyMaker func(context.Context, ValueGetter) (groupKey, error) @@ -85,6 +87,10 @@ func (xs *aggregate) Next(ctx context.Context) (more bool) { func (xs *aggregate) next(ctx context.Context) (more bool, err error) { var g *aggregateGroup for { + if xs.ctr >= int(xs.filter.limit) { + return false, nil + } + // Make sure it's cleared out and ready for fresh data xs.scanRow.Reset() @@ -121,6 +127,7 @@ func (xs *aggregate) next(ctx context.Context) (more bool, err error) { continue } + xs.ctr++ break } diff --git a/pkg/dal/exec_aggregate_test.go b/pkg/dal/exec_aggregate_test.go index 2afb525d5..3020cbf2e 100644 --- a/pkg/dal/exec_aggregate_test.go +++ b/pkg/dal/exec_aggregate_test.go @@ -32,7 +32,6 @@ func TestStepAggregate(t *testing.T) { ) baseBehavior := []testCase{ - // Basic behavior { name: "basic one key group", sourceAttributes: basicAttrs, @@ -1107,3 +1106,240 @@ func TestStepAggregate_more(t *testing.T) { }) } } + +func TestStepAggregate_paging(t *testing.T) { + basicAttrs := []simpleAttribute{ + {ident: "k1"}, + {ident: "k2"}, + {ident: "v1"}, + {ident: "txt"}, + } + + tcc := []struct { + name string + in []simpleRow + + sourceAttributes []simpleAttribute + group []simpleAttribute + outAttributes []simpleAttribute + + f internalFilter + + outF1 []simpleRow + outF2 []simpleRow + outB1 []simpleRow + }{ + { + name: "group asc", + sourceAttributes: basicAttrs, + in: []simpleRow{ + {"k1": "a", "v1": 1, "txt": "a1"}, + {"k1": "b", "v1": 2, "txt": "b2"}, + {"k1": "c", "v1": 3, "txt": "c3"}, + {"k1": "d", "v1": 4, "txt": "d4"}, + }, + + group: []simpleAttribute{ + {ident: "k1", source: "k1"}, + }, + outAttributes: []simpleAttribute{ + {ident: "cc", expr: "count(k1)"}, + }, + + f: internalFilter{ + limit: 2, + orderBy: filter.SortExprSet{{Column: "k1", Descending: false}}, + }, + + outF1: []simpleRow{ + {"k1": "a", "cc": float64(1)}, + {"k1": "b", "cc": float64(1)}, + }, + outF2: []simpleRow{ + {"k1": "c", "cc": float64(1)}, + {"k1": "d", "cc": float64(1)}, + }, + outB1: []simpleRow{ + {"k1": "a", "cc": float64(1)}, + {"k1": "b", "cc": float64(1)}, + }, + }, + { + name: "group desc", + sourceAttributes: basicAttrs, + in: []simpleRow{ + {"k1": "a", "v1": 1, "txt": "a1"}, + {"k1": "b", "v1": 2, "txt": "b2"}, + {"k1": "c", "v1": 3, "txt": "c3"}, + {"k1": "d", "v1": 4, "txt": "d4"}, + }, + + group: []simpleAttribute{ + {ident: "k1", source: "k1"}, + }, + outAttributes: []simpleAttribute{ + {ident: "cc", expr: "count(k1)"}, + }, + + f: internalFilter{ + limit: 2, + orderBy: filter.SortExprSet{{Column: "k1", Descending: true}}, + }, + + outF1: []simpleRow{ + {"k1": "d", "cc": float64(1)}, + {"k1": "c", "cc": float64(1)}, + }, + outF2: []simpleRow{ + {"k1": "b", "cc": float64(1)}, + {"k1": "a", "cc": float64(1)}, + }, + outB1: []simpleRow{ + {"k1": "d", "cc": float64(1)}, + {"k1": "c", "cc": float64(1)}, + }, + }, + + { + name: "agg asc", + sourceAttributes: basicAttrs, + in: []simpleRow{ + {"k1": "a", "v1": 1, "txt": "a1"}, + {"k1": "b", "v1": 2, "txt": "b2"}, + {"k1": "c", "v1": 3, "txt": "c3"}, + {"k1": "d", "v1": 4, "txt": "d4"}, + }, + + group: []simpleAttribute{ + {ident: "k1", source: "k1"}, + }, + outAttributes: []simpleAttribute{ + {ident: "mm", expr: "max(v1)"}, + }, + + f: internalFilter{ + limit: 2, + orderBy: filter.SortExprSet{{Column: "mm", Descending: false}}, + }, + + outF1: []simpleRow{ + {"k1": "a", "mm": float64(1)}, + {"k1": "b", "mm": float64(2)}, + }, + outF2: []simpleRow{ + {"k1": "c", "mm": float64(3)}, + {"k1": "d", "mm": float64(4)}, + }, + outB1: []simpleRow{ + {"k1": "a", "mm": float64(1)}, + {"k1": "b", "mm": float64(2)}, + }, + }, + { + name: "agg desc", + sourceAttributes: basicAttrs, + in: []simpleRow{ + {"k1": "a", "v1": 1, "txt": "a1"}, + {"k1": "b", "v1": 2, "txt": "b2"}, + {"k1": "c", "v1": 3, "txt": "c3"}, + {"k1": "d", "v1": 4, "txt": "d4"}, + }, + + group: []simpleAttribute{ + {ident: "k1", source: "k1"}, + }, + outAttributes: []simpleAttribute{ + {ident: "mm", expr: "max(v1)"}, + }, + + f: internalFilter{ + limit: 2, + orderBy: filter.SortExprSet{{Column: "mm", Descending: true}}, + }, + + outF1: []simpleRow{ + {"k1": "d", "mm": float64(4)}, + {"k1": "c", "mm": float64(3)}, + }, + outF2: []simpleRow{ + {"k1": "b", "mm": float64(2)}, + {"k1": "a", "mm": float64(1)}, + }, + outB1: []simpleRow{ + {"k1": "d", "mm": float64(4)}, + {"k1": "c", "mm": float64(3)}, + }, + }, + } + + ctx := context.Background() + for _, tc := range tcc { + t.Run(tc.name, func(t *testing.T) { + buff := InMemoryBuffer() + for _, r := range tc.in { + require.NoError(t, buff.Add(ctx, r)) + } + + var d Aggregate + + prep := func(f internalFilter) { + d = Aggregate{ + Filter: f, + Group: saToMapping(tc.group...), + OutAttributes: saToMapping(tc.outAttributes...), + SourceAttributes: saToMapping(tc.sourceAttributes...), + } + } + check := func(iter Iterator, assert []simpleRow) (first, last simpleRow) { + i := 0 + for iter.Next(ctx) { + out := simpleRow{} + + require.NoError(t, iter.Scan(out)) + + require.Equal(t, assert[i], out) + if i == 0 { + first = out + } + last = out + i++ + } + require.NoError(t, iter.Err()) + require.Equal(t, len(assert), i) + + return + } + + f := tc.f + var ( + first, last simpleRow + ) + + // First page, no cursor + prep(f) + aa, err := d.iterator(ctx, buff) + require.NoError(t, err) + _, last = check(aa, tc.outF1) + + // Second page, cursor + require.NoError(t, buff.Seek(ctx, 0)) + f.cursor, err = aa.ForwardCursor(last) + require.NoError(t, err) + + prep(f) + aa, err = d.iterator(ctx, buff) + require.NoError(t, err) + first, _ = check(aa, tc.outF2) + + // Third page, back, cursor + require.NoError(t, buff.Seek(ctx, 0)) + f.cursor, err = aa.BackCursor(first) + require.NoError(t, err) + + prep(f) + aa, err = d.iterator(ctx, buff) + require.NoError(t, err) + check(aa, tc.outB1) + }) + } +} diff --git a/pkg/dal/exec_join_left.go b/pkg/dal/exec_join_left.go index d0f34118d..b563da30a 100644 --- a/pkg/dal/exec_join_left.go +++ b/pkg/dal/exec_join_left.go @@ -162,6 +162,10 @@ func (xs *joinLeft) ForwardCursor(v ValueGetter) (pc *filter.PagingCursor, err e // next prepares the next scan row based on the defined join plan func (xs *joinLeft) next(ctx context.Context) (more bool, err error) { + if xs.i >= int(xs.filter.limit) { + return false, nil + } + more, err = xs.pullNext(ctx) if !more || err != nil { return more, err diff --git a/pkg/dal/exec_join_left_test.go b/pkg/dal/exec_join_left_test.go index f240f58b1..235ef92f4 100644 --- a/pkg/dal/exec_join_left_test.go +++ b/pkg/dal/exec_join_left_test.go @@ -1122,3 +1122,273 @@ func TestStepJoinLocal_more(t *testing.T) { }) } } + +func TestStepJoin_paging(t *testing.T) { + basicLeftSrcAttrs := []simpleAttribute{ + {ident: "l_pk", t: TypeID{}}, + {ident: "l_val", t: TypeText{}}, + } + basicRightSrcAttrs := []simpleAttribute{ + {ident: "f_pk", t: TypeID{}}, + {ident: "f_fk", t: TypeRef{}}, + {ident: "f_val", t: TypeText{}}, + } + + basicAttrs := []simpleAttribute{ + {ident: "l_pk", primary: true}, + {ident: "l_val"}, + {ident: "f_pk", primary: true}, + {ident: "f_fk"}, + {ident: "f_val"}, + } + + tcc := []struct { + name string + + outAttributes []simpleAttribute + leftAttributes []simpleAttribute + rightAttributes []simpleAttribute + joinPred JoinPredicate + + lIn []simpleRow + fIn []simpleRow + + f internalFilter + + outF1 []simpleRow + outF2 []simpleRow + outB1 []simpleRow + }{ + { + name: "key asc", + outAttributes: basicAttrs, + leftAttributes: basicLeftSrcAttrs, + rightAttributes: basicRightSrcAttrs, + + joinPred: JoinPredicate{Left: "l_pk", Right: "f_fk"}, + lIn: []simpleRow{ + {"l_pk": 1, "l_val": "l1 v1"}, + {"l_pk": 2, "l_val": "l2 v1"}, + {"l_pk": 3, "l_val": "l3 v1"}, + {"l_pk": 4, "l_val": "l4 v1"}, + }, + fIn: []simpleRow{ + {"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"}, + {"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"}, + {"f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + {"f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + }, + + f: internalFilter{ + limit: 2, + orderBy: filter.SortExprSet{{Column: "l_pk", Descending: false}}, + }, + + outF1: []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"}, + }, + outF2: []simpleRow{ + {"l_pk": 3, "l_val": "l3 v1", "f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + {"l_pk": 4, "l_val": "l4 v1", "f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + }, + outB1: []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: "key desc", + outAttributes: basicAttrs, + leftAttributes: basicLeftSrcAttrs, + rightAttributes: basicRightSrcAttrs, + + joinPred: JoinPredicate{Left: "l_pk", Right: "f_fk"}, + lIn: []simpleRow{ + {"l_pk": 1, "l_val": "l1 v1"}, + {"l_pk": 2, "l_val": "l2 v1"}, + {"l_pk": 3, "l_val": "l3 v1"}, + {"l_pk": 4, "l_val": "l4 v1"}, + }, + fIn: []simpleRow{ + {"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"}, + {"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"}, + {"f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + {"f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + }, + + f: internalFilter{ + limit: 2, + orderBy: filter.SortExprSet{{Column: "l_pk", Descending: true}}, + }, + + outF1: []simpleRow{ + {"l_pk": 4, "l_val": "l4 v1", "f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + {"l_pk": 3, "l_val": "l3 v1", "f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + }, + outF2: []simpleRow{ + {"l_pk": 2, "l_val": "l2 v1", "f_pk": 2, "f_fk": 2, "f_val": "f2 v1"}, + {"l_pk": 1, "l_val": "l1 v1", "f_pk": 1, "f_fk": 1, "f_val": "f1 v1"}, + }, + outB1: []simpleRow{ + {"l_pk": 4, "l_val": "l4 v1", "f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + {"l_pk": 3, "l_val": "l3 v1", "f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + }, + }, + + { + name: "val asc", + outAttributes: basicAttrs, + leftAttributes: basicLeftSrcAttrs, + rightAttributes: basicRightSrcAttrs, + + joinPred: JoinPredicate{Left: "l_pk", Right: "f_fk"}, + lIn: []simpleRow{ + {"l_pk": 1, "l_val": "l1 v1"}, + {"l_pk": 2, "l_val": "l2 v1"}, + {"l_pk": 3, "l_val": "l3 v1"}, + {"l_pk": 4, "l_val": "l4 v1"}, + }, + fIn: []simpleRow{ + {"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"}, + {"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"}, + {"f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + {"f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + }, + + f: internalFilter{ + limit: 2, + orderBy: filter.SortExprSet{{Column: "f_val", Descending: false}}, + }, + + outF1: []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"}, + }, + outF2: []simpleRow{ + {"l_pk": 3, "l_val": "l3 v1", "f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + {"l_pk": 4, "l_val": "l4 v1", "f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + }, + outB1: []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: "val desc", + outAttributes: basicAttrs, + leftAttributes: basicLeftSrcAttrs, + rightAttributes: basicRightSrcAttrs, + + joinPred: JoinPredicate{Left: "l_pk", Right: "f_fk"}, + lIn: []simpleRow{ + {"l_pk": 1, "l_val": "l1 v1"}, + {"l_pk": 2, "l_val": "l2 v1"}, + {"l_pk": 3, "l_val": "l3 v1"}, + {"l_pk": 4, "l_val": "l4 v1"}, + }, + fIn: []simpleRow{ + {"f_pk": 1, "f_fk": 1, "f_val": "f1 v1"}, + {"f_pk": 2, "f_fk": 2, "f_val": "f2 v1"}, + {"f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + {"f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + }, + + f: internalFilter{ + limit: 2, + orderBy: filter.SortExprSet{{Column: "f_val", Descending: true}}, + }, + + outF1: []simpleRow{ + {"l_pk": 4, "l_val": "l4 v1", "f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + {"l_pk": 3, "l_val": "l3 v1", "f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + }, + outF2: []simpleRow{ + {"l_pk": 2, "l_val": "l2 v1", "f_pk": 2, "f_fk": 2, "f_val": "f2 v1"}, + {"l_pk": 1, "l_val": "l1 v1", "f_pk": 1, "f_fk": 1, "f_val": "f1 v1"}, + }, + outB1: []simpleRow{ + {"l_pk": 4, "l_val": "l4 v1", "f_pk": 4, "f_fk": 4, "f_val": "f4 v1"}, + {"l_pk": 3, "l_val": "l3 v1", "f_pk": 3, "f_fk": 3, "f_val": "f3 v1"}, + }, + }, + } + + ctx := context.Background() + for _, tc := range tcc { + t.Run(tc.name, func(t *testing.T) { + buffL := InMemoryBuffer() + for _, r := range tc.lIn { + require.NoError(t, buffL.Add(ctx, r)) + } + buffR := InMemoryBuffer() + for _, r := range tc.fIn { + require.NoError(t, buffR.Add(ctx, r)) + } + + var d Join + + prep := func(f internalFilter) { + d = Join{ + Filter: f, + On: tc.joinPred, + OutAttributes: saToMapping(tc.outAttributes...), + LeftAttributes: saToMapping(tc.leftAttributes...), + RightAttributes: saToMapping(tc.rightAttributes...), + } + } + check := func(iter Iterator, assert []simpleRow) (first, last simpleRow) { + i := 0 + for iter.Next(ctx) { + out := simpleRow{} + + require.NoError(t, iter.Scan(out)) + + require.Equal(t, assert[i], out) + if i == 0 { + first = out + } + last = out + i++ + } + require.NoError(t, iter.Err()) + require.Equal(t, len(assert), i) + + return + } + + f := tc.f + var ( + first, last simpleRow + ) + + // First page, no cursor + prep(f) + aa, err := d.iterator(ctx, buffL, buffR) + require.NoError(t, err) + _, last = check(aa, tc.outF1) + + // Second page, cursor + require.NoError(t, buffL.Seek(ctx, 0)) + require.NoError(t, buffR.Seek(ctx, 0)) + f.cursor, err = aa.ForwardCursor(last) + require.NoError(t, err) + + prep(f) + aa, err = d.iterator(ctx, buffL, buffR) + require.NoError(t, err) + first, _ = check(aa, tc.outF2) + + // Third page, back, cursor + require.NoError(t, buffL.Seek(ctx, 0)) + require.NoError(t, buffR.Seek(ctx, 0)) + f.cursor, err = aa.BackCursor(first) + require.NoError(t, err) + + prep(f) + aa, err = d.iterator(ctx, buffL, buffR) + require.NoError(t, err) + check(aa, tc.outB1) + }) + } +}