mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-03-05 20:59:13 -08:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Rename for clarity and consistency.

Browse source 
Misc. changes for code cleanliness.

Signed-off-by: Goutham Veeramachaneni <cs14btech11014@iith.ac.in>
This commit is contained in:
Goutham Veeramachaneni 2017-05-22 16:42:36 +05:30
parent 7ccf9d0211
commit 244b73fce1
No known key found for this signature in database
GPG key ID: F1C217E8E9023CAD
10 changed files with 275 additions and 257 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
block.go
View file

@ -182,10 +182,10 @@ func newPersistedBlock(dir string) (*persistedBlock, error) {
return nil, err return nil, err
} }
ts := make(map[uint32][]trange) ts := make(map[uint32]intervals)
for tr.Next() { for tr.Next() {
s := tr.At() s := tr.At()
ts[s.ref] = s.ranges ts[s.ref] = s.intervals
} }
pb := &persistedBlock{ pb := &persistedBlock{
@ -238,7 +238,7 @@ func (pb *persistedBlock) Delete(mint, maxt int64, ms ...labels.Matcher) error {
ir := pb.indexr ir := pb.indexr
// Choose only valid postings which have chunks in the time-range. // Choose only valid postings which have chunks in the time-range.
delStones := map[uint32][]trange{} delStones := map[uint32]intervals{}
Outer: Outer:
for p.Next() { for p.Next() {
@ -260,7 +260,7 @@ Outer:
if maxtime > maxt { if maxtime > maxt {
maxtime = maxt maxtime = maxt
} }
delStones[p.At()] = []trange{{mint, maxtime}} delStones[p.At()] = intervals{{mint, maxtime}}
continue Outer continue Outer
} }
} }

6
chunks.go
View file

@ -59,7 +59,7 @@ func (cm *ChunkMeta) writeHash(h hash.Hash) error {
type deletedIterator struct { type deletedIterator struct {
it chunks.Iterator it chunks.Iterator
dranges []trange intervals intervals
} }
func (it *deletedIterator) At() (int64, float64) { func (it *deletedIterator) At() (int64, float64) {
@ -71,13 +71,13 @@ Outer:
for it.it.Next() { for it.it.Next() {
ts, _ := it.it.At() ts, _ := it.it.At()
for _, tr := range it.dranges { for _, tr := range it.intervals {
if tr.inBounds(ts) { if tr.inBounds(ts) {
continue Outer continue Outer
} }
if ts > tr.maxt { if ts > tr.maxt {
it.dranges = it.dranges[1:] it.intervals = it.intervals[1:]
continue continue
} }

24
chunks_test.go
View file

@ -50,23 +50,23 @@ func TestDeletedIterator(t *testing.T) {
} }
cases := []struct { cases := []struct {
r []trange r intervals
}{ }{
{r: []trange{{1, 20}}}, {r: intervals{{1, 20}}},
{r: []trange{{1, 10}, {12, 20}, {21, 23}, {25, 30}}}, {r: intervals{{1, 10}, {12, 20}, {21, 23}, {25, 30}}},
{r: []trange{{1, 10}, {12, 20}, {20, 30}}}, {r: intervals{{1, 10}, {12, 20}, {20, 30}}},
{r: []trange{{1, 10}, {12, 23}, {25, 30}}}, {r: intervals{{1, 10}, {12, 23}, {25, 30}}},
{r: []trange{{1, 23}, {12, 20}, {25, 30}}}, {r: intervals{{1, 23}, {12, 20}, {25, 30}}},
{r: []trange{{1, 23}, {12, 20}, {25, 3000}}}, {r: intervals{{1, 23}, {12, 20}, {25, 3000}}},
{r: []trange{{0, 2000}}}, {r: intervals{{0, 2000}}},
{r: []trange{{500, 2000}}}, {r: intervals{{500, 2000}}},
{r: []trange{{0, 200}}}, {r: intervals{{0, 200}}},
{r: []trange{{1000, 20000}}}, {r: intervals{{1000, 20000}}},
} }
for _, c := range cases { for _, c := range cases {
i := int64(-1) i := int64(-1)
it := &deletedIterator{it: chk.Iterator(), dranges: c.r[:]} it := &deletedIterator{it: chk.Iterator(), intervals: c.r[:]}
ranges := c.r[:] ranges := c.r[:]
for it.Next() { for it.Next() {
i++ i++

50
compact.go
View file

@ -323,19 +323,19 @@ func (c *compactor) populate(blocks []Block, indexw IndexWriter, chunkw ChunkWri
) )
for set.Next() { for set.Next() {
lset, chks, ranges := set.At() // The chunks here are not fully deleted. lset, chks, dranges := set.At() // The chunks here are not fully deleted.
if len(ranges) > 0 { if len(dranges) > 0 {
// Re-encode the chunk to not have deleted values. // Re-encode the chunk to not have deleted values.
for _, chk := range chks { for _, chk := range chks {
if intervalOverlap(ranges[0].mint, ranges[len(ranges)-1].maxt, chk.MinTime, chk.MaxTime) { if intervalOverlap(dranges[0].mint, dranges[len(dranges)-1].maxt, chk.MinTime, chk.MaxTime) {
newChunk := chunks.NewXORChunk() newChunk := chunks.NewXORChunk()
app, err := newChunk.Appender() app, err := newChunk.Appender()
if err != nil { if err != nil {
return nil, err return nil, err
} }
it := &deletedIterator{it: chk.Chunk.Iterator(), dranges: ranges} it := &deletedIterator{it: chk.Chunk.Iterator(), intervals: dranges}
for it.Next() { for it.Next() {
ts, v := it.At() ts, v := it.At()
app.Append(ts, v) app.Append(ts, v)
@ -402,7 +402,7 @@ func (c *compactor) populate(blocks []Block, indexw IndexWriter, chunkw ChunkWri
type compactionSet interface { type compactionSet interface {
Next() bool Next() bool
At() (labels.Labels, []*ChunkMeta, []trange) At() (labels.Labels, []*ChunkMeta, intervals)
Err() error Err() error
} }
@ -412,10 +412,10 @@ type compactionSeriesSet struct {
chunks ChunkReader chunks ChunkReader
tombstones TombstoneReader tombstones TombstoneReader
l labels.Labels l labels.Labels
c []*ChunkMeta c []*ChunkMeta
dranges []trange intervals intervals
err error err error
} }
func newCompactionSeriesSet(i IndexReader, c ChunkReader, t TombstoneReader, p Postings) *compactionSeriesSet { func newCompactionSeriesSet(i IndexReader, c ChunkReader, t TombstoneReader, p Postings) *compactionSeriesSet {
@ -435,9 +435,9 @@ func (c *compactionSeriesSet) Next() bool {
if c.tombstones.Seek(c.p.At()) { if c.tombstones.Seek(c.p.At()) {
s := c.tombstones.At() s := c.tombstones.At()
if c.p.At() == s.ref { if c.p.At() == s.ref {
c.dranges = s.ranges c.intervals = s.intervals
} else { } else {
c.dranges = nil c.intervals = nil
} }
} }
c.l, c.c, c.err = c.index.Series(c.p.At()) c.l, c.c, c.err = c.index.Series(c.p.At())
@ -446,10 +446,10 @@ func (c *compactionSeriesSet) Next() bool {
} }
// Remove completely deleted chunks. // Remove completely deleted chunks.
if len(c.dranges) > 0 { if len(c.intervals) > 0 {
chks := make([]*ChunkMeta, 0, len(c.c)) chks := make([]*ChunkMeta, 0, len(c.c))
for _, chk := range c.c { for _, chk := range c.c {
if !(trange{chk.MinTime, chk.MaxTime}.isSubrange(c.dranges)) { if !(interval{chk.MinTime, chk.MaxTime}.isSubrange(c.intervals)) {
chk.Chunk, c.err = c.chunks.Chunk(chk.Ref) chk.Chunk, c.err = c.chunks.Chunk(chk.Ref)
if c.err != nil { if c.err != nil {
return false return false
@ -472,17 +472,17 @@ func (c *compactionSeriesSet) Err() error {
return c.p.Err() return c.p.Err()
} }
func (c *compactionSeriesSet) At() (labels.Labels, []*ChunkMeta, []trange) { func (c *compactionSeriesSet) At() (labels.Labels, []*ChunkMeta, intervals) {
return c.l, c.c, c.dranges return c.l, c.c, c.intervals
} }
type compactionMerger struct { type compactionMerger struct {
a, b compactionSet a, b compactionSet
aok, bok bool aok, bok bool
l labels.Labels l labels.Labels
c []*ChunkMeta c []*ChunkMeta
dranges []trange intervals intervals
} }
type compactionSeries struct { type compactionSeries struct {
@ -523,21 +523,21 @@ func (c *compactionMerger) Next() bool {
d := c.compare() d := c.compare()
// Both sets contain the current series. Chain them into a single one. // Both sets contain the current series. Chain them into a single one.
if d > 0 { if d > 0 {
c.l, c.c, c.dranges = c.b.At() c.l, c.c, c.intervals = c.b.At()
c.bok = c.b.Next() c.bok = c.b.Next()
} else if d < 0 { } else if d < 0 {
c.l, c.c, c.dranges = c.a.At() c.l, c.c, c.intervals = c.a.At()
c.aok = c.a.Next() c.aok = c.a.Next()
} else { } else {
l, ca, ra := c.a.At() l, ca, ra := c.a.At()
_, cb, rb := c.b.At() _, cb, rb := c.b.At()
for _, r := range rb { for _, r := range rb {
ra = addNewInterval(ra, r) ra = ra.add(r)
} }
c.l = l c.l = l
c.c = append(ca, cb...) c.c = append(ca, cb...)
c.dranges = ra c.intervals = ra
c.aok = c.a.Next() c.aok = c.a.Next()
c.bok = c.b.Next() c.bok = c.b.Next()
@ -552,8 +552,8 @@ func (c *compactionMerger) Err() error {
return c.b.Err() return c.b.Err()
} }
func (c *compactionMerger) At() (labels.Labels, []*ChunkMeta, []trange) { func (c *compactionMerger) At() (labels.Labels, []*ChunkMeta, intervals) {
return c.l, c.c, c.dranges return c.l, c.c, c.intervals
} }
func renameFile(from, to string) error { func renameFile(from, to string) error {

4
head.go
View file

@ -162,7 +162,7 @@ func (h *HeadBlock) init() error {
for tr.Next() { for tr.Next() {
s := tr.At() s := tr.At()
h.tombstones.refs = append(h.tombstones.refs, s.ref) h.tombstones.refs = append(h.tombstones.refs, s.ref)
h.tombstones.stones[s.ref] = s.ranges h.tombstones.stones[s.ref] = s.intervals
} }
return errors.Wrap(err, "tombstones reader iteration") return errors.Wrap(err, "tombstones reader iteration")
} }
@ -245,7 +245,7 @@ Outer:
if maxtime > maxt { if maxtime > maxt {
maxtime = maxt maxtime = maxt
} }
h.tombstones.stones[ref] = addNewInterval(h.tombstones.stones[ref], trange{mint, maxtime}) h.tombstones.stones[ref] = h.tombstones.stones[ref].add(interval{mint, maxtime})
} }
if p.Err() != nil { if p.Err() != nil {

36
head_test.go
View file

@ -399,28 +399,28 @@ func TestDeleteSimple(t *testing.T) {
require.NoError(t, app.Commit()) require.NoError(t, app.Commit())
cases := []struct { cases := []struct {
dranges []trange intervals intervals
remaint []int64 remaint []int64
}{ }{
{ {
dranges: []trange{{0, 3}}, intervals: intervals{{0, 3}},
remaint: []int64{4, 5, 6, 7, 8, 9}, remaint: []int64{4, 5, 6, 7, 8, 9},
}, },
{ {
dranges: []trange{{1, 3}}, intervals: intervals{{1, 3}},
remaint: []int64{0, 4, 5, 6, 7, 8, 9}, remaint: []int64{0, 4, 5, 6, 7, 8, 9},
}, },
{ {
dranges: []trange{{1, 3}, {4, 7}}, intervals: intervals{{1, 3}, {4, 7}},
remaint: []int64{0, 8, 9}, remaint: []int64{0, 8, 9},
}, },
{ {
dranges: []trange{{1, 3}, {4, 700}}, intervals: intervals{{1, 3}, {4, 700}},
remaint: []int64{0}, remaint: []int64{0},
}, },
{ {
dranges: []trange{{0, 9}}, intervals: intervals{{0, 9}},
remaint: []int64{}, remaint: []int64{},
}, },
} }
@ -431,7 +431,7 @@ Outer:
hb.tombstones = newEmptyTombstoneReader() hb.tombstones = newEmptyTombstoneReader()
// Delete the ranges. // Delete the ranges.
for _, r := range c.dranges { for _, r := range c.intervals {
require.NoError(t, hb.Delete(r.mint, r.maxt, labels.NewEqualMatcher("a", "b"))) require.NoError(t, hb.Delete(r.mint, r.maxt, labels.NewEqualMatcher("a", "b")))
} }
@ -597,18 +597,18 @@ func TestDelete_e2e(t *testing.T) {
// Delete a time-range from each-selector. // Delete a time-range from each-selector.
dels := []struct { dels := []struct {
ms []labels.Matcher ms []labels.Matcher
drange []trange drange intervals
}{ }{
{ {
ms: []labels.Matcher{labels.NewEqualMatcher("a", "b")}, ms: []labels.Matcher{labels.NewEqualMatcher("a", "b")},
drange: []trange{{300, 500}, {600, 670}}, drange: intervals{{300, 500}, {600, 670}},
}, },
{ {
ms: []labels.Matcher{ ms: []labels.Matcher{
labels.NewEqualMatcher("a", "b"), labels.NewEqualMatcher("a", "b"),
labels.NewEqualMatcher("job", "prom-k8s"), labels.NewEqualMatcher("job", "prom-k8s"),
}, },
drange: []trange{{300, 500}, {100, 670}}, drange: intervals{{300, 500}, {100, 670}},
}, },
{ {
ms: []labels.Matcher{ ms: []labels.Matcher{
@ -616,7 +616,7 @@ func TestDelete_e2e(t *testing.T) {
labels.NewEqualMatcher("instance", "localhost:9090"), labels.NewEqualMatcher("instance", "localhost:9090"),
labels.NewEqualMatcher("job", "prometheus"), labels.NewEqualMatcher("job", "prometheus"),
}, },
drange: []trange{{300, 400}, {100, 6700}}, drange: intervals{{300, 400}, {100, 6700}},
}, },
// TODO: Add Regexp Matchers. // TODO: Add Regexp Matchers.
} }
@ -717,7 +717,7 @@ func boundedSamples(full []sample, mint, maxt int64) []sample {
return full return full
} }
func deletedSamples(full []sample, dranges []trange) []sample { func deletedSamples(full []sample, dranges intervals) []sample {
ds := make([]sample, 0, len(full)) ds := make([]sample, 0, len(full))
Outer: Outer:
for _, s := range full { for _, s := range full {

60
querier.go
View file

@ -369,7 +369,7 @@ func (s *mergedSeriesSet) Next() bool {
type chunkSeriesSet interface { type chunkSeriesSet interface {
Next() bool Next() bool
At() (labels.Labels, []*ChunkMeta, stone) At() (labels.Labels, []*ChunkMeta, intervals)
Err() error Err() error
} }
@ -381,14 +381,14 @@ type baseChunkSeries struct {
tombstones TombstoneReader tombstones TombstoneReader
absent []string // labels that must be unset in results. absent []string // labels that must be unset in results.
lset labels.Labels lset labels.Labels
chks []*ChunkMeta chks []*ChunkMeta
stone stone intervals intervals
err error err error
} }
func (s *baseChunkSeries) At() (labels.Labels, []*ChunkMeta, stone) { func (s *baseChunkSeries) At() (labels.Labels, []*ChunkMeta, intervals) {
return s.lset, s.chks, s.stone return s.lset, s.chks, s.intervals
} }
func (s *baseChunkSeries) Err() error { return s.err } func (s *baseChunkSeries) Err() error { return s.err }
@ -413,12 +413,12 @@ Outer:
s.lset = lset s.lset = lset
s.chks = chunks s.chks = chunks
if s.tombstones.Seek(ref) && s.tombstones.At().ref == ref { if s.tombstones.Seek(ref) && s.tombstones.At().ref == ref {
s.stone = s.tombstones.At() s.intervals = s.tombstones.At().intervals
// Only those chunks that are not entirely deleted. // Only those chunks that are not entirely deleted.
chks := make([]*ChunkMeta, 0, len(s.chks)) chks := make([]*ChunkMeta, 0, len(s.chks))
for _, chk := range s.chks { for _, chk := range s.chks {
if !(trange{chk.MinTime, chk.MaxTime}.isSubrange(s.stone.ranges)) { if !(interval{chk.MinTime, chk.MaxTime}.isSubrange(s.intervals)) {
chks = append(chks, chk) chks = append(chks, chk)
} }
} }
@ -442,20 +442,20 @@ type populatedChunkSeries struct {
chunks ChunkReader chunks ChunkReader
mint, maxt int64 mint, maxt int64
err error err error
chks []*ChunkMeta chks []*ChunkMeta
lset labels.Labels lset labels.Labels
stone stone intervals intervals
} }
func (s *populatedChunkSeries) At() (labels.Labels, []*ChunkMeta, stone) { func (s *populatedChunkSeries) At() (labels.Labels, []*ChunkMeta, intervals) {
return s.lset, s.chks, s.stone return s.lset, s.chks, s.intervals
} }
func (s *populatedChunkSeries) Err() error { return s.err } func (s *populatedChunkSeries) Err() error { return s.err }
func (s *populatedChunkSeries) Next() bool { func (s *populatedChunkSeries) Next() bool {
for s.set.Next() { for s.set.Next() {
lset, chks, stn := s.set.At() lset, chks, dranges := s.set.At()
for len(chks) > 0 { for len(chks) > 0 {
if chks[0].MaxTime >= s.mint { if chks[0].MaxTime >= s.mint {
@ -482,7 +482,7 @@ func (s *populatedChunkSeries) Next() bool {
s.lset = lset s.lset = lset
s.chks = chks s.chks = chks
s.stone = stn s.intervals = dranges
return true return true
} }
@ -503,14 +503,14 @@ type blockSeriesSet struct {
func (s *blockSeriesSet) Next() bool { func (s *blockSeriesSet) Next() bool {
for s.set.Next() { for s.set.Next() {
lset, chunks, stn := s.set.At() lset, chunks, dranges := s.set.At()
s.cur = &chunkSeries{ s.cur = &chunkSeries{
labels: lset, labels: lset,
chunks: chunks, chunks: chunks,
mint: s.mint, mint: s.mint,
maxt: s.maxt, maxt: s.maxt,
stone: stn, intervals: dranges,
} }
return true return true
} }
@ -531,7 +531,7 @@ type chunkSeries struct {
mint, maxt int64 mint, maxt int64
stone stone intervals intervals
} }
func (s *chunkSeries) Labels() labels.Labels { func (s *chunkSeries) Labels() labels.Labels {
@ -539,7 +539,7 @@ func (s *chunkSeries) Labels() labels.Labels {
} }
func (s *chunkSeries) Iterator() SeriesIterator { func (s *chunkSeries) Iterator() SeriesIterator {
return newChunkSeriesIterator(s.chunks, s.stone, s.mint, s.maxt) return newChunkSeriesIterator(s.chunks, s.intervals, s.mint, s.maxt)
} }
// SeriesIterator iterates over the data of a time series. // SeriesIterator iterates over the data of a time series.
@ -637,13 +637,13 @@ type chunkSeriesIterator struct {
maxt, mint int64 maxt, mint int64
stone stone intervals intervals
} }
func newChunkSeriesIterator(cs []*ChunkMeta, s stone, mint, maxt int64) *chunkSeriesIterator { func newChunkSeriesIterator(cs []*ChunkMeta, dranges intervals, mint, maxt int64) *chunkSeriesIterator {
it := cs[0].Chunk.Iterator() it := cs[0].Chunk.Iterator()
if len(s.ranges) > 0 { if len(dranges) > 0 {
it = &deletedIterator{it: it, dranges: s.ranges} it = &deletedIterator{it: it, intervals: dranges}
} }
return &chunkSeriesIterator{ return &chunkSeriesIterator{
chunks: cs, chunks: cs,
@ -653,7 +653,7 @@ func newChunkSeriesIterator(cs []*ChunkMeta, s stone, mint, maxt int64) *chunkSe
mint: mint, mint: mint,
maxt: maxt, maxt: maxt,
stone: s, intervals: dranges,
} }
} }
@ -688,8 +688,8 @@ func (it *chunkSeriesIterator) Seek(t int64) (ok bool) {
it.i = x it.i = x
it.cur = it.chunks[x].Chunk.Iterator() it.cur = it.chunks[x].Chunk.Iterator()
if len(it.stone.ranges) > 0 { if len(it.intervals) > 0 {
it.cur = &deletedIterator{it: it.cur, dranges: it.stone.ranges} it.cur = &deletedIterator{it: it.cur, intervals: it.intervals}
} }
for it.cur.Next() { for it.cur.Next() {
@ -722,8 +722,8 @@ func (it *chunkSeriesIterator) Next() bool {
it.i++ it.i++
it.cur = it.chunks[it.i].Chunk.Iterator() it.cur = it.chunks[it.i].Chunk.Iterator()
if len(it.stone.ranges) > 0 { if len(it.intervals) > 0 {
it.cur = &deletedIterator{it: it.cur, dranges: it.stone.ranges} it.cur = &deletedIterator{it: it.cur, intervals: it.intervals}
} }
return it.Next() return it.Next()

16
querier_test.go
View file

@ -481,10 +481,10 @@ func TestBlockQuerierDelete(t *testing.T) {
}, },
}, },
tombstones: newMapTombstoneReader( tombstones: newMapTombstoneReader(
map[uint32][]trange{ map[uint32]intervals{
1: []trange{{1, 3}}, 1: intervals{{1, 3}},
2: []trange{{1, 3}, {6, 10}}, 2: intervals{{1, 3}, {6, 10}},
3: []trange{{6, 10}}, 3: intervals{{6, 10}},
}, },
), ),
@ -876,7 +876,7 @@ func TestSeriesIterator(t *testing.T) {
chunkFromSamples(tc.b), chunkFromSamples(tc.b),
chunkFromSamples(tc.c), chunkFromSamples(tc.c),
} }
res := newChunkSeriesIterator(chkMetas, stone{}, tc.mint, tc.maxt) res := newChunkSeriesIterator(chkMetas, nil, tc.mint, tc.maxt)
smplValid := make([]sample, 0) smplValid := make([]sample, 0)
for _, s := range tc.exp { for _, s := range tc.exp {
@ -947,7 +947,7 @@ func TestSeriesIterator(t *testing.T) {
chunkFromSamples(tc.b), chunkFromSamples(tc.b),
chunkFromSamples(tc.c), chunkFromSamples(tc.c),
} }
res := newChunkSeriesIterator(chkMetas, stone{}, tc.mint, tc.maxt) res := newChunkSeriesIterator(chkMetas, nil, tc.mint, tc.maxt)
smplValid := make([]sample, 0) smplValid := make([]sample, 0)
for _, s := range tc.exp { for _, s := range tc.exp {
@ -1094,8 +1094,8 @@ func (m *mockChunkSeriesSet) Next() bool {
return m.i < len(m.l) return m.i < len(m.l)
} }
func (m *mockChunkSeriesSet) At() (labels.Labels, []*ChunkMeta, stone) { func (m *mockChunkSeriesSet) At() (labels.Labels, []*ChunkMeta, intervals) {
return m.l[m.i], m.cm[m.i], stone{} return m.l[m.i], m.cm[m.i], nil
} }
func (m *mockChunkSeriesSet) Err() error { func (m *mockChunkSeriesSet) Err() error {

82
tombstones.go
View file

@ -36,14 +36,14 @@ func writeTombstoneFile(dir string, tr TombstoneReader) error {
// Write the ranges. // Write the ranges.
buf.reset() buf.reset()
buf.putVarint64(int64(len(s.ranges))) buf.putVarint64(int64(len(s.intervals)))
n, err := f.Write(buf.get()) n, err := f.Write(buf.get())
if err != nil { if err != nil {
return err return err
} }
pos += int64(n) pos += int64(n)
for _, r := range s.ranges { for _, r := range s.intervals {
buf.reset() buf.reset()
buf.putVarint64(r.mint) buf.putVarint64(r.mint)
buf.putVarint64(r.maxt) buf.putVarint64(r.maxt)
@ -93,8 +93,8 @@ func writeTombstoneFile(dir string, tr TombstoneReader) error {
// stone holds the information on the posting and time-range // stone holds the information on the posting and time-range
// that is deleted. // that is deleted.
type stone struct { type stone struct {
ref uint32 ref uint32
ranges []trange intervals intervals
} }
// TombstoneReader is the iterator over tombstones. // TombstoneReader is the iterator over tombstones.
@ -164,7 +164,7 @@ func (t *tombstoneReader) Next() bool {
return false return false
} }
dranges := make([]trange, 0, numRanges) dranges := make(intervals, 0, numRanges)
for i := 0; i < int(numRanges); i++ { for i := 0; i < int(numRanges); i++ {
mint := d.varint64() mint := d.varint64()
maxt := d.varint64() maxt := d.varint64()
@ -173,11 +173,11 @@ func (t *tombstoneReader) Next() bool {
return false return false
} }
dranges = append(dranges, trange{mint, maxt}) dranges = append(dranges, interval{mint, maxt})
} }
t.stones = t.stones[12:] t.stones = t.stones[12:]
t.cur = stone{ref: ref, ranges: dranges} t.cur = stone{ref: ref, intervals: dranges}
return true return true
} }
@ -217,10 +217,10 @@ type mapTombstoneReader struct {
refs []uint32 refs []uint32
cur uint32 cur uint32
stones map[uint32][]trange stones map[uint32]intervals
} }
func newMapTombstoneReader(ts map[uint32][]trange) *mapTombstoneReader { func newMapTombstoneReader(ts map[uint32]intervals) *mapTombstoneReader {
refs := make([]uint32, 0, len(ts)) refs := make([]uint32, 0, len(ts))
for k := range ts { for k := range ts {
refs = append(refs, k) refs = append(refs, k)
@ -231,7 +231,7 @@ func newMapTombstoneReader(ts map[uint32][]trange) *mapTombstoneReader {
} }
func newEmptyTombstoneReader() *mapTombstoneReader { func newEmptyTombstoneReader() *mapTombstoneReader {
return &mapTombstoneReader{stones: make(map[uint32][]trange)} return &mapTombstoneReader{stones: make(map[uint32]intervals)}
} }
func (t *mapTombstoneReader) Next() bool { func (t *mapTombstoneReader) Next() bool {
@ -265,7 +265,7 @@ func (t *mapTombstoneReader) Seek(ref uint32) bool {
} }
func (t *mapTombstoneReader) At() stone { func (t *mapTombstoneReader) At() stone {
return stone{ref: t.cur, ranges: t.stones[t.cur]} return stone{ref: t.cur, intervals: t.stones[t.cur]}
} }
func (t *mapTombstoneReader) Copy() TombstoneReader { func (t *mapTombstoneReader) Copy() TombstoneReader {
@ -285,11 +285,11 @@ type simpleTombstoneReader struct {
refs []uint32 refs []uint32
cur uint32 cur uint32
ranges []trange intervals intervals
} }
func newSimpleTombstoneReader(refs []uint32, drange []trange) *simpleTombstoneReader { func newSimpleTombstoneReader(refs []uint32, dranges intervals) *simpleTombstoneReader {
return &simpleTombstoneReader{refs: refs, ranges: drange} return &simpleTombstoneReader{refs: refs, intervals: dranges}
} }
func (t *simpleTombstoneReader) Next() bool { func (t *simpleTombstoneReader) Next() bool {
@ -323,11 +323,11 @@ func (t *simpleTombstoneReader) Seek(ref uint32) bool {
} }
func (t *simpleTombstoneReader) At() stone { func (t *simpleTombstoneReader) At() stone {
return stone{ref: t.cur, ranges: t.ranges} return stone{ref: t.cur, intervals: t.intervals}
} }
func (t *simpleTombstoneReader) Copy() TombstoneReader { func (t *simpleTombstoneReader) Copy() TombstoneReader {
return &simpleTombstoneReader{refs: t.refs[:], cur: t.cur, ranges: t.ranges} return &simpleTombstoneReader{refs: t.refs[:], cur: t.cur, intervals: t.intervals}
} }
func (t *simpleTombstoneReader) Err() error { func (t *simpleTombstoneReader) Err() error {
@ -378,8 +378,8 @@ func (t *mergedTombstoneReader) Next() bool {
t.bok = t.b.Next() t.bok = t.b.Next()
} else { } else {
// Merge time ranges. // Merge time ranges.
for _, r := range bcur.ranges { for _, r := range bcur.intervals {
acur.ranges = addNewInterval(acur.ranges, r) acur.intervals = acur.intervals.add(r)
} }
t.cur = acur t.cur = acur
@ -424,16 +424,16 @@ func (t *mergedTombstoneReader) Err() error {
return t.b.Err() return t.b.Err()
} }
type trange struct { type interval struct {
mint, maxt int64 mint, maxt int64
} }
func (tr trange) inBounds(t int64) bool { func (tr interval) inBounds(t int64) bool {
return t >= tr.mint && t <= tr.maxt return t >= tr.mint && t <= tr.maxt
} }
func (tr trange) isSubrange(ranges []trange) bool { func (tr interval) isSubrange(dranges intervals) bool {
for _, r := range ranges { for _, r := range dranges {
if r.inBounds(tr.mint) && r.inBounds(tr.maxt) { if r.inBounds(tr.mint) && r.inBounds(tr.maxt) {
return true return true
} }
@ -442,49 +442,51 @@ func (tr trange) isSubrange(ranges []trange) bool {
return false return false
} }
type intervals []interval
// This adds the new time-range to the existing ones. // This adds the new time-range to the existing ones.
// The existing ones must be sorted. // The existing ones must be sorted.
// TODO(gouthamve): {1, 2}, {3, 4} can be merged into {1, 4}. func (itvs intervals) add(n interval) intervals {
func addNewInterval(existing []trange, n trange) []trange { for i, r := range itvs {
for i, r := range existing {
// TODO(gouthamve): Make this codepath easier to digest. // TODO(gouthamve): Make this codepath easier to digest.
if r.inBounds(n.mint) { if r.inBounds(n.mint-1) || r.inBounds(n.mint) {
if n.maxt > r.maxt { if n.maxt > r.maxt {
existing[i].maxt = n.maxt itvs[i].maxt = n.maxt
} }
j := 0 j := 0
for _, r2 := range existing[i+1:] { for _, r2 := range itvs[i+1:] {
if n.maxt < r2.mint { if n.maxt < r2.mint {
break break
} }
j++ j++
} }
if j != 0 { if j != 0 {
if existing[i+j].maxt > n.maxt { if itvs[i+j].maxt > n.maxt {
existing[i].maxt = existing[i+j].maxt itvs[i].maxt = itvs[i+j].maxt
} }
existing = append(existing[:i+1], existing[i+j+1:]...) itvs = append(itvs[:i+1], itvs[i+j+1:]...)
} }
return existing return itvs
} }
if r.inBounds(n.maxt) { if r.inBounds(n.maxt+1) || r.inBounds(n.maxt) {
if n.mint < r.maxt { if n.mint < r.maxt {
existing[i].mint = n.mint itvs[i].mint = n.mint
} }
return existing return itvs
} }
if n.mint < r.mint { if n.mint < r.mint {
newRange := make([]trange, i, len(existing[:i])+1) newRange := make(intervals, i, len(itvs[:i])+1)
copy(newRange, existing[:i]) copy(newRange, itvs[:i])
newRange = append(newRange, n) newRange = append(newRange, n)
newRange = append(newRange, existing[i:]...) newRange = append(newRange, itvs[i:]...)
return newRange return newRange
} }
} }
existing = append(existing, n) itvs = append(itvs, n)
return existing return itvs
} }

246
tombstones_test.go
View file

@ -17,15 +17,15 @@ func TestWriteAndReadbackTombStones(t *testing.T) {
ref := uint32(0) ref := uint32(0)
stones := make(map[uint32][]trange) stones := make(map[uint32]intervals)
// Generate the tombstones. // Generate the tombstones.
for i := 0; i < 100; i++ { for i := 0; i < 100; i++ {
ref += uint32(rand.Int31n(10)) + 1 ref += uint32(rand.Int31n(10)) + 1
numRanges := rand.Intn(5) numRanges := rand.Intn(5)
dranges := make([]trange, numRanges) dranges := make(intervals, numRanges)
mint := rand.Int63n(time.Now().UnixNano()) mint := rand.Int63n(time.Now().UnixNano())
for j := 0; j < numRanges; j++ { for j := 0; j < numRanges; j++ {
dranges[j] = trange{mint, mint + rand.Int63n(1000)} dranges[j] = interval{mint, mint + rand.Int63n(1000)}
mint += rand.Int63n(1000) + 1 mint += rand.Int63n(1000) + 1
} }
stones[ref] = dranges stones[ref] = dranges
@ -49,54 +49,70 @@ func TestWriteAndReadbackTombStones(t *testing.T) {
func TestAddingNewIntervals(t *testing.T) { func TestAddingNewIntervals(t *testing.T) {
cases := []struct { cases := []struct {
exist []trange exist intervals
new trange new interval
exp []trange exp intervals
}{ }{
{ {
new: trange{1, 2}, new: interval{1, 2},
exp: []trange{{1, 2}}, exp: intervals{{1, 2}},
}, },
{ {
exist: []trange{{1, 10}, {12, 20}, {25, 30}}, exist: intervals{{1, 2}},
new: trange{21, 23}, new: interval{1, 2},
exp: []trange{{1, 10}, {12, 20}, {21, 23}, {25, 30}}, exp: intervals{{1, 2}},
}, },
{ {
exist: []trange{{1, 2}, {3, 5}, {6, 7}}, exist: intervals{{1, 4}, {6, 6}},
new: trange{6, 7}, new: interval{5, 6},
exp: []trange{{1, 2}, {3, 5}, {6, 7}}, exp: intervals{{1, 6}},
}, },
{ {
exist: []trange{{1, 10}, {12, 20}, {25, 30}}, exist: intervals{{1, 10}, {12, 20}, {25, 30}},
new: trange{21, 25}, new: interval{21, 23},
exp: []trange{{1, 10}, {12, 20}, {21, 30}}, exp: intervals{{1, 10}, {12, 23}, {25, 30}},
}, },
{ {
exist: []trange{{1, 10}, {12, 20}, {25, 30}}, exist: intervals{{1, 2}, {3, 5}, {7, 7}},
new: trange{18, 23}, new: interval{6, 7},
exp: []trange{{1, 10}, {12, 23}, {25, 30}}, exp: intervals{{1, 2}, {3, 7}},
}, },
{ {
exist: []trange{{1, 10}, {12, 20}, {25, 30}}, exist: intervals{{1, 10}, {12, 20}, {25, 30}},
new: trange{9, 23}, new: interval{21, 25},
exp: []trange{{1, 23}, {25, 30}}, exp: intervals{{1, 10}, {12, 30}},
}, },
{ {
exist: []trange{{1, 10}, {12, 20}, {25, 30}}, exist: intervals{{1, 10}, {12, 20}, {25, 30}},
new: trange{9, 230}, new: interval{18, 23},
exp: []trange{{1, 230}}, exp: intervals{{1, 10}, {12, 23}, {25, 30}},
}, },
{ {
exist: []trange{{5, 10}, {12, 20}, {25, 30}}, exist: intervals{{1, 10}, {12, 20}, {25, 30}},
new: trange{1, 4}, new: interval{9, 23},
exp: []trange{{1, 4}, {5, 10}, {12, 20}, {25, 30}}, exp: intervals{{1, 23}, {25, 30}},
},
{
exist: intervals{{1, 10}, {12, 20}, {25, 30}},
new: interval{9, 230},
exp: intervals{{1, 230}},
},
{
exist: intervals{{5, 10}, {12, 20}, {25, 30}},
new: interval{1, 4},
exp: intervals{{1, 10}, {12, 20}, {25, 30}},
},
{
exist: intervals{{5, 10}, {12, 20}, {25, 30}},
new: interval{11, 14},
exp: intervals{{5, 20}, {25, 30}},
}, },
} }
for _, c := range cases { for _, c := range cases {
require.Equal(t, c.exp, addNewInterval(c.exist, c.new))
require.Equal(t, c.exp, c.exist.add(c.new))
} }
return return
} }
@ -104,20 +120,20 @@ func TestAddingNewIntervals(t *testing.T) {
func TestTombstoneReadersSeek(t *testing.T) { func TestTombstoneReadersSeek(t *testing.T) {
// This is assuming that the listPostings is perfect. // This is assuming that the listPostings is perfect.
table := struct { table := struct {
m map[uint32][]trange m map[uint32]intervals
cases []uint32 cases []uint32
}{ }{
m: map[uint32][]trange{ m: map[uint32]intervals{
2: []trange{{1, 2}}, 2: intervals{{1, 2}},
3: []trange{{1, 4}, {5, 6}}, 3: intervals{{1, 4}, {5, 6}},
4: []trange{{10, 15}, {16, 20}}, 4: intervals{{10, 15}, {16, 20}},
5: []trange{{1, 4}, {5, 6}}, 5: intervals{{1, 4}, {5, 6}},
50: []trange{{10, 20}, {35, 50}}, 50: intervals{{10, 20}, {35, 50}},
600: []trange{{100, 2000}}, 600: intervals{{100, 2000}},
1000: []trange{}, 1000: intervals{},
1500: []trange{{10000, 500000}}, 1500: intervals{{10000, 500000}},
1600: []trange{{1, 2}, {3, 4}, {4, 5}, {6, 7}}, 1600: intervals{{1, 2}, {3, 4}, {4, 5}, {6, 7}},
}, },
cases: []uint32{1, 10, 20, 40, 30, 20, 50, 599, 601, 1000, 1600, 1601, 2000}, cases: []uint32{1, 10, 20, 40, 30, 20, 50, 599, 601, 1000, 1600, 1601, 2000},
@ -138,13 +154,13 @@ func TestTombstoneReadersSeek(t *testing.T) {
require.Equal(t, pr.Seek(ref), trc.Seek(ref)) require.Equal(t, pr.Seek(ref), trc.Seek(ref))
if pr.Seek(ref) { if pr.Seek(ref) {
require.Equal(t, pr.At(), trc.At().ref) require.Equal(t, pr.At(), trc.At().ref)
require.Equal(t, table.m[pr.At()], trc.At().ranges) require.Equal(t, table.m[pr.At()], trc.At().intervals)
} }
for pr.Next() { for pr.Next() {
require.True(t, trc.Next()) require.True(t, trc.Next())
require.Equal(t, pr.At(), trc.At().ref) require.Equal(t, pr.At(), trc.At().ref)
require.Equal(t, table.m[pr.At()], trc.At().ranges) require.Equal(t, table.m[pr.At()], trc.At().intervals)
} }
require.False(t, trc.Next()) require.False(t, trc.Next())
@ -172,7 +188,7 @@ func TestTombstoneReadersSeek(t *testing.T) {
return return
}) })
t.Run("simpleTombstoneReader", func(t *testing.T) { t.Run("simpleTombstoneReader", func(t *testing.T) {
ranges := []trange{{1, 2}, {3, 4}, {5, 6}} dranges := intervals{{1, 2}, {3, 4}, {5, 6}}
for _, ref := range table.cases { for _, ref := range table.cases {
// Create the listPostings. // Create the listPostings.
@ -182,19 +198,19 @@ func TestTombstoneReadersSeek(t *testing.T) {
} }
sort.Sort(uint32slice(refs)) sort.Sort(uint32slice(refs))
pr := newListPostings(refs[:]) pr := newListPostings(refs[:])
tr := newSimpleTombstoneReader(refs[:], ranges) tr := newSimpleTombstoneReader(refs[:], dranges)
// Compare both. // Compare both.
trc := tr.Copy() trc := tr.Copy()
require.Equal(t, pr.Seek(ref), trc.Seek(ref)) require.Equal(t, pr.Seek(ref), trc.Seek(ref))
if pr.Seek(ref) { if pr.Seek(ref) {
require.Equal(t, pr.At(), trc.At().ref) require.Equal(t, pr.At(), trc.At().ref)
require.Equal(t, ranges, tr.At().ranges) require.Equal(t, dranges, tr.At().intervals)
} }
for pr.Next() { for pr.Next() {
require.True(t, trc.Next()) require.True(t, trc.Next())
require.Equal(t, pr.At(), trc.At().ref, "refs") require.Equal(t, pr.At(), trc.At().ref, "refs")
require.Equal(t, ranges, trc.At().ranges) require.Equal(t, dranges, trc.At().intervals)
} }
require.False(t, trc.Next()) require.False(t, trc.Next())
@ -213,93 +229,93 @@ func TestMergedTombstoneReader(t *testing.T) {
}{ }{
{ {
a: newMapTombstoneReader( a: newMapTombstoneReader(
map[uint32][]trange{ map[uint32]intervals{
2: []trange{{1, 2}}, 2: intervals{{1, 2}},
3: []trange{{1, 4}, {5, 6}}, 3: intervals{{1, 4}, {6, 6}},
4: []trange{{10, 15}, {16, 20}}, 4: intervals{{10, 15}, {16, 20}},
5: []trange{{1, 4}, {5, 6}}, 5: intervals{{1, 4}, {5, 6}},
50: []trange{{10, 20}, {35, 50}}, 50: intervals{{10, 20}, {35, 50}},
600: []trange{{100, 2000}}, 600: intervals{{100, 2000}},
1000: []trange{}, 1000: intervals{},
1500: []trange{{10000, 500000}}, 1500: intervals{{10000, 500000}},
1600: []trange{{1, 2}, {3, 4}, {4, 5}, {6, 7}}, 1600: intervals{{1, 2}, {3, 4}, {4, 5}, {6, 7}},
}, },
), ),
b: newMapTombstoneReader( b: newMapTombstoneReader(
map[uint32][]trange{ map[uint32]intervals{
2: []trange{{1, 2}}, 2: intervals{{1, 2}},
3: []trange{{1, 4}, {5, 6}}, 3: intervals{{5, 6}},
4: []trange{{10, 15}, {16, 20}}, 4: intervals{{10, 15}, {16, 20}},
5: []trange{{1, 4}, {5, 6}}, 5: intervals{{1, 4}, {5, 6}},
50: []trange{{10, 20}, {35, 50}}, 50: intervals{{10, 20}, {35, 50}},
600: []trange{{100, 2000}}, 600: intervals{{100, 2000}},
1000: []trange{}, 1000: intervals{},
1500: []trange{{10000, 500000}}, 1500: intervals{{10000, 500000}},
1600: []trange{{1, 2}, {3, 4}, {4, 5}, {6, 7}}, 1600: intervals{{1, 2}, {3, 4}, {4, 5}, {6, 7}},
}, },
), ),
exp: newMapTombstoneReader( exp: newMapTombstoneReader(
map[uint32][]trange{ map[uint32]intervals{
2: []trange{{1, 2}}, 2: intervals{{1, 2}},
3: []trange{{1, 4}, {5, 6}}, 3: intervals{{1, 6}},
4: []trange{{10, 15}, {16, 20}}, 4: intervals{{10, 20}},
5: []trange{{1, 4}, {5, 6}}, 5: intervals{{1, 6}},
50: []trange{{10, 20}, {35, 50}}, 50: intervals{{10, 20}, {35, 50}},
600: []trange{{100, 2000}}, 600: intervals{{100, 2000}},
1000: []trange{}, 1000: intervals{},
1500: []trange{{10000, 500000}}, 1500: intervals{{10000, 500000}},
1600: []trange{{1, 2}, {3, 5}, {6, 7}}, 1600: intervals{{1, 7}},
}, },
), ),
}, },
{ {
a: newMapTombstoneReader( a: newMapTombstoneReader(
map[uint32][]trange{ map[uint32]intervals{
2: []trange{{1, 2}}, 2: intervals{{1, 2}},
3: []trange{{1, 4}, {5, 6}}, 3: intervals{{1, 4}, {6, 6}},
4: []trange{{10, 15}, {16, 20}}, 4: intervals{{10, 15}, {17, 20}},
5: []trange{{1, 4}, {5, 6}}, 5: intervals{{1, 6}},
50: []trange{{10, 20}, {35, 50}}, 50: intervals{{10, 20}, {35, 50}},
600: []trange{{100, 2000}}, 600: intervals{{100, 2000}},
1000: []trange{}, 1000: intervals{},
1500: []trange{{10000, 500000}}, 1500: intervals{{10000, 500000}},
1600: []trange{{1, 2}, {3, 4}, {4, 5}, {6, 7}}, 1600: intervals{{1, 2}, {3, 4}, {4, 5}, {6, 7}},
}, },
), ),
b: newMapTombstoneReader( b: newMapTombstoneReader(
map[uint32][]trange{ map[uint32]intervals{
20: []trange{{1, 2}}, 20: intervals{{1, 2}},
30: []trange{{1, 4}, {5, 6}}, 30: intervals{{1, 4}, {5, 6}},
40: []trange{{10, 15}, {16, 20}}, 40: intervals{{10, 15}, {16, 20}},
60: []trange{{1, 4}, {5, 6}}, 60: intervals{{1, 4}, {5, 6}},
500: []trange{{10, 20}, {35, 50}}, 500: intervals{{10, 20}, {35, 50}},
6000: []trange{{100, 2000}}, 6000: intervals{{100, 2000}},
10000: []trange{}, 10000: intervals{},
15000: []trange{{10000, 500000}}, 15000: intervals{{10000, 500000}},
1600: []trange{{1, 2}, {3, 4}, {4, 5}, {6, 7}}, 1600: intervals{{1, 2}, {3, 4}, {4, 5}, {6, 7}},
}, },
), ),
exp: newMapTombstoneReader( exp: newMapTombstoneReader(
map[uint32][]trange{ map[uint32]intervals{
2: []trange{{1, 2}}, 2: intervals{{1, 2}},
3: []trange{{1, 4}, {5, 6}}, 3: intervals{{1, 4}, {6, 6}},
4: []trange{{10, 15}, {16, 20}}, 4: intervals{{10, 15}, {17, 20}},
5: []trange{{1, 4}, {5, 6}}, 5: intervals{{1, 6}},
50: []trange{{10, 20}, {35, 50}}, 50: intervals{{10, 20}, {35, 50}},
600: []trange{{100, 2000}}, 600: intervals{{100, 2000}},
1000: []trange{}, 1000: intervals{},
1500: []trange{{10000, 500000}}, 1500: intervals{{10000, 500000}},
20: []trange{{1, 2}}, 20: intervals{{1, 2}},
30: []trange{{1, 4}, {5, 6}}, 30: intervals{{1, 4}, {5, 6}},
40: []trange{{10, 15}, {16, 20}}, 40: intervals{{10, 15}, {16, 20}},
60: []trange{{1, 4}, {5, 6}}, 60: intervals{{1, 4}, {5, 6}},
500: []trange{{10, 20}, {35, 50}}, 500: intervals{{10, 20}, {35, 50}},
6000: []trange{{100, 2000}}, 6000: intervals{{100, 2000}},
10000: []trange{}, 10000: intervals{},
15000: []trange{{10000, 500000}}, 15000: intervals{{10000, 500000}},
1600: []trange{{1, 2}, {3, 5}, {6, 7}}, 1600: intervals{{1, 7}},
}, },
), ),
}, },