[query] Refactor block iterators, fix unconsolidated series iterator

src/query/README.md

@@ -89,6 +89,7 @@ Users can validate m3query's Prometheus results for a given query against input
 Query: `go_gc_duration_seconds{quantile="1"} * 2`
 
 Input JSON:
+
 ```
 {
   "input": {
@@ -145,6 +146,7 @@ Input JSON:
   }
 }
 ```
+
 Full request: `curl -X POST 'localhost:7201/api/v1/debug/validate_query?start=1543431465&end=1543435045&step=14s&query=go_gc_duration_seconds%7Bquantile%3D%221%22%7D*2' -d @<input_json_file> --header "Content-Type: application/json"`
 
 [doc-img]: https://godoc.org/github.com/m3db/m3/src/query?status.svg

src/query/api/v1/handler/prometheus/native/read.go

@@ -143,7 +143,10 @@ func (h *PromReadHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 // ServeHTTPWithEngine returns query results from the storage
-func (h *PromReadHandler) ServeHTTPWithEngine(w http.ResponseWriter, r *http.Request, engine *executor.Engine) ([]*ts.Series, models.RequestParams, *RespError) {
+func (h *PromReadHandler) ServeHTTPWithEngine(
+	w http.ResponseWriter,
+	r *http.Request, engine *executor.Engine,
+) ([]*ts.Series, models.RequestParams, *RespError) {
 	ctx := context.WithValue(r.Context(), handler.HeaderKey, r.Header)
 	logger := logging.WithContext(ctx)
 

src/query/api/v1/handler/prometheus/native/read_common.go

@@ -58,7 +58,6 @@ func read(
 	// Results is closed by execute
 	results := make(chan executor.Query)
 	go engine.ExecuteExpr(ctx, parser, opts, params, results)
-
 	// Block slices are sorted by start time
 	// TODO: Pooling
 	sortedBlockList := make([]blockWithMeta, 0, initialBlockAlloc)
@@ -143,11 +142,6 @@ func sortedBlocksToSeriesList(blockList []blockWithMeta) ([]*ts.Series, error) {
 	}
 
 	firstBlock := blockList[0].block
-	firstStepIter, err := firstBlock.StepIter()
-	if err != nil {
-		return nil, err
-	}
-
 	firstSeriesIter, err := firstBlock.SeriesIter()
 	if err != nil {
 		return nil, err
@@ -171,22 +165,35 @@ func sortedBlocksToSeriesList(blockList []blockWithMeta) ([]*ts.Series, error) {
 		seriesIters[i] = seriesIter
 	}
 
-	numValues := firstStepIter.StepCount() * len(blockList)
+	numValues := 0
+	for _, block := range blockList {
+		b, err := block.block.StepIter()
+		if err != nil {
+			return nil, err
+		}
+
+		numValues += b.StepCount()
+	}
+
 	for i := 0; i < numSeries; i++ {
 		values := ts.NewFixedStepValues(bounds.StepSize, numValues, math.NaN(), bounds.Start)
 		valIdx := 0
 		for idx, iter := range seriesIters {
 			if !iter.Next() {
+				if err = iter.Err(); err != nil {
+					return nil, err
+				}
+
 				return nil, fmt.Errorf("invalid number of datapoints for series: %d, block: %d", i, idx)
 			}
 
-			blockSeries, err := iter.Current()
-			if err != nil {
+			if err = iter.Err(); err != nil {
 				return nil, err
 			}
 
-			for i := 0; i < blockSeries.Len(); i++ {
-				values.SetValueAt(valIdx, blockSeries.ValueAtStep(i))
+			blockSeries := iter.Current()
+			for j := 0; j < blockSeries.Len(); j++ {
+				values.SetValueAt(valIdx, blockSeries.ValueAtStep(j))
 				valIdx++
 			}
 		}
@@ -224,27 +231,18 @@ func insertSortedBlock(
 			"the block, wanted: %d, found: %d", seriesCount, blockSeriesCount)
 	}
 
-	blockStepIter, err := b.StepIter()
-	if err != nil {
-		return nil, err
-	}
-
-	blockStepCount := blockStepIter.StepCount()
-	if stepCount != blockStepCount {
-		return nil, fmt.Errorf("mismatch in number of steps for the"+
-			"block, wanted: %d, found: %d", stepCount, blockStepCount)
-	}
-
 	// Binary search to keep the start times sorted
 	index := sort.Search(len(blockList), func(i int) bool {
-		return blockList[i].meta.Bounds.Start.Before(blockMeta.Bounds.Start)
+		return blockList[i].meta.Bounds.Start.After(blockMeta.Bounds.Start)
 	})
+
 	// Append here ensures enough size in the slice
 	blockList = append(blockList, blockWithMeta{})
 	copy(blockList[index+1:], blockList[index:])
 	blockList[index] = blockWithMeta{
 		block: b,
 		meta:  blockMeta,
 	}
+
 	return blockList, nil
 }

src/query/block/column.go

@@ -36,17 +36,14 @@ type columnBlock struct {
 	seriesMeta []SeriesMeta
 }
 
-// Unconsolidated returns the unconsolidated version for the block
 func (c *columnBlock) Unconsolidated() (UnconsolidatedBlock, error) {
 	return nil, fmt.Errorf("unconsolidated view not supported for block, meta: %s", c.meta)
 }
 
-// Meta returns the metadata for the block
 func (c *columnBlock) Meta() Metadata {
 	return c.meta
 }
 
-// StepIter returns a StepIterator
 func (c *columnBlock) StepIter() (StepIter, error) {
 	if len(c.columns) != c.meta.Bounds.Steps() {
 		return nil, fmt.Errorf("mismatch in block columns and meta bounds, columns: %d, bounds: %v", len(c.columns), c.meta.Bounds)
@@ -61,7 +58,6 @@ func (c *columnBlock) StepIter() (StepIter, error) {
 }
 
 // TODO: allow series iteration
-// SeriesIter returns a SeriesIterator
 func (c *columnBlock) SeriesIter() (SeriesIter, error) {
 	return newColumnBlockSeriesIter(c.columns, c.meta, c.seriesMeta), nil
 }
@@ -78,12 +74,10 @@ func (c *columnBlock) WithMetadata(
 }
 
 // TODO: allow series iteration
-// SeriesMeta returns the metadata for each series in the block
 func (c *columnBlock) SeriesMeta() []SeriesMeta {
 	return c.seriesMeta
 }
 
-// StepCount returns the total steps
 func (c *columnBlock) StepCount() int {
 	return len(c.columns)
 }
@@ -95,10 +89,12 @@ func (c *columnBlock) Close() error {
 }
 
 type colBlockIter struct {
-	columns    []column
-	seriesMeta []SeriesMeta
-	meta       Metadata
-	idx        int
+	idx         int
+	timeForStep time.Time
+	err         error
+	meta        Metadata
+	seriesMeta  []SeriesMeta
+	columns     []column
 }
 
 func (c *colBlockIter) SeriesMeta() []SeriesMeta {
@@ -113,30 +109,38 @@ func (c *colBlockIter) StepCount() int {
 	return len(c.columns)
 }
 
-// Next returns true if iterator has more values remaining
 func (c *colBlockIter) Next() bool {
+	if c.err != nil {
+		return false
+	}
+
 	c.idx++
-	return c.idx < len(c.columns)
-}
+	next := c.idx < len(c.columns)
+	if !next {
+		return false
+	}
 
-// Current returns the current step
-func (c *colBlockIter) Current() (Step, error) {
-	col := c.columns[c.idx]
-	t, err := c.meta.Bounds.TimeForIndex(c.idx)
-	// TODO: Test panic case
-	if err != nil {
-		panic(err)
+	c.timeForStep, c.err = c.meta.Bounds.TimeForIndex(c.idx)
+	if c.err != nil {
+		return false
 	}
 
+	return next
+}
+
+func (c *colBlockIter) Err() error {
+	return c.err
+}
+
+func (c *colBlockIter) Current() Step {
+	col := c.columns[c.idx]
 	return ColStep{
-		time:   t,
+		time:   c.timeForStep,
 		values: col.Values,
-	}, nil
+	}
 }
 
-// Close frees up resources
-func (c *colBlockIter) Close() {
-}
+func (c *colBlockIter) Close() { /*no-op*/ }
 
 // ColStep is a single column containing data from multiple series at a given time step
 type ColStep struct {
@@ -210,18 +214,29 @@ type column struct {
 
 // columnBlockSeriesIter is used to iterate over a column. Assumes that all columns have the same length
 type columnBlockSeriesIter struct {
-	columns    []column
 	idx        int
 	blockMeta  Metadata
+	values     []float64
+	columns    []column
 	seriesMeta []SeriesMeta
 }
 
 func (m *columnBlockSeriesIter) Meta() Metadata {
 	return m.blockMeta
 }
 
-func newColumnBlockSeriesIter(columns []column, blockMeta Metadata, seriesMeta []SeriesMeta) SeriesIter {
-	return &columnBlockSeriesIter{columns: columns, blockMeta: blockMeta, seriesMeta: seriesMeta, idx: -1}
+func newColumnBlockSeriesIter(
+	columns []column,
+	blockMeta Metadata,
+	seriesMeta []SeriesMeta,
+) SeriesIter {
+	return &columnBlockSeriesIter{
+		columns:    columns,
+		blockMeta:  blockMeta,
+		seriesMeta: seriesMeta,
+		idx:        -1,
+		values:     make([]float64, len(columns)),
+	}
 }
 
 func (m *columnBlockSeriesIter) SeriesMeta() []SeriesMeta {
@@ -237,19 +252,31 @@ func (m *columnBlockSeriesIter) SeriesCount() int {
 	return len(cols[0].Values)
 }
 
+func (m *columnBlockSeriesIter) Err() error {
+	// no-op
+	return nil
+}
+
 func (m *columnBlockSeriesIter) Next() bool {
 	m.idx++
-	return m.idx < m.SeriesCount()
-}
+	next := m.idx < m.SeriesCount()
+	if !next {
+		return false
+	}
 
-func (m *columnBlockSeriesIter) Current() (Series, error) {
 	cols := m.columns
-	values := make([]float64, len(cols))
 	for i := 0; i < len(cols); i++ {
-		values[i] = cols[i].Values[m.idx]
+		m.values[i] = cols[i].Values[m.idx]
 	}
 
-	return NewSeries(values, m.seriesMeta[m.idx]), nil
+	return next
+}
+
+func (m *columnBlockSeriesIter) Current() Series {
+	// TODO: pool these
+	vals := make([]float64, len(m.values))
+	copy(vals, m.values)
+	return NewSeries(vals, m.seriesMeta[m.idx])
 }
 
 // TODO: Actually free resources once we do pooling