DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Export dal.Row for reusability

Browse Source
This commit is contained in:
Tomaž Jerman
2022-08-29 13:35:40 +02:00
parent 377e07cbd9
commit dc70e2aeb8
9 changed files with 71 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/dal/buffer.go
+14 -14
View File
@@ -35,7 +35,7 @@ type (
Add(context.Context, ValueGetter) (err error)
}
row struct {
Row struct {
counters map[string]uint
values valueSet
@@ -48,7 +48,7 @@ type (
valueSet map[string][]any
)
func (r row) SelectGVal(ctx context.Context, k string) (interface{}, error) {
func (r Row) SelectGVal(ctx context.Context, k string) (interface{}, error) {
if r.values[k] == nil {
return nil, nil
}
@@ -61,13 +61,13 @@ func (r row) SelectGVal(ctx context.Context, k string) (interface{}, error) {
return o, nil
}
func (r *row) Reset() {
func (r *Row) Reset() {
for k := range r.counters {
r.counters[k] = 0
}
}
func (r *row) SetValue(name string, pos uint, v any) error {
func (r *Row) SetValue(name string, pos uint, v any) error {
if r.values == nil {
r.values = make(valueSet)
}
@@ -91,7 +91,7 @@ func (r *row) SetValue(name string, pos uint, v any) error {
// WithValue is a simple helper to construct rows with populated values
// The main use is for tests so restrain from using it in code.
func (r *row) WithValue(name string, pos uint, v any) *row {
func (r *Row) WithValue(name string, pos uint, v any) *Row {
err := r.SetValue(name, pos, v)
if err != nil {
panic(err)
@@ -100,15 +100,15 @@ func (r *row) WithValue(name string, pos uint, v any) *row {
return r
}
func (r *row) CountValues() map[string]uint {
func (r *Row) CountValues() map[string]uint {
return r.counters
}
func (r *row) GetValue(name string, pos uint) (any, error) {
func (r *Row) GetValue(name string, pos uint) (any, error) {
return r.values[name][pos], nil
}
func (r *row) String() string {
func (r *Row) String() string {
out := make([]string, 0, 20)
for k, vv := range r.values {
for i, v := range vv {
@@ -119,7 +119,7 @@ func (r *row) String() string {
return strings.Join(out, " | ")
}
func (r row) Copy() *row {
func (r Row) Copy() *Row {
out := &r
out.values = out.values.Copy()
@@ -137,7 +137,7 @@ func (vv valueSet) Copy() valueSet {
return out
}
func mergeRows(mapping []AttributeMapping, dst *row, ss ...*row) (err error) {
func mergeRows(mapping []AttributeMapping, dst *Row, ss ...*Row) (err error) {
if len(mapping) == 0 {
return mergeRowsFull(dst, ss...)
}
@@ -145,7 +145,7 @@ func mergeRows(mapping []AttributeMapping, dst *row, ss ...*row) (err error) {
return mergeRowsMapped(mapping, dst, ss...)
}
func mergeRowsFull(dst *row, rows ...*row) (err error) {
func mergeRowsFull(dst *Row, rows ...*Row) (err error) {
for _, r := range rows {
for name, vv := range r.values {
for i, values := range vv {
@@ -170,7 +170,7 @@ func mergeRowsFull(dst *row, rows ...*row) (err error) {
return
}
func mergeRowsMapped(mapping []AttributeMapping, out *row, rows ...*row) (err error) {
func mergeRowsMapped(mapping []AttributeMapping, out *Row, rows ...*Row) (err error) {
for _, mp := range mapping {
name := mp.Source()
@@ -193,8 +193,8 @@ func mergeRowsMapped(mapping []AttributeMapping, out *row, rows ...*row) (err er
// makeRowComparator is a utility for easily making a row comparator for
// the given sort expression
func makeRowComparator(ss ...*filter.SortExpr) func(a, b *row) bool {
return func(a, b *row) bool {
func makeRowComparator(ss ...*filter.SortExpr) func(a, b *Row) bool {
return func(a, b *Row) bool {
for _, s := range ss {
cmp := compareGetters(a, b, a.counters, b.counters, s.Column)
pkg/dal/buffer_test.go
+16 -16
View File
@@ -9,27 +9,27 @@ import (
func TestMergeRows(t *testing.T) {
tcc := []struct {
name string
a *row
b *row
a *Row
b *Row
mapping []AttributeMapping
out *row
out *Row
}{{
name: "full merge; no mapping",
a: (&row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello"),
b: (&row{}).WithValue("attr3", 0, true).WithValue("attr4", 0, "ee").WithValue("attr4", 1, 25),
out: (&row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello").WithValue("attr3", 0, true).WithValue("attr4", 0, "ee").WithValue("attr4", 1, 25),
a: (&Row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello"),
b: (&Row{}).WithValue("attr3", 0, true).WithValue("attr4", 0, "ee").WithValue("attr4", 1, 25),
out: (&Row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello").WithValue("attr3", 0, true).WithValue("attr4", 0, "ee").WithValue("attr4", 1, 25),
}, {
name: "full merge; no mapping; collision",
a: (&row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello"),
b: (&row{}).WithValue("attr2", 0, true).WithValue("attr3", 0, "ee").WithValue("attr3", 1, 25),
out: (&row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, true).WithValue("attr3", 0, "ee").WithValue("attr3", 1, 25),
a: (&Row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello"),
b: (&Row{}).WithValue("attr2", 0, true).WithValue("attr3", 0, "ee").WithValue("attr3", 1, 25),
out: (&Row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, true).WithValue("attr3", 0, "ee").WithValue("attr3", 1, 25),
},
{
name: "mapped merge",
a: (&row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello"),
b: (&row{}).WithValue("attr3", 0, true).WithValue("attr4", 0, "ee").WithValue("attr4", 1, 25),
out: (&row{}).WithValue("a", 0, 10).WithValue("b", 0, "hi").WithValue("b", 1, "hello").WithValue("c", 0, true).WithValue("d", 0, "ee").WithValue("d", 1, 25),
a: (&Row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello"),
b: (&Row{}).WithValue("attr3", 0, true).WithValue("attr4", 0, "ee").WithValue("attr4", 1, 25),
out: (&Row{}).WithValue("a", 0, 10).WithValue("b", 0, "hi").WithValue("b", 1, "hello").WithValue("c", 0, true).WithValue("d", 0, "ee").WithValue("d", 1, 25),
mapping: saToMapping([]simpleAttribute{{
ident: "a",
source: "attr1",
@@ -45,9 +45,9 @@ func TestMergeRows(t *testing.T) {
}}...),
}, {
name: "mapped merge with conflicts",
a: (&row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello"),
b: (&row{}).WithValue("attr3", 0, true).WithValue("attr4", 0, "ee").WithValue("attr4", 1, 25),
out: (&row{}).WithValue("a", 0, 10).WithValue("b", 0, true).WithValue("c", 0, "ee").WithValue("c", 1, 25),
a: (&Row{}).WithValue("attr1", 0, 10).WithValue("attr2", 0, "hi").WithValue("attr2", 1, "hello"),
b: (&Row{}).WithValue("attr3", 0, true).WithValue("attr4", 0, "ee").WithValue("attr4", 1, 25),
out: (&Row{}).WithValue("a", 0, 10).WithValue("b", 0, true).WithValue("c", 0, "ee").WithValue("c", 1, 25),
mapping: saToMapping([]simpleAttribute{{
ident: "a",
source: "attr1",
@@ -65,7 +65,7 @@ func TestMergeRows(t *testing.T) {
for _, c := range tcc {
t.Run(c.name, func(t *testing.T) {
out := &row{}
out := &Row{}
err := mergeRows(c.mapping, out, c.a, c.b)
require.NoError(t, err)
require.Equal(t, c.out, out)
pkg/dal/exec_aggregate.go
+6 -6
View File
@@ -16,7 +16,7 @@ type (
source Iterator
err error
scanRow *row
scanRow *Row
planned bool
rowTester tester
@@ -218,7 +218,7 @@ func (xs *aggregate) pullEntireSource(ctx context.Context) (err error) {
}
// @todo consider pre-populating the hashmaps
r := &row{
r := &Row{
counters: make(map[string]uint),
values: make(valueSet),
}
@@ -341,7 +341,7 @@ func (xs *aggregate) nextGroup(ctx context.Context) (_ *aggregateGroup, err erro
return xs.groups[xs.i-1], nil
}
func (s *aggregate) scanKey(g *aggregateGroup, dst *row) (err error) {
func (s *aggregate) scanKey(g *aggregateGroup, dst *Row) (err error) {
for i, attr := range s.def.Group {
// @todo multi value support?
dst.SetValue(attr.Identifier(), 0, g.key[i])
@@ -351,7 +351,7 @@ func (s *aggregate) scanKey(g *aggregateGroup, dst *row) (err error) {
return nil
}
func (s *aggregate) keep(ctx context.Context, r *row) bool {
func (s *aggregate) keep(ctx context.Context, r *Row) bool {
if s.rowTester == nil {
return true
}
@@ -416,9 +416,9 @@ func inKeys(kk []AttributeMapping, ident string) int {
return -1
}
func (xs *aggregate) initScanRow() (out *row) {
func (xs *aggregate) initScanRow() (out *Row) {
// base
out = &row{
out = &Row{
counters: make(map[string]uint),
values: make(valueSet),
}
pkg/dal/exec_join_left.go
+9 -9
View File
@@ -27,7 +27,7 @@ type (
leftSource Iterator
rightSource Iterator
err error
scanRow *row
scanRow *Row
planned bool
filtered bool
@@ -48,7 +48,7 @@ type (
// @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]
outSorted *btree.Generic[*Row]
i int
}
)
@@ -99,7 +99,7 @@ func (xs *joinLeft) More(limit uint, v ValueGetter) (err error) {
// 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.outSorted = btree.NewGenericOptions[*Row](makeRowComparator(xs.filter.OrderBy()...), btree.Options{NoLocks: true})
xs.scanRow = nil
xs.planned = false
xs.i = 0
@@ -237,7 +237,7 @@ func (xs *joinLeft) pullEntireSource(ctx context.Context) (err error) {
// 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{
r := &Row{
counters: make(map[string]uint),
values: make(valueSet),
}
@@ -259,7 +259,7 @@ func (xs *joinLeft) pullEntireRightSource(ctx context.Context) (err error) {
// work on this stage
func (xs *joinLeft) pullEntireLeftSource(ctx context.Context) (err error) {
for xs.leftSource.Next(ctx) {
l := &row{
l := &Row{
counters: make(map[string]uint),
values: make(valueSet),
}
@@ -281,7 +281,7 @@ func (xs *joinLeft) pullEntireLeftSource(ctx context.Context) (err error) {
//
// @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) {
func (xs *joinLeft) joinRight(ctx context.Context, left *Row) (err error) {
bb, ok, err := xs.getRelatedBuffers(left)
if err != nil || !ok {
return
@@ -308,7 +308,7 @@ func (xs *joinLeft) joinRight(ctx context.Context, left *row) (err error) {
}
// 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) {
func (xs *joinLeft) getRelatedBuffers(l *Row) (out []*relIndexBuffer, ok bool, err error) {
attrIdent := xs.joinLeftAttr.Identifier()
attrType := xs.joinLeftAttr.Properties().Type
var aux *relIndexBuffer
@@ -355,7 +355,7 @@ func (xs *joinLeft) getRelatedBuffers(l *row) (out []*relIndexBuffer, ok bool, e
// 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) {
func (xs *joinLeft) indexRightRow(r *Row) (err error) {
attrIdent := xs.joinRightAttr.Identifier()
attrType := xs.joinRightAttr.Properties().Type
@@ -390,7 +390,7 @@ func (xs *joinLeft) indexRightRow(r *row) (err error) {
}
// keep checks if the row should be kept or discarded
func (xs *joinLeft) keep(ctx context.Context, r *row) bool {
func (xs *joinLeft) keep(ctx context.Context, r *Row) bool {
if xs.rowTester == nil {
return true
}
pkg/dal/exec_link_left.go
+10 -10
View File
@@ -18,7 +18,7 @@ type (
leftSource Iterator
rightSource Iterator
err error
scanRow *row
scanRow *Row
planned bool
filtered bool
@@ -35,15 +35,15 @@ type (
rowTester tester
// Buffer to keep track of pulled left rows
leftRows []*row
leftRows []*Row
relIndex *relIndex
keepLeft bool
leftIndex int
rightIndex int
// Some helper fields for temporary data
leftRow *row
rightRow *row
leftRow *Row
rightRow *Row
relScanBuffer *relIndexBuffer
}
)
@@ -97,7 +97,7 @@ func (xs *linkLeft) More(limit uint, v ValueGetter) (err error) {
// Redo the state
// @todo adjust based on aggregation plan; reuse buffered, etc.
xs.relIndex = newRelIndex(xs.rightSortAttrs...)
xs.leftRows = make([]*row, 0, 128)
xs.leftRows = make([]*Row, 0, 128)
xs.scanRow = nil
xs.planned = false
xs.keepLeft = false
@@ -259,7 +259,7 @@ func (xs *linkLeft) nextBuffered() (more bool, err error) {
}
// getRelatedBuffer returns all of the right rows corresponding to the given left row
func (xs *linkLeft) getRelatedBuffer(l *row) (out *relIndexBuffer, ok bool, err error) {
func (xs *linkLeft) getRelatedBuffer(l *Row) (out *relIndexBuffer, ok bool, err error) {
attrIdent := xs.linkLeftAttr.Identifier()
attrType := xs.linkLeftAttr.Properties().Type
@@ -328,7 +328,7 @@ func (xs *linkLeft) pullEntireSource(ctx context.Context) (err error) {
func (xs *linkLeft) pullEntireRightSource(ctx context.Context) (err error) {
// Drain the source
for xs.rightSource.Next(ctx) {
r := &row{
r := &Row{
counters: make(map[string]uint),
values: make(valueSet),
}
@@ -355,7 +355,7 @@ func (xs *linkLeft) pullEntireLeftSource(ctx context.Context) (err error) {
// Drain the source
for xs.leftSource.Next(ctx) {
l := &row{
l := &Row{
counters: make(map[string]uint),
values: make(valueSet),
}
@@ -384,7 +384,7 @@ func (xs *linkLeft) pullEntireLeftSource(ctx context.Context) (err error) {
// indexRightRow pushes the provided row onto the rel index
// @todo consider moving most of this logic to the relIndex struct.
func (xs *linkLeft) indexRightRow(r *row) (err error) {
func (xs *linkLeft) indexRightRow(r *Row) (err error) {
attrIdent := xs.linkRightAttr.Identifier()
attrType := xs.linkRightAttr.Properties().Type
@@ -490,7 +490,7 @@ func (xs *linkLeft) sortLeftRows() (err error) {
// keep checks if the row should be kept or discarded
//
// Link's keep is a bit more complicated and it looks at the related buffer as well.
func (xs *linkLeft) keep(ctx context.Context, left *row, buffer *relIndexBuffer) (keep bool) {
func (xs *linkLeft) keep(ctx context.Context, left *Row, buffer *relIndexBuffer) (keep bool) {
// If no buffer, we won't keep -- left inner join like behavior
if buffer == nil {
return false
pkg/dal/rel_index.go
+3 -3
View File
@@ -34,7 +34,7 @@ func newRelIndex(tt ...string) *relIndex {
}
// AddInt adds a new row under the int key
func (ri *relIndex) AddInt(k int64, r *row) {
func (ri *relIndex) AddInt(k int64, r *Row) {
c, ok := ri.GetInt(k)
if !ok {
c = newRelIndexBuffer(ri.track...)
@@ -48,7 +48,7 @@ func (ri *relIndex) GetInt(k int64) (out *relIndexBuffer, ok bool) {
return
}
func (ri *relIndex) AddString(k string, r *row) {
func (ri *relIndex) AddString(k string, r *Row) {
c, ok := ri.GetString(k)
if !ok {
c = newRelIndexBuffer(ri.track...)
@@ -62,7 +62,7 @@ func (ri *relIndex) GetString(k string) (out *relIndexBuffer, ok bool) {
return
}
func (ri *relIndex) AddID(k uint64, r *row) {
func (ri *relIndex) AddID(k uint64, r *Row) {
c, ok := ri.GetID(k)
if !ok {
c = newRelIndexBuffer(ri.track...)
pkg/dal/rel_index_buffer.go
+4 -4
View File
@@ -15,7 +15,7 @@ type (
// should be ok ig...
//
// Probably don't need to sort yet
rows []*row
rows []*Row
}
)
@@ -29,7 +29,7 @@ func newRelIndexBuffer(tt ...string) *relIndexBuffer {
}
// add adds a new *row to the buffer
func (lc *relIndexBuffer) add(r *row) {
func (lc *relIndexBuffer) add(r *Row) {
if len(lc.rows) == 0 {
for _, ix := range lc.track {
lc.min[ix] = r.values[ix][0]
@@ -47,7 +47,7 @@ func (lc *relIndexBuffer) add(r *row) {
}
// updMin updates the min stat
func (lc *relIndexBuffer) updMin(r *row) {
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]
@@ -59,7 +59,7 @@ func (lc *relIndexBuffer) updMin(r *row) {
}
// updMax updates the max stat
func (lc *relIndexBuffer) updMax(r *row) {
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]
pkg/dal/rel_index_buffer_test.go
+3 -3
View File
@@ -9,7 +9,7 @@ import (
func TestRelIndexBuffer(t *testing.T) {
cc := newRelIndexBuffer("a", "b")
cc.add((&row{}).WithValue("a", 0, 1).WithValue("b", 0, "a"))
cc.add((&Row{}).WithValue("a", 0, 1).WithValue("b", 0, "a"))
require.Len(t, cc.rows, 1)
require.Equal(t, 1, cc.min["a"])
@@ -17,13 +17,13 @@ func TestRelIndexBuffer(t *testing.T) {
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"))
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"))
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"])
pkg/dal/runner_test.go
+6 -6
View File
@@ -17,30 +17,30 @@ func TestRowEvaluatorTest(t *testing.T) {
name: "single row ok",
expr: `row.test == 10`,
in: map[string]ValueGetter{
"row": (&row{}).WithValue("test", 0, 10),
"row": (&Row{}).WithValue("test", 0, 10),
},
out: true,
}, {
name: "single row nok",
expr: `row.test == 11`,
in: map[string]ValueGetter{
"row": (&row{}).WithValue("test", 0, 10),
"row": (&Row{}).WithValue("test", 0, 10),
},
out: false,
}, {
name: "two rows ok",
expr: `local.key == foreign.ref`,
in: map[string]ValueGetter{
"local": (&row{}).WithValue("key", 0, 10),
"foreign": (&row{}).WithValue("ref", 0, 10),
"local": (&Row{}).WithValue("key", 0, 10),
"foreign": (&Row{}).WithValue("ref", 0, 10),
},
out: true,
}, {
name: "two rows nok",
expr: `local.key == foreign.ref`,
in: map[string]ValueGetter{
"local": (&row{}).WithValue("key", 0, 10),
"foreign": (&row{}).WithValue("ref", 0, 11),
"local": (&Row{}).WithValue("key", 0, 10),
"foreign": (&Row{}).WithValue("ref", 0, 11),
},
out: false,
}}