Add StepIter to blocks #778

benraskin92 · 2018-07-02T20:53:59Z

No description provided.

codecov · 2018-07-02T23:55:48Z

Codecov Report

Merging #778 into master will decrease coverage by 0.14%.
The diff coverage is 67.21%.

@@            Coverage Diff             @@
##           master     #778      +/-   ##
==========================================
- Coverage   78.08%   77.94%   -0.15%     
==========================================
  Files         364      367       +3     
  Lines       31058    31238     +180     
==========================================
+ Hits        24253    24348      +95     
- Misses       5185     5250      +65     
- Partials     1620     1640      +20

Flag	Coverage Δ
#coordinator	`60.11% <67.21%> (+0.06%)`	⬆️
#dbnode	`81.66% <ø> (-0.07%)`	⬇️
#m3ninx	`72.7% <ø> (ø)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update f94b19a...6f11dc6. Read the comment docs.

arnikola · 2018-07-03T14:54:37Z

src/coordinator/storage/local/storage.go

 	if err != nil {
 		return block.Result{}, err
 	}

-	res, err := storage.FetchResultToBlockResult(fetchResult, query)
+	iterAlloc := func(r io.Reader) encoding.ReaderIterator {


Pull this out into an init{} block since we only need a single iterAlloc.

You may also want to populate this with iterator pools from m3db like this too (should be able to use encoding.IteratorPools() to get the pools out):

encodingOpts := encoding.NewOptions(). SetEncoderPool(encoderPool). SetReaderIteratorPool(iteratorPool). SetBytesPool(bytesPool). SetSegmentReaderPool(segmentReaderPool)

arnikola · 2018-07-03T15:00:32Z

src/coordinator/storage/local/storage.go

+	iterAlloc := func(r io.Reader) encoding.ReaderIterator {
+		return m3tsz.NewReaderIterator(r, m3tsz.DefaultIntOptimizationEnabled, encoding.NewOptions())
+	}
+	sliceOfSeriesBlocks, err := m3db.ConvertM3DBSeriesIterators(seriesIters, iterAlloc)
 	if err != nil {
 		return block.Result{}, err


nit: var emptyBlock = block.Result{} outside the function, then return emptyBlock instead of block.Result{}

arnikola · 2018-07-03T15:02:29Z

src/coordinator/storage/local/storage.go

+		return block.Result{}, err
+	}
+
+	var res block.Result


nit

return block.Result { Blocks: multiSeriesBlocks, }, nil

arnikola · 2018-07-03T15:03:24Z

src/coordinator/ts/m3db/block_iteration.go

+
+// StepIter creates a new step iterator for a given MultiSeriesBlock
+func (m MultiSeriesBlock) StepIter() block.StepIter {
+	return &multiSeriesBlockStepIter{seriesIters: newConsolidatedSeriesBlockIters(m.Blocks),


nit: would be easier to read with a newline between { and seriesIters

arnikola · 2018-07-03T15:15:53Z

src/coordinator/ts/m3db/block_iteration.go

+	for i, block := range blocks {
+		consolidatedNSBlockIters := make([]*consolidatedNSBlockIter, len(blocks[0].ConsolidatedNSBlocks))
+		for j, jblock := range block.ConsolidatedNSBlocks {
+


nit newline

arnikola · 2018-07-03T18:14:09Z

src/coordinator/ts/m3db/block_iteration_test.go

+	for i, expectedSlice := range expectedResults {
+		for j, exp := range expectedSlice {
+			if math.IsNaN(exp) {
+				require.True(t, math.IsNaN(actualResults[i][j]))


nit: these can be asserts

arnikola · 2018-07-03T18:20:38Z

src/coordinator/ts/m3db/blocks.go

@@ -33,7 +36,13 @@ var (
 // SeriesBlockToMultiSeriesBlocks converts M3DB blocks to multi series blocks
 func SeriesBlockToMultiSeriesBlocks(multiNamespaceSeriesList []MultiNamespaceSeries, seriesIteratorsPool encoding.MutableSeriesIteratorsPool) (MultiSeriesBlocks, error) {


nit: split this with newlines, it's pretty long

arnikola · 2018-07-03T18:23:55Z

src/coordinator/ts/m3db/blocks.go

+		commonTags      = make(map[string]string)
+		firstSeriesTags = make(map[string]string)
+	)
+
 	for multiNamespaceSeriesIdx, multiNamespaceSeries := range multiNamespaceSeriesList {


nit: multiNamespaceSeriesIdx is a bit verbose for an index we're only using to check if we're on the first run through the loop

arnikola · 2018-07-03T18:29:57Z

src/coordinator/ts/m3db/blocks.go

@@ -44,13 +53,34 @@ func SeriesBlockToMultiSeriesBlocks(multiNamespaceSeriesList []MultiNamespaceSer
 		// MultiSeriesBlocks list with the proper size
 		if multiNamespaceSeriesIdx == 0 {
 			multiSeriesBlocks = make(MultiSeriesBlocks, len(consolidatedSeriesBlocks))
+			commonTags, err = storage.FromIdentTagIteratorToTags(consolidatedSeriesBlocks[0].ConsolidatedNSBlocks[0].SeriesIterators.Iters()[0].Tags())


This is unsafe. Can't you just get tags out of the mutliNamespaceSeries?

arnikola · 2018-07-03T18:46:48Z

src/coordinator/ts/m3db/blocks.go

+			Bounds: block.Bounds{
+				Start:    seriesBlock.start,
+				End:      seriesBlock.end,
+				StepSize: time.Minute, // get actual step size!!


nikunjgit · 2018-07-03T21:41:25Z

src/coordinator/storage/local/storage.go

 	if err != nil {
 		return block.Result{}, err
 	}

+	iterAlloc := func(r io.Reader) encoding.ReaderIterator {


do this when you create the storage and keep the function reference ?

nikunjgit · 2018-07-03T21:44:22Z

src/coordinator/storage/local/storage.go

+	iterAlloc := func(r io.Reader) encoding.ReaderIterator {
+		return m3tsz.NewReaderIterator(r, m3tsz.DefaultIntOptimizationEnabled, encoding.NewOptions())
+	}
+	sliceOfSeriesBlocks, err := m3db.ConvertM3DBSeriesIterators(seriesIters, iterAlloc)


call it M3DBIteratorsToSeriesBlocks ?

also the variable can be seriesBlockList

nikunjgit · 2018-07-03T21:46:30Z

src/coordinator/storage/local/storage.go

+	}
+
+	var (
+		res block.Result


res := block.Result{
Blocks: make([]block.Block, len(multiSeriesBlocks))
}

nikunjgit · 2018-07-03T21:48:32Z

src/coordinator/storage/local/storage_test.go

+
+	iter, err := test.BuildTestSeriesIterator()
+	require.NoError(t, err)
+	iterators := encoding.NewSeriesIterators([]encoding.SeriesIterator{iter}, nil)


also add a test with multiple iterators ?

nikunjgit · 2018-07-03T21:50:22Z

src/coordinator/ts/m3db/block_iteration.go

+// // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// // THE SOFTWARE.
+
+package m3db


consider adding a package src/coordinator/ts/m3db/block/ ? A lot of your files can be moved inside it

arnikola · 2018-07-10T13:47:16Z

src/coordinator/ts/m3db/types.go

@@ -88,19 +88,24 @@ type ConsolidatedSeriesBlock struct {
 	consolidationFunc    ConsolidationFunc // nolint
 }

-func (c ConsolidatedSeriesBlock) beyondBounds(multiSeriesBlock MultiSeriesBlock) bool {
-	if c.Metadata.Bounds.Start != multiSeriesBlock.Metadata.Bounds.Start || c.Metadata.Bounds.End != multiSeriesBlock.Metadata.Bounds.End {
+// func (c ConsolidatedSeriesBlock) equalBounds(multiSeriesBlock MultiSeriesBlock) bool {


Delete this?

arnikola · 2018-07-10T13:48:06Z

src/coordinator/ts/m3db/block_iteration.go

 			return true
 		}
+		fmt.Println("seriesIndex: ", c.seriesIndex, "indexTime: ", c.indexTime, "bounds: ", c.bounds)
+		// curr, _, _ := c.consolidatedNSBlockSeriesIters[c.seriesIndex].Current()


arnikola · 2018-07-10T13:48:19Z

src/coordinator/ts/m3db/block_iteration.go

+		lastDP, _, _ = c.consolidatedNSBlockSeriesIters[c.seriesIndex].Current()
+		c.lastDP = lastDP
+	}
+	// if c.indexTime.Add(c.bounds.StepSize).Before(lastDP.Timestamp) {


arnikola · 2018-07-10T13:48:25Z

src/coordinator/ts/m3db/block_iteration.go

 func (c *consolidatedNSBlockIter) Next() bool {
-	if len(c.consolidatesNSBlockSeriesIters) == 0 {
+	c.indexTime = c.indexTime.Add(c.bounds.StepSize)
+	fmt.Println("index time: ", c.indexTime)


Delete println

arnikola · 2018-07-10T13:48:32Z

src/coordinator/ts/m3db/block_iteration.go

+	}
+	c.indexTime = c.indexTime.Add(c.bounds.StepSize)
+
+	fmt.Println("last dp: ", lastDP)


Delete println

arnikola · 2018-07-10T13:53:19Z

src/coordinator/ts/m3db/types.go

-	}
-	return true
-}
+// func (c ConsolidatedNSBlock) beyondBounds(consolidatedSeriesBlock ConsolidatedSeriesBlock) bool {


arnikola · 2018-07-10T13:53:45Z

src/coordinator/ts/m3db/blocks.go

 			multiSeriesBlocks = make(MultiSeriesBlocks, len(consolidatedSeriesBlocks))
-			commonTags, err = storage.FromIdentTagIteratorToTags(consolidatedSeriesBlocks[0].ConsolidatedNSBlocks[0].SeriesIterators.Iters()[0].Tags())
+			commonTags, err = storage.FromIdentTagIteratorToTags(multiNamespaceSeries[0].Tags)
+			// commonTags, err = storage.FromIdentTagIteratorToTags(consolidatedSeriesBlocks[0].ConsolidatedNSBlocks[0].SeriesIterators.Iters()[0].Tags())


arnikola · 2018-07-10T13:53:58Z

src/coordinator/ts/m3db/block_iteration.go

-	}
-	if c.currentTime.Add(c.bounds.StepSize).Before(lastDP.Timestamp) || c.lastVal {
-		return math.NaN()
+		fmt.Println("last dp current: ", c.lastDP)


Delete println

arnikola · 2018-07-10T13:54:46Z

src/coordinator/ts/m3db/block_iteration.go


 	return false
 }

 func (c *consolidatedNSBlockIter) Current() float64 {
-	lastDP := c.lastDP
+	if !c.indexTime.After(c.lastDP.Timestamp) && c.indexTime.Add(c.bounds.StepSize).After(c.lastDP.Timestamp) {


nit: Might read better if you pull out c.indexTime and c.lastDP.Timestamp as variables first

arnikola · 2018-07-10T13:56:56Z

src/coordinator/ts/m3db/blocks.go

-			commonTags, err = storage.FromIdentTagIteratorToTags(consolidatedSeriesBlocks[0].ConsolidatedNSBlocks[0].SeriesIterators.Iters()[0].Tags())
+			commonTags, err = storage.FromIdentTagIteratorToTags(multiNamespaceSeries[0].Tags)
+			// commonTags, err = storage.FromIdentTagIteratorToTags(consolidatedSeriesBlocks[0].ConsolidatedNSBlocks[0].SeriesIterators.Iters()[0].Tags())
+			fmt.Println("tags: ", commonTags)


Delete println

benraskin92 · 2018-07-10T15:19:54Z

I'll get rid of all of the println's and commented out code

arnikola · 2018-07-10T16:51:03Z

src/coordinator/storage/local/storage.go

+func (s *localStorage) fetchRaw(
+	query index.Query,
+	opts index.QueryOptions,
+) (encoding.SeriesIterators, bool, error) { // nolint: unparam


nit: nolint usually goes above the function declaration

arnikola · 2018-07-10T18:23:22Z

src/coordinator/ts/m3db/block/block_conversion.go

-				multiSeriesBlocks[consolidatedSeriesBlockIdx].Start = consolidatedSeriesBlock.Start
-				multiSeriesBlocks[consolidatedSeriesBlockIdx].End = consolidatedSeriesBlock.End
+			if seriesIdx == 0 {
+				multiSeriesBlocks[consolidatedSeriesBlockIdx].Metadata.Bounds.StepSize = consolidatedSeriesBlock.Metadata.Bounds.StepSize


nit: can you make these names shorter? It's quite difficult to read at the moment

Also, can pull out multiSeriesBlocks[consolidatedSeriesBlockIdx].Metadata.Bounds into a var?

nikunjgit · 2018-07-10T16:46:53Z

src/coordinator/storage/local/storage.go

+	query index.Query,
+	opts index.QueryOptions,
+) (encoding.SeriesIterators, bool, error) { // nolint: unparam
+	iters, exhaustive, err := s.session.FetchTagged(s.namespace, query, opts)


just return s.session.FetchTagged(...) ?

Also, wondering why you need this function since you're just calling another function inside

nikunjgit · 2018-07-10T17:15:10Z

src/coordinator/storage/local/storage.go

-	res, err := storage.FetchResultToBlockResult(fetchResult, query)
+	// NB/todo(braskin): because we are only support querying one namespace now, we can just create
+	// a multiNamespaceList with one element. However, once we support querying multiple namespaces,
+	// we will need to append each namespace sliceOfSeriesBlocks to the multiNamespaceList


old comment ? Might need to change the names

nikunjgit · 2018-07-10T17:16:40Z

src/coordinator/storage/local/storage_test.go

@@ -123,6 +125,25 @@ func TestLocalRead(t *testing.T) {
 	assert.Equal(t, tags, results.SeriesList[0].Tags)
 }

+func TestLocalFetchBlocks(t *testing.T) {


also test values ?

nikunjgit · 2018-07-10T17:25:17Z

src/coordinator/ts/m3db/block/block_conversion.go

@@ -18,11 +18,14 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN


just call it convertor.go ? Its already in the block package

nikunjgit · 2018-07-10T17:28:37Z

src/coordinator/ts/m3db/block/block_conversion.go

@@ -31,45 +34,81 @@ var (
 )

 // SeriesBlockToMultiSeriesBlocks converts M3DB blocks to multi series blocks


can you add comments to highlight the flow ?
something like:

m3db block -> consolidated series block (per namespace) -> ....

nikunjgit · 2018-07-10T18:49:25Z

src/coordinator/ts/m3db/block/block_iteration.go

+	return consolidatedSeriesBlockIters
+}
+
+func newConsolidatedNSBlockIter(nsBlock ConsolidatedNSBlock) *consolidatedNSBlockIter {


comment about what you're iterating ?

nikunjgit · 2018-07-10T19:13:04Z

src/coordinator/ts/m3db/block/block_iteration.go

+
+func (c *consolidatedNSBlockIter) Next() bool {
+	c.indexTime = c.indexTime.Add(c.bounds.StepSize)
+	lastDP := c.lastDP


initialize it right before you use it

nikunjgit · 2018-07-10T19:13:40Z

src/coordinator/ts/m3db/block/block_iteration.go

+	return true
+}
+
+func (c *consolidatedNSBlockIter) next() bool {


call it nextIterator or something ?

nikunjgit · 2018-07-10T19:14:06Z

src/coordinator/ts/m3db/block/block_iteration.go

+}
+
+func (c *consolidatedNSBlockIter) Current() float64 {
+	if !c.indexTime.After(c.lastDP.Timestamp) && c.indexTime.Add(c.bounds.StepSize).After(c.lastDP.Timestamp) {


extract lastDP ??

also can you add some comments about what you're checking

nikunjgit · 2018-07-10T19:14:58Z

src/coordinator/ts/m3db/block/block_iteration_test.go

+	return results
+}
+
+func TestStepIteration(t *testing.T) {


probably need more tests, also should write down conditions you are testing. This test is pretty opaque

nikunjgit

Its a bit confusing right now jumping back and forth between iterators. My suggestion is:

block/multiSeries.go
block/consolidatedSeries.go
block/nsSeries.go

You can also split their tests in a similar way.

nikunjgit · 2018-07-12T16:01:16Z

src/coordinator/block/types.go

@@ -104,6 +104,12 @@ type StepIter interface {
 	Meta() Metadata
 }

+// ValueIterator iterates through a generic iterator


better comment ?

nikunjgit · 2018-07-12T16:03:16Z

src/coordinator/ts/m3db/block/convertor.go


-			if !consolidatedSeriesBlock.beyondBounds(multiSeriesBlocks[consolidatedSeriesBlockIdx]) {
+			if !equalBounds(consolidatedSeriesBlock.Metadata.Bounds, multiSeriesBlocks[consolidatedSeriesBlockIdx].Metadata.Bounds) {


define Equals method on the bounds ?

nikunjgit · 2018-07-12T16:03:47Z

src/coordinator/ts/m3db/block/convertor.go

+
+func (m MultiSeriesBlocks) commonTags() map[string]string {
+	commonTags := make(map[string]string)
+	for tag, val := range m[0].Blocks[0].Metadata.Tags {


length checks ?

nikunjgit · 2018-07-12T16:04:36Z

src/coordinator/ts/m3db/block/convertor.go

+}
+
+func (m MultiSeriesBlocks) commonTags() map[string]string {
+	commonTags := make(map[string]string)


should be models.Tags ?

also, might be cleaner to define a Clone() method on models.Tags which returns a copy ?

nikunjgit · 2018-07-12T17:05:17Z

src/coordinator/ts/m3db/block/convertor.go

+	// NB(braskin): all series are the same across MultiSeriesBlocks so only need
+	// to look at one
+	for _, series := range m[0].Blocks {
+		seriesTags := series.Metadata.Tags


skip the first one ?

nikunjgit · 2018-07-12T17:30:04Z

src/coordinator/ts/m3db/block/iterator.go

+	}
+
+	lastDP := c.lastDP
+	for c.indexTime.After(lastDP.Timestamp) && c.nextIterator() {


add a comment

nikunjgit · 2018-07-12T17:34:20Z

src/coordinator/ts/m3db/block/iterator.go

+	}
+
+	for i, seriesBlock := range blocks {
+		consolidatedNSBlockIters := make([]block.ValueIterator, len(blocks[0].ConsolidatedNSBlocks))


move out len(blocks[0].ConsolidatedNSBlocks)

nikunjgit · 2018-07-12T17:39:22Z

src/coordinator/ts/m3db/block/iterator.go

+func (m *multiSeriesBlockStepIter) SeriesMeta() []block.SeriesMeta {
+	metas := make([]block.SeriesMeta, len(m.blocks))
+	for i, s := range m.blocks {
+		// todo(braskin): add name


metas[i].Name = s.Metadata.Tags[models.MetricName] ?

nikunjgit · 2018-07-12T17:39:42Z

src/coordinator/ts/m3db/block/iterator.go

+	if len(m.blocks) == 0 {
+		return 0
+	}
+	return m.blocks[0].Metadata.Bounds.Steps()


FYI: this is inclusive of the end

nikunjgit · 2018-07-12T17:40:43Z

src/coordinator/ts/m3db/block/iterator.go

+	}
+
+	bounds := m.meta.Bounds
+	t := bounds.Start.Add(time.Duration(m.index) * bounds.StepSize)


use bounds.TimeForIndex(m.index)

arnikola · 2018-07-12T15:59:59Z

src/coordinator/block/types.go

@@ -104,6 +104,12 @@ type StepIter interface {
 	Meta() Metadata
 }

+// ValueIterator iterates through a generic iterator


nit ... is a generic float value iterator

arnikola · 2018-07-12T16:00:56Z

src/coordinator/storage/local/storage.go

+		return emptyBlock, err
+	}
+
+	// NB/todo(braskin): because we are only support querying one namespace now, we can just create


Probably worth rebasing on master now that #785 is merged

arnikola · 2018-07-12T16:06:13Z

src/coordinator/storage/local/storage_test.go

+	assert.NoError(t, err)
+	require.NotNil(t, results)
+	require.Len(t, results.Blocks, 2)
+	require.Equal(t, models.Tags{"foo": "bar", "baz": "qux"}, results.Blocks[0].Meta().Tags)


Worth testing results.Blocks[1].Meta().Tags ?

arnikola · 2018-07-12T16:07:59Z

src/coordinator/ts/m3db/block/convertor.go

@@ -18,11 +18,14 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 // THE SOFTWARE.



nit: should be converter.go

arnikola · 2018-07-12T16:09:16Z

src/coordinator/ts/m3db/block/convertor.go

@@ -31,45 +34,84 @@ var (
 )

 // SeriesBlockToMultiSeriesBlocks converts M3DB blocks to multi series blocks
-func SeriesBlockToMultiSeriesBlocks(multiNamespaceSeriesList []MultiNamespaceSeries, seriesIteratorsPool encoding.MutableSeriesIteratorsPool) (MultiSeriesBlocks, error) {
+func SeriesBlockToMultiSeriesBlocks(multiNamespaceSeriesList []MultiNamespaceSeries, seriesIteratorsPool encoding.MutableSeriesIteratorsPool, stepSize time.Duration) (MultiSeriesBlocks, error) {


nit1: Split args over newlines
nit2: can we use shorter names? Easy to get lost with lines being this long

arnikola · 2018-07-12T18:07:13Z

src/coordinator/ts/m3db/block/iterator_test.go

+			End:      end,
+			StepSize: stepSize,
+		},
+		indexTime: start.Add(-1 * stepSize),


nit start.Sub(stepSize)

arnikola · 2018-07-12T18:08:22Z

src/coordinator/ts/m3db/block/iterator_test.go

+			},
+			start:           now,
+			end:             now.Add(600 * time.Second),
+			stepSize:        60 * time.Second,


nit time.Minute?

arnikola · 2018-07-12T18:09:57Z

src/coordinator/ts/m3db/block/iterator_test.go

+	start, end                     time.Time
+	stepSize                       time.Duration
+	dps                            [][]ts.Datapoint
+	expectedResults, actualResults []float64


nit: would prefer to have actualResults be a variable built in the test, rather than added to the struct itself

arnikola · 2018-07-12T18:10:31Z

src/coordinator/ts/m3db/block/iterator_test.go

+type multiSeriesBlockStepIterTestCase struct {
+	dps             [][]float64
+	expectedResults [][]float64
+	actualResults   [][]float64


Same here re actualResults

arnikola · 2018-07-12T18:10:47Z

src/coordinator/ts/m3db/block/iterator_test.go

+
+type mockBlockStepIterTestCase struct {
+	dps                            [][]float64
+	actualResults, expectedResults [][]float64


Same here re: actualResults

nikunjgit · 2018-07-16T02:43:09Z

src/coordinator/block/types.go

@@ -50,6 +50,14 @@ type Bounds struct {
 	StepSize time.Duration
 }

+// Equals determines whether two bounds (start and end) are equal


fix comment: whether the bound (start and end) is same as the current bound ? Also, do you need to check the step size as well ?

nikunjgit · 2018-07-16T02:43:28Z

src/coordinator/models/tag.go

+// Clone returns a clone of the Tags
+func (t Tags) Clone() Tags {
+	tags := make(Tags)
+


remove line ?

nikunjgit · 2018-07-16T02:49:04Z

src/coordinator/ts/m3db/block/block_converter.go

@@ -18,11 +18,15 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN


converter.go ?

nikunjgit · 2018-07-16T02:49:50Z

src/coordinator/ts/m3db/block/block_converter.go

-				multiSeriesBlocks[consolidatedSeriesBlockIdx].Start = consolidatedSeriesBlock.Start
-				multiSeriesBlocks[consolidatedSeriesBlockIdx].End = consolidatedSeriesBlock.End
+			if seriesIdx == 0 {
+				multiSeriesBlocks[consolidatedSeriesBlockIdx].Metadata.Bounds.StepSize = consolidatedSeriesBlock.Metadata.Bounds.StepSize


extract multiSeriesBlocks[consolidatedSeriesBlockIdx] into a variable ?

nikunjgit · 2018-07-16T02:51:06Z

src/coordinator/ts/m3db/block/block_converter.go

@@ -79,11 +123,12 @@ func newConsolidatedSeriesBlocks(multiNamespaceSeries MultiNamespaceSeries, seri
 		for consolidatedNSBlockIdx, consolidatedNSBlock := range consolidatedNSBlocks {
 			// we only want to set the start and end times once
 			if seriesBlocksIdx == 0 {
-				consolidatedSeriesBlocks[consolidatedNSBlockIdx].Start = consolidatedNSBlock.Start
-				consolidatedSeriesBlocks[consolidatedNSBlockIdx].End = consolidatedNSBlock.End
+				consolidatedSeriesBlocks[consolidatedNSBlockIdx].Metadata.Bounds.StepSize = consolidatedNSBlock.Bounds.StepSize


extract consolidatedSeriesBlocks[consolidatedNSBlockIdx] into variable ?

nikunjgit · 2018-07-16T03:05:02Z

src/coordinator/ts/m3db/block/multi_series.go

+	}
+
+	bounds := m.meta.Bounds
+	t, err := bounds.TimeForIndex(m.index)


do this before creating values ?

nikunjgit · 2018-07-16T03:07:14Z

src/coordinator/ts/m3db/block/ns_series.go

+// Current returns the float64 value for that step
+func (c *consolidatedNSBlockIter) Current() float64 {
+	lastDP := c.lastDP
+	// NB(braskin): if the last datapoint is after the current step, but before the (current step+1),


move comment down ?

arnikola · 2018-07-16T12:25:47Z

src/coordinator/block/types.go

@@ -50,6 +50,14 @@ type Bounds struct {
 	StepSize time.Duration
 }

+// Equals determines whether two bounds (start and end) are equal
+func (b Bounds) Equals(compareBound Bounds) bool {
+	if !b.Start.Equal(compareBound.Start) || !b.End.Equal(compareBound.End) {


nit: cleaner to return b.Start.Equal(compareBound.Start) && b.End.Equal(compareBound.End); to Nikunj's point, if we don't care about stepsize here we can rename it from Equalsto EqualEndpoints or something

arnikola · 2018-07-16T12:34:12Z

src/coordinator/ts/m3db/block/consolidated_series.go

+type ConsolidatedSeriesBlock struct {
+	Metadata             block.Metadata
+	ConsolidatedNSBlocks []ConsolidatedNSBlock
+	consolidationFunc    ConsolidationFunc // nolint


nit:// nolint usually goes on top of struct

Even if it's just for one field in the struct?

I guess it doesn't really matter since it shouldn't be // nolint much longer :p

arnikola · 2018-07-16T12:39:19Z

src/coordinator/ts/m3db/block/consolidated_series.go

+
+// ConsolidatedSeriesBlocks contain all of the consolidated blocks for
+// a single timeseries across namespaces.
+// Each ConsolidatedBlockIterator will have the same size


nit: misplaced comment?

arnikola · 2018-07-16T12:42:10Z

src/coordinator/ts/m3db/block/consolidated_series.go

+	}
+
+	// todo(braskin): until we have consolidation
+	return values[0]


Still a bit dangerous; no guarantee there will be one value

arnikola · 2018-07-16T13:31:34Z

src/coordinator/ts/m3db/block/iterator_test.go

@@ -96,7 +96,7 @@ func TestConsolidatedNSBlockIter(t *testing.T) {
 			end:      now.Add(1200 * time.Second),
 			stepSize: 60 * time.Second,
 			expectedResults: []float64{1, 2, nan, nan, nan, 3, nan, nan, nan, nan,
-				nan, nan, nan, nan, nan, nan, nan, 5, nan, 6},
+				nan, nan, nan, nan, nan, nan, nan, 5, nan, 6, nan},


Why did the expected change? Was there a logic issue previously?

Yeah logic changed to be inclusive of the last step.

arnikola · 2018-07-16T13:32:14Z

src/coordinator/ts/m3db/block/iterator.go

+	if err != nil {
+		return math.NaN()
+	}
+	if !indexTime.After(lastDP.Timestamp) && indexTime.Add(c.bounds.StepSize).After(lastDP.Timestamp) {


nit: newline

arnikola

Is the .ci change intentional btw?

arnikola · 2018-07-16T18:50:32Z

src/coordinator/ts/m3db/block/consolidated_series_test.go

+		},
+	}
+
+	for _, test := range testCases {


This is a little different from the recommended way of writing table tests which gets you a few nice things like splitting each of your test cases into a separate test run, allowing them to fail independently

Your choice if you want to convert it to that style or not here

arnikola · 2018-07-16T18:51:47Z

src/coordinator/ts/m3db/block/converter.go

@@ -31,45 +35,88 @@ var (
 )

 // SeriesBlockToMultiSeriesBlocks converts M3DB blocks to multi series blocks
-func SeriesBlockToMultiSeriesBlocks(multiNamespaceSeriesList []MultiNamespaceSeries, seriesIteratorsPool encoding.MutableSeriesIteratorsPool) (MultiSeriesBlocks, error) {
+func SeriesBlockToMultiSeriesBlocks(
+	multiNamespaceSeriesList []MultiNamespaceSeries, seriesIteratorsPool encoding.MutableSeriesIteratorsPool, stepSize time.Duration) (MultiSeriesBlocks, error) {


nit: can you add newlines between these params to reduce the line length?

arnikola · 2018-07-16T18:53:04Z

src/coordinator/ts/m3db/block/converter.go

-				multiSeriesBlocks[consolidatedSeriesBlockIdx].Start = consolidatedSeriesBlock.Start
-				multiSeriesBlocks[consolidatedSeriesBlockIdx].End = consolidatedSeriesBlock.End
+			if seriesIdx == 0 {
+				blockBounds := multiSeriesBlocks[consolidatedSeriesBlockIdx].Metadata.Bounds


Can you do this instead? �

blockBounds := multiSeriesBlocks[consolidatedSeriesBlockIdx].Metadata.Bounds multiSeriesBlocks[consolidatedSeriesBlockIdx].Metadata.Bounds = blockBounds �```

I don't think that works because the underlaying structs aren't pointers.

Does it matter if we're just making a copy though?

arnikola · 2018-07-16T18:53:41Z

src/coordinator/ts/m3db/block/converter.go

 // newConsolidatedSeriesBlocks creates consolidated blocks by timeseries across namespaces
-func newConsolidatedSeriesBlocks(multiNamespaceSeries MultiNamespaceSeries, seriesIteratorsPool encoding.MutableSeriesIteratorsPool) (ConsolidatedSeriesBlocks, error) {
+func newConsolidatedSeriesBlocks(
+	multiNamespaceSeries MultiNamespaceSeries, seriesIteratorsPool encoding.MutableSeriesIteratorsPool, stepSize time.Duration) (ConsolidatedSeriesBlocks, error) {


nit: can you add newlines to the arguement list?

arnikola · 2018-07-16T18:56:26Z

src/coordinator/ts/m3db/block/types.go

+type MultiNamespaceSeries []SeriesBlocks
+
+// ID enforces the same ID across namespaces
+func (n MultiNamespaceSeries) ID() ident.ID { return n[0].ID }


Potentially dangerous if len(n) == 0

arnikola · 2018-07-16T19:25:56Z

src/coordinator/ts/m3db/block/multi_series.go

+			nsBlockIter := newConsolidatedNSBlockIter(nsBlock)
+			consolidatedNSBlockIters[j] = nsBlockIter
+		}
+		seriesBlockIters[i] = &consolidatedSeriesBlockIter{


nit: newline

arnikola · 2018-07-16T19:30:58Z

src/coordinator/ts/m3db/block/converter_test.go

+	seriesIterator.EXPECT().Next().Return(false)
+
+	for _, dp := range dps {
+		seriesIterator.EXPECT().Current().Return(ts.Datapoint{Timestamp: dp.Timestamp, Value: dp.Value}, xtime.Millisecond, nil).Times(1)


nit: might be a bit easier to read if you generate the point separately
dp := ts.Datapoint{Timestamp: dp.Timestamp, Value: dp.Value}

arnikola · 2018-07-16T19:32:20Z

src/coordinator/ts/m3db/block/converter.go

@@ -96,16 +146,20 @@ func newConsolidatedSeriesBlocks(multiNamespaceSeries MultiNamespaceSeries, seri

 // newConsolidatedNSBlocks creates a slice of consolidated blocks per namespace for a single timeseries
 // nolint: unparam
-func newConsolidatedNSBlocks(seriesBlocks SeriesBlocks, seriesIteratorsPool encoding.MutableSeriesIteratorsPool) []ConsolidatedNSBlock {
+func newConsolidatedNSBlocks(
+	seriesBlocks SeriesBlocks, seriesIteratorsPool encoding.MutableSeriesIteratorsPool, stepSize time.Duration) []ConsolidatedNSBlock {


nit: newlines in args list

arnikola · 2018-07-16T19:33:48Z

src/coordinator/ts/m3db/block/converter.go

+				multiSeriesBlocks[consolidatedSeriesBlockIdx].Metadata.Bounds = blockBounds
+			}
+
+			dupedTags := consolidatedSeriesBlock.ConsolidatedNSBlocks[0].SeriesIterators.Iters()[0].Tags().Duplicate()


Can you add comments here, or add a method on consolidatedNSBlocks that gives tags? There's a lot of indexing here without any clear guarantees that it's safe to do

arnikola · 2018-07-16T19:34:51Z

src/coordinator/ts/m3db/block/converter.go

+		return models.Tags{}
+	}
+
+	commonTags := m[0].Blocks[0].Metadata.Tags.Clone()


Is the clone necessary?

nikunjgit · 2018-07-16T20:30:42Z

src/coordinator/block/types.go

@@ -50,6 +50,17 @@ type Bounds struct {
 	StepSize time.Duration
 }

+// Equals determines whether the bound (start and end) are equal
+func (b Bounds) Equals(compareBound Bounds) bool {
+	equalTimes := b.Start.Equal(compareBound.Start) && b.End.Equal(compareBound.End)


Can be simplified:

if !equalTimes { return false } return b.StepSize == compareBound.StepSize

nikunjgit · 2018-07-16T20:38:29Z

src/coordinator/storage/local/storage.go

+	return session.FetchTagged(namespaceID, query, opts)
+}
+
+func (s *localStorage) FetchBlocks(ctx context.Context, query *storage.FetchQuery, options *storage.FetchOptions) (block.Result, error) {


add a todo here to merge this with Fetch()

nikunjgit · 2018-07-16T21:26:00Z

src/coordinator/ts/m3db/block/multi_series.go

+// SeriesIter creates a new series iterator for a given MultiSeriesBlock
+func (m MultiSeriesBlock) SeriesIter() (block.SeriesIter, error) {
+	// todo(braskin): implement SeriesIter()
+	return nil, errors.New("SeriesIter not implemented")


as we discussed, you need to implement this so that you can test e2e

nikunjgit · 2018-07-20T14:36:13Z

src/coordinator/ts/m3db/block/consolidated_series.go

+type ConsolidatedBlocks []ConsolidatedBlock
+
+func (c *consolidatedBlockStepIter) Current() float64 {
+	values := make([]float64, 0, 1)


make([]float64, 0, len(c.nsBlockSeriesIters))

nikunjgit · 2018-07-20T14:38:13Z

src/coordinator/ts/m3db/block/consolidated_series.go

+		values = append(values, dp)
+	}
+
+	if len(values) > 0 {


just return values[0], I don't think you need to check since Current() is only called when Next() returns true. You should never have to return math.NaN()

nikunjgit · 2018-07-20T14:40:34Z

src/coordinator/ts/m3db/block/consolidated_series.go

+}
+
+// Current returns a slice of values for the current series
+func (c *consolidatedBlockSeriesIter) Current() []float64 {


actually at conslidatedBlock level, doesn't look like you have a concept of series and steps. I think you should just have one iterator : consolidatedBlockIter which should return a single float64 on current. The concept of Step and Series probably comes at a higher level

nikunjgit · 2018-07-20T14:42:42Z

src/coordinator/ts/m3db/block/ns_series.go

+// NSBlock is a single block for a given timeseries and namespace
+// which contains all of the necessary SeriesIterators so that consolidation can
+// happen across namespaces
+type NSBlock struct {


same comment as above, at this level you don't have a concept of step and series since you just are representing a single series. Having a single iter with Current returning float64 ?

nikunjgit · 2018-12-17T16:53:48Z

Are you still working on it or close it out ?

CLAassistant · 2021-01-27T04:38:20Z

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.

Benjamin Raskin seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

arnikola reviewed Jul 3, 2018

View reviewed changes

nikunjgit reviewed Jul 5, 2018

View reviewed changes

arnikola reviewed Jul 10, 2018

View reviewed changes

benraskin92 force-pushed the braskin/block_iters branch from 9bb0aec to 1c91fa0 Compare July 10, 2018 15:53

arnikola reviewed Jul 10, 2018

View reviewed changes

nikunjgit reviewed Jul 10, 2018

View reviewed changes

benraskin92 force-pushed the braskin/block_iters branch from e764178 to af57f10 Compare July 11, 2018 22:37

benraskin92 changed the base branch from master to lazyEval July 11, 2018 22:37

nikunjgit reviewed Jul 12, 2018

View reviewed changes

arnikola reviewed Jul 12, 2018

View reviewed changes

nikunjgit force-pushed the lazyEval branch from 5a6b110 to 0313679 Compare July 12, 2018 18:15

benraskin92 force-pushed the braskin/block_iters branch from d870ef2 to 61c0185 Compare July 14, 2018 12:56

benraskin92 changed the base branch from lazyEval to master July 14, 2018 12:57

nikunjgit reviewed Jul 16, 2018

View reviewed changes

arnikola reviewed Jul 16, 2018

View reviewed changes

Benjamin Raskin added 11 commits July 16, 2018 17:15

Add StepIter to blocks

c100ab0

Remove prints

71c36b9

Add tests and satisfy block interface

565605a

Clean up PR

61aaed5

Add nolint

9dfd527

Skip read tests

cd96117

Address comments

05f4cf5

More comments

ae5de4d

Remove lastVal

629f745

Address more comments

fead6c0

Add more tests

9ed418f

Benjamin Raskin added 9 commits July 16, 2018 17:15

Update iterators

96819d0

Clean up code

a34592d

Address more comments

8f29ce1

Change structure

86db7fe

Fix storage tests

cb6c682

Fix lint issues

496e110

Address comments

122d00a

More comments

d0c6893

More comments

e6a0d81

nikunjgit reviewed Jul 16, 2018

View reviewed changes

Add series iterator to block

ece562c

benraskin92 force-pushed the braskin/block_iters branch from 5e8ecda to ece562c Compare July 19, 2018 21:03

nikunjgit reviewed Jul 20, 2018

View reviewed changes

Benjamin Raskin added 5 commits July 20, 2018 16:42

WIP

a939b5f

Fix iters

76cf626

Simplify bound equality func

3a1b9a6

Remove print lines

4b853ec

Stash

6f11dc6

robskillington force-pushed the master branch from fdbef61 to 0c6a0fe Compare October 15, 2019 06:42

robskillington force-pushed the master branch from e130b7e to 3cf877e Compare November 12, 2020 20:21

		@@ -33,7 +36,13 @@ var (
		// SeriesBlockToMultiSeriesBlocks converts M3DB blocks to multi series blocks
		func SeriesBlockToMultiSeriesBlocks(multiNamespaceSeriesList []MultiNamespaceSeries, seriesIteratorsPool encoding.MutableSeriesIteratorsPool) (MultiSeriesBlocks, error) {

		@@ -18,11 +18,14 @@
		// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN

		@@ -31,45 +34,81 @@ var (
		)

		// SeriesBlockToMultiSeriesBlocks converts M3DB blocks to multi series blocks


		if !consolidatedSeriesBlock.beyondBounds(multiSeriesBlocks[consolidatedSeriesBlockIdx]) {
		if !equalBounds(consolidatedSeriesBlock.Metadata.Bounds, multiSeriesBlocks[consolidatedSeriesBlockIdx].Metadata.Bounds) {

		@@ -18,11 +18,14 @@
		// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
		// THE SOFTWARE.

		@@ -18,11 +18,15 @@
		// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN

Add StepIter to blocks #778

Are you sure you want to change the base?

Add StepIter to blocks #778

Conversation

benraskin92 commented Jul 2, 2018

codecov bot commented Jul 2, 2018 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benraskin92 commented Jul 10, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nikunjgit left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nikunjgit Jul 16, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Jul 2, 2018 •

edited

Loading

nikunjgit Jul 16, 2018 •

edited

Loading