Add an inmemory buffer for easier testing
This commit is contained in:
Tomaž Jerman
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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]
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
Reference in New Issue
Block a user