Merge pull request #465 from dinal/master

dev -> master

Author: dinal

Diff for: go.mod

@@ -6,24 +6,23 @@ require (
 	github.com/cespare/xxhash v1.1.0
 	github.com/ghodss/yaml v1.0.0
 	github.com/imdario/mergo v0.3.7
+	github.com/inconshreveable/mousetrap v1.0.0 // indirect
+	github.com/kr/pretty v0.2.0 // indirect
 	github.com/nuclio/logger v0.0.1
 	github.com/nuclio/nuclio-sdk-go v0.0.0-20190205170814-3b507fbd0324
 	github.com/nuclio/nuclio-test-go v0.0.0-20180704132150-0ce6587f8e37
 	github.com/nuclio/zap v0.0.2
-	github.com/pavius/impi v0.0.0-20200212064320-5db7efa5f87b // indirect
 	github.com/pkg/errors v0.8.1
 	github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a
-	github.com/russross/blackfriday v1.5.2+incompatible // indirect
 	github.com/spf13/cobra v0.0.3
 	github.com/stretchr/testify v1.4.0
 	github.com/v3io/frames v0.6.8-v0.9.11
-	github.com/v3io/v3io-go v0.0.5-0.20191205125653-9003ae83f0b6
+	github.com/v3io/v3io-go v0.1.5-0.20200301152134-6880d30985de
 	github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2
 )
 
 replace (
-	github.com/v3io/frames => github.com/v3io/frames v0.6.9-v0.9.12.0.20200219120609-981ffb872c73
-	github.com/v3io/v3io-go => github.com/v3io/v3io-go v0.1.5-0.20200224125003-964a745e51aa
+	github.com/v3io/frames => github.com/v3io/frames v0.6.11-v0.9.15.0.20200301102651-c1e7acf2c501
 	github.com/xwb1989/sqlparser => github.com/v3io/sqlparser v0.0.0-20190306105200-4d7273501871
 	labix.org/v2/mgo => github.com/go-mgo/mgo v0.0.0-20180705113738-7446a0344b7872c067b3d6e1b7642571eafbae17
 	launchpad.net/gocheck => github.com/go-check/check v0.0.0-20180628173108-788fd78401277ebd861206a03c884797c6ec5541

Diff for: go.sum

@@ -62,7 +62,7 @@ func (mc *MetricsCache) metricFeed(index int) {
 				mc.logger.Debug(`Complete update cycle - "in-flight requests"=%d; "metric queue length"=%d\n`, inFlight, length)
 
 				// If data was sent and the queue is empty, mark as completion
-				if length == 0 && gotData {
+				if length == 0 && gotData && len(mc.asyncAppendChan) == 0 {
 					gotCompletion = true
 					if completeChan != nil {
 						completeChan <- 0
@@ -80,7 +80,7 @@ func (mc *MetricsCache) metricFeed(index int) {
 						completeChan = app.resp
 
 						length := mc.metricQueue.Length()
-						if gotCompletion && length == 0 {
+						if gotCompletion && length == 0 && len(mc.asyncAppendChan) == 0 {
 							completeChan <- 0
 							gotCompletion = false
 							gotData = false
@@ -209,7 +209,7 @@ func (mc *MetricsCache) metricsUpdateLoop(index int) {
 				}
 
 				// Notify the metric feeder when all in-flight tasks are done
-				if mc.updatesInFlight == 0 {
+				if mc.updatesInFlight == 0 && len(mc.asyncAppendChan) == 0 {
 					mc.logger.Debug("Return to feed. Metric queue length: %d", mc.metricQueue.Length())
 					mc.updatesComplete <- 0
 				}

Diff for: pkg/appender/ingest.go

@@ -160,6 +160,8 @@ type xorAppender struct {
 
 	leading  uint8
 	trailing uint8
+
+	isPreviousNewSeries bool
 }
 
 func (a *xorAppender) Encoding() Encoding {
@@ -192,20 +194,22 @@ func (a *xorAppender) Append(t int64, vvar interface{}) {
 		return
 	}
 
-	if num == 0 {
+	// We write time deltas as 32 bits (for compression) if the delta is too large we'll start a new series
+	tDelta = uint64(t - a.t)
+	shouldStartNewSeries := num == 0 || bits.Len64(tDelta) >= 32
+
+	if shouldStartNewSeries {
 		// add a signature 11111 to indicate start of cseries in case we put few in the same chunk (append to existing)
 		a.b.writeBits(0x1f, 5)
 		a.b.writeBits(uint64(t), 51)
 		a.b.writeBits(math.Float64bits(v), 64)
-
-	} else if num == 1 {
-		tDelta = uint64(t - a.t)
-
+		a.isPreviousNewSeries = true
+		tDelta = 0 // saving time delta for the first element is redundant
+	} else if a.isPreviousNewSeries {
 		a.b.writeBits(tDelta, 32)
 		a.writeVDelta(v)
-
+		a.isPreviousNewSeries = false
 	} else {
-		tDelta = uint64(t - a.t)
 		dod := int64(tDelta - a.tDelta)
 
 		// Gorilla has a max resolution of seconds, Prometheus milliseconds.
@@ -228,6 +232,7 @@ func (a *xorAppender) Append(t int64, vvar interface{}) {
 		}
 
 		a.writeVDelta(v)
+
 	}
 
 	a.t = t

Diff for: pkg/chunkenc/xor.go

@@ -62,14 +62,16 @@ const (
 	DefaultUseServerAggregateCoefficient = 3
 
 	// KV attribute names
-	MaxTimeAttrName     = "_maxtime"
-	LabelSetAttrName    = "_lset"
-	EncodingAttrName    = "_enc"
-	OutOfOrderAttrName  = "_ooo"
-	MetricNameAttrName  = "_name"
-	ObjectNameAttrName  = "__name"
-	ChunkAttrPrefix     = "_v"
-	AggregateAttrPrefix = "_v_"
+	MaxTimeAttrName         = "_maxtime"
+	LabelSetAttrName        = "_lset"
+	EncodingAttrName        = "_enc"
+	OutOfOrderAttrName      = "_ooo"
+	MetricNameAttrName      = "_name"
+	ObjectNameAttrName      = "__name"
+	ChunkAttrPrefix         = "_v"
+	AggregateAttrPrefix     = "_v_"
+	MtimeSecsAttributeName  = "__mtime_secs"
+	MtimeNSecsAttributeName = "__mtime_nsecs"
 
 	PrometheusMetricNameAttribute = "__name__"
 

Diff for: pkg/config/config.go

@@ -195,7 +195,7 @@ func (p *PartitionManager) updateSchema() error {
 			}
 
 			input := &v3io.PutItemInput{Path: schemaFilePath, Attributes: attributes}
-			err := p.container.PutItemSync(input)
+			_, err := p.container.PutItemSync(input)
 
 			if err != nil {
 				outerError = errors.Wrap(err, "failed to update partitions table.")
@@ -235,7 +235,7 @@ func (p *PartitionManager) DeletePartitionsFromSchema(partitionsToDelete []*DBPa
 			deletePartitionExpression.WriteString(");")
 		}
 		expression := deletePartitionExpression.String()
-		err := p.container.UpdateItemSync(&v3io.UpdateItemInput{Path: p.GetSchemaFilePath(), Expression: &expression})
+		_, err := p.container.UpdateItemSync(&v3io.UpdateItemInput{Path: p.GetSchemaFilePath(), Expression: &expression})
 		if err != nil {
 			return err
 		}
@@ -589,6 +589,33 @@ func (p *DBPartition) Time2Bucket(t int64) int {
 	return int((t - p.startTime) / p.rollupTime)
 }
 
+// Return the start time of an aggregation bucket by id
+func (p *DBPartition) GetAggregationBucketStartTime(id int) int64 {
+	return p.startTime + int64(id)*p.rollupTime
+}
+
+// Return the end time of an aggregation bucket by id
+func (p *DBPartition) GetAggregationBucketEndTime(id int) int64 {
+	return p.startTime + int64(id+1)*p.rollupTime - 1
+}
+
+func (p *DBPartition) Times2BucketRange(start, end int64) []int {
+	var buckets []int
+
+	if start > p.GetEndTime() || end < p.startTime {
+		return buckets
+	}
+
+	startingAggrBucket := p.Time2Bucket(start)
+	endAggrBucket := p.Time2Bucket(end)
+
+	for bucketID := startingAggrBucket; bucketID <= endAggrBucket; bucketID++ {
+		buckets = append(buckets, bucketID)
+	}
+
+	return buckets
+}
+
 // Return the nearest chunk start time for the specified time
 func (p *DBPartition) GetChunkMint(t int64) int64 {
 	if t > p.GetEndTime() {
@@ -618,6 +645,37 @@ func (p *DBPartition) TimeToChunkID(tmilli int64) (int, error) {
 	return -1, errors.Errorf("Time %d isn't within the range of this partition.", tmilli)
 }
 
+// Check if a chunk (by attribute name) is in the given time range.
+func (p *DBPartition) IsChunkInRangeByAttr(attr string, mint, maxt int64) bool {
+
+	// Discard '_v' prefix
+	chunkIDStr := attr[2:]
+	chunkID, err := strconv.ParseInt(chunkIDStr, 10, 64)
+	if err != nil {
+		return false
+	}
+
+	chunkStartTime := p.startTime + (chunkID-1)*p.chunkInterval
+	chunkEndTime := chunkStartTime + p.chunkInterval - 1
+
+	return mint <= chunkStartTime && maxt >= chunkEndTime
+}
+
+// Get a chunk's start time by it's attribute name
+func (p *DBPartition) GetChunkStartTimeByAttr(attr string) (int64, error) {
+
+	// Discard '_v' prefix
+	chunkIDStr := attr[2:]
+	chunkID, err := strconv.ParseInt(chunkIDStr, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+
+	chunkStartTime := p.startTime + (chunkID-1)*p.chunkInterval
+
+	return chunkStartTime, nil
+}
+
 // Check whether the specified time is within the range of this partition
 func (p *DBPartition) InRange(t int64) bool {
 	if p.manager.cyclic {

Diff for: pkg/partmgr/partmgr.go

@@ -1206,10 +1206,14 @@ func (suite *testSelectDataframeSuite) TestSparseNumericColumnsWithNotExistingMe
 		suite.basicQueryTime + 3*tsdbtest.MinuteInMillis,
 		suite.basicQueryTime + 4*tsdbtest.MinuteInMillis}
 	expectedColumns := map[string][]interface{}{
-		"cpu_0": {math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN()},
-		"cpu_1": {10.0, 20.0, 30.0, 40.0, 50.0, math.NaN(), 22.0, 33.0, math.NaN(), 55.0},
-		"cpu_2": {math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), 10.0, 20.0, math.NaN(), 40.0, 50.0},
-		"fake":  {math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN()},
+		"cpu_0-linux":   {math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN()},
+		"cpu_1-linux":   {10.0, 20.0, 30.0, 40.0, 50.0},
+		"cpu_2-linux":   {math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN()},
+		"fake-linux":    {math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN()},
+		"cpu_0-windows": {math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN()},
+		"cpu_1-windows": {math.NaN(), 22.0, 33.0, math.NaN(), 55.0},
+		"cpu_2-windows": {10.0, 20.0, math.NaN(), 40.0, 50.0},
+		"fake-windows":  {math.NaN(), math.NaN(), math.NaN(), math.NaN(), math.NaN()},
 	}
 
 	testParams := tsdbtest.NewTestParams(suite.T(),
@@ -1263,23 +1267,22 @@ func (suite *testSelectDataframeSuite) TestSparseNumericColumnsWithNotExistingMe
 	iter, err := querierV2.SelectDataFrame(params)
 	requireCtx.NoError(err, "failed to execute query")
 
-	rowId := -1
 	var seriesCount int
 	for iter.NextFrame() {
 		seriesCount++
 		frame, err := iter.GetFrame()
 		requireCtx.NoError(err)
 		indexCol := frame.Indices()[0]
-
+		osLabel := frame.Labels()["os"]
 		nullValuesMap := frame.NullValuesMap()
 		requireCtx.NotNil(nullValuesMap, "null value map should not be empty")
 
 		for i := 0; i < indexCol.Len(); i++ {
-			rowId++
 			t, _ := indexCol.TimeAt(i)
 			timeMillis := t.UnixNano() / int64(time.Millisecond)
 			requireCtx.Equal(expectedTimeColumn[i], timeMillis, "time column does not match at index %d", i)
 			for _, columnName := range frame.Names() {
+				key := fmt.Sprintf("%v-%v", columnName, osLabel)
 				var v interface{}
 				column, err := frame.Column(columnName)
 				requireCtx.NoError(err)
@@ -1288,7 +1291,8 @@ func (suite *testSelectDataframeSuite) TestSparseNumericColumnsWithNotExistingMe
 					if v == math.NaN() {
 						requireCtx.True(nullValuesMap[i].NullColumns[columnName])
 					}
-					bothNaN := math.IsNaN(expectedColumns[column.Name()][i].(float64)) && math.IsNaN(v.(float64))
+
+					bothNaN := math.IsNaN(expectedColumns[key][i].(float64)) && math.IsNaN(v.(float64))
 					if bothNaN {
 						continue
 					}
@@ -1301,9 +1305,9 @@ func (suite *testSelectDataframeSuite) TestSparseNumericColumnsWithNotExistingMe
 					suite.Fail(fmt.Sprintf("column type is not as expected: %v", column.DType()))
 				}
 
-				expectedValue := expectedColumns[columnName][rowId]
+				expectedValue := expectedColumns[key][i]
 				if !math.IsNaN(expectedValue.(float64)) || !math.IsNaN(v.(float64)) {
-					requireCtx.Equal(expectedValue, v, "column %v does not match at index %d", columnName, rowId)
+					requireCtx.Equal(expectedValue, v, "column %v does not match at index %d", columnName, i)
 				}
 			}
 		}

Diff for: pkg/pquerier/pqueriertest/dataframe_query_integration_test.go

@@ -3,6 +3,7 @@
 package pqueriertest
 
 import (
+	"github.com/v3io/v3io-tsdb/pkg/aggregate"
 	"testing"
 	"time"
 
@@ -151,3 +152,72 @@ func (suite *testVariantTypeSuite) TestVariantTypeQueryWithSeries() {
 
 	assert.Equal(suite.T(), 1, seriesCount, "series count didn't match expected")
 }
+
+func (suite *testVariantTypeSuite) TestCountAggregationForVariantTypeQueryWithSeries() {
+	adapter, err := tsdb.NewV3ioAdapter(suite.v3ioConfig, nil, nil)
+	if err != nil {
+		suite.T().Fatalf("failed to create v3io adapter. reason: %s", err)
+	}
+
+	metricName := "log"
+	labels := utils.LabelsFromStringList("os", "linux", "__name__", metricName)
+
+	dataToIngest := []string{"a", "b", "c", "d", "e", "f"}
+	numberOfEvents := len(dataToIngest)
+	var expectedTimeColumn []int64
+	for i := 0; i < numberOfEvents; i++ {
+		expectedTimeColumn = append(expectedTimeColumn, suite.basicQueryTime+int64(i)*tsdbtest.MinuteInMillis)
+	}
+
+	expected := map[string][]tsdbtest.DataPoint{"count": {{Time: suite.basicQueryTime - 5*tsdbtest.MinuteInMillis, Value: numberOfEvents}}}
+
+	appender, err := adapter.Appender()
+	if err != nil {
+		suite.T().Fatalf("failed to create v3io appender. reason: %s", err)
+	}
+
+	ref, err := appender.Add(labels, expectedTimeColumn[0], dataToIngest[0])
+	if err != nil {
+		suite.T().Fatalf("Failed to add data to the TSDB appender. Reason: %s", err)
+	}
+	for i := 1; i < numberOfEvents; i++ {
+		appender.AddFast(labels, ref, expectedTimeColumn[i], dataToIngest[i])
+	}
+
+	if _, err := appender.WaitForCompletion(0); err != nil {
+		suite.T().Fatalf("Failed to wait for TSDB append completion. Reason: %s", err)
+	}
+
+	querierV2, err := adapter.QuerierV2()
+	if err != nil {
+		suite.T().Fatalf("Failed to create querier v2, err: %v", err)
+	}
+
+	params := &pquerier.SelectParams{
+		From:      suite.basicQueryTime - tsdbtest.DaysInMillis,
+		To:        suite.basicQueryTime + tsdbtest.DaysInMillis,
+		Functions: "count",
+		Step:      10 * tsdbtest.MinuteInMillis}
+
+	set, err := querierV2.Select(params)
+	if err != nil {
+		suite.T().Fatalf("Failed to exeute query, err: %v", err)
+	}
+
+	var seriesCount int
+	for set.Next() {
+		seriesCount++
+		iter := set.At().Iterator()
+
+		data, err := tsdbtest.IteratorToSlice(iter)
+		if err != nil {
+			suite.T().Fatal(err)
+		}
+		labels := set.At().Labels()
+		agg := labels.Get(aggregate.AggregateLabel)
+
+		suite.compareSingleMetricWithAggregator(data, expected, agg)
+	}
+
+	assert.Equal(suite.T(), 1, seriesCount, "series count didn't match expected")
+}