diff --git a/api.go b/api.go index d8cdef042..8d3bfcea5 100644 --- a/api.go +++ b/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...) } diff --git a/api_test.go b/api_test.go index f592afb6e..88bb65510 100644 --- a/api_test.go +++ b/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)) + } +} diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 8ef6d5909..d033a24e6 100644 --- a/fragment_internal_test.go +++ b/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") + } + } +} diff --git a/lmdb.go b/lmdb.go index 9e0d30ed9..b67436abf 100644 --- a/lmdb.go +++ b/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. diff --git a/tx_test.go b/tx_test.go index 300e3444b..ae5e0b984 100644 --- a/tx_test.go +++ b/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") } }