From f80a0b810721611c278f04d5ab22b083e8da5b6f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Mon, 22 Aug 2022 11:32:47 +0200 Subject: [PATCH] Add an inmemory buffer for easier testing --- pkg/dal/buffer.go | 220 ++++++++++++++++++++++++++ pkg/dal/buffer_test.go | 74 +++++++++ pkg/dal/inmem_buffer.go | 220 ++++++++++++++++++++++++++ pkg/dal/inmem_buffer_test.go | 298 +++++++++++++++++++++++++++++++++++ 4 files changed, 812 insertions(+) create mode 100644 pkg/dal/buffer.go create mode 100644 pkg/dal/buffer_test.go create mode 100644 pkg/dal/inmem_buffer.go create mode 100644 pkg/dal/inmem_buffer_test.go diff --git a/pkg/dal/buffer.go b/pkg/dal/buffer.go new file mode 100644 index 000000000..71f2654b4 --- /dev/null +++ b/pkg/dal/buffer.go @@ -0,0 +1,220 @@ +package dal + +import ( + "context" + "fmt" + "strings" + + "github.com/cortezaproject/corteza-server/pkg/filter" +) + +type ( + // OrderedBuffer provides the buffered data in the specified order + OrderedBuffer interface { + Buffer + + // InOrder makes the buffer provide the stored data in the specified order + InOrder(ss ...*filter.SortExpr) (err error) + } + + // Buffer provides a place where you can buffer the data provided by DAL + Buffer interface { + // Seek moves the index pointer to the specified location + // After the Seek call, a Next() call is required + Seek(context.Context, int) error + + // Len returns the number of elements in the buffer + Len() int + + Iterator + Adder + } + + Adder interface { + // Add adds a new ValueGetter to the buffer + Add(context.Context, ValueGetter) (err error) + } + + row struct { + counters map[string]uint + values valueSet + + // ... + + // Metadata to make it easier to work with + // @todo add when needed + } + + valueSet map[string][]any +) + +func (r row) SelectGVal(ctx context.Context, k string) (interface{}, error) { + if r.values[k] == nil { + return nil, nil + } + + if len(r.values[k]) == 0 { + return nil, nil + } + + o := r.values[k][0] + return o, nil +} + +func (r *row) Reset() { + for k := range r.counters { + r.counters[k] = 0 + } +} + +func (r *row) SetValue(name string, pos uint, v any) error { + if r.values == nil { + r.values = make(valueSet) + } + if r.counters == nil { + r.counters = make(map[string]uint) + } + + // Make sure there is space for it + // @note benchmarking proves that the rest of the function introduces + // a lot of memory pressure. + // Investigate options on reworking this/reducing allocations. + if int(pos)+1 > len(r.values[name]) { + r.values[name] = append(r.values[name], make([]any, (int(pos)+1)-len(r.values[name]))...) + } + + r.values[name][pos] = v + r.counters[name]++ + + return nil +} + +// 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 { + err := r.SetValue(name, pos, v) + if err != nil { + panic(err) + } + + return r +} + +func (r *row) CountValues() map[string]uint { + return r.counters +} + +func (r *row) GetValue(name string, pos uint) (any, error) { + return r.values[name][pos], nil +} + +func (r *row) String() string { + out := make([]string, 0, 20) + for k, vv := range r.values { + for i, v := range vv { + out = append(out, fmt.Sprintf("%s [%d] %v", k, i, v)) + } + } + + return strings.Join(out, " | ") +} + +func (r row) Copy() *row { + out := &r + + out.values = out.values.Copy() + + return out +} + +func (vv valueSet) Copy() valueSet { + out := make(valueSet) + + for n, vv := range vv { + out[n] = vv + } + + return out +} + +func mergeRows(mapping []AttributeMapping, dst *row, ss ...*row) (err error) { + if len(mapping) == 0 { + return mergeRowsFull(dst, ss...) + } + + return mergeRowsMapped(mapping, dst, ss...) +} + +func mergeRowsFull(dst *row, rows ...*row) (err error) { + for _, r := range rows { + for name, vv := range r.values { + for i, values := range vv { + if dst.values == nil { + dst.values = make(valueSet) + dst.counters = make(map[string]uint) + } + + if i == 0 { + dst.values[name] = make([]any, len(vv)) + dst.counters[name] = 0 + } + + err = dst.SetValue(name, uint(i), values) + if err != nil { + return + } + } + } + } + + return +} + +func mergeRowsMapped(mapping []AttributeMapping, out *row, rows ...*row) (err error) { + + for _, mp := range mapping { + name := mp.Source() + for _, r := range rows { + if r.values[name] != nil { + if out.values == nil { + out.values = make(valueSet) + out.counters = make(map[string]uint) + } + + out.values[mp.Identifier()] = r.values[name] + out.counters[mp.Identifier()] = r.counters[name] + break + } + } + } + + return +} + +// 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 { + for _, s := range ss { + cmp := compareGetters(a, b, a.counters, b.counters, s.Column) + + less, skip := evalCmpResult(cmp, s) + if !skip { + return less + } + } + + return false + } +} + +func evalCmpResult(cmp int, s *filter.SortExpr) (less, skip bool) { + if cmp != 0 { + if s.Descending { + return cmp > 0, false + } + return cmp < 0, false + } + + return false, true +} diff --git a/pkg/dal/buffer_test.go b/pkg/dal/buffer_test.go new file mode 100644 index 000000000..91535fc3e --- /dev/null +++ b/pkg/dal/buffer_test.go @@ -0,0 +1,74 @@ +package dal + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestMergeRows(t *testing.T) { + tcc := []struct { + name string + a *row + b *row + mapping []AttributeMapping + 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), + }, { + 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), + }, + + { + 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), + mapping: saToMapping([]simpleAttribute{{ + ident: "a", + source: "attr1", + }, { + ident: "b", + source: "attr2", + }, { + ident: "c", + source: "attr3", + }, { + ident: "d", + source: "attr4", + }}...), + }, { + 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), + mapping: saToMapping([]simpleAttribute{{ + ident: "a", + source: "attr1", + }, { + ident: "b", + source: "attr2", + }, { + ident: "b", + source: "attr3", + }, { + ident: "c", + source: "attr4", + }}...), + }} + + for _, c := range tcc { + t.Run(c.name, func(t *testing.T) { + out := &row{} + err := mergeRows(c.mapping, out, c.a, c.b) + require.NoError(t, err) + require.Equal(t, c.out, out) + }) + } +} diff --git a/pkg/dal/inmem_buffer.go b/pkg/dal/inmem_buffer.go new file mode 100644 index 000000000..9b2ac2d90 --- /dev/null +++ b/pkg/dal/inmem_buffer.go @@ -0,0 +1,220 @@ +package dal + +import ( + "context" + "fmt" + "sort" + + "github.com/cortezaproject/corteza-server/pkg/filter" +) + +type ( + // inmemBuffer isa simple DAL buffer which holds all of the data in memory + // + // This buffer should be used for small datasets + inmemBuffer struct { + rows []ValueGetter + ctrs []map[string]uint + + i int + cap int + closed bool + + more bool + err error + + sort filter.SortExprSet + } +) + +// InMemoryBuffer initializes a new DAL buffer where the data is kept in memory +func InMemoryBuffer() *inmemBuffer { + return &inmemBuffer{ + // @note we'll buffer the value counters along side the actual value counter + // because constant hashmap initialization is quite memory intensive and introduces + // a bit of a bottleneck. + // + // @todo investigate if this can be optimized at a lower level + rows: make([]ValueGetter, 0, 100), + ctrs: make([]map[string]uint, 0, 100), + + // i starts off as -1 so we don't need an extra state flag when doing the first Next + i: -1, + more: true, + } +} + +// InMemoryBufferWith returns a new buffer with the given value getters +func InMemoryBufferWith(ctx context.Context, vv ...ValueGetter) (Buffer, error) { + b := InMemoryBuffer() + for _, v := range vv { + if err := b.Add(ctx, v); err != nil { + return nil, err + } + } + return b, nil +} + +func (b *inmemBuffer) InOrder(ss ...*filter.SortExpr) (err error) { + b.sort = ss + + return nil +} + +// Single makes the buffer only keep one element at the time +// @todo make this option convert the buffer into a circular buffer +func (b *inmemBuffer) Single() { + b.cap = 1 + b.rows = make([]ValueGetter, 1) + b.ctrs = make([]map[string]uint, 1) + b.i = -1 +} + +func (b *inmemBuffer) Seek(_ context.Context, i int) (err error) { + // we go one level further and start at -1 + b.i = i - 1 + + b.more = i < b.Len() + return +} + +func (b *inmemBuffer) isSingle() bool { + return b.cap == 1 +} + +func (b *inmemBuffer) isSorted() bool { + return len(b.sort) > 0 +} + +func (b *inmemBuffer) Add(ctx context.Context, v ValueGetter) (err error) { + if b.i != -1 && b.isSorted() { + return fmt.Errorf("cannot buffer items after an access occurred and the buffer is ordered") + } + + if b.isSingle() { + b.rows[0] = v + b.ctrs[0] = v.CountValues() + } else { + b.rows = append(b.rows, v) + b.ctrs = append(b.ctrs, v.CountValues()) + } + + b.more = true + return +} + +func (b *inmemBuffer) Next(ctx context.Context) bool { + if !b.more { + return false + } + if b.closed { + panic("cannot call Next on closed buffer") + } + + // @todo use something like a b-tree to sort while inserting + if b.i == -1 && len(b.sort) > 0 { + sort.Sort(b) + } + + b.i++ + if b.isSingle() { + b.i = b.i % b.cap + } + + b.more = !b.isSingle() && b.i+1 < b.Len() + + return b.i < b.Len() +} + +func (b *inmemBuffer) More(uint, ValueGetter) error { + return nil +} + +func (b *inmemBuffer) Err() error { + return b.err +} + +func (b *inmemBuffer) Scan(s ValueSetter) (err error) { + if b.closed { + panic("cannot call Scan on closed buffer") + } + + r, ctrs, err := b.rowAt(b.i) + if err != nil { + return + } + + var v any + for name, count := range ctrs { + for i := uint(0); i < count; i++ { + v, err = r.GetValue(name, i) + if err != nil { + return + } + err = s.SetValue(name, i, v) + if err != nil { + return + } + } + } + return +} + +// @todo should this get a context for things like IO ops +func (b *inmemBuffer) Close() error { + b.ctrs = nil + b.rows = nil + b.closed = true + + return nil +} + +func (b *inmemBuffer) BackCursor(ValueGetter) (*filter.PagingCursor, error) { + return nil, fmt.Errorf("not supported") +} + +func (b *inmemBuffer) ForwardCursor(ValueGetter) (*filter.PagingCursor, error) { + return nil, fmt.Errorf("not supported") +} + +// rowAt returns the row at a given index +func (b *inmemBuffer) rowAt(i int) (ValueGetter, map[string]uint, error) { + if i >= b.Len() { + return nil, nil, nil + } + + return b.rows[i], b.ctrs[i], nil +} + +// sort.Interface methods + +func (b *inmemBuffer) Len() int { + return len(b.rows) +} + +func (b *inmemBuffer) Less(i, j int) bool { + var ( + err error + ra ValueGetter + rb ValueGetter + rac map[string]uint + rbc map[string]uint + ) + + ra, rac, err = b.rowAt(i) + if err != nil { + panic(err) + } + + rb, rbc, err = b.rowAt(j) + if err != nil { + panic(err) + } + + return valueGetterCounterComparator(b.sort, ra, rb, rac, rbc) +} + +func (b *inmemBuffer) Swap(i, j int) { + b.rows[i], b.rows[j] = b.rows[j], b.rows[i] + b.ctrs[i], b.ctrs[j] = b.ctrs[j], b.ctrs[i] +} diff --git a/pkg/dal/inmem_buffer_test.go b/pkg/dal/inmem_buffer_test.go new file mode 100644 index 000000000..2e1d181e5 --- /dev/null +++ b/pkg/dal/inmem_buffer_test.go @@ -0,0 +1,298 @@ +package dal + +import ( + "context" + "testing" + + "github.com/cortezaproject/corteza-server/pkg/filter" + "github.com/stretchr/testify/require" +) + +func TestInmemBuffer_rw(t *testing.T) { + gBuff := InMemoryBuffer() + ctx := context.Background() + + tcc := []struct { + name string + prep func() *inmemBuffer + in []simpleRow + test func(t *testing.T, in []simpleRow, buff *inmemBuffer) + }{{ + name: "fresh insert", + prep: func() *inmemBuffer { + return gBuff + }, + in: []simpleRow{ + { + "r1_k1": "r1_k1", + "r1_k2": "r1_k2", + "r1_k3": "r1_k3", + }}, + test: func(t *testing.T, in []simpleRow, buff *inmemBuffer) { + r := make(simpleRow) + + require.True(t, buff.Next(ctx)) + require.NoError(t, buff.Scan(r)) + require.Equal(t, in[0], r) + + require.False(t, buff.Next(ctx)) + }, + }, { + name: "insert into existing", + prep: func() *inmemBuffer { + return gBuff + }, + in: []simpleRow{ + { + "r2_k1": "r2_k1", + "r2_k2": "r2_k2", + "r2_k3": "r2_k3", + }}, + test: func(t *testing.T, in []simpleRow, buff *inmemBuffer) { + r := make(simpleRow) + + require.True(t, buff.Next(ctx)) + require.NoError(t, buff.Scan(r)) + require.Equal(t, in[0], r) + + require.False(t, buff.Next(ctx)) + }, + }} + + for _, c := range tcc { + t.Run(c.name, func(t *testing.T) { + buff := c.prep() + for _, in := range c.in { + require.NoError(t, buff.Add(ctx, in)) + } + c.test(t, c.in, buff) + }) + } +} + +func TestInmemBuffer_single_rw(t *testing.T) { + mk := func() *inmemBuffer { + buff := InMemoryBuffer() + buff.Single() + return buff + } + ctx := context.Background() + + tcc := []struct { + name string + prep func() *inmemBuffer + in []simpleRow + test func(t *testing.T, in []simpleRow, buff *inmemBuffer) + }{{ + name: "one", + prep: func() *inmemBuffer { + return mk() + }, + in: []simpleRow{ + { + "r1_k1": "r1_k1", + "r1_k2": "r1_k2", + "r1_k3": "r1_k3", + }}, + test: func(t *testing.T, in []simpleRow, buff *inmemBuffer) { + r := make(simpleRow) + + require.True(t, buff.Next(ctx)) + require.NoError(t, buff.Scan(r)) + require.Equal(t, in[0], r) + + require.False(t, buff.Next(ctx)) + }, + }, { + name: "two", + prep: func() *inmemBuffer { + return mk() + }, + in: []simpleRow{ + { + "r1_k1": "r1_k1", + "r1_k2": "r1_k2", + "r1_k3": "r1_k3", + }, { + "r2_k1": "r2_k1", + "r2_k2": "r2_k2", + "r2_k3": "r2_k3", + }}, + test: func(t *testing.T, in []simpleRow, buff *inmemBuffer) { + r := make(simpleRow) + + require.True(t, buff.Next(ctx)) + require.NoError(t, buff.Scan(r)) + require.Equal(t, in[1], r) + + require.False(t, buff.Next(ctx)) + }, + }} + + for _, c := range tcc { + t.Run(c.name, func(t *testing.T) { + buff := c.prep() + for _, in := range c.in { + require.NoError(t, buff.Add(ctx, in)) + } + c.test(t, c.in, buff) + }) + } +} + +func TestInmemBuffer_single_writeMidRead(t *testing.T) { + ctx := context.Background() + + buff := InMemoryBuffer() + buff.Single() + + a := simpleRow{ + "r1_k1": "r1_k1", + "r1_k2": "r1_k2", + "r1_k3": "r1_k3", + } + + b := simpleRow{ + "r2_k1": "r2_k1", + "r2_k2": "r2_k2", + "r2_k3": "r2_k3", + } + + c := simpleRow{ + "r3_k1": "r3_k1", + "r3_k2": "r3_k2", + "r3_k3": "r3_k3", + } + + // Write + require.NoError(t, buff.Add(ctx, a)) + + // Read + tmp := simpleRow{} + require.True(t, buff.Next(ctx)) + require.NoError(t, buff.Err()) + require.NoError(t, buff.Scan(tmp)) + require.Equal(t, a, tmp) + + // Next write + require.NoError(t, buff.Add(ctx, b)) + // Next write + require.NoError(t, buff.Add(ctx, c)) + + // Read + tmp = simpleRow{} + require.True(t, buff.Next(ctx)) + require.NoError(t, buff.Err()) + require.NoError(t, buff.Scan(tmp)) + require.Equal(t, c, tmp) + + // Read empty + require.False(t, buff.Next(ctx)) + require.NoError(t, buff.Err()) +} + +func TestInmemBuffer_inOrder(t *testing.T) { + mk := func() *inmemBuffer { + buff := InMemoryBuffer() + return buff + } + ctx := context.Background() + + tcc := []struct { + name string + prep func() *inmemBuffer + in []simpleRow + out []simpleRow + sort filter.SortExprSet + }{{ + name: "in order", + prep: func() *inmemBuffer { + return mk() + }, + in: []simpleRow{ + { + "order": 3, + }, { + "order": 2, + }, { + "order": 4, + }, { + "order": 1, + }}, + out: []simpleRow{ + { + "order": 1, + }, { + "order": 2, + }, { + "order": 3, + }, { + "order": 4, + }}, + sort: filter.SortExprSet{{ + Column: "order", + Descending: false, + }}, + }, + } + + for _, c := range tcc { + t.Run(c.name, func(t *testing.T) { + buff := c.prep() + buff.InOrder(c.sort...) + for _, in := range c.in { + require.NoError(t, buff.Add(ctx, in)) + } + + out := make([]simpleRow, 0, 4) + for buff.Next(ctx) { + require.NoError(t, buff.Err()) + + r := simpleRow{} + require.NoError(t, buff.Scan(r)) + out = append(out, r) + } + + require.Equal(t, c.out, out) + }) + } +} + +func TestInmemBuffer_seek(t *testing.T) { + ctx := context.Background() + + buff := InMemoryBuffer() + + a := simpleRow{ + "k": "k1", + } + b := simpleRow{ + "k": "k2", + } + c := simpleRow{ + "k": "k3", + } + + tmp := simpleRow{} + + // Write + require.NoError(t, buff.Add(ctx, a)) + require.NoError(t, buff.Add(ctx, b)) + require.NoError(t, buff.Add(ctx, c)) + + // Seek to end + require.NoError(t, buff.Seek(ctx, 4)) + require.False(t, buff.Next(ctx)) + + // Seek to start + require.NoError(t, buff.Seek(ctx, 0)) + require.True(t, buff.Next(ctx)) + require.NoError(t, buff.Scan(tmp)) + require.Equal(t, a, tmp) + + // Seek to middle + require.NoError(t, buff.Seek(ctx, 2)) + require.True(t, buff.Next(ctx)) + require.NoError(t, buff.Scan(tmp)) + require.Equal(t, c, tmp) +}