ImportRequest.Clear and ImportValuesRequest.Clear respected by api.Import() and api.ImportValues()

- tested in TestAPI_ClearFlagForImportAndImportValues api_test.go

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")
+		}
+	}
+}

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.

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")
 	}
 }