Merge branch 'master' into pg-sql

api.go

@@ -1114,7 +1114,28 @@ func (api *API) ImportAtomicRecord(ctx context.Context, req *AtomicRecord, opts
 	return tx.Commit()
 }
 
+// This is a hide your face ugly hack, forced upon
+// us by the horrible invention of function based options
+// by the usually brilliant Rob Pike. - JEA
+func addClearToImportOptions(opts []ImportOption) []ImportOption {
+	var opt ImportOptions
+	for _, o := range opts {
+		// check for side-effect of setting io.Clear; that is
+		// how we know it is present.
+		_ = o(&opt)
+		if opt.Clear {
+			// we already have the clear flag set, so nothing more to do.
+			return opts
+		}
+	}
+	// no clear flag being set, add that option now.
+	return append(opts, OptImportOptionsClear(true))
+}
+
 func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOption) error {
+	if req.Clear {
+		opts = addClearToImportOptions(opts)
+	}
 	return api.ImportWithTx(ctx, nil, req, opts...)
 }
 
@@ -1254,6 +1275,9 @@ func (api *API) ImportWithTx(ctx context.Context, tx Tx, req *ImportRequest, opt
 }
 
 func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts ...ImportOption) error {
+	if req.Clear {
+		opts = addClearToImportOptions(opts)
+	}
 	return api.ImportValueWithTx(ctx, nil, req, opts...)
 }
 

api_test.go

@@ -474,3 +474,127 @@ type offsetModHasher struct{}
 func (*offsetModHasher) Hash(key uint64, n int) int {
 	return int(key+1) % n
 }
+
+func TestAPI_ClearFlagForImportAndImportValues(t *testing.T) {
+	c := test.MustRunCluster(t, 1,
+		[]server.CommandOption{
+			server.OptCommandServerOptions(
+				pilosa.OptServerNodeID("node0"),
+				pilosa.OptServerClusterHasher(&offsetModHasher{}),
+				pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
+			)},
+	)
+	defer c.Close()
+
+	// plan:
+	//  1. set a bit
+	//  2. clear with Import() using the ImportRequest.Clear flag
+	//  3. verifiy the clear is done.
+	//  repeat for ImportValueRequest and ImportValues()
+
+	m0 := c[0]
+	m0api := m0.API
+
+	ctx := context.Background()
+	index := "i"
+	fieldAcct0 := "acct0"
+
+	opts := pilosa.OptFieldTypeInt(-1000, 1000)
+
+	_, err := m0api.CreateIndex(ctx, index, pilosa.IndexOptions{})
+	if err != nil {
+		t.Fatalf("creating index: %v", err)
+	}
+	_, err = m0api.CreateField(ctx, index, fieldAcct0, opts)
+	if err != nil {
+		t.Fatalf("creating fieldAcct0: %v", err)
+	}
+
+	iraField := "ira" // set field.
+	iraRowID := uint64(3)
+	_, err = m0api.CreateField(ctx, index, iraField)
+	if err != nil {
+		t.Fatalf("creating fieldIRA: %v", err)
+	}
+
+	acctOwnerID := uint64(78) // ColumnID
+	shard := acctOwnerID / ShardWidth
+	acct0bal := int64(500)
+
+	ivr0 := &pilosa.ImportValueRequest{
+		Index:     index,
+		Field:     fieldAcct0,
+		Shard:     shard,
+		ColumnIDs: []uint64{acctOwnerID},
+		Values:    []int64{acct0bal},
+	}
+	ir0 := &pilosa.ImportRequest{
+		Index:     index,
+		Field:     iraField,
+		Shard:     shard,
+		ColumnIDs: []uint64{acctOwnerID},
+		RowIDs:    []uint64{iraRowID},
+	}
+
+	if err := m0api.Import(ctx, ir0); err != nil {
+		t.Fatal(err)
+	}
+	if err := m0api.ImportValue(ctx, ivr0); err != nil {
+		t.Fatal(err)
+	}
+
+	bitIsSet := func() bool {
+		query := fmt.Sprintf("Row(%v=%v)", iraField, iraRowID)
+		res, err := m0api.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query})
+		panicOn(err)
+		cols := res.Results[0].(*pilosa.Row).Columns()
+		for i := range cols {
+			if cols[i] == acctOwnerID {
+				return true
+			}
+		}
+		return false
+	}
+
+	if !bitIsSet() {
+		panic("IRA bit should have been set")
+	}
+
+	queryAcct := func(m0api *pilosa.API, acctOwnerID uint64, fieldAcct0, index string) (acctBal int64) {
+		query := fmt.Sprintf("FieldValue(field=%v, column=%v)", fieldAcct0, acctOwnerID)
+		res, err := m0api.Query(context.Background(), &pilosa.QueryRequest{Index: index, Query: query})
+		panicOn(err)
+
+		if len(res.Results) == 0 {
+			return 0
+		}
+		valCount := res.Results[0].(pilosa.ValCount)
+		return valCount.Val
+	}
+
+	bal := queryAcct(m0api, acctOwnerID, fieldAcct0, index)
+
+	if bal != acct0bal {
+		panic(fmt.Sprintf("expected %v, observed %v starting acct0 balance", acct0bal, bal))
+	}
+
+	// clear the bit
+	ir0.Clear = true
+	if err := m0api.Import(ctx, ir0); err != nil {
+		t.Fatal(err)
+	}
+
+	if bitIsSet() {
+		panic("IRA bit should have been cleared")
+	}
+
+	// clear the BSI
+	ivr0.Clear = true
+	if err := m0api.ImportValue(ctx, ivr0); err != nil {
+		t.Fatal(err)
+	}
+	bal = queryAcct(m0api, acctOwnerID, fieldAcct0, index)
+	if bal != 0 {
+		panic(fmt.Sprintf("expected %v, observed %v starting acct0 balance", acct0bal, 0))
+	}
+}

fragment_internal_test.go

@@ -29,6 +29,7 @@ import (
 	"runtime"
 	"runtime/debug"
 	"sort"
+	"strings"
 	"testing"
 	"testing/quick"
 
@@ -190,6 +191,8 @@ func TestFragment_RowcacheMap(t *testing.T) {
 
 // Ensure a fragment can clear a row.
 func TestFragment_ClearRow(t *testing.T) {
+	notBlueGreenTest(t)
+
 	f, idx := mustOpenFragment("i", "f", viewStandard, 0, "")
 	_ = idx
 	defer f.Clean(t)
@@ -225,6 +228,7 @@ func TestFragment_ClearRow(t *testing.T) {
 
 // Ensure a fragment can set a row.
 func TestFragment_SetRow(t *testing.T) {
+	notBlueGreenTest(t)
 	f, idx := mustOpenFragment("i", "f", viewStandard, 7, "")
 	_ = idx
 	defer f.Clean(t)
@@ -5644,3 +5648,12 @@ func TestFragment_Bug_Q2DoubleDelete(t *testing.T) {
 		t.Fatalf("expected nothing got %v", res)
 	}
 }
+
+func notBlueGreenTest(t *testing.T) {
+	src := os.Getenv("PILOSA_TXSRC")
+	if strings.Contains(src, "_") {
+		if strings.Contains(src, "roaring") {
+			t.Skip("skip under blue green with roaring")
+		}
+	}
+}

http/translator.go

@@ -22,19 +22,26 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"sync"
 
 	"github.com/pilosa/pilosa/v2"
 	"github.com/pilosa/pilosa/v2/logger"
 )
 
 func GetOpenTranslateReaderFunc(client *http.Client) pilosa.OpenTranslateReaderFunc {
+	return GetOpenTranslateReaderWithLockerFunc(client, nopLocker{})
+}
+
+func GetOpenTranslateReaderWithLockerFunc(client *http.Client, locker sync.Locker) pilosa.OpenTranslateReaderFunc {
 	return func(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap) (pilosa.TranslateEntryReader, error) {
-		return openTranslateReader(ctx, nodeURL, offsets, client)
+		return openTranslateReader(ctx, nodeURL, offsets, client, locker)
 	}
 }
 
-func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap, client *http.Client) (pilosa.TranslateEntryReader, error) {
+func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap, client *http.Client, locker sync.Locker) (pilosa.TranslateEntryReader, error) {
 	r := NewTranslateEntryReader(ctx, client)
+	r.locker = locker
+
 	r.URL = nodeURL + "/internal/translate/data"
 	r.Offsets = offsets
 	if err := r.Open(); err != nil {
@@ -43,9 +50,16 @@ func openTranslateReader(ctx context.Context, nodeURL string, offsets pilosa.Tra
 	return r, nil
 }
 
+type nopLocker struct{}
+
+func (nopLocker) Lock()   {}
+func (nopLocker) Unlock() {}
+
 // TranslateEntryReader represents an implementation of pilosa.TranslateEntryReader.
 // It consolidates all index & field translate entries into a single reader.
 type TranslateEntryReader struct {
+	locker sync.Locker
+
 	ctx    context.Context
 	cancel func()
 
@@ -70,7 +84,7 @@ func NewTranslateEntryReader(ctx context.Context, client *http.Client) *Translat
 	if client == nil {
 		client = http.DefaultClient
 	}
-	r := &TranslateEntryReader{HTTPClient: client, Logger: logger.NopLogger}
+	r := &TranslateEntryReader{locker: nopLocker{}, HTTPClient: client, Logger: logger.NopLogger}
 	r.ctx, r.cancel = context.WithCancel(ctx)
 	return r
 }
@@ -116,7 +130,10 @@ func (r *TranslateEntryReader) Close() error {
 		r.cancel()
 	}
 	if r.body != nil {
-		return r.body.Close()
+		r.locker.Lock()
+		err := r.body.Close()
+		r.locker.Unlock()
+		return err
 	}
 	return nil
 }
@@ -124,5 +141,8 @@ func (r *TranslateEntryReader) Close() error {
 // ReadEntry reads the next entry from the stream into entry.
 // Returns io.EOF at the end of the stream.
 func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
+	r.locker.Lock()
+	defer r.locker.Unlock()
+
 	return r.dec.Decode(&entry)
 }

http/translator_test.go

@@ -17,6 +17,7 @@ package http_test
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -151,3 +152,84 @@ func TestTranslateStore_EntryReader(t *testing.T) {
 		})
 	*/
 }
+
+func benchmarkSetup(b *testing.B, ctx context.Context, key string, nkeys int) (string, pilosa.TranslateOffsetMap, func()) {
+	b.Helper()
+
+	cluster := test.MustRunCluster(b, 1)
+	primary := cluster[0]
+
+	idx := primary.MustCreateIndex(b, "i", pilosa.IndexOptions{})
+	fld := primary.MustCreateField(b, idx.Name(), "f", pilosa.OptFieldKeys())
+	offset := make(pilosa.TranslateOffsetMap)
+	offset.SetIndexPartitionOffset(idx.Name(), 0, 1)
+	offset.SetFieldOffset(idx.Name(), fld.Name(), 1)
+
+	// Set data on the primary node.
+	for k := 0; k < nkeys; k++ {
+		if _, err := primary.API.Query(ctx, &pilosa.QueryRequest{
+			Index: idx.Name(),
+			Query: fmt.Sprintf(`Set(%d, %s="%s%[1]d")`, k, fld.Name(), key),
+		}); err != nil {
+			b.Fatalf("quering api: %+v", err)
+		}
+	}
+
+	return primary.URL(), offset, func() {
+		b.Helper()
+
+		if err := primary.API.DeleteIndex(ctx, idx.Name()); err != nil {
+			panic(err)
+		}
+		if err := cluster.Close(); err != nil {
+			panic(err)
+		}
+	}
+}
+
+func benchmarkReadEntry(b *testing.B, r pilosa.TranslateEntryReader, key string, nkeys int) {
+	var entry pilosa.TranslateEntry
+	for k := 0; k < nkeys; k++ {
+		if err := r.ReadEntry(&entry); err != nil {
+			b.Fatalf("reading entry: %+v", err)
+		}
+		if entry.Key != fmt.Sprintf("%s%d", key, k) {
+			b.Fatalf("got: %s, expected: %s%d", entry.Key, key, k)
+		}
+	}
+}
+
+const (
+	key   = "foo"
+	nkeys = 1000
+)
+
+func BenchmarkReadEntryNoMutex(b *testing.B) {
+	ctx := context.Background()
+	url, offset, teardown := benchmarkSetup(b, ctx, key, nkeys)
+	defer teardown()
+
+	for n := 0; n < b.N; n++ {
+		r, err := http.GetOpenTranslateReaderFunc(nil)(ctx, url, offset)
+		if err != nil {
+			b.Fatalf("opening translate reader: %+v", err)
+		}
+		benchmarkReadEntry(b, r, key, nkeys)
+		r.Close()
+	}
+}
+
+func BenchmarkReadEntryWithMutex(b *testing.B) {
+	ctx := context.Background()
+	url, offset, teardown := benchmarkSetup(b, ctx, key, nkeys)
+	defer teardown()
+
+	for n := 0; n < b.N; n++ {
+		r, err := http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})(ctx, url, offset)
+		if err != nil {
+			b.Fatalf("opening translate reader: %+v", err)
+		}
+		benchmarkReadEntry(b, r, key, nkeys)
+		r.Close()
+	}
+}

lmdb.go

@@ -147,9 +147,9 @@ func (r *lmdbRegistrar) openLMDBWrapper(path0 string) (*LMDBWrapper, error) {
 
 	flags = flags |
 		lmdb.WriteMap | // Use a writable memory map.
-		lmdb.NoMetaSync | // Don't fsync metapage after commit.
-		lmdb.NoSync | // Don't fsync after commit.
-		lmdb.MapAsync | // Flush asynchronously when using the WriteMap flag.
+		//lmdb.NoMetaSync | // Don't fsync metapage after commit.
+		//lmdb.NoSync | // Don't fsync after commit.
+		//lmdb.MapAsync | // Flush asynchronously when using the WriteMap flag.
 		lmdb.NoMemInit // Disable LMDB memory initialization
 
 	err = env.Open(path, flags, 0644)
@@ -344,7 +344,7 @@ func (tx *LMDBTx) Type() string {
 }
 
 func (tx *LMDBTx) UseRowCache() bool {
-	return false
+	return true
 }
 
 // Pointer gives us a memory address for the underlying transaction for debugging.

pg/server_test.go

@@ -59,7 +59,7 @@ func TestStartupInvalidLength(t *testing.T) {
 	res := testing.Benchmark(func(b *testing.B) {
 		connect, shutdown, err := pgtest.ServeMem(&pg.Server{
 			MaxStartupSize: 1024,
-			Logger:         logger.NewLogfLogger(t),
+			Logger:         logger.NopLogger,
 		})
 		if err != nil {
 			t.Fatalf("starting in-memory postgres server: %v", err)

server/handler_test.go

@@ -27,6 +27,7 @@ import (
 	"net/http/httptest"
 	"reflect"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -1300,7 +1301,7 @@ func TestCluster_TranslateStore(t *testing.T) {
 	cluster[0] = test.NewCommandNode(true,
 		server.OptCommandServerOptions(
 			pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
-			pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
+			pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})),
 		),
 	)
 	cluster[0].Config.Gossip.Port = "0"
@@ -1329,7 +1330,7 @@ func TestClusterTranslator(t *testing.T) {
 	cluster[1] = test.NewCommandNode(false,
 		server.OptCommandServerOptions(
 			pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
-			pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
+			pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(nil, &sync.Mutex{})),
 		),
 	)
 	cluster[1].Config.Gossip.Port = "0"

server/server.go

@@ -31,6 +31,7 @@ import (
 	"os/signal"
 	"runtime"
 	"strconv"
+	"sync"
 	"syscall"
 	"time"
 
@@ -389,7 +390,7 @@ func (m *Command) SetupServer() error {
 		pilosa.OptServerDiagnosticsInterval(diagnosticsInterval),
 		pilosa.OptServerExecutorPoolSize(m.Config.WorkerPoolSize),
 		pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
-		pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(c)),
+		pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderWithLockerFunc(c, &sync.Mutex{})),
 		pilosa.OptServerLogger(m.logger),
 		pilosa.OptServerAttrStoreFunc(boltdb.NewAttrStore),
 		pilosa.OptServerSystemInfo(gopsutil.NewSystemInfo()),

tx_test.go

@@ -59,7 +59,10 @@ func queryBalances(m0api *pilosa.API, acctOwnerID uint64, fldAcct0, fldAcct1, in
 	return
 }
 func skipForRoaring(t *testing.T) {
-	if strings.Contains(os.Getenv("PILOSA_TXSRC"), "roaring") {
+	src := os.Getenv("PILOSA_TXSRC")
+	// once txfactory.go DefaultTxsrc != RoaringTxn, this
+	// will break, of course. Take out the src == "" below.
+	if src == "" || strings.Contains(src, "roaring") {
 		t.Skip("skip if roaring pseudo-txn involved -- won't show transactional rollback")
 	}
 }