Implement pipeline join (left) execution step
This commit is contained in:
Tomaž Jerman
@@ -0,0 +1,434 @@
|
||||
package dal
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/filter"
|
||||
"github.com/spf13/cast"
|
||||
"github.com/tidwall/btree"
|
||||
)
|
||||
|
||||
type (
|
||||
// Considerations for optimizations
|
||||
// - Rework how values are stored internally.
|
||||
// Potentially rework the pipeline input/output interfaces, how ValueGetter/Setter work.
|
||||
// - Skip initial left row scan/sort when sorting not requested.
|
||||
// - Rework row struct to use slices instead of hashmaps.
|
||||
// With how values are now the performance gain is not that impactful and the
|
||||
// rework complexity is a bit too high for now.
|
||||
// - Use `.More` to filter rows by key.
|
||||
// - When data is provided in a satisfactory order, use that to pull data in chunks.
|
||||
joinLeft struct {
|
||||
def Join
|
||||
filter internalFilter
|
||||
|
||||
leftSource Iterator
|
||||
rightSource Iterator
|
||||
err error
|
||||
scanRow *row
|
||||
planned bool
|
||||
filtered bool
|
||||
|
||||
// Index the attributes for easier lookups later on
|
||||
outAttrIndex map[string]int
|
||||
leftAttrIndex map[string]int
|
||||
rightAttrIndex map[string]int
|
||||
|
||||
joinRightAttr AttributeMapping
|
||||
joinLeftAttr AttributeMapping
|
||||
|
||||
rowTester tester
|
||||
|
||||
// Index to keep track of related rows
|
||||
relIndex *relIndex
|
||||
|
||||
// Output placeholder for sorted rows
|
||||
// @todo consider a generic slice for cases when sorting is not needed.
|
||||
// This will probably save up on memory/time since we don't even need
|
||||
// to pull everything.
|
||||
outSorted *btree.Generic[*row]
|
||||
i int
|
||||
}
|
||||
)
|
||||
|
||||
func (xs *joinLeft) init(ctx context.Context) (err error) {
|
||||
xs.relIndex = newRelIndex()
|
||||
xs.indexAttributes()
|
||||
|
||||
xs.rowTester, err = prepareGenericRowTester(xs.def.Filter)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// @note careful here if you throw routines into the mix; see the NoLocks flag.
|
||||
// Enabling locks does have a performance impact so you might be better off by
|
||||
// constructing multiple of these but then you'll also need to complicate
|
||||
// the .Next methods a bit.
|
||||
xs.outSorted = btree.NewGenericOptions[*row](makeRowComparator(xs.filter.OrderBy()...), btree.Options{NoLocks: true})
|
||||
|
||||
xs.joinLeftAttr = xs.def.LeftAttributes[xs.leftAttrIndex[xs.def.On.Left]]
|
||||
xs.joinRightAttr = xs.def.RightAttributes[xs.rightAttrIndex[xs.def.On.Right]]
|
||||
|
||||
return xs.applyPlan(ctx)
|
||||
}
|
||||
|
||||
func (xs *joinLeft) 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 *joinLeft) More(limit uint, v ValueGetter) (err error) {
|
||||
xs.def.Filter.cursor, err = filter.PagingCursorFrom(xs.def.Filter.OrderBy(), v, xs.collectPrimaryAttributes()...)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Redo row tester
|
||||
xs.rowTester, err = prepareGenericRowTester(xs.def.Filter)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Redo the state
|
||||
// @todo adjust based on aggregation plan; reuse buffered, etc.
|
||||
xs.relIndex = newRelIndex()
|
||||
xs.outSorted = btree.NewGenericOptions[*row](makeRowComparator(xs.filter.OrderBy()...), btree.Options{NoLocks: true})
|
||||
xs.scanRow = nil
|
||||
xs.planned = false
|
||||
xs.i = 0
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (xs *joinLeft) Err() error { return xs.err }
|
||||
|
||||
func (xs *joinLeft) 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 *joinLeft) 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 *joinLeft) BackCursor(v ValueGetter) (pc *filter.PagingCursor, err error) {
|
||||
pc, err = filter.PagingCursorFrom(xs.def.Filter.OrderBy(), v, xs.collectPrimaryAttributes()...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pc.ROrder = true
|
||||
pc.LThen = xs.def.Filter.OrderBy().Reversed()
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (xs *joinLeft) ForwardCursor(v ValueGetter) (pc *filter.PagingCursor, err error) {
|
||||
pc, err = filter.PagingCursorFrom(xs.def.Filter.OrderBy(), v, xs.collectPrimaryAttributes()...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// // // // // // // // // // // // // // // // // // // // // // // // //
|
||||
// Utility methods
|
||||
|
||||
// next prepares the next scan row based on the defined join plan
|
||||
func (xs *joinLeft) next(ctx context.Context) (more bool, err error) {
|
||||
more, err = xs.pullNext(ctx)
|
||||
if !more || err != nil {
|
||||
return more, err
|
||||
}
|
||||
|
||||
xs.scanRow, _ = xs.outSorted.GetAt(xs.i)
|
||||
xs.i++
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// pullNext pulls additional data so we can produce more
|
||||
//
|
||||
// This step may be omitted based on the join plan.
|
||||
func (xs *joinLeft) pullNext(ctx context.Context) (more bool, err error) {
|
||||
// Pull next chunk from source if not entirely buffered
|
||||
if xs.def.plan.partialScan {
|
||||
// @todo this case is currently not implemented so we're erroring it out
|
||||
return false, fmt.Errorf("partialScan join plan strategy not implemented")
|
||||
}
|
||||
|
||||
// Check if buffer has more
|
||||
if xs.i >= xs.outSorted.Len() {
|
||||
return false, nil
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// applyPlan runs plan specific logic to prepare the state
|
||||
func (xs *joinLeft) 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 *joinLeft) 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
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// pullEntireRightSource pulls and indexes all of the right bits
|
||||
func (xs *joinLeft) pullEntireRightSource(ctx context.Context) (err error) {
|
||||
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 left bits and attempts to do as much of the joining
|
||||
// work on this stage
|
||||
func (xs *joinLeft) pullEntireLeftSource(ctx context.Context) (err error) {
|
||||
for xs.leftSource.Next(ctx) {
|
||||
l := &row{
|
||||
counters: make(map[string]uint),
|
||||
values: make(valueSet),
|
||||
}
|
||||
|
||||
err = xs.leftSource.Scan(l)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err = xs.joinRight(ctx, l)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return xs.leftSource.Err()
|
||||
}
|
||||
|
||||
// joinRight finds related right rows for the given left row and matches them up
|
||||
//
|
||||
// @note for sorting, we use a b-tree as it's self sorting.
|
||||
// Benchmarking shows that using a slice is negligibly faster if faster at all.
|
||||
func (xs *joinLeft) joinRight(ctx context.Context, left *row) (err error) {
|
||||
bb, ok, err := xs.getRelatedBuffers(left)
|
||||
if err != nil || !ok {
|
||||
return
|
||||
}
|
||||
|
||||
for _, b := range bb {
|
||||
for _, r := range b.rows {
|
||||
// Merge the two
|
||||
err = mergeRows(xs.def.OutAttributes, r, left, r)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Assert if we want to keep
|
||||
if !xs.keep(ctx, r) {
|
||||
continue
|
||||
}
|
||||
|
||||
xs.outSorted.Set(r)
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// getRelatedBuffers returns all of the right rows corresponding to the given left row
|
||||
func (xs *joinLeft) getRelatedBuffers(l *row) (out []*relIndexBuffer, ok bool, err error) {
|
||||
attrIdent := xs.joinLeftAttr.Identifier()
|
||||
attrType := xs.joinLeftAttr.Properties().Type
|
||||
var aux *relIndexBuffer
|
||||
|
||||
for c := uint(0); c < l.counters[attrIdent]; c++ {
|
||||
v, _ := l.GetValue(attrIdent, c)
|
||||
|
||||
switch attrType.(type) {
|
||||
case TypeNumber:
|
||||
aux, ok = xs.relIndex.GetInt(cast.ToInt64(v))
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
out = append(out, aux)
|
||||
continue
|
||||
|
||||
case TypeText:
|
||||
aux, ok = xs.relIndex.GetString(cast.ToString(v))
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
out = append(out, aux)
|
||||
continue
|
||||
|
||||
case TypeID,
|
||||
TypeRef:
|
||||
aux, ok = xs.relIndex.GetID(cast.ToUint64(v))
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
out = append(out, aux)
|
||||
continue
|
||||
|
||||
default:
|
||||
// @note this should be validated way before
|
||||
err = fmt.Errorf("cannot use type %s ad join predicate", attrType.Type())
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// indexRightRow pushes the provided row onto the rel index
|
||||
// @todo consider moving most of this logic to the relIndex struct.
|
||||
func (xs *joinLeft) indexRightRow(r *row) (err error) {
|
||||
attrIdent := xs.joinRightAttr.Identifier()
|
||||
attrType := xs.joinRightAttr.Properties().Type
|
||||
|
||||
for i := uint(0); i < r.CountValues()[attrIdent]; i++ {
|
||||
v, err := r.GetValue(attrIdent, i)
|
||||
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)
|
||||
continue
|
||||
|
||||
case TypeText:
|
||||
xs.relIndex.AddString(cast.ToString(v), r)
|
||||
continue
|
||||
|
||||
case TypeID,
|
||||
TypeRef:
|
||||
xs.relIndex.AddID(cast.ToUint64(v), r)
|
||||
continue
|
||||
|
||||
default:
|
||||
// @note this should be validated way before
|
||||
return fmt.Errorf("cannot use type %s as join predicate", attrType.Type())
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// keep checks if the row should be kept or discarded
|
||||
func (xs *joinLeft) keep(ctx context.Context, r *row) bool {
|
||||
if xs.rowTester == nil {
|
||||
return true
|
||||
}
|
||||
|
||||
return xs.rowTester.Test(ctx, r)
|
||||
}
|
||||
|
||||
// collectPrimaryAttributes returns all of the attributes of the composited key
|
||||
//
|
||||
// For joins, all primary attributes from both of the sides should be in here
|
||||
// since that is what always uniquely identifies a joined row.
|
||||
//
|
||||
// @todo consider applying PK candidates and filter out some of these. I don't
|
||||
// think it'll provide much of a performance boost but worth a shot later on.
|
||||
func (xs *joinLeft) collectPrimaryAttributes() (out []string) {
|
||||
out = make([]string, 0, 2)
|
||||
for _, m := range xs.def.OutAttributes {
|
||||
if m.Properties().IsPrimary {
|
||||
out = append(out, m.Identifier())
|
||||
}
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func (xs *joinLeft) indexAttributes() {
|
||||
xs.outAttrIndex = make(map[string]int)
|
||||
for i, a := range xs.def.OutAttributes {
|
||||
xs.outAttrIndex[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,116 @@
|
||||
package dal
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math/rand"
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/filter"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func benchmarkExecJoin_local(b *testing.B, n int) {
|
||||
ctx := context.Background()
|
||||
attrs := []simpleAttribute{
|
||||
{ident: "l_k"},
|
||||
{ident: "l_v1"},
|
||||
{ident: "l_v2"},
|
||||
{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 := Join{
|
||||
Ident: "join",
|
||||
OutAttributes: saToMapping(attrs...),
|
||||
LeftAttributes: saToMapping(la...),
|
||||
RightAttributes: saToMapping(fa...),
|
||||
Filter: internalFilter{orderBy: filter.SortExprSet{{Column: "f_k"}}},
|
||||
On: JoinPredicate{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
|
||||
// BenchmarkExecJoin_local_200-12 1620 718632 ns/op
|
||||
// BenchmarkExecJoin_local_400-12 801 1474321 ns/op
|
||||
// BenchmarkExecJoin_local_600-12 504 2344433 ns/op
|
||||
// BenchmarkExecJoin_local_800-12 388 3098124 ns/op
|
||||
// BenchmarkExecJoin_local_1000-12 304 3876453 ns/op
|
||||
// BenchmarkExecJoin_local_1200-12 258 4653349 ns/op
|
||||
// BenchmarkExecJoin_local_1400-12 218 5406631 ns/op
|
||||
// BenchmarkExecJoin_local_1600-12 192 6215687 ns/op
|
||||
// BenchmarkExecJoin_local_1800-12 168 7185540 ns/op
|
||||
// BenchmarkExecJoin_local_2000-12 148 8021597 ns/op
|
||||
// BenchmarkExecJoin_local_2200-12 122 9350466 ns/op
|
||||
// BenchmarkExecJoin_local_2400-12 100 10035371 ns/op
|
||||
// BenchmarkExecJoin_local_2600-12 117 10247223 ns/op
|
||||
// BenchmarkExecJoin_local_2800-12 100 11282374 ns/op
|
||||
// BenchmarkExecJoin_local_3000-12 99 12591232 ns/op
|
||||
// BenchmarkExecJoin_local_3200-12 86 13047472 ns/op
|
||||
// BenchmarkExecJoin_local_3400-12 80 13859590 ns/op
|
||||
// BenchmarkExecJoin_local_3600-12 76 14986832 ns/op
|
||||
// BenchmarkExecJoin_local_3800-12 70 15826467 ns/op
|
||||
// BenchmarkExecJoin_local_4000-12 61 16537483 ns/op
|
||||
// BenchmarkExecJoin_local_4200-12 68 17226297 ns/op
|
||||
// BenchmarkExecJoin_local_4400-12 67 17996956 ns/op
|
||||
// BenchmarkExecJoin_local_4600-12 62 18622211 ns/op
|
||||
// BenchmarkExecJoin_local_4800-12 61 19948914 ns/op
|
||||
// BenchmarkExecJoin_local_5000-12 57 20614964 ns/op
|
||||
func BenchmarkExecJoin_local_200(b *testing.B) { benchmarkExecJoin_local(b, 200) }
|
||||
func BenchmarkExecJoin_local_400(b *testing.B) { benchmarkExecJoin_local(b, 400) }
|
||||
func BenchmarkExecJoin_local_600(b *testing.B) { benchmarkExecJoin_local(b, 600) }
|
||||
func BenchmarkExecJoin_local_800(b *testing.B) { benchmarkExecJoin_local(b, 800) }
|
||||
func BenchmarkExecJoin_local_1000(b *testing.B) { benchmarkExecJoin_local(b, 1000) }
|
||||
func BenchmarkExecJoin_local_1200(b *testing.B) { benchmarkExecJoin_local(b, 1200) }
|
||||
func BenchmarkExecJoin_local_1400(b *testing.B) { benchmarkExecJoin_local(b, 1400) }
|
||||
func BenchmarkExecJoin_local_1600(b *testing.B) { benchmarkExecJoin_local(b, 1600) }
|
||||
func BenchmarkExecJoin_local_1800(b *testing.B) { benchmarkExecJoin_local(b, 1800) }
|
||||
func BenchmarkExecJoin_local_2000(b *testing.B) { benchmarkExecJoin_local(b, 2000) }
|
||||
func BenchmarkExecJoin_local_2200(b *testing.B) { benchmarkExecJoin_local(b, 2200) }
|
||||
func BenchmarkExecJoin_local_2400(b *testing.B) { benchmarkExecJoin_local(b, 2400) }
|
||||
func BenchmarkExecJoin_local_2600(b *testing.B) { benchmarkExecJoin_local(b, 2600) }
|
||||
func BenchmarkExecJoin_local_2800(b *testing.B) { benchmarkExecJoin_local(b, 2800) }
|
||||
func BenchmarkExecJoin_local_3000(b *testing.B) { benchmarkExecJoin_local(b, 3000) }
|
||||
func BenchmarkExecJoin_local_3200(b *testing.B) { benchmarkExecJoin_local(b, 3200) }
|
||||
func BenchmarkExecJoin_local_3400(b *testing.B) { benchmarkExecJoin_local(b, 3400) }
|
||||
func BenchmarkExecJoin_local_3600(b *testing.B) { benchmarkExecJoin_local(b, 3600) }
|
||||
func BenchmarkExecJoin_local_3800(b *testing.B) { benchmarkExecJoin_local(b, 3800) }
|
||||
func BenchmarkExecJoin_local_4000(b *testing.B) { benchmarkExecJoin_local(b, 4000) }
|
||||
func BenchmarkExecJoin_local_4200(b *testing.B) { benchmarkExecJoin_local(b, 4200) }
|
||||
func BenchmarkExecJoin_local_4400(b *testing.B) { benchmarkExecJoin_local(b, 4400) }
|
||||
func BenchmarkExecJoin_local_4600(b *testing.B) { benchmarkExecJoin_local(b, 4600) }
|
||||
func BenchmarkExecJoin_local_4800(b *testing.B) { benchmarkExecJoin_local(b, 4800) }
|
||||
func BenchmarkExecJoin_local_5000(b *testing.B) { benchmarkExecJoin_local(b, 5000) }
|
||||
@@ -0,0 +1,655 @@
|
||||
package dal
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/cortezaproject/corteza-server/pkg/filter"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestStepJoinLocal(t *testing.T) {
|
||||
crs1 := &filter.PagingCursor{}
|
||||
crs1.Set("l_pk", 1, false)
|
||||
crs1.Set("l_val", "l1 v1", false)
|
||||
crs1.Set("f_pk", 1, false)
|
||||
crs1.Set("f_fk", 1, false)
|
||||
crs1.Set("f_val", "f1 v1", false)
|
||||
|
||||
basicAttrs := []simpleAttribute{
|
||||
{ident: "l_pk", t: TypeID{}},
|
||||
{ident: "l_val", t: TypeText{}},
|
||||
{ident: "f_pk", t: TypeID{}},
|
||||
{ident: "f_fk", t: TypeRef{}},
|
||||
{ident: "f_val", t: TypeText{}},
|
||||
}
|
||||
basicLocalAttrs := []simpleAttribute{
|
||||
{ident: "l_pk", t: TypeID{}},
|
||||
{ident: "l_val", t: TypeText{}},
|
||||
}
|
||||
basicForeignAttrs := []simpleAttribute{
|
||||
{ident: "f_pk", t: TypeID{}},
|
||||
{ident: "f_fk", t: TypeRef{}},
|
||||
{ident: "f_val", t: TypeText{}},
|
||||
}
|
||||
|
||||
tcc := []struct {
|
||||
name string
|
||||
|
||||
outAttributes []simpleAttribute
|
||||
leftAttributes []simpleAttribute
|
||||
rightAttributes []simpleAttribute
|
||||
joinPred JoinPredicate
|
||||
|
||||
lIn []simpleRow
|
||||
fIn []simpleRow
|
||||
out []simpleRow
|
||||
|
||||
f internalFilter
|
||||
}{
|
||||
// Basic behavior
|
||||
{
|
||||
name: "basic link",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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 omit missing rows",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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 foreign",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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 local",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{Left: "l_pk", Right: "f_fk"},
|
||||
|
||||
lIn: []simpleRow{},
|
||||
fIn: []simpleRow{},
|
||||
out: []simpleRow{},
|
||||
},
|
||||
|
||||
// Filtering
|
||||
{
|
||||
name: "filtering constraints single attr",
|
||||
outAttributes: append(basicAttrs, simpleAttribute{ident: "l_const"}),
|
||||
leftAttributes: append(basicLocalAttrs, simpleAttribute{ident: "l_const"}),
|
||||
rightAttributes: basicForeignAttrs,
|
||||
|
||||
joinPred: JoinPredicate{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 multiple attrs",
|
||||
outAttributes: append(basicAttrs, simpleAttribute{ident: "l_const_a"}, simpleAttribute{ident: "l_const_b"}),
|
||||
leftAttributes: append(basicLocalAttrs, simpleAttribute{ident: "l_const_a"}, simpleAttribute{ident: "l_const_b"}),
|
||||
rightAttributes: basicForeignAttrs,
|
||||
|
||||
joinPred: JoinPredicate{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",
|
||||
outAttributes: append(basicAttrs, simpleAttribute{ident: "l_const_a"}, simpleAttribute{ident: "l_const_b"}),
|
||||
leftAttributes: append(basicLocalAttrs, simpleAttribute{ident: "l_const_a"}, simpleAttribute{ident: "l_const_b"}),
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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'",
|
||||
},
|
||||
},
|
||||
|
||||
// Paging
|
||||
{
|
||||
name: "paging cut off first entry",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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{
|
||||
cursor: crs1,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "paging cut off last entry with constant true",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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: "true",
|
||||
cursor: crs1,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "paging cut off last entry with constant false",
|
||||
outAttributes: basicAttrs,
|
||||
leftAttributes: basicLocalAttrs,
|
||||
rightAttributes: basicForeignAttrs,
|
||||
joinPred: JoinPredicate{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",
|
||||
cursor: crs1,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
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))
|
||||
}
|
||||
|
||||
tc.f.orderBy = filter.SortExprSet{
|
||||
{Column: "l_pk"},
|
||||
{Column: "f_pk"},
|
||||
}
|
||||
|
||||
def := Join{
|
||||
Ident: "foo",
|
||||
On: tc.joinPred,
|
||||
OutAttributes: saToMapping(tc.outAttributes...),
|
||||
LeftAttributes: saToMapping(tc.leftAttributes...),
|
||||
RightAttributes: saToMapping(tc.rightAttributes...),
|
||||
Filter: tc.f,
|
||||
|
||||
plan: joinPlan{},
|
||||
}
|
||||
|
||||
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 TestStepJoinLocal_cursorCollect_forward(t *testing.T) {
|
||||
tcc := []struct {
|
||||
name string
|
||||
ss filter.SortExprSet
|
||||
in simpleRow
|
||||
attrs []simpleAttribute
|
||||
out func() *filter.PagingCursor
|
||||
err bool
|
||||
}{
|
||||
{
|
||||
name: "simple",
|
||||
in: simpleRow{"pk1": 1, "f1": "v1"},
|
||||
attrs: []simpleAttribute{{
|
||||
ident: "pk1",
|
||||
primary: true,
|
||||
}},
|
||||
out: func() *filter.PagingCursor {
|
||||
pc := &filter.PagingCursor{}
|
||||
pc.Set("pk1", 1, false)
|
||||
return pc
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, c := range tcc {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
|
||||
jj := &joinLeft{
|
||||
def: Join{
|
||||
Filter: internalFilter{
|
||||
orderBy: c.ss,
|
||||
},
|
||||
OutAttributes: saToMapping(c.attrs...),
|
||||
},
|
||||
}
|
||||
|
||||
out, err := jj.ForwardCursor(c.in)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Equal(t, c.out(), out)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStepJoinLocal_cursorCollect_back(t *testing.T) {
|
||||
tcc := []struct {
|
||||
name string
|
||||
ss filter.SortExprSet
|
||||
in simpleRow
|
||||
attrs []simpleAttribute
|
||||
out func() *filter.PagingCursor
|
||||
err bool
|
||||
}{
|
||||
{
|
||||
name: "simple",
|
||||
in: simpleRow{"pk1": 1, "f1": "v1"},
|
||||
attrs: []simpleAttribute{{
|
||||
ident: "pk1",
|
||||
primary: true,
|
||||
}},
|
||||
out: func() *filter.PagingCursor {
|
||||
pc := &filter.PagingCursor{}
|
||||
pc.Set("pk1", 1, false)
|
||||
pc.ROrder = true
|
||||
return pc
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, c := range tcc {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
|
||||
jj := &joinLeft{
|
||||
def: Join{
|
||||
Filter: internalFilter{
|
||||
orderBy: c.ss,
|
||||
},
|
||||
OutAttributes: saToMapping(c.attrs...),
|
||||
},
|
||||
}
|
||||
|
||||
out, err := jj.BackCursor(c.in)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Equal(t, c.out(), out)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStepJoinLocal_more(t *testing.T) {
|
||||
tcc := []struct {
|
||||
name string
|
||||
|
||||
attributes []simpleAttribute
|
||||
localAttributes []simpleAttribute
|
||||
foreignAttributes []simpleAttribute
|
||||
joinPred JoinPredicate
|
||||
|
||||
lIn []simpleRow
|
||||
fIn []simpleRow
|
||||
|
||||
out1 []simpleRow
|
||||
out2 []simpleRow
|
||||
|
||||
f internalFilter
|
||||
}{
|
||||
{
|
||||
name: "one",
|
||||
attributes: []simpleAttribute{
|
||||
{ident: "l_pk", primary: true},
|
||||
{ident: "l_val"},
|
||||
{ident: "f_pk", primary: true},
|
||||
{ident: "f_fk"},
|
||||
{ident: "f_val"},
|
||||
},
|
||||
localAttributes: []simpleAttribute{
|
||||
{ident: "l_pk", t: TypeID{}},
|
||||
{ident: "l_val", t: TypeText{}},
|
||||
},
|
||||
foreignAttributes: []simpleAttribute{
|
||||
{ident: "f_pk", t: TypeID{}},
|
||||
{ident: "f_fk", t: TypeRef{}},
|
||||
{ident: "f_val", t: TypeText{}},
|
||||
},
|
||||
|
||||
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"},
|
||||
},
|
||||
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"},
|
||||
},
|
||||
|
||||
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": 2, "f_fk": 2, "f_val": "f2 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) {
|
||||
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))
|
||||
}
|
||||
|
||||
tc.f.orderBy = filter.SortExprSet{
|
||||
{Column: "l_pk"},
|
||||
{Column: "f_pk"},
|
||||
}
|
||||
|
||||
def := Join{
|
||||
Ident: "foo",
|
||||
On: tc.joinPred,
|
||||
OutAttributes: saToMapping(tc.attributes...),
|
||||
LeftAttributes: saToMapping(tc.localAttributes...),
|
||||
RightAttributes: saToMapping(tc.foreignAttributes...),
|
||||
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.NoError(t, xs.More(0, out))
|
||||
|
||||
l.Seek(ctx, 0)
|
||||
f.Seek(ctx, 0)
|
||||
|
||||
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)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package dal
|
||||
|
||||
type (
|
||||
// relIndex is a generic struct for indexing data which join/link can use
|
||||
//
|
||||
// The current index implementation utilizes a series of hashmaps based on what
|
||||
// type of values we're indexing on.
|
||||
// As we currently only support single value predicate, the hashmaps proved
|
||||
// to be a bit more efficient then b-trees (other consideration) when testing
|
||||
// on larger datasets.
|
||||
//
|
||||
// @todo when we support multiple join predicates, this should probably change
|
||||
// into a b-tree as it might be a bit faster thennested hashmaps.
|
||||
//
|
||||
// @todo do some benchmarks in regards to using generics for key
|
||||
relIndex struct {
|
||||
track []string
|
||||
|
||||
ints map[int64]*relIndexBuffer
|
||||
strings map[string]*relIndexBuffer
|
||||
ids map[uint64]*relIndexBuffer
|
||||
}
|
||||
)
|
||||
|
||||
// newRelIndex initializes a new relIndex with the specified tracked attributes
|
||||
// @todo benchmark with generics as key
|
||||
func newRelIndex(tt ...string) *relIndex {
|
||||
return &relIndex{
|
||||
track: tt,
|
||||
ints: make(map[int64]*relIndexBuffer),
|
||||
strings: make(map[string]*relIndexBuffer),
|
||||
ids: make(map[uint64]*relIndexBuffer),
|
||||
}
|
||||
}
|
||||
|
||||
// AddInt adds a new row under the int key
|
||||
func (ri *relIndex) AddInt(k int64, r *row) {
|
||||
c, ok := ri.GetInt(k)
|
||||
if !ok {
|
||||
c = newRelIndexBuffer(ri.track...)
|
||||
ri.ints[k] = c
|
||||
}
|
||||
c.add(r)
|
||||
}
|
||||
|
||||
func (ri *relIndex) GetInt(k int64) (out *relIndexBuffer, ok bool) {
|
||||
out, ok = ri.ints[k]
|
||||
return
|
||||
}
|
||||
|
||||
func (ri *relIndex) AddString(k string, r *row) {
|
||||
c, ok := ri.GetString(k)
|
||||
if !ok {
|
||||
c = newRelIndexBuffer(ri.track...)
|
||||
ri.strings[k] = c
|
||||
}
|
||||
c.add(r)
|
||||
}
|
||||
|
||||
func (ri *relIndex) GetString(k string) (out *relIndexBuffer, ok bool) {
|
||||
out, ok = ri.strings[k]
|
||||
return
|
||||
}
|
||||
|
||||
func (ri *relIndex) AddID(k uint64, r *row) {
|
||||
c, ok := ri.GetID(k)
|
||||
if !ok {
|
||||
c = newRelIndexBuffer(ri.track...)
|
||||
ri.ids[k] = c
|
||||
}
|
||||
c.add(r)
|
||||
}
|
||||
|
||||
func (ri *relIndex) GetID(k uint64) (out *relIndexBuffer, ok bool) {
|
||||
out, ok = ri.ids[k]
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package dal
|
||||
|
||||
type (
|
||||
relIndexBuffer struct {
|
||||
// attribute statistics to track
|
||||
//
|
||||
// These stats can be used when performing operations like sorting and
|
||||
// binary searching for specific values.
|
||||
track []string
|
||||
|
||||
min map[string]any
|
||||
max map[string]any
|
||||
|
||||
// @todo make b-tree for sorting? will probably be small so simple slice sort
|
||||
// should be ok ig...
|
||||
//
|
||||
// Probably don't need to sort yet
|
||||
rows []*row
|
||||
}
|
||||
)
|
||||
|
||||
// newRelIndexBuffer initializes a new relIndexBuffer tracking the given attributes
|
||||
func newRelIndexBuffer(tt ...string) *relIndexBuffer {
|
||||
return &relIndexBuffer{
|
||||
min: make(map[string]any),
|
||||
max: make(map[string]any),
|
||||
track: tt,
|
||||
}
|
||||
}
|
||||
|
||||
// add adds a new *row to the buffer
|
||||
func (lc *relIndexBuffer) add(r *row) {
|
||||
if len(lc.rows) == 0 {
|
||||
for _, ix := range lc.track {
|
||||
lc.min[ix] = r.values[ix][0]
|
||||
lc.max[ix] = r.values[ix][0]
|
||||
}
|
||||
|
||||
lc.rows = append(lc.rows, r)
|
||||
return
|
||||
}
|
||||
|
||||
lc.updMin(r)
|
||||
lc.updMax(r)
|
||||
|
||||
lc.rows = append(lc.rows, r)
|
||||
}
|
||||
|
||||
// updMin updates the min stat
|
||||
func (lc *relIndexBuffer) updMin(r *row) {
|
||||
for _, ix := range lc.track {
|
||||
for i := uint(0); i < r.CountValues()[ix]; i++ {
|
||||
v := r.values[ix][i]
|
||||
if compareValues(v, lc.min[ix]) < 0 {
|
||||
lc.min[ix] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// updMax updates the max stat
|
||||
func (lc *relIndexBuffer) updMax(r *row) {
|
||||
for _, ix := range lc.track {
|
||||
for i := uint(0); i < r.CountValues()[ix]; i++ {
|
||||
v := r.values[ix][i]
|
||||
if compareValues(v, lc.max[ix]) > 0 {
|
||||
lc.max[ix] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package dal
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestRelIndexBuffer(t *testing.T) {
|
||||
cc := newRelIndexBuffer("a", "b")
|
||||
|
||||
cc.add((&row{}).WithValue("a", 0, 1).WithValue("b", 0, "a"))
|
||||
|
||||
require.Len(t, cc.rows, 1)
|
||||
require.Equal(t, 1, cc.min["a"])
|
||||
require.Equal(t, 1, cc.max["a"])
|
||||
require.Equal(t, "a", cc.min["b"])
|
||||
require.Equal(t, "a", cc.max["b"])
|
||||
|
||||
cc.add((&row{}).WithValue("a", 0, -1).WithValue("b", 0, "a"))
|
||||
require.Equal(t, -1, cc.min["a"])
|
||||
require.Equal(t, 1, cc.max["a"])
|
||||
require.Equal(t, "a", cc.min["b"])
|
||||
require.Equal(t, "a", cc.max["b"])
|
||||
|
||||
cc.add((&row{}).WithValue("a", 0, 2).WithValue("b", 0, "aa"))
|
||||
require.Equal(t, -1, cc.min["a"])
|
||||
require.Equal(t, 2, cc.max["a"])
|
||||
require.Equal(t, "a", cc.min["b"])
|
||||
require.Equal(t, "aa", cc.max["b"])
|
||||
}
|
||||
Reference in New Issue
Block a user