[query] Fix temporal function tag application bug

Makefile

@@ -332,11 +332,11 @@ define SUBDIR_RULES
 # bundle.
 # ifeq ($(SUBDIR), kube)
 
-# Builds the single kube bundle from individual manifest files.
+# Builds the single kube bundle from individual manifest files. 
 # all-gen-kube: install-tools
 # 	@echo "--- Generating kube bundle"
 # 	@./kube/scripts/build_bundle.sh
-# 	find kube -name '*.yaml' -print0 | PATH=$(combined_bin_paths):$(PATH) xargs -0 kubeval -v=1.12.0
+# 	find kube -name '*.yaml' -print0 | PATH=$(combined_bin_paths):$(PATH) xargs -0 kubeval -v=1.12.0 
 
 # else
 

src/cmd/services/m3comparator/main/main.go

@@ -36,24 +36,39 @@ import (
 	"go.uber.org/zap"
 )
 
-func main() {
-	var (
-		iterPools = pools.BuildIteratorPools(
-			pools.BuildIteratorPoolsOptions{})
-		poolWrapper = pools.NewPoolsWrapper(iterPools)
+var (
+	iterPools = pools.BuildIteratorPools(
+		pools.BuildIteratorPoolsOptions{})
+	poolWrapper = pools.NewPoolsWrapper(iterPools)
 
-		iOpts      = instrument.NewOptions()
-		logger     = iOpts.Logger()
-		tagOptions = models.NewTagOptions()
+	iOpts      = instrument.NewOptions()
+	logger     = iOpts.Logger()
+	tagOptions = models.NewTagOptions()
 
-		encoderPoolOpts = pool.NewObjectPoolOptions()
-		encoderPool     = encoding.NewEncoderPool(encoderPoolOpts)
-	)
+	encoderPoolOpts = pool.NewObjectPoolOptions()
+	encoderPool     = encoding.NewEncoderPool(encoderPoolOpts)
+
+	checkedBytesPool pool.CheckedBytesPool
+	encodingOpts     encoding.Options
+)
+
+func init() {
+	buckets := []pool.Bucket{{Capacity: 10, Count: 10}}
+	newBackingBytesPool := func(s []pool.Bucket) pool.BytesPool {
+		return pool.NewBytesPool(s, nil)
+	}
+
+	checkedBytesPool = pool.NewCheckedBytesPool(buckets, nil, newBackingBytesPool)
+	checkedBytesPool.Init()
+
+	encodingOpts = encoding.NewOptions().SetEncoderPool(encoderPool).SetBytesPool(checkedBytesPool)
 
 	encoderPool.Init(func() encoding.Encoder {
-		return m3tsz.NewEncoder(time.Time{}, nil, true, encoding.NewOptions())
+		return m3tsz.NewEncoder(time.Time{}, nil, true, encodingOpts)
 	})
+}
 
+func main() {
 	opts := iteratorOptions{
 		encoderPool:   encoderPool,
 		iteratorPools: iterPools,

src/cmd/services/m3comparator/main/parser/parser.go

@@ -21,8 +21,10 @@
 package parser
 
 import (
+	"encoding/json"
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -45,10 +47,35 @@ type Datapoints []Datapoint
 
 // Datapoint is a JSON serializeable datapoint for the series.
 type Datapoint struct {
-	Value     float64   `json:"val"`
+	Value     Value     `json:"val"`
 	Timestamp time.Time `json:"ts"`
 }
 
+// Value is a JSON serizlizable float64 that allows NaNs.
+type Value float64
+
+// MarshalJSON returns state as the JSON encoding of a Value.
+func (v Value) MarshalJSON() ([]byte, error) {
+	return json.Marshal(fmt.Sprintf("%g", float64(v)))
+}
+
+// UnmarshalJSON unmarshals JSON-encoded data into a Value.
+func (v *Value) UnmarshalJSON(data []byte) error {
+	var str string
+	err := json.Unmarshal(data, &str)
+	if err != nil {
+		return err
+	}
+
+	f, err := strconv.ParseFloat(str, 64)
+	if err != nil {
+		return err
+	}
+
+	*v = Value(f)
+	return nil
+}
+
 // IDOrGenID gets the ID for this result.
 func (r *Series) IDOrGenID() string {
 	if len(r.id) == 0 {

src/cmd/services/m3comparator/main/parser/parser_test.go

@@ -0,0 +1,59 @@
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The abovale copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package parser
+
+import (
+	"encoding/json"
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestElectionStateJSONMarshal(t *testing.T) {
+	for _, input := range []struct {
+		val Value
+		str string
+	}{
+		{val: 1231243.123123, str: `"1.231243123123e+06"`},
+		{val: 0.000000001, str: `"1e-09"`},
+		{val: Value(math.NaN()), str: `"NaN"`},
+		{val: Value(math.Inf(1)), str: `"+Inf"`},
+		{val: Value(math.Inf(-1)), str: `"-Inf"`},
+	} {
+		b, err := json.Marshal(input.val)
+		require.NoError(t, err)
+		assert.Equal(t, input.str, string(b))
+
+		var val Value
+		json.Unmarshal([]byte(input.str), &val)
+		if math.IsNaN(float64(input.val)) {
+			assert.True(t, math.IsNaN(float64(val)))
+		} else if math.IsInf(float64(input.val), 1) {
+			assert.True(t, math.IsInf(float64(val), 1))
+		} else if math.IsInf(float64(input.val), -1) {
+			assert.True(t, math.IsInf(float64(val), -1))
+		} else {
+			assert.Equal(t, input.val, val)
+		}
+	}
+}

src/cmd/services/m3comparator/main/querier.go

@@ -35,6 +35,7 @@ import (
 	"github.com/m3db/m3/src/query/block"
 	"github.com/m3db/m3/src/query/storage"
 	"github.com/m3db/m3/src/query/storage/m3"
+	xtime "github.com/m3db/m3/src/x/time"
 )
 
 var _ m3.Querier = (*querier)(nil)
@@ -62,9 +63,11 @@ func generateSeriesBlock(
 	numPoints := int(blockSize / resolution)
 	dps := make(seriesBlock, 0, numPoints)
 	for i := 0; i < numPoints; i++ {
+		stamp := start.Add(resolution * time.Duration(i))
 		dp := ts.Datapoint{
-			Timestamp: start.Add(resolution * time.Duration(i)),
-			Value:     rand.Float64(),
+			Timestamp:      stamp,
+			TimestampNanos: xtime.ToUnixNano(stamp),
+			Value:          rand.Float64(),
 		}
 
 		dps = append(dps, dp)
@@ -116,13 +119,17 @@ func (q *querier) FetchCompressed(
 	name := q.opts.tagOptions.MetricName()
 	for _, matcher := range query.TagMatchers {
 		if bytes.Equal(name, matcher.Name) {
-			iters = q.handler.getSeriesIterators(string(matcher.Value))
+			iters, err = q.handler.getSeriesIterators(string(matcher.Value))
+			if err != nil {
+				return m3.SeriesFetchResult{}, noop, err
+			}
+
 			break
 		}
 	}
 
 	if iters == nil || iters.Len() == 0 {
-		iters, err = q.genIters(query)
+		iters, err = q.generateRandomIters(query)
 		if err != nil {
 			return m3.SeriesFetchResult{}, noop, err
 		}
@@ -139,7 +146,7 @@ func (q *querier) FetchCompressed(
 	}, cleanup, nil
 }
 
-func (q *querier) genIters(
+func (q *querier) generateRandomIters(
 	query *storage.FetchQuery,
 ) (encoding.SeriesIterators, error) {
 	var (

src/cmd/services/m3comparator/main/series_load_handler.go

@@ -76,32 +76,37 @@ func (l *seriesLoadHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 func (l *seriesLoadHandler) getSeriesIterators(
-	name string) encoding.SeriesIterators {
+	name string) (encoding.SeriesIterators, error) {
 	l.RLock()
 	defer l.RUnlock()
 
 	logger := l.iterOpts.iOpts.Logger()
 	seriesMap, found := l.nameIDSeriesMap[name]
 	if !found || len(seriesMap.series) == 0 {
-		return nil
+		return nil, nil
 	}
 
 	iters := make([]encoding.SeriesIterator, 0, len(seriesMap.series))
 	for _, series := range seriesMap.series {
 		encoder := l.iterOpts.encoderPool.Get()
 		dps := series.Datapoints
-		encoder.Reset(time.Now(), len(dps), nil)
+		startTime := time.Time{}
+		if len(dps) > 0 {
+			startTime = dps[0].Timestamp.Truncate(time.Hour)
+		}
+
+		encoder.Reset(startTime, len(dps), nil)
 		for _, dp := range dps {
 			err := encoder.Encode(ts.Datapoint{
 				Timestamp:      dp.Timestamp,
-				Value:          dp.Value,
+				Value:          float64(dp.Value),
 				TimestampNanos: xtime.ToUnixNano(dp.Timestamp),
-			}, xtime.Second, nil)
+			}, xtime.Nanosecond, nil)
 
 			if err != nil {
 				encoder.Close()
 				logger.Error("error encoding datapoints", zap.Error(err))
-				return nil
+				return nil, err
 			}
 		}
 
@@ -138,7 +143,7 @@ func (l *seriesLoadHandler) getSeriesIterators(
 	return encoding.NewSeriesIterators(
 		iters,
 		l.iterOpts.iteratorPools.MutableSeriesIterators(),
-	)
+	), nil
 }
 
 func calculateSeriesRange(seriesList []parser.Series) (time.Time, time.Time) {