DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Add paging tests for pipeline aggregate and join

Browse Source
This commit is contained in:
Tomaž Jerman
2022-09-21 13:24:10 +02:00
parent c373bb7bb3
commit b3cdbd060f
4 changed files with 518 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/dal/exec_aggregate.go
+7
View File
@@ -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
}
pkg/dal/exec_aggregate_test.go
+237 -1
View File
@@ -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)
})
}
}
pkg/dal/exec_join_left.go
+4
View File
@@ -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
pkg/dal/exec_join_left_test.go
+270
View File
@@ -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)
})
}
}