Start sketching on a rolling window function

Author: tobgu

cmd/qfgenerate/main.go

@@ -7,13 +7,13 @@ import (
 	"go/format"
 	"os"
 
-	"github.com/tobgu/qframe/qerrors"
 	bgenerator "github.com/tobgu/qframe/internal/bcolumn"
 	egenerator "github.com/tobgu/qframe/internal/ecolumn"
 	fgenerator "github.com/tobgu/qframe/internal/fcolumn"
 	igenerator "github.com/tobgu/qframe/internal/icolumn"
 	qfgenerator "github.com/tobgu/qframe/internal/qframe/generator"
 	sgenerator "github.com/tobgu/qframe/internal/scolumn"
+	"github.com/tobgu/qframe/qerrors"
 )
 
 /*

config/csv/config.go

@@ -95,7 +95,6 @@ func RowCountHint(rowCount int) ConfigFunc {
 	}
 }
 
-
 // Headers can be used to specify the header names for a CSV file without header.
 //
 // header - Slice with column names.

config/eval/context.go

@@ -6,9 +6,9 @@ import (
 	"reflect"
 	"strings"
 
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/function"
 	qfstrings "github.com/tobgu/qframe/internal/strings"
+	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/types"
 )
 

config/rolling/config.go

@@ -0,0 +1,92 @@
+package rolling
+
+import "github.com/tobgu/qframe/qerrors"
+
+// DataValue can be any of int/float/*string/bool, eg. any type that a column may take.
+type DataValue = interface{}
+
+// IntervalFunc is a function taking two parameters of the same DataValue and returning boolean stating if
+// the two values are part of the same interval or not.
+//
+// For example, x and y within one unit from each other (with x assumed to be <= y):
+type IntervalFunc = interface{}
+
+// It should be considered a private implementation detail and should never be
+// referenced or used directly outside of the QFrame code. To manipulate it
+// use the functions returning ConfigFunc below.
+type Config struct {
+	PadValue        DataValue
+	IntervalColName string
+	IntervalFunc    IntervalFunc
+	WindowSize      int
+	Position        string // center/start/end
+}
+
+// ConfigFunc is a function that operates on a Config object.
+type ConfigFunc func(c *Config)
+
+func NewConfig(ff []ConfigFunc) (Config, error) {
+	c := Config{
+		WindowSize: 1,
+		Position:   "center",
+	}
+
+	for _, fn := range ff {
+		fn(&c)
+	}
+
+	if c.WindowSize <= 0 {
+		return c, qerrors.New("Rolling config", "Window size must be positive, was %d", c.WindowSize)
+	}
+
+	if c.Position != "center" && c.Position != "start" && c.Position != "end" {
+		return c, qerrors.New("Rolling config", "Position must be center/start/end, was %s", c.Position)
+	}
+
+	if c.IntervalFunc != nil && c.WindowSize != 1 {
+		return c, qerrors.New("Rolling config", "Cannot set both interval function and window size")
+	}
+
+	return c, nil
+}
+
+// PadValue can be used to set the value to use in the beginning and/or end of the column to fill out any values
+// where fewer than WindowSize values are available.
+func PadValue(v DataValue) ConfigFunc {
+	return func(c *Config) {
+		c.PadValue = v
+	}
+}
+
+// IntervalFunction can be used to set a dynamic interval based on the content of another column.
+// QFrame will include all rows from the start row of the window until (but not including) the first row that is
+// not part of the interval according to 'fn'. The first parameter passed to 'fn' is always the value at the start
+// of the window.
+//
+// For example, lets say that you have a time series with millisecond resolution integer timestamps in column 'ts'
+// and values in column 'value' that you would like to compute a rolling average over a minute for.
+//
+// In this case:
+// col = "ts", fn = func(tsStart, tsEnd int) bool { return tsEnd < tsStart + int(time.Minute / time.Millisecond)}
+func IntervalFunction(colName string, fn IntervalFunc) ConfigFunc {
+	return func(c *Config) {
+		c.IntervalColName = colName
+		c.IntervalFunc = fn
+	}
+}
+
+// WindowSize is used to set the size of the Window. By default this is 1.
+func WindowSize(s int) ConfigFunc {
+	return func(c *Config) {
+		c.WindowSize = s
+	}
+}
+
+// Position is used to set where in window the resulting value should be inserted.
+// Valid values: start/center/end
+// Default value: center
+func Position(p string) ConfigFunc {
+	return func(c *Config) {
+		c.Position = p
+	}
+}

filter.go

@@ -4,10 +4,10 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/filter"
 	"github.com/tobgu/qframe/internal/index"
 	"github.com/tobgu/qframe/internal/math/integer"
+	"github.com/tobgu/qframe/qerrors"
 )
 
 // FilterClause is an internal interface representing a filter of some kind that can be applied on a QFrame.
@@ -22,8 +22,8 @@ type FilterClause interface {
 type Filter filter.Filter
 
 type comboClause struct {
-	err        error                //nolint:structcheck
-	subClauses []FilterClause       //nolint:structcheck
+	err        error          //nolint:structcheck
+	subClauses []FilterClause //nolint:structcheck
 }
 
 // AndClause represents the logical conjunction of multiple clauses.

grouper.go

@@ -1,10 +1,10 @@
 package qframe
 
 import (
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/internal/grouper"
 	"github.com/tobgu/qframe/internal/icolumn"
 	"github.com/tobgu/qframe/internal/index"
+	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/types"
 )
 

internal/bcolumn/column.go

@@ -4,10 +4,10 @@ import (
 	"reflect"
 	"strconv"
 
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/internal/column"
 	"github.com/tobgu/qframe/internal/hash"
 	"github.com/tobgu/qframe/internal/index"
+	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/types"
 )
 

internal/bcolumn/column_gen.go

@@ -2,6 +2,7 @@ package column
 
 import (
 	"fmt"
+	"github.com/tobgu/qframe/config/rolling"
 
 	"github.com/tobgu/qframe/internal/hash"
 	"github.com/tobgu/qframe/internal/index"
@@ -23,6 +24,8 @@ type Column interface {
 	Apply1(fn interface{}, ix index.Int) (interface{}, error)
 	Apply2(fn interface{}, s2 Column, ix index.Int) (Column, error)
 
+	Rolling(fn interface{}, ix index.Int, config rolling.Config) (Column, error)
+
 	FunctionType() types.FunctionType
 	DataType() types.DataType
 }

internal/column/column.go

@@ -2,16 +2,17 @@ package ecolumn
 
 import (
 	"fmt"
+	"github.com/tobgu/qframe/config/rolling"
 	"reflect"
 	"strings"
 
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/filter"
 	"github.com/tobgu/qframe/internal/column"
 	"github.com/tobgu/qframe/internal/hash"
 	"github.com/tobgu/qframe/internal/index"
 	"github.com/tobgu/qframe/internal/scolumn"
 	qfstrings "github.com/tobgu/qframe/internal/strings"
+	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/types"
 )
 
@@ -567,6 +568,10 @@ func (c Column) View(ix index.Int) View {
 	return View{column: c, index: ix}
 }
 
+func (c Column) Rolling(fn interface{}, ix index.Int, config rolling.Config) (column.Column, error) {
+	return c, nil
+}
+
 func (c Column) FunctionType() types.FunctionType {
 	return types.FunctionTypeString
 }

internal/ecolumn/column.go

@@ -1,10 +1,10 @@
 package ecolumn
 
 import (
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/filter"
 	"github.com/tobgu/qframe/internal/index"
 	qfstrings "github.com/tobgu/qframe/internal/strings"
+	"github.com/tobgu/qframe/qerrors"
 )
 
 var filterFuncs0 = map[string]func(index.Int, []enumVal, index.Bool){

internal/ecolumn/filters.go

@@ -6,10 +6,10 @@ import (
 	"strconv"
 	"unsafe"
 
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/internal/column"
 	"github.com/tobgu/qframe/internal/hash"
 	"github.com/tobgu/qframe/internal/index"
+	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/types"
 )
 

internal/fcolumn/column.go

@@ -27,13 +27,13 @@ type tableEntry struct {
 }
 
 type table struct {
-	entries       []tableEntry
-	comparables   []column.Comparable
-	stats         GroupStats
-	hashBuf       *hash.MemHash
-	loadFactor    float64
-	groupCount    uint32
-	collectIx     bool
+	entries     []tableEntry
+	comparables []column.Comparable
+	stats       GroupStats
+	hashBuf     *hash.MemHash
+	loadFactor  float64
+	groupCount  uint32
+	collectIx   bool
 }
 
 const growthFactor = 2

internal/fcolumn/column_gen.go

@@ -16,7 +16,7 @@ type stringStruct struct {
 
 // MemHash holds a buffer of bytes and allows calculation of a hash over the bytes.
 type MemHash struct {
- 	buf []byte
+	buf []byte
 }
 
 // Reset resets the memory buffer

internal/grouper/grouper.go

@@ -80,7 +80,7 @@ func Test_SmallIntDistribution(t *testing.T) {
 	hasher := hash.MemHash{}
 	for i := 1; i < 177; i++ {
 		hasher.Reset()
-		hasher.Write([]byte{0,0,0,0,0,0,0,byte(i)})
+		hasher.Write([]byte{0, 0, 0, 0, 0, 0, 0, byte(i)})
 		hash := hasher.Hash()
 		result[hash] = result[hash] + 1
 	}

internal/hash/memhash.go

@@ -1,10 +1,10 @@
 package icolumn
 
 import (
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/internal/column"
 	"github.com/tobgu/qframe/internal/hash"
 	"github.com/tobgu/qframe/internal/index"
+	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/types"
 	"reflect"
 	"strconv"

internal/hash/memhash_test.go

@@ -4,11 +4,11 @@ import (
 	"io"
 	"math"
 
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/internal/ecolumn"
 	"github.com/tobgu/qframe/internal/fastcsv"
 	"github.com/tobgu/qframe/internal/ncolumn"
 	"github.com/tobgu/qframe/internal/strings"
+	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/types"
 )
 

internal/icolumn/column.go

@@ -4,14 +4,14 @@ import (
 	"fmt"
 	"reflect"
 
-	"github.com/tobgu/qframe/qerrors"
 	"github.com/tobgu/qframe/internal/bcolumn"
 	"github.com/tobgu/qframe/internal/column"
 	"github.com/tobgu/qframe/internal/ecolumn"
 	"github.com/tobgu/qframe/internal/fcolumn"
 	"github.com/tobgu/qframe/internal/icolumn"
 	"github.com/tobgu/qframe/internal/index"
 	"github.com/tobgu/qframe/internal/scolumn"
+	"github.com/tobgu/qframe/qerrors"
 )
 
 type SQLConfig struct {