From a009247ab7d37c866ca08b86b0396c1301c1e744 Mon Sep 17 00:00:00 2001
From: Fabian Reinartz <fab.reinartz@gmail.com>
Date: Fri, 30 Dec 2016 19:36:28 +0100
Subject: [PATCH] Add matching of empty label

---
 db.go      |  2 +-
 querier.go | 44 ++++++++++++++++++++++++++++++++------------
 wal.go     |  7 +++++++
 3 files changed, 40 insertions(+), 13 deletions(-)

diff --git a/db.go b/db.go
index d43cbd00f..24d2cd598 100644
--- a/db.go
+++ b/db.go
@@ -251,7 +251,7 @@ func (s *Shard) appendBatch(samples []hashedSample) error {
 	err := s.head.appendBatch(samples)
 
 	// TODO(fabxc): randomize over time and use better scoring function.
-	if s.head.stats.SampleCount/(uint64(s.head.stats.ChunkCount)+1) > 400 {
+	if s.head.stats.SampleCount/(uint64(s.head.stats.ChunkCount)+1) > 24000 {
 		select {
 		case s.persistCh <- struct{}{}:
 			go func() {
diff --git a/querier.go b/querier.go
index 3f3f1e213..5e54020fc 100644
--- a/querier.go
+++ b/querier.go
@@ -220,16 +220,26 @@ func newBlockQuerier(ix IndexReader, s SeriesReader, mint, maxt int64) *blockQue
 }
 
 func (q *blockQuerier) Select(ms ...labels.Matcher) SeriesSet {
-	var its []Postings
+	var (
+		its    []Postings
+		absent []string
+	)
 	for _, m := range ms {
+		// If the matcher checks absence of a label, don't select them
+		// but propagate the check into the series set.
+		if _, ok := m.(*labels.EqualMatcher); ok && m.Matches("") {
+			absent = append(absent, m.Name())
+			continue
+		}
 		its = append(its, q.selectSingle(m))
 	}
 
 	return &blockSeriesSet{
-		index: q.index,
-		it:    Intersect(its...),
-		mint:  q.mint,
-		maxt:  q.maxt,
+		index:  q.index,
+		it:     Intersect(its...),
+		absent: absent,
+		mint:   q.mint,
+		maxt:   q.maxt,
 	}
 }
 
@@ -412,8 +422,9 @@ func (s *shardSeriesSet) Next() bool {
 // blockSeriesSet is a set of series from an inverted index query.
 type blockSeriesSet struct {
 	index      IndexReader
-	it         Postings
-	mint, maxt int64
+	it         Postings // postings list referencing series
+	absent     []string // labels that must not be set for result series
+	mint, maxt int64    // considered time range
 
 	err error
 	cur Series
@@ -421,18 +432,27 @@ type blockSeriesSet struct {
 
 func (s *blockSeriesSet) Next() bool {
 	// Step through the postings iterator to find potential series.
-	// Resolving series may return nil if no applicable data for the
-	// time range exists and we can skip to the next series.
+outer:
 	for s.it.Next() {
 		series, err := s.index.Series(s.it.Value(), s.mint, s.maxt)
 		if err != nil {
 			s.err = err
 			return false
 		}
-		if series != nil {
-			s.cur = series
-			return true
+		// Resolving series may return nil if no applicable data for the
+		// time range exists and we can skip to the next series.
+		if series == nil {
+			continue
 		}
+		// If a series contains a label that must be absent, it is skipped as well.
+		for _, abs := range s.absent {
+			if series.Labels().Get(abs) != "" {
+				continue outer
+			}
+		}
+
+		s.cur = series
+		return true
 	}
 	if s.it.Err() != nil {
 		s.err = s.it.Err()
diff --git a/wal.go b/wal.go
index 18f66edb9..dd06f8766 100644
--- a/wal.go
+++ b/wal.go
@@ -188,7 +188,14 @@ func (e *walEncoder) encodeSeries(series []labels.Labels) error {
 		e.buf = append(e.buf, b[:n]...)
 
 		for _, l := range lset {
+			// func() {
+			// 	defer func() {
+			// 		if recover() != nil {
+			// 			fmt.Println(l)
+			// 		}
+			// 	}()
 			n = binary.PutUvarint(b, uint64(len(l.Name)))
+			// }()
 			e.buf = append(e.buf, b[:n]...)
 			e.buf = append(e.buf, l.Name...)