diff --git a/pkg/dal/exec_join_left.go b/pkg/dal/exec_join_left.go new file mode 100644 index 000000000..08dcebb91 --- /dev/null +++ b/pkg/dal/exec_join_left.go @@ -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 + } +} diff --git a/pkg/dal/exec_join_left_bench_test.go b/pkg/dal/exec_join_left_bench_test.go new file mode 100644 index 000000000..0f45d8a15 --- /dev/null +++ b/pkg/dal/exec_join_left_bench_test.go @@ -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) } diff --git a/pkg/dal/exec_join_left_test.go b/pkg/dal/exec_join_left_test.go new file mode 100644 index 000000000..f4c6ed09a --- /dev/null +++ b/pkg/dal/exec_join_left_test.go @@ -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) + }) + } +} diff --git a/pkg/dal/rel_index.go b/pkg/dal/rel_index.go new file mode 100644 index 000000000..bfbae6209 --- /dev/null +++ b/pkg/dal/rel_index.go @@ -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 +} diff --git a/pkg/dal/rel_index_buffer.go b/pkg/dal/rel_index_buffer.go new file mode 100644 index 000000000..8be74390e --- /dev/null +++ b/pkg/dal/rel_index_buffer.go @@ -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 + } + } + } +} diff --git a/pkg/dal/rel_index_buffer_test.go b/pkg/dal/rel_index_buffer_test.go new file mode 100644 index 000000000..8df5e56eb --- /dev/null +++ b/pkg/dal/rel_index_buffer_test.go @@ -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"]) +}