|
|
@ -16,6 +16,7 @@ package storage |
|
|
|
import ( |
|
|
|
import ( |
|
|
|
"bytes" |
|
|
|
"bytes" |
|
|
|
"container/heap" |
|
|
|
"container/heap" |
|
|
|
|
|
|
|
"fmt" |
|
|
|
"sort" |
|
|
|
"sort" |
|
|
|
"strings" |
|
|
|
"strings" |
|
|
|
"sync" |
|
|
|
"sync" |
|
|
@ -418,8 +419,7 @@ func (h *genericSeriesSetHeap) Pop() interface{} { |
|
|
|
// with "almost" the same data, e.g. from 2 Prometheus HA replicas. This is fine, since from the Prometheus perspective
|
|
|
|
// with "almost" the same data, e.g. from 2 Prometheus HA replicas. This is fine, since from the Prometheus perspective
|
|
|
|
// this never happens.
|
|
|
|
// this never happens.
|
|
|
|
//
|
|
|
|
//
|
|
|
|
// NOTE: Use this merge function only when you see potentially overlapping series, as this introduces a small overhead
|
|
|
|
// It's optimized for non-overlap cases as well.
|
|
|
|
// to handle overlaps between series.
|
|
|
|
|
|
|
|
func ChainedSeriesMerge(series ...Series) Series { |
|
|
|
func ChainedSeriesMerge(series ...Series) Series { |
|
|
|
if len(series) == 0 { |
|
|
|
if len(series) == 0 { |
|
|
|
return nil |
|
|
|
return nil |
|
|
@ -438,10 +438,12 @@ func ChainedSeriesMerge(series ...Series) Series { |
|
|
|
|
|
|
|
|
|
|
|
// chainSampleIterator is responsible to iterate over samples from different iterators of the same time series in timestamps
|
|
|
|
// chainSampleIterator is responsible to iterate over samples from different iterators of the same time series in timestamps
|
|
|
|
// order. If one or more samples overlap, one sample from random overlapped ones is kept and all others with the same
|
|
|
|
// order. If one or more samples overlap, one sample from random overlapped ones is kept and all others with the same
|
|
|
|
// timestamp are dropped.
|
|
|
|
// timestamp are dropped. It's optimized for non-overlap cases as well.
|
|
|
|
type chainSampleIterator struct { |
|
|
|
type chainSampleIterator struct { |
|
|
|
iterators []chunkenc.Iterator |
|
|
|
iterators []chunkenc.Iterator |
|
|
|
h samplesIteratorHeap |
|
|
|
h samplesIteratorHeap |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
curr chunkenc.Iterator |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func newChainSampleIterator(iterators []chunkenc.Iterator) chunkenc.Iterator { |
|
|
|
func newChainSampleIterator(iterators []chunkenc.Iterator) chunkenc.Iterator { |
|
|
@ -462,43 +464,42 @@ func (c *chainSampleIterator) Seek(t int64) bool { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (c *chainSampleIterator) At() (t int64, v float64) { |
|
|
|
func (c *chainSampleIterator) At() (t int64, v float64) { |
|
|
|
if len(c.h) == 0 { |
|
|
|
if c.h == nil { |
|
|
|
panic("chainSampleIterator.At() called after .Next() returned false.") |
|
|
|
panic("chainSampleIterator.At() called after .Next() returned false.") |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
return c.curr.At() |
|
|
|
return c.h[0].At() |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (c *chainSampleIterator) Next() bool { |
|
|
|
func (c *chainSampleIterator) Next() bool { |
|
|
|
if c.h == nil { |
|
|
|
if c.h == nil { |
|
|
|
for _, iter := range c.iterators { |
|
|
|
c.curr = c.iterators[0] |
|
|
|
|
|
|
|
for _, iter := range c.iterators[1:] { |
|
|
|
if iter.Next() { |
|
|
|
if iter.Next() { |
|
|
|
heap.Push(&c.h, iter) |
|
|
|
heap.Push(&c.h, iter) |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
} else if len(c.h) == 0 { |
|
|
|
return len(c.h) > 0 |
|
|
|
return c.curr != nil && c.curr.Next() |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if len(c.h) == 0 { |
|
|
|
|
|
|
|
return false |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
currt, _ := c.At() |
|
|
|
for { |
|
|
|
for len(c.h) > 0 { |
|
|
|
if c.curr.Next() { |
|
|
|
|
|
|
|
currt, _ := c.curr.At() |
|
|
|
nextt, _ := c.h[0].At() |
|
|
|
nextt, _ := c.h[0].At() |
|
|
|
// All but one of the overlapping samples will be dropped.
|
|
|
|
if currt < nextt { |
|
|
|
if nextt != currt { |
|
|
|
return true |
|
|
|
break |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
if currt == nextt { |
|
|
|
iter := heap.Pop(&c.h).(chunkenc.Iterator) |
|
|
|
fmt.Println("same ts", currt, nextt) |
|
|
|
if iter.Next() { |
|
|
|
// Ignoring sample.
|
|
|
|
heap.Push(&c.h, iter) |
|
|
|
continue |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
heap.Push(&c.h, c.curr) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
return len(c.h) > 0 |
|
|
|
c.curr = heap.Pop(&c.h).(chunkenc.Iterator) |
|
|
|
|
|
|
|
return true |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (c *chainSampleIterator) Err() error { |
|
|
|
func (c *chainSampleIterator) Err() error { |
|
|
|