Comments

src/query/block/types.go

@@ -158,11 +158,13 @@ type Result struct {
 // ConsolidationFunc consolidates a bunch of datapoints into a single float value
 type ConsolidationFunc func(datapoints ts.Datapoints) float64
 
-// TakeLast is a consolidation function which takes the last datapoint
+// TakeLast is a consolidation function which takes the last datapoint which has non nan value
 func TakeLast(values ts.Datapoints) float64 {
-	if len(values) == 0 {
-		return math.NaN()
+	for i := len(values) - 1; i >= 0; i-- {
+		if !math.IsNaN(values[i].Value) {
+			return values[i].Value
+		}
 	}
 
-	return values[len(values)-1].Value
+	return math.NaN()
 }

src/query/executor/transform/lazy.go

@@ -137,19 +137,23 @@ func (s *seriesIter) Next() bool {
 }
 
 type lazyBlock struct {
-	mu sync.Mutex
-
+	mu             sync.Mutex
 	rawBlock       block.Block
 	lazyNode       *lazyNode
 	ID             parser.NodeID
 	processedBlock block.Block
+	processError   error
 }
 
 // Unconsolidated returns the unconsolidated version for the block
 func (f *lazyBlock) Unconsolidated() (block.UnconsolidatedBlock, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 
+	if f.processError != nil {
+		return nil, f.processError
+	}
+
 	if f.processedBlock != nil {
 		return f.processedBlock.Unconsolidated()
 	}
@@ -165,6 +169,10 @@ func (f *lazyBlock) StepIter() (block.StepIter, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 
+	if f.processError != nil {
+		return nil, f.processError
+	}
+
 	if f.processedBlock != nil {
 		return f.processedBlock.StepIter()
 	}
@@ -194,6 +202,10 @@ func (f *lazyBlock) SeriesIter() (block.SeriesIter, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 
+	if f.processError != nil {
+		return nil, f.processError
+	}
+
 	if f.processedBlock != nil {
 		return f.processedBlock.SeriesIter()
 	}
@@ -226,6 +238,7 @@ func (f *lazyBlock) Close() error {
 func (f *lazyBlock) process() error {
 	err := f.lazyNode.fNode.Process(f.ID, f.rawBlock)
 	if err != nil {
+		f.processError = err
 		return err
 	}
 

src/query/functions/temporal/base.go

@@ -111,7 +111,7 @@ func (c *baseNode) Process(ID parser.NodeID, b block.Block) error {
 	}
 
 	if unconsolidatedBlock == nil {
-		return fmt.Errorf("block needs to be unconsolidated for the op: %s", c.op)
+		return fmt.Errorf("block needs to be unconsolidated for temporal operations: %s", c.op)
 	}
 
 	iter, err := unconsolidatedBlock.StepIter()
@@ -304,15 +304,20 @@ func (c *baseNode) processSingleRequest(request processRequest) error {
 			// TODO: Consider using a rotating slice since this is inefficient
 			if desiredLength <= len(values) {
 				values = values[len(values)-desiredLength:]
-				flattenedValues := make(ts.Datapoints, 0)
+				flattenedValues := make(ts.Datapoints, 0, len(values))
 				for _, dps := range values {
-					for _, dp := range dps {
-						if dp.Timestamp.Before(oldestDatapointTimestamp) {
-							continue
+					var (
+						idx int
+						dp  ts.Datapoint
+					)
+					for idx, dp = range dps {
+						// Keep going until we find datapoints at the oldest timestamp for window
+						if !dp.Timestamp.Before(oldestDatapointTimestamp) {
+							break
 						}
-
-						flattenedValues = append(flattenedValues, dp)
 					}
+
+					flattenedValues = append(flattenedValues, dps[idx:]...)
 				}
 
 				newVal = c.processor.Process(flattenedValues)

src/query/storage/block.go

@@ -49,41 +49,41 @@ func NewMultiBlockWrapper(unconsolidatedBlock block.UnconsolidatedBlock) block.B
 }
 
 type multiBlockWrapper struct {
-	unconsolidated block.UnconsolidatedBlock
-	consolidated   block.Block
-	mu             sync.Mutex
+	unconsolidated   block.UnconsolidatedBlock
+	consolidated     block.Block
+	consolidateError error
+	once             sync.Once
 }
 
 func (m *multiBlockWrapper) Unconsolidated() (block.UnconsolidatedBlock, error) {
 	return m.unconsolidated, nil
 }
 
 func (m *multiBlockWrapper) StepIter() (block.StepIter, error) {
-	if err := m.tryConsolidate(); err != nil {
+	if err := m.consolidate(); err != nil {
 		return nil, err
 	}
 
 	return m.consolidated.StepIter()
 }
 
-func (m *multiBlockWrapper) tryConsolidate() error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	if m.consolidated == nil {
+func (m *multiBlockWrapper) consolidate() error {
+	m.once.Do(func() {
 		consolidated, err := m.unconsolidated.Consolidate()
 		if err != nil {
-			return err
+			m.consolidateError = err
+			return
 		}
 
 		m.consolidated = consolidated
-	}
 
-	return nil
+	})
+
+	return m.consolidateError
 }
 
 func (m *multiBlockWrapper) SeriesIter() (block.SeriesIter, error) {
-	if err := m.tryConsolidate(); err != nil {
+	if err := m.consolidate(); err != nil {
 		return nil, err
 	}
 

src/query/ts/values.go

@@ -84,9 +84,10 @@ func (d Datapoints) AlignToBounds(bounds models.Bounds) []Datapoints {
 	steps := bounds.Steps()
 	stepValues := make([]Datapoints, steps)
 	dpIdx := 0
+	stepSize := bounds.StepSize
+	t := bounds.Start
 	for i := 0; i < steps; i++ {
 		startIdx := dpIdx
-		t, _ := bounds.TimeForIndex(i)
 		for dpIdx < numDatapoints && d[dpIdx].Timestamp.Before(t) {
 			dpIdx++
 		}
@@ -97,6 +98,7 @@ func (d Datapoints) AlignToBounds(bounds models.Bounds) []Datapoints {
 		}
 
 		stepValues[i] = singleStepValues
+		t = t.Add(stepSize)
 	}
 
 	return stepValues