partitions PR fixes (#45)

* update part manager API

* add multi-partition query (WIP)

* add multi-partition query (WIP)

* partition manager impl (#31)

* new schema

* fix recursive imports

* use chunk and partition format interval

* remove prints

* format

* fix utils

* implement getShardingBuckets

* add multi-partition query - working

* add multi-partition query - working (#34)

* Merge Dev to Partitions @ Yaron (#1)

* Make BatchSize configurable; Update BenchIngest test

* Suppot `--stdin` and constant memory when reading CSV. (#27)

Previously, the entire CSV file was read into memory before anything else could run.

* Add validation

* Refactor: move stuff to common; Add wait and validation to Nuclio bench

* init configuration from testdata

* address PR comments

* Add tests (#28)

* add aggregators test

* add ingest+query+create tests

* rename tests description, add negative aggregator test.

* add integration target to make file + pr comments

* add bug id to ignored tests

* go format

* add a test

* format

* create nuclio build tag, create test util

* format

* run every table test case as a sub test

* more pr comments + move tests util to tsdbtest

* move test models to tsdbtest

* support ignoring tests

* create unique DB name on test setup

* format

* add delete TSDB test

* minor style fix

* fix nuclio test

* format

* style changes

* improve tests + use testify

* add build tags to go get

* fix tags on go get

* add race condition check on test

* add multi-partition query - working (#34)

* Fix CI

* partition manager fixes (#38)

* code review

* fix utils

* cleanup

* fmt

* fix test

* code review

* invalid aggr list

* tests

* Fix skipping of start partition in query. (#42)

* pr fixes, mostly style & readability

* minor fix

* edit comment

* minor fix

* fixed pr comments,propogate errors and other minor fixes

* revert sharding bucket feature, change default to 1 bucket

* revert w,M,Y support

* Minor usability and code improvements.

* logger -> println because logger is not initialized...

* go fmt

Author: talIguaz

pkg/aggregate/aggregate.go

@@ -22,6 +22,7 @@ package aggregate
 
 import (
 	"fmt"
+	"github.com/v3io/v3io-tsdb/pkg/config"
 	"math"
 	"strings"
 )
@@ -58,18 +59,62 @@ var aggrToString = map[AggrType]string{
 	aggrTypeStddev: "stddev", aggrTypeStdvar: "stdvar", aggrTypeAll: "*",
 }
 
+var aggrToSchemaField = map[string]config.SchemaField{
+	"count":  {Name: "count", Type: "array", Nullable: true, Items: "double"},
+	"sum":    {Name: "sum", Type: "array", Nullable: true, Items: "double"},
+	"sqr":    {Name: "sqr", Type: "array", Nullable: true, Items: "double"},
+	"max":    {Name: "max", Type: "array", Nullable: true, Items: "double"},
+	"min":    {Name: "min", Type: "array", Nullable: true, Items: "double"},
+	"last":   {Name: "last", Type: "array", Nullable: true, Items: "double"},
+	"avg":    {Name: "avg", Type: "array", Nullable: true, Items: "double"},
+	"rate":   {Name: "rate", Type: "array", Nullable: true, Items: "double"},
+	"stddev": {Name: "stddev", Type: "array", Nullable: true, Items: "double"},
+	"stdvar": {Name: "stdvar", Type: "array", Nullable: true, Items: "double"},
+}
+
+func SchemaFieldFromString(aggregators []string, col string) ([]config.SchemaField, error) {
+	fieldList := make([]config.SchemaField, 0, len(aggregators))
+	for _, s := range aggregators {
+		trimmed := strings.TrimSpace(s)
+		if trimmed != "" {
+			if trimmed == "*" {
+				fieldList = make([]config.SchemaField, 0, len(aggrToSchemaField))
+				for _, val := range aggrToSchemaField {
+					fieldList = append(fieldList, getAggrFullName(val, col))
+				}
+				return fieldList, nil
+			} else {
+				field, ok := aggrToSchemaField[trimmed]
+				if !ok {
+					return nil, fmt.Errorf("invalid aggragator type '%s'", trimmed)
+				}
+				fieldList = append(fieldList, getAggrFullName(field, col))
+			}
+		}
+	}
+	return fieldList, nil
+}
+
+func getAggrFullName(field config.SchemaField, col string) config.SchemaField {
+	fullName := fmt.Sprintf("_%s_%s", col, field.Name)
+	field.Name = fullName
+	return field
+}
+
 func (a AggrType) String() string { return aggrToString[a] }
 
 // convert comma separated string to aggregator mask
-func AggrsFromString(list string) (AggrType, error) {
-	split := strings.Split(list, ",")
+func AggrsFromString(split []string) (AggrType, error) {
 	var aggrList AggrType
 	for _, s := range split {
-		aggr, ok := aggrTypeString[s]
-		if !ok {
-			return aggrList, fmt.Errorf("Invalid aggragator type %s", s)
+		trimmed := strings.TrimSpace(s)
+		if trimmed != "" {
+			aggr, ok := aggrTypeString[trimmed]
+			if !ok {
+				return aggrList, fmt.Errorf("invalid aggragator type '%s'", s)
+			}
+			aggrList = aggrList | aggr
 		}
-		aggrList = aggrList | aggr
 	}
 	return aggrList, nil
 }

pkg/aggregate/aggregate_test.go

@@ -70,6 +70,20 @@ func TestAggregators(t *testing.T) {
 			data:      map[int64]float64{1: 7.5, 2: 2.5},
 			exprCol:   "v", bucket: 1,
 			expectedUpdateExpr: "_v_count[1]=_v_count[1]+2;", expectedSetExpr: "_v_count[1]=2;", expectFail: true},
+
+		{desc: "Should aggregate data when specifying aggregators with sapces",
+			aggString: "min , max   ",
+			data:      map[int64]float64{1: 7.5, 2: 2.5},
+			exprCol:   "v", bucket: 1,
+			expectedUpdateExpr: fmt.Sprintf("_v_min[1]=min(_v_min[1],%f);_v_max[1]=max(_v_max[1],%f);", 2.5, 7.5),
+			expectedSetExpr:    fmt.Sprintf("_v_min[1]=%f;_v_max[1]=%f;", 2.5, 7.5)},
+
+		{desc: "Should aggregate data when specifying aggregators with empty values",
+			aggString: "min , ,max   ",
+			data:      map[int64]float64{1: 7.5, 2: 2.5},
+			exprCol:   "v", bucket: 1,
+			expectedUpdateExpr: fmt.Sprintf("_v_min[1]=min(_v_min[1],%f);_v_max[1]=max(_v_max[1],%f);", 2.5, 7.5),
+			expectedSetExpr:    fmt.Sprintf("_v_min[1]=%f;_v_max[1]=%f;", 2.5, 7.5)},
 	}
 
 	for _, test := range testCases {
@@ -87,7 +101,7 @@ func TestAggregators(t *testing.T) {
 func testAggregatorCase(t *testing.T, aggString string, data map[int64]float64, exprCol string, bucket int,
 	expectedUpdateExpr string, expectedSetExpr string, expectFail bool) {
 
-	aggregator, err := AggrsFromString(aggString)
+	aggregator, err := AggrsFromString(strings.Split(aggString, ","))
 	if err != nil {
 		if !expectFail {
 			t.Fatal(err)

pkg/appender/store.go

@@ -44,7 +44,7 @@ func NewChunkStore() *chunkStore {
 // chunkStore store state & latest + previous chunk appenders
 type chunkStore struct {
 	curChunk int
-	lastTid  int
+	lastTid  int64
 	chunks   [2]*attrAppender
 
 	aggrList      *aggregate.AggregatorList
@@ -115,7 +115,10 @@ func (cs *chunkStore) GetChunksState(mc *MetricsCache, metric *MetricState) (boo
 
 	// init chunk and create aggregation list object based on partition policy
 	t := cs.pending[0].t
-	part := mc.partitionMngr.TimeToPart(t)
+	part, err := mc.partitionMngr.TimeToPart(t)
+	if err != nil {
+		return false, err
+	}
 	cs.chunks[0].initialize(part, t)
 	cs.aggrList = aggregate.NewAggregatorList(part.AggrType())
 
@@ -182,7 +185,7 @@ func (cs *chunkStore) ProcessGetResp(mc *MetricsCache, metric *MetricState, resp
 	}
 
 	// set Last TableId, indicate that there is no need to create metric object
-	cs.lastTid = cs.chunks[0].partition.GetId()
+	cs.lastTid = cs.chunks[0].partition.GetStartTime()
 
 }
 
@@ -217,7 +220,8 @@ func (cs *chunkStore) chunkByTime(t int64) *attrAppender {
 		if err != nil {
 			return nil
 		}
-		cur.initialize(part.NextPart(t), t) // TODO: next part
+		nextPart, _ := part.NextPart(t)
+		cur.initialize(nextPart, t)
 		cur.appender = app
 		cs.curChunk = cs.curChunk ^ 1
 
@@ -251,10 +255,13 @@ func (cs *chunkStore) WriteChunks(mc *MetricsCache, metric *MetricState) (bool,
 
 	// init partition info and find if we need to init the metric headers (labels, ..) in case of new partition
 	t0 := cs.pending[0].t
-	partition := mc.partitionMngr.TimeToPart(t0)
-	if partition.GetId() > cs.lastTid {
+	partition, err := mc.partitionMngr.TimeToPart(t0)
+	if err != nil {
+		return false, err
+	}
+	if partition.GetStartTime() > cs.lastTid {
 		notInitialized = true
-		cs.lastTid = partition.GetId()
+		cs.lastTid = partition.GetStartTime()
 	}
 
 	// init aggregation buckets info

pkg/config/config.go

@@ -27,6 +27,7 @@ import (
 )
 
 const DefaultConfigurationFileName = "v3io.yaml"
+const SCHEMA_CONFIG = ".schema"
 
 type V3ioConfig struct {
 	// V3IO Connection details: Url, Data container, relative path for this dataset, credentials
@@ -54,42 +55,52 @@ type V3ioConfig struct {
 	BatchSize int `json:"batchSize,omitempty"`
 }
 
-type DBPartConfig struct {
-	// Indicating this is a valid Partition file, Signature == 'TSDB'
-	Signature string `json:"signature"`
-	// Version of the config
-	Version string `json:"version"`
-	// Description of this TSDB
-	Description string `json:"description,omitempty"`
-	// Partition Key, __name__ by default
-	ShardingKey string `json:"partitionKey,omitempty"`
-	// Sorting Key, dimensions used for sorting per DB shard
-	SortingKey string `json:"sortingKey,omitempty"`
-	// indicate if it is cyclic (single partition, return to first chunk after the last)
-	IsCyclic bool `json:"isCyclic,omitempty"`
-	// Number of hours per chunk (1hr default)
-	HrInChunk int `json:"hrInChunk,omitempty"`
-	// Days per table/object (in a partition), after N days will use a new table or go to start (Cyclic partition)
-	// this is used only for the Head configuration, per partition we look at StartTime & EndTime
-	DaysPerObj int `json:"daysPerObj,omitempty"`
-	// How many days to save samples
-	DaysRetention int `json:"daysRetention,omitempty"`
-	// Start from time/date in Unix milisec
-	StartTime int64 `json:"startTime,omitempty"`
-	// End by time/date in Unix milisec
-	EndTime int64 `json:"endTime,omitempty"`
-	// Partition name format e.g. 'dd-mm-yy'
-	PartFormat string `json:"partFormat,omitempty"`
-
-	// Comma seperated list of default aggregation functions e.g. 'count,sum,avg,max'
-	DefaultRollups string `json:"defaultRollups,omitempty"`
-	// Number of minutes per aggregation bucket (aggregation interval)
-	RollupMin int `json:"rollupMin,omitempty"`
-	// If true, dont save raw samples/chunks, only aggregates
-	DelRawSamples bool `json:"delRawSamples,omitempty"`
-
-	// Metric specific policy
-	MetricsConfig map[string]MetricConfig `json:"metricsConfig,omitempty"`
+type Rollup struct {
+	Aggregators            string `json:"aggregators"`
+	AggregatorsGranularity string `json:"aggregatorsGranularity"`
+	//["cloud","local"] for the aggregators and sample chucks
+	StorageClass string `json:"storageClass"`
+	//in hours. 0  means no need to save samples
+	SampleRetention int `json:"sampleRetention"`
+	// format : 1m, 7d, 3h . Possible intervals: m/d/h
+	LayerRetentionTime string `json:"layerRetentionTime"`
+}
+
+type TableSchema struct {
+	Version             int      `json:"version"`
+	RollupLayers        []Rollup `json:"rollupLayers"`
+	ShardingBuckets     int      `json:"shardingBuckets"`
+	PartitionerInterval string   `json:"partitionerInterval"`
+	ChunckerInterval    string   `json:"chunckerInterval"`
+}
+
+type PartitionSchema struct {
+	Version                int      `json:"version"`
+	Aggregators            []string `json:"aggregators"`
+	AggregatorsGranularity string   `json:"aggregatorsGranularity"`
+	StorageClass           string   `json:"storageClass"`
+	SampleRetention        int      `json:"sampleRetention"`
+	PartitionerInterval    string   `json:"partitionerInterval"`
+	ChunckerInterval       string   `json:"chunckerInterval"`
+}
+
+type Partition struct {
+	StartTime  int64           `json:"startTime"`
+	SchemaInfo PartitionSchema `json:"schemaInfo"`
+}
+
+type SchemaField struct {
+	Name     string `json:"name"`
+	Type     string `json:"type"`
+	Nullable bool   `json:"nullable"`
+	Items    string `json:"items,omitempty"`
+}
+
+type Schema struct {
+	TableSchemaInfo     TableSchema     `json:"tableSchemaInfo"`
+	PartitionSchemaInfo PartitionSchema `json:"partitionSchemaInfo"`
+	Partitions          []Partition     `json:"partitions"`
+	Fields              []SchemaField   `json:"fields"`
 }
 
 type MetricConfig struct {