Isolation

block.go

@@ -130,12 +130,6 @@ type BlockReader interface {
 	Tombstones() (TombstoneReader, error)
 }
 
-// Appendable defines an entity to which data can be appended.
-type Appendable interface {
-	// Appender returns a new Appender against an underlying store.
-	Appender() Appender
-}
-
 // BlockMeta provides meta information about a block.
 type BlockMeta struct {
 	// Unique identifier for the block and its contents. Changes on compaction.

db.go

@@ -355,6 +355,7 @@ func (a dbAppender) Commit() error {
 		default:
 		}
 	}
+
 	return err
 }
 
@@ -683,7 +684,9 @@ func (db *DB) Querier(mint, maxt int64) (Querier, error) {
 
 	sq := &querier{
 		blocks: make([]Querier, 0, len(blocks)),
+		db:     db,
 	}
+
 	for _, b := range blocks {
 		q, err := NewBlockQuerier(b, mint, maxt)
 		if err == nil {

db_test.go

@@ -19,11 +19,13 @@ import (
 	"math/rand"
 	"os"
 	"sort"
+	"strconv"
 	"testing"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/tsdb/labels"
 	"github.com/prometheus/tsdb/testutil"
+	"github.com/stretchr/testify/require"
 )
 
 func openTestDB(t testing.TB, opts *Options) (db *DB, close func()) {
@@ -55,6 +57,10 @@ func query(t testing.TB, q Querier, matchers ...labels.Matcher) map[string][]sam
 		}
 		testutil.Ok(t, it.Err())
 
+		if len(samples) == 0 {
+			continue
+		}
+
 		name := series.Labels().String()
 		result[name] = samples
 	}
@@ -892,3 +898,128 @@ func expandSeriesSet(ss SeriesSet) ([]labels.Labels, error) {
 
 	return result, ss.Err()
 }
+
+func TestDBCannotSeePartialCommits(t *testing.T) {
+	tmpdir, _ := ioutil.TempDir("", "test")
+	defer os.RemoveAll(tmpdir)
+
+	db, err := Open(tmpdir, nil, nil, nil)
+	require.NoError(t, err)
+	defer db.Close()
+
+	stop := make(chan struct{})
+	firstInsert := make(chan struct{})
+
+	// Insert data in batches.
+	go func() {
+		iter := 0
+		for {
+			app := db.Appender()
+
+			for j := 0; j < 100; j++ {
+				_, err := app.Add(labels.FromStrings("foo", "bar", "a", strconv.Itoa(j)), int64(iter), float64(iter))
+				require.NoError(t, err)
+			}
+			err = app.Commit()
+			require.NoError(t, err)
+
+			if iter == 0 {
+				close(firstInsert)
+			}
+			iter++
+
+			select {
+			case <-stop:
+				return
+			default:
+			}
+		}
+	}()
+
+	<-firstInsert
+
+	// This is a race condition, so do a few tests to tickle it.
+	// Usually most will fail.
+	inconsistencies := 0
+	for i := 0; i < 10; i++ {
+		func() {
+			querier, err := db.Querier(0, 1000000)
+			testutil.Ok(t, err)
+			defer querier.Close()
+
+			ss, err := querier.Select(labels.NewEqualMatcher("foo", "bar"))
+			testutil.Ok(t, err)
+
+			seriesSet := readSeriesSet(t, ss)
+
+			require.NoError(t, err)
+			values := map[float64]struct{}{}
+			for _, series := range seriesSet {
+				values[series[len(series)-1].v] = struct{}{}
+			}
+			if len(values) != 1 {
+				inconsistencies++
+			}
+		}()
+	}
+	stop <- struct{}{}
+
+	require.Equal(t, 0, inconsistencies, "Some queries saw inconsistent results.")
+}
+
+func TestDBQueryDoesntSeeAppendsAfterCreation(t *testing.T) {
+	tmpdir, _ := ioutil.TempDir("", "test")
+	defer os.RemoveAll(tmpdir)
+
+	db, err := Open(tmpdir, nil, nil, nil)
+	require.NoError(t, err)
+	defer db.Close()
+
+	querier, err := db.Querier(0, 1000000)
+	testutil.Ok(t, err)
+	defer querier.Close()
+
+	app := db.Appender()
+	_, err = app.Add(labels.FromStrings("foo", "bar"), 0, 0)
+	require.NoError(t, err)
+	// This commit is after the querier is created, so should not be returned.
+	err = app.Commit()
+	require.NoError(t, err)
+
+	ss, err := querier.Select(labels.NewEqualMatcher("foo", "bar"))
+	testutil.Ok(t, err)
+
+	seriesSet := readSeriesSet(t, ss)
+	require.Equal(t, map[string][]sample{}, seriesSet)
+
+	querier, err = db.Querier(0, 1000000)
+	testutil.Ok(t, err)
+	defer querier.Close()
+
+	ss, err = querier.Select(labels.NewEqualMatcher("foo", "bar"))
+	seriesSet = readSeriesSet(t, ss)
+	require.Equal(t, seriesSet, map[string][]sample{`{foo="bar"}`: []sample{{t: 0, v: 0}}})
+}
+
+func readSeriesSet(t *testing.T, ss SeriesSet) map[string][]sample {
+	seriesSet := make(map[string][]sample)
+	for ss.Next() {
+		series := ss.At()
+
+		samples := []sample{}
+		it := series.Iterator()
+		for it.Next() {
+			t, v := it.At()
+			samples = append(samples, sample{t: t, v: v})
+		}
+		if len(samples) == 0 {
+			continue
+		}
+
+		name := series.Labels().String()
+		seriesSet[name] = samples
+	}
+	testutil.Ok(t, ss.Err())
+
+	return seriesSet
+}