scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1897 from molecula/fb-1114-rip-roaring

Browse source 
rip out roaring backend support
This commit is contained in:
Matthew Jaffee 2022-02-03 11:16:23 -06:00 • committed by GitHub
commit 063bdaf41e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
38 changed files with 214 additions and 3209 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
.gitlab/.gitlab-ci.yml
View file

@ -14,6 +14,16 @@ stages:
- gauntlet
- post build
smoke build:
image: golang:$GOVERSION
stage: lint
allow_failure: false
rules:
- if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
script:
- echo "Let's just see if it compiles... (sometimes the linter gives unclear errors if it doesn't)"
- go build ./...
golangci-lint:
image: golangci/golangci-lint:v1.39.0
stage: lint
@ -22,7 +32,7 @@ golangci-lint:
- if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
script:
- echo "Checking for issues in new code"
- golangci-lint run -v
- golangci-lint run
build lattice:
stage: test

14
Makefile
View file

@ -79,9 +79,6 @@ testvsub-race:
cd ..; \
done
tour:
./tournament.sh
bench:
$(GO) test ./... -bench=. -run=NoneZ -timeout=127m $(TESTFLAGS)
@ -159,7 +156,7 @@ clustertests: vendor
$(DOCKER_COMPOSE) -f internal/clustertests/docker-compose.yml down
# Run the cluster tests with authentication enabled
AUTH_ARGS="-c /go/src/github.com/molecula/featurebase/internal/authclustertests/testdata/featurebase.conf"
AUTH_ARGS="-c /go/src/github.com/molecula/featurebase/internal/clustertests/testdata/featurebase.conf"
authclustertests: vendor
CLUSTERTESTS_FB_ARGS=$(AUTH_ARGS) $(DOCKER_COMPOSE) -f internal/clustertests/docker-compose.yml down
CLUSTERTESTS_FB_ARGS=$(AUTH_ARGS) $(DOCKER_COMPOSE) -f internal/clustertests/docker-compose.yml build
@ -349,14 +346,5 @@ install-gometalinter:
GO111MODULE=off gometalinter --install
GO111MODULE=off $(GO) get github.com/remyoudompheng/go-misc/deadcode
test-txstore-rbf:
PILOSA_STORAGE_BACKEND=rbf $(MAKE) testv-race
# WARNING: This feature is no longer being tested regularly in CI. The test is
# very slow and very expensive, and we're not sure it actually provides useful
# information now.
test-txstore-rbf_bolt:
PILOSA_STORAGE_BACKEND=rbf_bolt $(MAKE) testv-race
test-external-lookup:
$(GO) test . -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) -run ^TestExternalLookup$$ -externalLookupDSN $(EXTERNAL_LOOKUP_DSN)

33
cmd/check.go
View file

@ -1,33 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package cmd
import (
"context"
"fmt"
"io"
"github.com/spf13/cobra"
"github.com/molecula/featurebase/v3/ctl"
)
var checker *ctl.CheckCommand
func newCheckCommand(stdin io.Reader, stdout io.Writer, stderr io.Writer) *cobra.Command {
checker = ctl.NewCheckCommand(stdin, stdout, stderr)
checkCmd := &cobra.Command{
Use: "check <path> [path2]...",
Short: "Do a consistency check on a FeatureBase data file.",
Long: `
Performs a consistency check on data files.
`,
RunE: func(cmd *cobra.Command, args []string) error {
if len(args) == 0 {
return fmt.Errorf("path required")
}
checker.Paths = args
return checker.Run(context.Background())
},
}
return checkCmd
}

23
cmd/check_test.go
View file

@ -1,23 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package cmd_test
import (
"strings"
"testing"
)
func TestCheckHelp(t *testing.T) {
output, err := ExecNewRootCommand(t, "check", "--help")
if !strings.Contains(output, "Usage:") ||
!strings.Contains(output, "Flags:") ||
!strings.Contains(output, "featurebase check") || err != nil {
t.Fatalf("Command 'check --help' not working, err: '%v', output: '%s'", err, output)
}
}
func TestCheckNoPath(t *testing.T) {
output, err := ExecNewRootCommand(t, "check")
if !strings.Contains(err.Error(), "path required") {
t.Fatalf("Command 'check' without args should error but: err: '%v', output: '%v'", err, output)
}
}

1
cmd/root.go
View file

@ -53,7 +53,6 @@ at https://docs.molecula.cloud/.
rc.AddCommand(newChkSumCommand(stdin, stdout, stderr))
rc.AddCommand(newBackupCommand(stdin, stdout, stderr))
rc.AddCommand(newRestoreCommand(stdin, stdout, stderr))
rc.AddCommand(newCheckCommand(stdin, stdout, stderr))
rc.AddCommand(newConfigCommand(stdin, stdout, stderr))
rc.AddCommand(newExportCommand(stdin, stdout, stderr))
rc.AddCommand(newGenerateConfigCommand(stdin, stdout, stderr))

122
ctl/check.go
View file

@ -1,122 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package ctl
import (
"context"
"fmt"
"io"
"os"
"path/filepath"
"syscall"
"github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/roaring"
"github.com/pkg/errors"
)
// CheckCommand represents a command for performing consistency checks on data files.
type CheckCommand struct {
// Data file paths.
Paths []string
// Standard input/output
*pilosa.CmdIO
}
// NewCheckCommand returns a new instance of CheckCommand.
func NewCheckCommand(stdin io.Reader, stdout, stderr io.Writer) *CheckCommand {
return &CheckCommand{
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
}
}
// Run executes the check command.
func (cmd *CheckCommand) Run(_ context.Context) error {
for _, path := range cmd.Paths {
switch filepath.Ext(path) {
case "":
if err := cmd.checkBitmapFile(path); err != nil {
return errors.Wrap(err, "checking bitmap")
}
case ".cache":
if err := cmd.checkCacheFile(path); err != nil {
return errors.Wrap(err, "checking cache")
}
case ".snapshotting":
if err := cmd.checkSnapshotFile(path); err != nil {
return errors.Wrap(err, "checking snapshot")
}
}
}
return nil
}
// checkBitmapFile performs a consistency check on path for a roaring bitmap file.
func (cmd *CheckCommand) checkBitmapFile(path string) (err error) {
// Open file handle.
f, err := os.Open(path)
if err != nil {
return errors.Wrap(err, "opening file")
}
defer f.Close()
fi, err := f.Stat()
if err != nil {
return errors.Wrap(err, "statting file")
}
// Memory map the file.
data, err := syscall.Mmap(int(f.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
if err != nil {
return errors.Wrap(err, "mmapping")
}
defer func() {
e := syscall.Munmap(data)
if e != nil {
fmt.Fprintf(cmd.Stderr, "WARNING: munmap failed: %v", e)
}
// don't overwrite another error with this, but also indicate
// this error.
if err == nil {
err = e
}
}()
// Attach the mmap file to the bitmap.
bm := roaring.NewBitmap()
if err := bm.UnmarshalBinary(data); err != nil {
return errors.Wrap(err, "unmarshalling")
}
// Perform consistency check.
if err := bm.Check(); err != nil {
// Print returned errors.
switch err := err.(type) {
case roaring.ErrorList:
for i := range err {
fmt.Fprintf(cmd.Stdout, "%s: %s\n", path, err[i].Error())
}
default:
fmt.Fprintf(cmd.Stdout, "%s: %s\n", path, err.Error())
}
}
// Print success message if no errors were found.
fmt.Fprintf(cmd.Stdout, "%s: ok\n", path)
return nil
}
// checkCacheFile performs a consistency check on path for a cache file.
func (cmd *CheckCommand) checkCacheFile(path string) error {
fmt.Fprintf(cmd.Stderr, "%s: ignoring cache file\n", path)
return nil
}
// checkSnapshotFile performs a consistency check on path for a snapshot file.
func (cmd *CheckCommand) checkSnapshotFile(path string) error {
fmt.Fprintf(cmd.Stderr, "%s: ignoring snapshot file\n", path)
return nil
}

95
ctl/check_test.go
View file

@ -1,95 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package ctl
import (
"bytes"
"io"
"os"
"strings"
"testing"
"context"
"github.com/molecula/featurebase/v3/testhook"
)
func TestCheckCommand_RunCacheFile(t *testing.T) {
fi, err := testhook.TempFile(t, "test*.cache")
if err != nil {
t.Fatalf("creating test file: %v", err)
}
cacheFile := fi.Name()
rder := []byte{}
stdin := bytes.NewReader(rder)
r, w, _ := os.Pipe()
cm := NewCheckCommand(stdin, w, w)
cm.Paths = []string{cacheFile}
err = cm.Run(context.Background())
w.Close()
var buf bytes.Buffer
if _, err := io.Copy(&buf, r); err != nil {
t.Fatalf("copy: %v", err)
}
if !strings.Contains(buf.String(), "ignoring cache file") {
t.Fatalf("expect: ignoring cache file, actual: '%s'", err)
}
}
func TestCheckCommand_RunSnapshot(t *testing.T) {
fi, err := testhook.TempFile(t, "test*.snapshotting")
if err != nil {
t.Fatalf("creating test file: %v", err)
}
snapshotFile := fi.Name()
rder := []byte{}
stdin := bytes.NewReader(rder)
r, w, _ := os.Pipe()
cm := NewCheckCommand(stdin, w, w)
cm.Paths = []string{snapshotFile}
err = cm.Run(context.Background())
w.Close()
var buf bytes.Buffer
if _, err := io.Copy(&buf, r); err != nil {
t.Fatalf("copy: %v", err)
}
if !strings.Contains(buf.String(), "ignoring snapshot file") {
t.Fatalf("expect: ignoring snapshot file, actual: '%s'", err)
}
}
func TestCheckCommand_Run(t *testing.T) {
file, err := testhook.TempFile(t, "run-command")
if err != nil {
t.Fatal(err)
}
fname := file.Name()
if _, err := file.Write([]byte("1234,1223")); err != nil {
t.Fatalf("writing to temp file: %v", err)
}
file.Close()
rder := []byte{}
stdin := bytes.NewReader(rder)
r, w, _ := os.Pipe()
cm := NewCheckCommand(stdin, w, w)
cm.Paths = []string{fname}
err = cm.Run(context.Background())
w.Close()
var buf bytes.Buffer
if _, err := io.Copy(&buf, r); err != nil {
t.Fatalf("copy: %v", err)
}
expectedPrefix := "checking bitmap: unmarshalling: "
if !strings.HasPrefix(err.Error(), expectedPrefix) {
t.Fatalf("expect error: '%s...', actual: '%s'", expectedPrefix, err)
}
// Todo: need correct roaring file for happy path
}

9
ctl/server.go
View file

@ -2,7 +2,6 @@
package ctl
import (
"fmt"
"time"
"github.com/molecula/featurebase/v3/server"
@ -75,13 +74,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.IntVar(&srv.Config.Profile.BlockRate, "profile.block-rate", srv.Config.Profile.BlockRate, "Sampling rate for goroutine blocking profiler. One sample per <rate> ns.")
flags.IntVar(&srv.Config.Profile.MutexFraction, "profile.mutex-fraction", srv.Config.Profile.MutexFraction, "Sampling fraction for mutex contention profiling. Sample 1/<rate> of events.")
// Storage
// Note: the default for --storage.backend must be kept "" empty string.
// Otherwise we cannot detect and honor the PILOSA_STORAGE_BACKEND env var
// over-ride.
// TODO: the comment above was carried over from the PILOSA_TXSRC flag, but
// we should confirm that this still applies.
flags.StringVar(&srv.Config.Storage.Backend, "storage.backend", storage.DefaultBackend, fmt.Sprintf("transaction/storage to use: one of roaring or rbf. The default is: %v. The env var PILOSA_STORAGE_BACKEND is over-ridden by --storage.backend option on the command line.", storage.DefaultBackend))
flags.StringVar(&srv.Config.Storage.Backend, "storage.backend", storage.DefaultBackend, "Storage backend to use: 'rbf' is only supported value.")
flags.BoolVar(&srv.Config.Storage.FsyncEnabled, "storage.fsync", true, "enable fsync fully safe flush-to-disk")
// RBF specific flags. See pilosa/rbf/cfg/cfg.go for definitions.

55
dbshard.go
View file

@ -67,9 +67,8 @@ type DBShard struct {
Shard uint64
Open bool
typ txtype
styp string
hasRoaring bool // if either of the types is roaringTxn
typ txtype
styp string
W DBWrapper
ParentDBIndex *DBIndex
@ -131,8 +130,7 @@ type DBPerShard struct {
// Easily see how many we have.
Flatmap map[flatkey]*DBShard
typ txtype
hasRoaring bool
typ txtype
txf *TxFactory
holder *Holder
@ -269,11 +267,6 @@ func (txf *TxFactory) NewDBPerShard(typ txtype, holderDir string, holder *Holder
vprint.PanicOn("must have holder.cfg.RBFConfig and holder.cfg.StorageConfig set here")
}
hasRoaring := false
if typ == roaringTxn {
hasRoaring = true
}
d = &DBPerShard{
typ: typ,
HolderDir: holderDir,
@ -281,7 +274,6 @@ func (txf *TxFactory) NewDBPerShard(typ txtype, holderDir string, holder *Holder
dbh: NewDBHolder(),
Flatmap: make(map[flatkey]*DBShard),
txf: txf,
hasRoaring: hasRoaring,
index2shards: newIndex2Shards(),
StorageConfig: holder.cfg.StorageConfig,
RBFConfig: holder.cfg.RBFConfig,
@ -407,10 +399,7 @@ func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *In
}
dbs, ok = dbi.Shard[shard]
if dbs != nil && dbs.closed {
// roaring txn are nil/fake anyway. Don't freak out.
if per.typ != roaringTxn {
vprint.PanicOn(fmt.Sprintf("cannot retain closed dbs across holder ReOpen dbs='%p'; per.typ='%v'", dbs, per.typ))
}
vprint.PanicOn(fmt.Sprintf("cannot retain closed dbs across holder ReOpen dbs='%p'; per.typ='%v'", dbs, per.typ))
}
if !ok {
dbs = &DBShard{
@ -421,7 +410,6 @@ func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *In
HolderPath: per.HolderDir,
idx: idx,
per: per,
hasRoaring: per.hasRoaring,
}
dbs.styp = per.typ.String()
dbi.Shard[shard] = dbs
@ -430,8 +418,6 @@ func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *In
if !dbs.Open {
var registry DBRegistry
switch dbs.typ {
case roaringTxn:
registry = globalRoaringReg
case rbfTxn:
registry = globalRbfDBReg
registry.(*rbfDBRegistrar).SetRBFConfig(per.RBFConfig)
@ -470,8 +456,6 @@ func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requir
return f.dbPerShard.TypedDBPerShardGetShardsForIndex(f.typ, idx, roaringViewPath, requireData)
}
// if roaringViewPath is "" then for ty == roaringTxn we go to disk to discover
// all the view paths under idx for type ty.
// requireData means open the database file and verify that at least one key is set.
// The returned sliceOfShards should not be modified. We will cache it for subsequent
// queries.
@ -485,14 +469,6 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
per.Mu.Lock()
defer per.Mu.Unlock()
if ty == roaringTxn && roaringViewPath != "" {
shardMap, err := roaringMapOfShards(roaringViewPath)
if err != nil {
return nil, err
}
return shardMap, nil
}
i2ss := per.index2shards
ss, ok := i2ss[idx.name]
@ -507,27 +483,6 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
// Upon return, cache the setOfShards value and reuse it next time
if ty == roaringTxn {
// INVAR: roaringViewPath == "", because the other case is
// handled above.
fields := idx.Fields()
for _, field := range fields {
for _, view := range field.views() {
shardMap, err := roaringMapOfShards(view.path)
if err != nil {
return nil,
errors.Wrap(err, fmt.Sprintf(
"TypedDBPerShardGetLocalShardsForIndex roaringTxn view.path='%v'", view.path))
}
for shard := range shardMap {
setOfShards.add(shard)
}
}
}
return setOfShards.CloneMaybe(), nil
}
// INVAR: not-roaring.
path := per.prefixForType(idx, ty)
ignoreEmpty := false
@ -730,8 +685,6 @@ func (per *DBPerShard) GetFieldView2ShardsMapForIndex(idx *Index) (vs *FieldView
ty := per.typ
switch ty {
case roaringTxn:
return roaringGetFieldView2Shards(idx)
default:
vs = NewFieldView2Shards()

71
dbshard_internal_test.go
View file

@ -71,7 +71,7 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
v2s.addViewShardSet(txkey.FieldView{Field: field, View: "standard"}, stdShardSet)
}
for _, src := range []string{"roaring", "rbf"} {
for _, src := range []string{"rbf"} {
cfg := mustHolderConfig()
cfg.StorageConfig.Backend = src
holder := NewHolder(tmpdir, cfg)
@ -82,7 +82,6 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
idx, err = NewIndex(holder, filepath.Join(tmpdir, index), index)
PanicOn(err)
}
estd := "rick/fields/_exists/views/standard"
std := "rick/fields/f/views/standard"
shards, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std, false)
@ -93,65 +92,23 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
panic(fmt.Sprintf("missing shard=%v from shards='%#v'", shard, shards))
}
}
if src == "roaring" {
// check estd too
shards, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd, false)
for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: shard})
fvs, err := tx.GetSortedFieldViewList(idx, shard)
PanicOn(err)
for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
if !shards[shard] {
panic(fmt.Sprintf("missing shard=%v from shards='%#v'", shard, shards))
}
// expect these same two field/views for all 6 shards
expect0 := txkey.FieldView{Field: "_exists", View: "standard"}
expect1 := txkey.FieldView{Field: "f", View: "standard"}
if len(fvs) != 2 {
panic(fmt.Sprintf("fvs should be len 2, got '%#v' (%s)", fvs, src))
}
// check GetSortedFieldViewList() and roaringGetFieldView2Shards()
vs, err := roaringGetFieldView2Shards(idx)
PanicOn(err)
for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: shard})
fvs, err := tx.GetSortedFieldViewList(idx, shard)
PanicOn(err)
// expect these same two field/views for all 6 shards
expect0 := txkey.FieldView{Field: "_exists", View: "standard"}
expect1 := txkey.FieldView{Field: "f", View: "standard"}
if len(fvs) != 2 {
panic(fmt.Sprintf("fvs should be len 2, got '%#v' (%s)", fvs, src))
}
if fvs[0] != expect0 {
panic(fmt.Sprintf("expected fvs[0]='%#v', but got '%#v'", expect0, fvs[0]))
}
if fvs[1] != expect1 {
panic(fmt.Sprintf("expected fvs[1]='%#v', but got '%#v'", expect1, fvs[1]))
}
for _, fv := range fvs {
if !vs.has(fv.Field, fv.View, shard) {
panic(fmt.Sprintf("vs did not contain fv='%#v' for shard %v", fv, shard))
}
}
tx.Rollback()
if fvs[0] != expect0 {
panic(fmt.Sprintf("expected fvs[0]='%#v', but got '%#v'", expect0, fvs[0]))
}
} else {
// non-roaring: rbf
for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: shard})
fvs, err := tx.GetSortedFieldViewList(idx, shard)
PanicOn(err)
// expect these same two field/views for all 6 shards
expect0 := txkey.FieldView{Field: "_exists", View: "standard"}
expect1 := txkey.FieldView{Field: "f", View: "standard"}
if len(fvs) != 2 {
panic(fmt.Sprintf("fvs should be len 2, got '%#v' (%s)", fvs, src))
}
if fvs[0] != expect0 {
panic(fmt.Sprintf("expected fvs[0]='%#v', but got '%#v'", expect0, fvs[0]))
}
if fvs[1] != expect1 {
panic(fmt.Sprintf("expected fvs[1]='%#v', but got '%#v'", expect1, fvs[1]))
}
tx.Rollback()
if fvs[1] != expect1 {
panic(fmt.Sprintf("expected fvs[1]='%#v', but got '%#v'", expect1, fvs[1]))
}
tx.Rollback()
}
holder.Close()
}

3
executor.go
View file

@ -1651,6 +1651,9 @@ func (d *DistinctTimestamp) Union(other DistinctTimestamp) DistinctTimestamp {
return DistinctTimestamp{Name: d.Name, Values: vals}
}
const ViewNotFound = Error("view not found")
const FragmentNotFound = Error("fragment not found")
func executeDistinctShardSet(ctx context.Context, qcx *Qcx, idx *Index, fieldName string, shard uint64, filterBitmap *roaring.Bitmap) (result *Row, err0 error) {
index := idx.Name()
tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard})

10
executor_test.go
View file

@ -33,7 +33,6 @@ import (
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/proto"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/storage"
"github.com/molecula/featurebase/v3/test"
"github.com/molecula/featurebase/v3/testhook"
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
@ -1277,15 +1276,6 @@ func TestExecutor_Execute_Count(t *testing.T) {
}
func roaringOnlyTest(t *testing.T) {
src := pilosa.CurrentBackend()
if src == pilosa.RoaringTxn || (storage.DefaultBackend == pilosa.RoaringTxn && src == "") {
// okay to run, we are under roaring only
} else {
t.Skip("skip for everything but roaring")
}
}
// Ensure a set query can be executed.
func TestExecutor_Execute_Set(t *testing.T) {
t.Run("RowIDColumnID", func(t *testing.T) {

1
field_internal_test.go
View file

@ -247,7 +247,6 @@ func NewTestField(t testing.TB, opts FieldOption) *TestField {
}
cfg := DefaultHolderConfig()
cfg.StorageConfig.Backend = CurrentBackendOrDefault()
cfg.StorageConfig.FsyncEnabled = false
cfg.RBFConfig.FsyncEnabled = false
h := NewHolder(path, cfg)

320
fragment.go
View file

@ -3,7 +3,6 @@ package pilosa
import (
"archive/tar"
"bufio"
"bytes"
"container/heap"
"context"
@ -16,7 +15,6 @@ import (
"math/bits"
"os"
"path/filepath"
"runtime/debug"
"sort"
"strconv"
"strings"
@ -59,18 +57,12 @@ const (
// width of roaring containers is 2^16
containerWidth = 1 << 16
// snapshotExt is the file extension used for an in-process snapshot.
snapshotExt = ".snapshotting"
// cacheExt is the file extension for persisted cache ids.
cacheExt = ".cache"
// HashBlockSize is the number of rows in a merkle hash block.
HashBlockSize = 100
// defaultFragmentMaxOpN is the default value for Fragment.MaxOpN.
defaultFragmentMaxOpN = 10000
// Row ids used for boolean fields.
falseRowID = uint64(0)
trueRowID = uint64(1)
@ -132,22 +124,11 @@ type fragment struct {
// idx cached to avoid repeatedly looking it up everywhere.
idx *Index
// parent holder, used to find snapshot queue, etc.
// parent holder
holder *Holder
// debugging tool: addresses of current and previous maps
prevdata, currdata struct{ from, to uintptr }
// File-backed storage
flags byte // user-defined flags passed to roaring
storage *roaring.Bitmap
opN int // number of ops since snapshot (may be approximate for imports)
ops int // number of higher-level operations, as opposed to bit changes
snapshotPending bool // set to true when requesting a snapshot, set to false after snapshot completes
snapshotCond sync.Cond
snapshotErr error // error yielded by the last snapshot operation
snapshotStamp time.Time // timestamp of last snapshot
open bool // is this fragment actually open?
storage *roaring.Bitmap
// Cache for row counts.
CacheType string // passed in by field
@ -164,11 +145,6 @@ type fragment struct {
// Cached checksums for each block.
checksums map[int][]byte
// Number of operations performed before performing a snapshot.
// This limits the size of fragments on the heap and flushes them to disk
// so that they can be mmapped and heap utilization can be kept low.
MaxOpN int
// Logger used for out-of-band log entries.
Logger logger.Logger
@ -194,18 +170,15 @@ func newFragment(holder *Holder, spec fragSpec, shard uint64, flags byte) *fragm
fieldstr: spec.fieldstr,
fld: spec.field,
shard: shard,
flags: flags,
idx: idx,
CacheType: DefaultCacheType,
CacheSize: DefaultCacheSize,
holder: holder,
MaxOpN: defaultFragmentMaxOpN,
stats: stats.NopStatsClient,
}
f.snapshotCond = sync.Cond{L: &f.mu}
return f
}
@ -259,12 +232,6 @@ func (f *fragment) Open() error {
defer f.mu.Unlock()
if err := func() error {
// Initialize storage in a function so we can close if anything goes wrong.
f.holder.Logger.Debugf("open storage for index/field/view/fragment: %s/%s/%s/%d", f.index(), f.field(), f.view(), f.shard)
if err := f.openStorage(true); err != nil {
return errors.Wrap(err, "opening storage")
}
// Fill cache with rows persisted to disk.
f.holder.Logger.Debugf("open cache for index/field/view/fragment: %s/%s/%s/%d", f.index(), f.field(), f.view(), f.shard)
if err := f.openCache(); err != nil {
@ -278,57 +245,12 @@ func (f *fragment) Open() error {
f.close()
return err
}
f.open = true
_ = testhook.Opened(f.holder.Auditor, f, nil)
f.holder.Logger.Debugf("successfully opened index/field/view/fragment: %s/%s/%s/%d", f.index(), f.field(), f.view(), f.shard)
return nil
}
// emptyStorage is the common case for importStorage/applyStorage where they
// get no data. It tries to write the current storage to the provided file,
// which is assumed to be the file they didn't get any data from.
func (f *fragment) emptyStorage(file *os.File) (bool, error) {
if f.holder.Opts.ReadOnly {
return false, errors.New("can't flush/create storage for read-only holder")
}
// No data. We'll mark this for no mapping, clear any existing
// mapped containers, and set the Source to nil. We also have no
// ops.
f.opN = 0
f.ops = 0
f.storage.SetOps(0, 0)
f.storage.PreferMapping(false)
_, err := f.storage.RemapRoaringStorage(nil)
f.storage.SetSource(nil)
if err != nil {
return false, fmt.Errorf("applying/importing storage: no data, and clearing old mapping also failed: %v", err)
}
// Write the existing storage out to the file so it's
// a valid Roaring file thereafter. nothing to unmarshal.
// In the unlikely event that this happened even though we
// had significant data, we're not mapping it, but that's
// harmless even if it's not maximally efficient.
bi := bufio.NewWriter(file)
if _, err = f.storage.WriteTo(bi); err != nil {
return false, fmt.Errorf("init storage file: %s", err)
}
bi.Flush()
return false, nil
}
// openStorage opens the storage bitmap. Does nothing in RBF-world and will be removed soon.
func (f *fragment) openStorage(unmarshalData bool) error {
if !f.idx.NeedsSnapshot() {
f.currdata = struct{ from, to uintptr }{}
f.prevdata = f.currdata
return nil // openStorage becomes a noop under RBF, Badger, etc.
}
return nil
}
// openCache initializes the cache from row ids persisted to disk.
func (f *fragment) openCache() error {
// Determine cache type from field name.
@ -384,12 +306,6 @@ func (f *fragment) Close() error {
defer func() {
_ = testhook.Closed(f.holder.Auditor, f, nil)
}()
for f.snapshotPending {
f.snapshotCond.Wait()
}
// Note: snapshots won't progress on a closed fragment, so we
// wait until after a possible pending snapshot to close.
f.open = false
return f.close()
}
@ -400,28 +316,12 @@ func (f *fragment) close() error {
return errors.Wrap(err, "flushing cache")
}
// Close underlying storage.
if err := f.closeStorage(); err != nil {
f.holder.Logger.Errorf("fragment: error closing storage: err=%s, path=%s", err, f.path())
return errors.Wrap(err, "closing storage")
}
// Remove checksums.
f.checksums = nil
return nil
}
// closeStorage is essentially a no-op and will go away soon.
func (f *fragment) closeStorage() error {
// opN is determined by how many bit set/clear operations are in the storage
// write log, so once the storage is closed it should be 0. Opening new
// storage will set opN appropriately.
f.opN = 0
return nil
}
// mutexCheck checks for any entries in fragment which violate the mutex
// property of having only one value set for a given column ID.
func (f *fragment) mutexCheck(tx Tx, details bool, limit int) (map[uint64][]uint64, error) {
@ -544,9 +444,6 @@ func (f *fragment) unprotectedSetBit(tx Tx, rowID, columnID uint64) (changed boo
// Invalidate block checksum.
delete(f.checksums, int(rowID/HashBlockSize))
// Increment number of operations until snapshot is required.
f.incrementOpN(1)
// If we're using a cache, update it. Otherwise skip the
// possibly-expensive count operation.
if f.CacheType != CacheTypeNone {
@ -596,9 +493,6 @@ func (f *fragment) unprotectedClearBit(tx Tx, rowID, columnID uint64) (changed b
// Invalidate block checksum.
delete(f.checksums, int(rowID/HashBlockSize))
// Increment number of operations until snapshot is required.
f.incrementOpN(1)
// If we're using a cache, update it. Otherwise skip the
// possibly-expensive count operation.
if f.CacheType != CacheTypeNone {
@ -665,8 +559,6 @@ func (f *fragment) unprotectedSetRow(tx Tx, row *Row, rowID uint64) (changed boo
}
}
// Snapshot storage.
f.holder.SnapshotQueue.Enqueue(f)
f.stats.Count("setRow", 1, 1.0)
return changed, nil
@ -705,9 +597,6 @@ func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err e
// Clear the row in cache.
f.cache.Add(rowID, 0)
// Snapshot storage.
f.holder.SnapshotQueue.Enqueue(f)
return changed, nil
}
@ -1962,7 +1851,7 @@ func (f *fragment) mergeBlock(tx Tx, id int, data []pairSet) (sets, clears []pai
return sets[1:], clears[1:], err
}
// bulkImport bulk imports a set of bits and then snapshots the storage.
// bulkImport bulk imports a set of bits.
// The cache is updated to reflect the new data.
func (f *fragment) bulkImport(tx Tx, rowIDs, columnIDs []uint64, options *ImportOptions) error {
// Verify that there are an equal number of row ids and column ids.
@ -2175,68 +2064,48 @@ func (p parallelSlices) Swap(i, j int) {
// snapshot of the fragment or just do in-memory updates while appending
// operations to the op log.
func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64]struct{}) error {
//tx.AddN()
doFunc := func() error {
if len(set) > 0 {
f.stats.Count(MetricImportingN, int64(len(set)), 1)
if len(set) > 0 {
f.stats.Count(MetricImportingN, int64(len(set)), 1)
// TODO benchmark Add/RemoveN behavior with sorted/unsorted positions
changedN, err := tx.Add(f.index(), f.field(), f.view(), f.shard, set...)
if err != nil {
return errors.Wrap(err, "adding positions")
}
f.stats.Count(MetricImportedN, int64(changedN), 1)
f.incrementOpN(changedN)
// TODO benchmark Add/RemoveN behavior with sorted/unsorted positions
changedN, err := tx.Add(f.index(), f.field(), f.view(), f.shard, set...)
if err != nil {
return errors.Wrap(err, "adding positions")
}
f.stats.Count(MetricImportedN, int64(changedN), 1)
}
if len(clear) > 0 {
f.stats.Count(MetricClearingN, int64(len(clear)), 1)
changedN, err := tx.Remove(f.index(), f.field(), f.view(), f.shard, clear...)
if err != nil {
return errors.Wrap(err, "clearing positions")
}
f.stats.Count(MetricClearedN, int64(changedN), 1)
f.incrementOpN(changedN)
if len(clear) > 0 {
f.stats.Count(MetricClearingN, int64(len(clear)), 1)
changedN, err := tx.Remove(f.index(), f.field(), f.view(), f.shard, clear...)
if err != nil {
return errors.Wrap(err, "clearing positions")
}
f.stats.Count(MetricClearedN, int64(changedN), 1)
}
// Update cache counts for all affected rows.
for rowID := range rowSet {
// Invalidate block checksum.
delete(f.checksums, int(rowID/HashBlockSize))
if f.CacheType != CacheTypeNone {
start := rowID * ShardWidth
end := (rowID + 1) * ShardWidth
n, err := tx.CountRange(f.index(), f.field(), f.view(), f.shard, start, end)
if err != nil {
return errors.Wrap(err, "CountRange")
}
f.cache.BulkAdd(rowID, n)
}
}
// Update cache counts for all affected rows.
for rowID := range rowSet {
// Invalidate block checksum.
delete(f.checksums, int(rowID/HashBlockSize))
if f.CacheType != CacheTypeNone {
f.cache.Invalidate()
}
return nil
}
err := doFunc()
if err != nil && f.storage != nil {
// we got an error. it's possible that the error indicates that something went wrong.
mappedIn, mappedOut, unmappedIn, errs, e2 := f.storage.SanityCheckMapping(f.currdata.from, f.currdata.to)
if errs != 0 {
f.holder.Logger.Errorf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v",
f.path(), mappedIn, mappedOut, unmappedIn, errs, e2)
if f.prevdata.from != f.currdata.from {
mappedIn, mappedOut, unmappedIn, errs, e2 = f.storage.SanityCheckMapping(f.prevdata.from, f.prevdata.to)
f.holder.Logger.Errorf("with previous map, storage would have %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total, last %v",
mappedIn, mappedOut, unmappedIn, errs, e2)
start := rowID * ShardWidth
end := (rowID + 1) * ShardWidth
n, err := tx.CountRange(f.index(), f.field(), f.view(), f.shard, start, end)
if err != nil {
return errors.Wrap(err, "CountRange")
}
f.cache.BulkAdd(rowID, n)
}
}
return err
if f.CacheType != CacheTypeNone {
f.cache.Invalidate()
}
return nil
}
// sliceDifference removes everything from original that's found in remove,
@ -2536,125 +2405,6 @@ func (f *fragment) importRoaringOverwrite(ctx context.Context, tx Tx, data []byt
return f.importRoaring(ctx, tx, data, false)
}
// incrementOpN increase the operation count by one.
// If the count exceeds the maximum allowed then a snapshot is performed.
func (f *fragment) incrementOpN(changed int) {
if changed <= 0 {
return
}
// don't count opN or ops if our index doesn't want snapshots
if !f.idx.NeedsSnapshot() {
return
}
f.opN += changed
f.ops++
if f.opN > f.MaxOpN {
f.holder.SnapshotQueue.Enqueue(f)
}
}
// Snapshot writes the storage bitmap to disk and reopens it. This may
// coexist with existing background-queue snapshotting; it does not remove
// things from the queue. You probably don't want to do this; use
// the snapshotQueue's Enqueue/Await.
func (f *fragment) Snapshot() error {
f.mu.Lock()
defer f.mu.Unlock()
return f.snapshot()
}
func track(start time.Time, message string, stats stats.StatsClient, logger logger.Logger) {
elapsed := time.Since(start)
logger.Debugf("%s took %s", message, elapsed)
stats.Timing(MetricSnapshotDurationSeconds, elapsed, 1.0)
}
// snapshot does the actual snapshot operation. it does not check or care
// about f.snapshotPending.
func (f *fragment) snapshot() (err error) {
if !f.idx.NeedsSnapshot() {
return nil
}
if !f.open {
return errors.New("snapshot request on closed fragment")
}
wouldPanic := debug.SetPanicOnFault(true)
defer func() {
debug.SetPanicOnFault(wouldPanic)
if r := recover(); r != nil {
if e2, ok := r.(error); ok {
err = e2
// special case: if we caught a page fault, we diagnose that directly. sadly,
// we can't see the actual values that were used to generate this, probably.
if e2.Error() == "runtime error: invalid memory address or nil pointer dereference" {
mappedIn, mappedOut, unmappedIn, errs, _ := f.storage.SanityCheckMapping(f.currdata.from, f.currdata.to)
f.holder.Logger.Errorf("transaction failed on %s. storage has %d mapped in range, %d mapped out of range, %d unmapped in range, %d errors total",
f.path(), mappedIn, mappedOut, unmappedIn, errs)
}
} else {
err = fmt.Errorf("non-error PanicOn: %v", r)
}
}
}()
_, err = unprotectedWriteToFragment(f, f.storage)
if err == nil {
f.snapshotStamp = time.Now()
}
return err
}
// unprotectedWriteToFragment writes the fragment f with bm as the data. It is unprotected, and
// f.mu must be locked when calling it.
func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) (n int64, err error) { // nolint: interfacer
completeMessage := fmt.Sprintf("fragment: snapshot complete %s/%s/%s/%d", f.index(), f.field(), f.view(), f.shard)
start := time.Now()
defer track(start, completeMessage, f.stats, f.holder.Logger)
// Create a temporary file to snapshot to.
snapshotPath := f.path() + snapshotExt
file, err := os.Create(snapshotPath)
if err != nil {
return n, fmt.Errorf("create snapshot file: %s", err)
}
// No deferred close, because we want to close it sooner than the
// end of this function.
// Write storage to snapshot.
bw := bufio.NewWriter(file)
if n, err = bm.WriteTo(bw); err != nil {
file.Close()
return n, fmt.Errorf("snapshot write to: %s", err)
}
if err := bw.Flush(); err != nil {
file.Close()
return n, fmt.Errorf("flush: %s", err)
}
// we close the file here so we don't still have it open when trying
// to open it in a moment.
file.Close()
// Move snapshot to data file location.
if err := os.Rename(snapshotPath, f.path()); err != nil {
return n, fmt.Errorf("rename snapshot: %s", err)
}
// if we reloaded from the file, we'd end up with this bitmap
// as our storage. so... let's use this bitmap. as our storage.
f.storage = bm
// Reopen storage.
if err := f.openStorage(false); err != nil {
return n, fmt.Errorf("open storage: %s", err)
}
// Reset operation count.
f.opN = 0
return n, nil
}
// RecalculateCache rebuilds the cache regardless of invalidate time delay.
func (f *fragment) RecalculateCache() {
f.mu.Lock()

999
fragment_internal_test.go
View file

File diff suppressed because it is too large Load diff

28
holder.go
View file

@ -85,8 +85,7 @@ type Holder struct {
// The interval at which the cached row ids are persisted to disk.
cacheFlushInterval time.Duration
Logger logger.Logger
SnapshotQueue SnapshotQueue
Logger logger.Logger
// Instantiates new translation stores
OpenTranslateStore OpenTranslateStoreFunc
@ -271,8 +270,6 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
Logger: cfg.Logger,
Opts: HolderOpts{StorageBackend: cfg.StorageConfig.Backend},
SnapshotQueue: defaultSnapshotQueue,
Auditor: NewAuditor(),
path: path,
@ -734,16 +731,15 @@ func (h *Holder) maybeSpool(msg Message) bool {
return true
}
// Activate runs the background tasks relevant to keeping a holder in a stable
// state, such as scanning it for needed snapshots, or flushing caches. This
// is separate from opening because, while a server would nearly always want
// to do this, other use cases (like consistency checks of a data directory)
// Activate runs the background tasks relevant to keeping a holder in
// a stable state, such as flushing caches. This is separate from
// opening because, while a server would nearly always want to do
// this, other use cases (like consistency checks of a data directory)
// need to avoid it even getting started.
func (h *Holder) Activate() {
// Periodically flush cache.
h.wg.Add(2)
h.wg.Add(1)
go func() { defer h.wg.Done(); h.monitorCacheFlush() }()
go func() { defer h.wg.Done(); h.SnapshotQueue.ScanHolder(h, h.closing) }()
}
// checkForeignIndex is a check before applying a foreign
@ -791,7 +787,6 @@ func (h *Holder) Close() error {
// Notify goroutines of closing and wait for completion.
close(h.closing)
h.wg.Wait()
for _, index := range h.Indexes() {
if err := index.Close(); err != nil {
return errors.Wrap(err, "closing index")
@ -809,10 +804,6 @@ func (h *Holder) Close() error {
h.opened.mu.Lock()
h.opened.ch = make(chan struct{})
h.opened.mu.Unlock()
if h.SnapshotQueue != nil {
h.SnapshotQueue.Stop()
h.SnapshotQueue = nil
}
if h.lookupDB != nil {
err := h.lookupDB.Close()
@ -827,13 +818,6 @@ func (h *Holder) Close() error {
return nil
}
func (h *Holder) NeedsSnapshot() bool {
h.mu.RLock()
defer h.mu.RUnlock()
return h.txf.NeedsSnapshot()
}
// HasData returns true if Holder contains at least one index.
// This is used to determine if the rebalancing of data is necessary
// when a node joins the cluster.

10
holder_internal_test.go
View file

@ -174,17 +174,9 @@ func TestHolderOperatorCancel(t *testing.T) {
}
}
// mustHolderConfig is meant to help minimize the number of places in the code
// where we're reading the PILOSA_STORAGE_BACKEND environment variable for
// testing purposes. Ideally we would handle this differently, but this is a
// first attempt at improving things. Note: the actual os.Getenv() call was
// moved to the CurrentBackend() function.
// mustHolderConfig sets up a default holder config for tests.
func mustHolderConfig() *HolderConfig {
cfg := DefaultHolderConfig()
if backend := CurrentBackend(); backend != "" {
_ = MustBackendToTxtype(backend)
cfg.StorageConfig.Backend = backend
}
cfg.StorageConfig.FsyncEnabled = false
cfg.RBFConfig.FsyncEnabled = false
cfg.Schemator = disco.InMemSchemator

111
holder_test.go
View file

@ -5,13 +5,12 @@ import (
"context"
"math"
"os"
"path/filepath"
"reflect"
"strings"
"testing"
"time"
"github.com/molecula/featurebase/v3"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/test"
@ -21,10 +20,7 @@ import (
// mustHolderConfig provides a default test-friendly holder config.
func mustHolderConfig() *pilosa.HolderConfig {
cfg := pilosa.DefaultHolderConfig()
if backend := pilosa.CurrentBackend(); backend != "" {
_ = pilosa.MustBackendToTxtype(backend)
cfg.StorageConfig.Backend = backend
}
cfg.StorageConfig.Backend = "rbf"
cfg.StorageConfig.FsyncEnabled = false
cfg.RBFConfig.FsyncEnabled = false
cfg.Schemator = disco.InMemSchemator
@ -55,109 +51,6 @@ func TestHolder_Open(t *testing.T) {
t.Fatalf("unexpected error: %v", err)
}
})
t.Run("ErrFragmentStoragePermission", func(t *testing.T) {
roaringOnlyTest(t)
if os.Geteuid() == 0 {
t.Skip("Skipping permissions test since user is root.")
}
h := test.MustOpenHolder(t)
defer h.Close()
var idx *pilosa.Index
var err error
if idx, err = h.CreateIndex("foo", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
}
var shard uint64
tx := idx.Txf().NewTx(pilosa.Txo{Write: writable, Index: idx, Shard: shard})
defer tx.Rollback()
if field, err := idx.CreateField("bar", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(tx, 0, 0, nil); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
} else if err := os.Chmod(filepath.Join(h.Path(), "foo", "bar", "views", "standard", "fragments", "0"), 0000); err != nil {
t.Fatal(err)
}
defer func() {
_ = os.Chmod(filepath.Join(h.Path(), "foo", "bar", "views", "standard", "fragments", "0"), 0644)
}()
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "permission denied") {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ErrFragmentStorageCorrupt", func(t *testing.T) {
roaringOnlyTest(t)
h := test.MustOpenHolder(t)
defer h.Close()
var idx *pilosa.Index
var err error
if idx, err = h.CreateIndex("foo", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
}
var shard uint64
tx := idx.Txf().NewTx(pilosa.Txo{Write: writable, Index: idx, Shard: shard})
if err != nil {
t.Fatal(err)
}
defer tx.Rollback()
if field, err := idx.CreateField("bar", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(tx, 0, 0, nil); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
} else if err := os.Truncate(filepath.Join(h.Path(), "foo", "bar", "views", "standard", "fragments", "0"), 2); err != nil {
t.Fatal(err)
}
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open fragment: shard=0, err=opening storage: unmarshal storage") {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ErrFragmentStorageRecoverable", func(t *testing.T) {
roaringOnlyTest(t)
h := test.MustOpenHolder(t)
defer h.Close()
idx, err := h.CreateIndex("foo", pilosa.IndexOptions{})
if err != nil {
t.Fatal(err)
}
var shard uint64
tx := idx.Txf().NewTx(pilosa.Txo{Write: writable, Index: idx, Shard: shard})
defer tx.Rollback()
if field, err := idx.CreateField("bar", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(tx, 0, 0, nil); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
} else if err := os.Truncate(filepath.Join(h.IndexesPath(), "foo", "bar", "views", "standard", "fragments", "0"), 20); err != nil {
t.Fatal(err)
}
if err := h.Reopen(); err != nil {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ForeignIndex", func(t *testing.T) {
t.Run("ErrForeignIndexNotFound", func(t *testing.T) {
h := test.MustOpenHolder(t)

3
http/handler.go
View file

@ -37,6 +37,7 @@ import (
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/pql"
"github.com/molecula/featurebase/v3/rbf"
"github.com/molecula/featurebase/v3/storage"
"github.com/molecula/featurebase/v3/topology"
"github.com/molecula/featurebase/v3/tracing"
"github.com/pkg/errors"
@ -1087,7 +1088,7 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
req, ok := qreq.(*pilosa.QueryRequest)
if DoPerQueryProfiling {
backend := pilosa.CurrentBackend()
backend := storage.DefaultBackend
reqHash := hash(req.Query)
qlen := len(req.Query)

4
index.go
View file

@ -93,10 +93,6 @@ func (i *Index) NewTx(txo Txo) Tx {
return i.holder.txf.NewTx(txo)
}
func (i *Index) NeedsSnapshot() bool {
return i.holder.txf.NeedsSnapshot()
}
// CreatedAt is an timestamp for a specific version of an index.
func (i *Index) CreatedAt() int64 {
i.mu.RLock()

2
internal/clustertests/testdata/featurebase.conf vendored
View file

@ -378,6 +378,6 @@
logout-url = "https://login.microsoftonline.com/common/oauth2/v2.0/logout"
scopes = ["https://graph.microsoft.com/.default", "offline_access"]
secret-key = "DEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF"
permissions = "/go/src/github.com/molecula/featurebase/internal/clustertestsx/testdata/permissions.yaml"
permissions = "/go/src/github.com/molecula/featurebase/internal/clustertests/testdata/permissions.yaml"
query-log-path = "query-log-test.log"
redirect-base-url = "https://localhost:10101"

68
mmap_test.go
View file

@ -1,68 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package pilosa
import (
"math/rand"
"runtime"
"testing"
"github.com/molecula/featurebase/v3/logger"
)
type cv struct {
cols []uint64
vals []int64
}
func forceSnapshotsCheckMapping(t *testing.T) {
depth := uint64(6)
f, idx, tx := mustOpenBSIFragment(t, "i", "f", viewStandard, 0)
tx.Rollback()
f.Logger = logger.NewLogfLogger(t)
defer f.Clean(t)
tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard})
defer tx.Rollback()
for i := 0; i < f.MaxOpN; i++ {
_, _ = f.setBit(tx, 0, uint64(32*i))
}
// force snapshot so we get a mmapped row...
err := f.Snapshot()
if err != nil {
t.Fatalf("initial snapshot error: %v", err)
}
values := make([]cv, 1024)
for i := range values {
cols := make([]uint64, 128)
vals := make([]int64, 128)
for j := range cols {
// pick values in the first 16 cols of each of the 16
// shards in a default shardwidth, so each set will
// probably change some values from the previous one.
cols[j] = uint64(((rand.Int63n(16) & int64(i>>2)) << 16) + rand.Int63n(16))
vals[j] = int64(rand.Int63n(1 << depth))
}
values[i] = cv{cols, vals}
}
// modify the original bitmap, until it causes a snapshot, which
// then invalidates the other map...
for i := 0; i < 32; i++ {
cv := values[i%len(values)]
// periodically force gc, so if we have a small pool of maps
// we'll go in and out of mapping mode
if i%5 == 0 {
runtime.GC()
}
err := f.importValue(tx, cv.cols, cv.vals, depth, (i%3 == 1))
if err != nil {
t.Fatalf("importValue[%d]: %v", i, err)
}
err = f.Snapshot()
if err != nil {
t.Fatalf("snapshot[%d]: %v", i, err)
}
}
}

19
pilosa.go
View file

@ -2,13 +2,11 @@
package pilosa
import (
"os"
"regexp"
"time"
"github.com/molecula/featurebase/v3/disco"
pnet "github.com/molecula/featurebase/v3/net"
"github.com/molecula/featurebase/v3/storage"
"github.com/pkg/errors"
)
@ -157,20 +155,3 @@ func AddressWithDefaults(addr string) (*pnet.URI, error) {
}
return pnet.NewURIFromAddress(addr)
}
// CurrentBackend is one step in an attempt to centralize (and either minimize
// or completely remove), the calls to environment variables throughout the
// tests. Ideally we could get rid of this and rely completely on the
// configuration parameters.
func CurrentBackend() string {
return os.Getenv("PILOSA_STORAGE_BACKEND")
}
// CurrentBackendOrDefault tries the environment variable first, but falls back
// to the default backend if the environment variable is empty.
func CurrentBackendOrDefault() string {
if backend := os.Getenv("PILOSA_STORAGE_BACKEND"); backend != "" {
return backend
}
return storage.DefaultBackend
}

10
pprof.go
View file

@ -19,10 +19,7 @@ import (
// commented out—in holder.go.
func CPUProfileForDur(dur time.Duration, outpath string) {
// per-query pprof output:
backend := CurrentBackend()
if backend == "" {
backend = storage.DefaultBackend
}
backend := storage.DefaultBackend
path := outpath + "." + backend
f, err := os.Create(path)
vprint.PanicOn(err)
@ -45,10 +42,7 @@ func CPUProfileForDur(dur time.Duration, outpath string) {
// commented out—in holder.go.
func MemProfileForDur(dur time.Duration, outpath string) {
// per-query pprof output:
backend := CurrentBackend()
if backend == "" {
backend = storage.DefaultBackend
}
backend := storage.DefaultBackend
path := outpath + "." + backend
f, err := os.Create(path)
vprint.PanicOn(err)

2
roaring/filter.go
View file

@ -579,7 +579,6 @@ func (b *BitmapRowFilterMultiFilter) ConsiderData(key FilterKey, data *Container
// offsets the input bitmap's containers have, it matches them against
// corresponding keys.
type BitmapBitmapFilter struct {
filter *Bitmap // We don't use this while iterating, but in ludicrous edge cases it might be holding a generation we need. TODO @seebs I don't understand why this mentions generations
containers []*Container
nextOffsets []uint64
callback func(uint64) error
@ -629,7 +628,6 @@ func (b *BitmapBitmapFilter) ConsiderData(key FilterKey, data *Container) Filter
// because offset-within-row is what we care about.
func NewBitmapBitmapFilter(filter *Bitmap, callback func(uint64) error) *BitmapBitmapFilter {
b := &BitmapBitmapFilter{
filter: filter,
callback: callback,
containers: make([]*Container, rowWidth),
nextOffsets: make([]uint64, rowWidth),

53
roaring/roaring.go
View file

@ -167,7 +167,6 @@ type ContainerIterator interface {
// Bitmap represents a roaring bitmap.
type Bitmap struct {
Containers Containers
Source Source
// User-defined flags.
Flags byte
@ -248,7 +247,6 @@ func (b *Bitmap) Freeze() *Bitmap {
// Create a copy of the bitmap structure.
other := &Bitmap{
Containers: b.Containers.Freeze(),
Source: b.Source,
}
return other
@ -609,20 +607,13 @@ func (b *Bitmap) OffsetRange(offset, start, end uint64) *Bitmap {
hi0, hi1 := highbits(start), highbits(end)
citer, _ := b.Containers.Iterator(hi0)
other := NewSliceBitmap()
mappedAny := false
for citer.Next() {
k, c := citer.Value()
if k >= hi1 {
break
}
if c.Mapped() {
mappedAny = true
}
other.Containers.Put(off+(k-hi0), c.Freeze())
}
if b.Source != nil && mappedAny {
other.Source = b.Source
}
return other
}
@ -661,7 +652,6 @@ func (b *Bitmap) IntersectionCount(other *Bitmap) uint64 {
// Intersect returns the intersection of b and other.
func (b *Bitmap) Intersect(other *Bitmap) *Bitmap {
output := NewBitmap()
usedB, usedOther := false, false
iiter, _ := b.Containers.Iterator(0)
jiter, _ := other.Containers.Iterator(0)
i, j := iiter.Next(), jiter.Next()
@ -676,26 +666,12 @@ func (b *Bitmap) Intersect(other *Bitmap) *Bitmap {
kj, cj = jiter.Value()
} else { // ki == kj
newC := intersect(ci, cj)
if newC == ci {
usedB = true
}
if newC == cj {
usedOther = true
}
output.Containers.Put(ki, newC)
i, j = iiter.Next(), jiter.Next()
ki, ci = iiter.Value()
kj, cj = jiter.Value()
}
}
switch {
case usedB && usedOther:
output.Source = MergeSources(b.Source, other.Source)
case usedB:
output.Source = b.Source
case usedOther:
output.Source = other.Source
}
return output
}
@ -1192,43 +1168,26 @@ func (b *Bitmap) UnionInPlace(others ...*Bitmap) {
func (b *Bitmap) unionIntoTargetSingle(target *Bitmap, other *Bitmap) {
iiter, _ := b.Containers.Iterator(0)
jiter, _ := other.Containers.Iterator(0)
usedB, usedOther := false, false
i, j := iiter.Next(), jiter.Next()
ki, ci := iiter.Value()
kj, cj := jiter.Value()
for i || j {
if i && (!j || ki < kj) {
target.Containers.Put(ki, ci.Freeze())
usedB = true
i = iiter.Next()
ki, ci = iiter.Value()
} else if j && (!i || ki > kj) {
target.Containers.Put(kj, cj.Freeze())
usedOther = true
j = jiter.Next()
kj, cj = jiter.Value()
} else { // ki == kj
newC := union(ci, cj)
target.Containers.Put(ki, newC)
if newC == ci {
usedB = true
}
if newC == cj {
usedOther = true
}
i, j = iiter.Next(), jiter.Next()
ki, ci = iiter.Value()
kj, cj = jiter.Value()
}
}
switch {
case usedB && usedOther:
target.Source = MergeSources(b.Source, other.Source)
case usedB:
target.Source = b.Source
case usedOther:
target.Source = other.Source
}
}
// unionInPlace stores the union of b and others into b. The others will
@ -1324,14 +1283,7 @@ func (b *Bitmap) unionInPlace(others ...*Bitmap) {
bitmapIters = make(handledIters, 0, requiredSliceSize)
}
var sources []Source
if b.Source != nil {
sources = append(sources, b.Source)
}
for _, other := range others {
if other.Source != nil {
sources = append(sources, other.Source)
}
otherIter, _ := other.Containers.Iterator(0)
if otherIter.Next() {
bitmapIters = append(bitmapIters, handledIter{
@ -1341,8 +1293,6 @@ func (b *Bitmap) unionInPlace(others ...*Bitmap) {
})
}
}
// new bitmap might have containers from any of those bitmaps in it
b.Source = MergeSources(sources...)
// Loop until we've exhausted every iter.
hasNext := true
@ -1505,9 +1455,6 @@ func (b *Bitmap) singleDifference(other *Bitmap) *Bitmap {
// Xor returns the bitwise exclusive or of b and other.
func (b *Bitmap) Xor(other *Bitmap) *Bitmap {
output := NewBitmap()
// Xor can end up with containers from either parent if the other
// had no container or an empty container.
output.Source = MergeSources(b.Source, other.Source)
iiter, _ := b.Containers.Iterator(0)
jiter, _ := other.Containers.Iterator(0)

85
roaring/source.go
View file

@ -1,85 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package roaring
import (
"strings"
)
// A Source represents the source a given bitmap gets its data from,
// such as a memory-mapped file. When combining bitmaps, we might
// track them together in a single combined-source of some sort.
type Source interface {
ID() string
Dead() bool
}
// MergeSources combines sources. If you have two bitmaps, and you're
// combining them, then the combination's source is a combination of
// those two sources.
func MergeSources(sources ...Source) Source {
sourceCount := 0
totalCount := 0
var lastSource Source
for _, s := range sources {
if s == nil {
continue
}
lastSource = s
if s, ok := s.(combinedSource); ok {
sourceCount++
totalCount += len(s)
} else {
sourceCount++
totalCount++
}
}
// if there's no sources (this includes all sources being
// empty combinedSources), we don't have a source.
if totalCount == 0 {
return nil
}
// if there's exactly one source, combined or otherwise, that's
// fine, we'll just return it.
if sourceCount == 1 {
return lastSource
}
// make a new combinedSource, flattening any combinedSources
// already present.
newSources := make([]Source, 0, totalCount)
for _, s := range sources {
if s == nil {
continue
}
if s, ok := s.(combinedSource); ok {
newSources = append(newSources, s...)
} else {
newSources = append(newSources, s)
}
}
return combinedSource(newSources)
}
// SetSource tells the bitmap what source to associate with new things it
// creates. This is possibly logically incorrect.
func (b *Bitmap) SetSource(s Source) {
b.Source = s
}
type combinedSource []Source
func (c combinedSource) ID() string {
ids := make([]string, len(c))
for i := range c {
ids[i] = c[i].ID()
}
return strings.Join(ids, ",")
}
func (c combinedSource) Dead() bool {
for i := range c {
if c[i].Dead() {
return true
}
}
return false
}

662
rrtx.go
View file

@ -1,662 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package pilosa
import (
"fmt"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"sync/atomic"
"github.com/molecula/featurebase/v3/roaring"
txkey "github.com/molecula/featurebase/v3/short_txkey"
"github.com/molecula/featurebase/v3/storage"
"github.com/molecula/featurebase/v3/vprint"
"github.com/pkg/errors"
)
// RoaringTx represents a fake transaction object for Roaring storage.
type RoaringTx struct {
write bool
Index *Index
Field *Field
fragment *fragment
o Txo
sn int64 // serial number
done bool
mu sync.Mutex // protect done as it changes state
w *RoaringWrapper
}
func (tx *RoaringTx) Type() string {
return RoaringTxn
}
// based on view.openFragments()
func roaringMapOfShards(optionalViewPath string) (shardMap map[uint64]bool, err error) {
shardMap = make(map[uint64]bool)
path := filepath.Join(optionalViewPath, "fragments")
file, err := os.Open(path)
if os.IsNotExist(err) {
return
} else if err != nil {
return nil, errors.Wrap(err, "opening fragments directory")
}
defer file.Close()
fis, err := file.Readdir(0)
if err != nil {
return nil, errors.Wrap(err, "reading fragments directory")
}
for _, fi := range fis {
//vv("rrtx next fi = '%v'", fi.Name())
if fi.IsDir() {
continue
}
name := fi.Name()
if strings.HasSuffix(name, ".cache") {
continue
}
// Parse filename into integer.
shard, err := strconv.ParseUint(filepath.Base(name), 10, 64)
if err != nil {
//vv("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", index, field, view, fi.Name())
//panic(fmt.Sprintf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", index, field, view, fi.Name()))
//tx.Index.holder.Logger.Debugf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", index, field, view, fi.Name())
continue
}
shardMap[shard] = true
}
return
}
// NewTxIterator returns a *roaring.Iterator that MUST have Close() called on it BEFORE
// the transaction Commits or Rollsback.
func (tx *RoaringTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator {
b, err := tx.bitmap(index, field, view, shard)
vprint.PanicOn(err)
return b.Iterator()
}
// ImportRoaringBits return values changed and rowSet will be inaccurate if
// the data []byte is supplied. This mimics the traditional roaring-per-file
// and should be faster.
func (tx *RoaringTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) {
f, err := tx.getFragment(index, field, view, shard)
if err != nil {
return 0, nil, err
}
changed, rowSet, err = f.storage.ImportRoaringRawIterator(rit, clear, true, rowSize)
return
}
func (c *RoaringTx) ApplyFilter(index, field, view string, shard uint64, ckey uint64, filter roaring.BitmapFilter) (err error) {
return GenericApplyFilter(c, index, field, view, shard, ckey, filter)
}
// Rollback
func (tx *RoaringTx) Rollback() {
tx.w.CleanupTx(tx)
}
// Commit
func (tx *RoaringTx) Commit() error {
tx.w.CleanupTx(tx)
return nil
}
func (tx *RoaringTx) RoaringBitmap(index, field, view string, shard uint64) (*roaring.Bitmap, error) {
return tx.bitmap(index, field, view, shard)
}
func (tx *RoaringTx) Container(index, field, view string, shard uint64, key uint64) (*roaring.Container, error) {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return nil, err
}
return b.Containers.Get(key), nil
}
func (tx *RoaringTx) PutContainer(index, field, view string, shard uint64, key uint64, c *roaring.Container) error {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return err
}
b.Containers.Put(key, c)
return nil
}
func (tx *RoaringTx) RemoveContainer(index, field, view string, shard uint64, key uint64) error {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return err
}
b.Containers.Remove(key)
return nil
}
func (tx *RoaringTx) Add(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
//vv("RoaringTx.Add(index='%v', shard='%v') stack=\n%v", index, shard, stack())
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return 0, err
}
// Note: do not replace b.AddN() with b.DirectAddN().
// DirectAddN() does not do op-log operations inside roaring, so the
// on-disk representation no longer matches the in-memory operations.
count, err := b.AddN(a...)
return count, err
}
func (tx *RoaringTx) Remove(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return 0, err
}
return b.RemoveN(a...)
}
func (tx *RoaringTx) Contains(index, field, view string, shard uint64, v uint64) (exists bool, err error) {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return false, err
}
return b.Contains(v), nil
}
func (tx *RoaringTx) ContainerIterator(index, field, view string, shard uint64, key uint64) (citer roaring.ContainerIterator, found bool, err error) {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return nil, false, errors.Wrap(err, "getting bitmap")
}
//vv("b bitmap back from bitmap(index='%v', field='%v', view='%v', shard='%v')='%#v'", index, field, view, shard, b.Slice())
citer, found = b.Containers.Iterator(key)
return citer, found, nil
}
func (tx *RoaringTx) ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return err
}
return b.ForEach(fn)
}
func (tx *RoaringTx) ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return err
}
return b.ForEachRange(start, end, fn)
}
func (tx *RoaringTx) Count(index, field, view string, shard uint64) (uint64, error) {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return 0, err
}
return b.Count(), nil
}
func (tx *RoaringTx) Max(index, field, view string, shard uint64) (uint64, error) {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return 0, err
}
return b.Max(), nil
}
func (tx *RoaringTx) Min(index, field, view string, shard uint64) (uint64, bool, error) {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return 0, false, err
}
v, ok := b.Min()
return v, ok, nil
}
func (tx *RoaringTx) CountRange(index, field, view string, shard uint64, start, end uint64) (uint64, error) {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return 0, err
}
return b.CountRange(start, end), nil
}
func (tx *RoaringTx) OffsetRange(index, field, view string, shard uint64, offset, start, end uint64) (*roaring.Bitmap, error) {
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return nil, err
}
return b.OffsetRange(offset, start, end), nil
}
// getFragment is used by IncrementOpN() and by bitmap()
func (tx *RoaringTx) getFragment(index, field, view string, shard uint64) (*fragment, error) {
// If a fragment is attached, always use it. Since it was set at Tx creation,
// it is highly likely to be correct.
if tx.fragment != nil {
// but still a basic sanity check.
if tx.fragment.index() != index ||
tx.fragment.field() != field ||
tx.fragment.view() != view ||
tx.fragment.shard != shard {
// still insist that index and shard match, since that is the current scope of all Tx.
if tx.fragment.index() != index ||
tx.fragment.shard != shard {
panic(fmt.Sprintf("different fragment cached vs requested. index='%v', field='%v'; view='%v'; shard='%v'; tx.fragment='%#v'", index, field, view, shard, tx.fragment))
}
// cannot use this fragment.
tx.fragment = nil
} else {
return tx.fragment, nil
}
}
// If a field is attached, start from there.
// Otherwise look up the field from the index.
f := tx.Field
if f == nil {
// we cannot assume that the tx.Index that we "started" on is the same
// as the index we are being queried; it might be foreign: TestExecutor_ForeignIndex
// So go through the holder
idx := tx.Index.holder.Index(index)
if idx == nil {
// only thing we can try is the cached index, and hope we aren't being asked for a foreign index.
f = tx.Index.Field(field)
if f == nil {
return nil, newNotFoundError(ErrFieldNotFound, field)
}
} else {
if f = idx.Field(field); f == nil {
return nil, newNotFoundError(ErrFieldNotFound, field)
}
}
}
// INVAR: f is not nil.
v := f.view(view)
if v == nil {
return nil, errors.Wrapf(ViewNotFound, "getting %s", view)
}
frag := v.Fragment(shard)
if frag == nil {
return nil, errors.Wrapf(FragmentNotFound, "field:%q, view:%q, shard:%d", field, view, shard)
}
// Note: we cannot cache frag into tx.fragment.
// Empirically, it breaks 245 top-level pilosa tests.
// tx.fragment = frag // breaks the world.
return frag, nil
}
const ViewNotFound = Error("view not found")
const FragmentNotFound = Error("fragment not found")
func (tx *RoaringTx) bitmap(index, field, view string, shard uint64) (*roaring.Bitmap, error) {
frag, err := tx.getFragment(index, field, view, shard)
if err != nil {
return nil, errors.Wrap(err, "getFragment")
}
return frag.storage, nil
}
func roaringGetFieldView2Shards(idx *Index) (vs *FieldView2Shards, err error) {
vs = NewFieldView2Shards()
// A) open the index directory
f, err := os.Open(idx.FieldsPath())
if err != nil {
return nil, errors.Wrap(err, "opening directory")
}
defer f.Close()
fieldFIs, err := f.Readdir(0)
if err != nil {
return nil, errors.Wrap(err, "reading directory")
}
//vv("roaringGetFieldView2Shards A) opened index path '%v'", idx.path)
// B) read the name of each field under the index
for _, loopFieldFi := range fieldFIs {
fieldFI := loopFieldFi
if !fieldFI.IsDir() {
continue
}
field := fieldFI.Name()
//vv("roaringGetFieldView2Shards B) on field '%v'", field)
fieldPath := filepath.Join(idx.FieldsPath(), field)
viewsDir := filepath.Join(fieldPath, "views")
file, err := os.Open(viewsDir)
if os.IsNotExist(err) {
//return nil
continue
} else if err != nil {
return nil, errors.Wrapf(err, "opening view directory '%v'", viewsDir)
}
defer file.Close()
// C) read the name of each view under the field
viewFIs, err := file.Readdir(0)
if err != nil {
return nil, errors.Wrapf(err, "reading views directory '%v'", viewsDir)
}
for _, viewFI := range viewFIs {
if !viewFI.IsDir() {
continue
}
view := viewFI.Name()
roaringViewPath := filepath.Join(viewsDir, view)
shardMap, err := roaringMapOfShards(roaringViewPath)
if err != nil {
return nil, errors.Wrapf(err, "reading view path directory '%v'", roaringViewPath)
}
if len(shardMap) == 0 {
//vv("roaringGetFieldView2Shards C) SAVED SPACE! field '%v' view '%v' had no shards", field, view)
continue
}
ss := newShardSetFromMap(shardMap)
fv := txkey.FieldView{Field: field, View: view}
vs.addViewShardSet(fv, ss)
//vv("roaringGetFieldView2Shards C) added field '%v' view '%v' with shards '%#v'", field, view, ss.shards)
}
}
return
}
// inefficient for roaring. Instead use the roaringGetFieldView2Shards() above.
func (tx *RoaringTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) {
// A) open the index directory
f, err := os.Open(idx.FieldsPath())
if err != nil {
return nil, errors.Wrap(err, "opening directory")
}
defer f.Close()
fieldFIs, err := f.Readdir(0)
if err != nil {
return nil, errors.Wrap(err, "reading directory")
}
//vv("A) shard %v, opened index path '%v'", shard, idx.path)
// B) read the name of each field under the index
for _, loopFieldFi := range fieldFIs {
fieldFI := loopFieldFi
if !fieldFI.IsDir() {
continue
}
field := fieldFI.Name()
//vv("B) on field '%v'", field)
fieldPath := filepath.Join(idx.FieldsPath(), field)
viewsDir := filepath.Join(fieldPath, "views")
file, err := os.Open(viewsDir)
if os.IsNotExist(err) {
//return nil
continue
} else if err != nil {
return nil, errors.Wrapf(err, "opening view directory '%v'", viewsDir)
}
defer file.Close()
// C) read the name of each view under the field
viewFIs, err := file.Readdir(0)
if err != nil {
return nil, errors.Wrapf(err, "reading views directory '%v'", viewsDir)
}
for _, viewFI := range viewFIs {
if !viewFI.IsDir() {
continue
}
view := viewFI.Name()
roaringViewPath := filepath.Join(viewsDir, view)
shardMap, err := roaringMapOfShards(roaringViewPath)
if err != nil {
return nil, errors.Wrapf(err, "reading view path directory '%v'", roaringViewPath)
}
if len(shardMap) == 0 {
continue
}
// once we know we have data for this shard!
if shardMap[shard] {
fv := txkey.FieldView{Field: field, View: view}
//vv("C) adding fv '%#v'", fv)
fvs = append(fvs, fv)
}
}
}
// directory stuff isn't returned in sorted order, we must sort.
sort.Slice(fvs, func(i, j int) bool {
if fvs[i].Field < fvs[j].Field {
return true
}
if fvs[i].Field > fvs[j].Field {
return false
}
return fvs[i].View < fvs[j].View
})
return
}
func (tx *RoaringTx) GetFieldSizeBytes(index, field string) (uint64, error) {
return 0, nil
}
//////// registrar and wrapper machinery
// roaringRegistrar mirrors the machinery expected
// for all backends for the roaring files approach.
//
type roaringRegistrar struct {
mu sync.Mutex
mp map[*RoaringWrapper]bool
path2db map[string]*RoaringWrapper
}
func (r *roaringRegistrar) Size() int {
r.mu.Lock()
defer r.mu.Unlock()
nmp := len(r.mp)
npa := len(r.path2db)
if nmp != npa {
panic(fmt.Sprintf("nmp=%v, vs npa=%v", nmp, npa))
}
return nmp
}
var globalRoaringReg *roaringRegistrar = newRoaringRegistrar()
func newRoaringRegistrar() *roaringRegistrar {
return &roaringRegistrar{
mp: make(map[*RoaringWrapper]bool),
path2db: make(map[string]*RoaringWrapper),
}
}
func (r *roaringRegistrar) unprotectedRegister(w *RoaringWrapper) {
r.mp[w] = true
r.path2db[w.path] = w
}
// unregister removes w from r
func (r *roaringRegistrar) unregister(w *RoaringWrapper) {
r.mu.Lock()
delete(r.mp, w)
delete(r.path2db, w.path)
r.mu.Unlock()
}
// openRoaringDB will check the registry and make a new instance only
// if one does not exist for its path0. Otherwise it returns
// the existing instance.
func (r *roaringRegistrar) OpenDBWrapper(path string, doAllocZero bool, _ *storage.Config) (DBWrapper, error) {
r.mu.Lock()
defer r.mu.Unlock()
w, ok := r.path2db[path]
if ok {
return w, nil
}
// otherwise, make a new roaring and store it in globalRoaringReg
w = &RoaringWrapper{
reg: r,
path: path,
}
r.unprotectedRegister(w)
return w, nil
}
func (w *RoaringWrapper) SetHolder(h *Holder) {
w.h = h
}
func (w *RoaringWrapper) Path() string {
return w.path
}
func (w *RoaringWrapper) HasData() (has bool, err error) {
return w.h.HasRoaringData()
}
func (w *RoaringWrapper) CleanupTx(tx Tx) {
r := tx.(*RoaringTx)
r.mu.Lock()
defer r.mu.Unlock()
if r.done {
return
}
r.done = true
}
func (w *RoaringWrapper) OpenListString() (r string) {
return "RoaringWrapper.OpenListString() not yet implemented"
}
func (w *RoaringWrapper) CloseDB() error {
return errors.New("CloseDB not supported in roaring")
}
func (w *RoaringWrapper) OpenDB() error {
return errors.New("OpenDB not supported in roaring")
}
// statically confirm that RoaringTx satisfies the Tx interface.
var _ Tx = (*RoaringTx)(nil)
// RoaringWrapper provides the NewTx() method.
type RoaringWrapper struct {
muDb sync.Mutex
path string
h *Holder
reg *roaringRegistrar
// make RoaringWrapper.Close() idempotent, avoiding panic on double Close()
closed bool
}
var globalNextTxSnRoaring int64
func (w *RoaringWrapper) NewTx(write bool, initialIndexName string, o Txo) (tx Tx, err error) {
sn := atomic.AddInt64(&globalNextTxSnRoaring, 1)
return &RoaringTx{
write: o.Write,
Field: o.Field,
Index: o.Index,
fragment: o.Fragment,
o: o,
sn: sn,
w: w,
}, nil
}
// Close shuts down the Roaring database.
func (w *RoaringWrapper) Close() (err error) {
w.muDb.Lock()
defer w.muDb.Unlock()
if !w.closed {
w.reg.unregister(w)
w.closed = true
}
return nil
}
func (w *RoaringWrapper) IsClosed() (closed bool) {
w.muDb.Lock()
closed = w.closed
w.muDb.Unlock()
return
}
func (w *RoaringWrapper) DeleteField(index, field, fieldPath string) error {
//vv("RoaringWrapper.DeleteField(index = '%v', field = '%v', fieldPath = '%v'", index, field, fieldPath)
// match txn sn count vs lmdb/etc.
atomic.AddInt64(&globalNextTxSnRoaring, 1)
err := os.RemoveAll(fieldPath)
if err != nil {
return errors.Wrap(err, "removing directory")
}
return nil
}
func (w *RoaringWrapper) DeleteFragment(index, field, view string, shard uint64, frag interface{}) error {
// match txn sn count vs lmdb/etc.
atomic.AddInt64(&globalNextTxSnRoaring, 1)
fragment, ok := frag.(*fragment)
if !ok {
return fmt.Errorf("RoaringStore.DeleteFragment must get frag of type *fragment, but got '%T'", frag)
}
// Delete fragment file.
if err := os.Remove(fragment.path()); err != nil {
return errors.Wrap(err, "deleting fragment file")
}
// Delete fragment cache file.
if err := os.Remove(fragment.cachePath()); err != nil {
return errors.Wrap(err, fmt.Sprintf("no cache file to delete for shard %d", fragment.shard))
}
return nil
}

22
server.go
View file

@ -64,11 +64,10 @@ type Server struct { // nolint: maligned
schemator disco.Schemator
// External
systemInfo SystemInfo
gcNotifier GCNotifier
logger logger.Logger
queryLogger logger.Logger
snapshotQueue SnapshotQueue
systemInfo SystemInfo
gcNotifier GCNotifier
logger logger.Logger
queryLogger logger.Logger
nodeID string
uri pnet.URI
@ -544,13 +543,6 @@ func (s *Server) UpAndDown() error {
func (s *Server) Open() error {
s.logger.Infof("open server. PID %v", os.Getpid())
if s.holder.NeedsSnapshot() {
// Start background monitoring.
s.snapshotQueue = newSnapshotQueue(10, 2, s.logger)
} else {
s.snapshotQueue = defaultSnapshotQueue //TODO (twg) rethink this
}
// Log startup
err := s.holder.logStartup()
if err != nil {
@ -612,7 +604,6 @@ func (s *Server) Open() error {
return errors.Wrap(err, "opening Holder")
}
// bring up the background tasks for the holder.
s.holder.SnapshotQueue = s.snapshotQueue
s.holder.Activate()
// if we joined existing cluster then broadcast "resize on add" message
if initState == disco.InitialClusterStateExisting {
@ -743,11 +734,6 @@ func (s *Server) Close() error {
if s.holder != nil {
errh = s.holder.Close()
}
if s.snapshotQueue != nil {
s.holder.SnapshotQueue = nil
s.snapshotQueue.Stop()
s.snapshotQueue = nil
}
// prefer to return holder error over cluster
// error. This order is somewhat arbitrary. It would be better if we had

18
server_internal_test.go
View file

@ -2,7 +2,6 @@
package pilosa
import (
"runtime"
"testing"
"time"
@ -10,23 +9,6 @@ import (
"github.com/molecula/featurebase/v3/testhook"
)
// Ensure the file handle count is working
func TestCountOpenFiles(t *testing.T) {
roaringOnlyTest(t)
// Windows is not supported yet
if runtime.GOOS == "windows" {
t.Skip("Skipping unsupported countOpenFiles test on Windows.")
}
count, err := countOpenFiles()
if err != nil {
t.Errorf("countOpenFiles failed: %s", err)
}
if count == 0 {
t.Error("countOpenFiles returned invalid value 0.")
}
}
func TestMonitorAntiEntropyZero(t *testing.T) {
td, err := testhook.TempDirInDir(t, *TempDir, "")

495
snapshotqueue.go
View file

@ -1,495 +0,0 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package pilosa
import (
"context"
"fmt"
"io"
"math/bits"
"os"
"sync"
"sync/atomic"
"time"
"github.com/molecula/featurebase/v3/logger"
"github.com/molecula/featurebase/v3/testhook"
"github.com/pkg/errors"
)
// snapshotQueue is a thing which can handle enqueuing snapshots. A snapshot
// queue distinguishes between high-priority requests, which get satisfied
// by the next available worker, and regular requests, which get enqueued
// if there's space in the queue, and otherwise dropped. There's also a
// separate background task to scan a holder for fragments which may need
// snapshots, but which is processed only when the queue is empty, and only
// slowly. "Await" awaits an existing snapshot if one is already enqueued.
// "Immediate" tries to do one right away. (If one's already enqueued, this
// can leave it in the queue, which will ignore anything that shows up with
// the request flag cleared.)
//
// Await, Enqueue, and Immediate should be called only with the fragment lock
// held.
//
// If you create a queue, it should get stopped at some point. The
// atomicSnapshotQueue implementation used as defaultSnapshotQueue has
// a Start function which will tell you whether it actually started a
// queue. This logic exists because in a normal server case, you probably
// want the queue to be shut down as part of server shutdown, but if you're
// running cluster tests, you probably want to start and shop the queue as
// part of the test, not stop it when any server terminates.
//
// It's less likely to be desireable to start/stop individual queues,
// because fragments use the defaultSnapshotQueue anyway. This design
// needs revisiting.
type SnapshotQueue interface {
Immediate(*fragment) error
Enqueue(*fragment)
Await(*fragment) error
ScanHolder(*Holder, chan struct{})
Stop()
}
// queuelessSnapshotQueue isn't a snapshot queue, but it satisfies the
// interface.
type queuelessSnapshotQueue struct{}
func (q *queuelessSnapshotQueue) Enqueue(f *fragment) {
// We don't actually try to enqueue the snapshot; it breaks things
// if a snapshot gets caused during a transaction.
}
func (q *queuelessSnapshotQueue) Await(f *fragment) error {
return nil
}
func (q *queuelessSnapshotQueue) Immediate(f *fragment) error {
return f.snapshot()
}
func (q *queuelessSnapshotQueue) ScanHolder(h *Holder, done chan struct{}) {
}
func (q *queuelessSnapshotQueue) Stop() {
}
var defaultSnapshotQueue = &queuelessSnapshotQueue{}
// newSnapshotQueue makes a new snapshot queue, of depth N, with
// w worker threads.
func newSnapshotQueue(n int, w int, l logger.Logger) SnapshotQueue {
ctx, cancel := context.WithCancel(context.Background())
sq := &prioritySnapshotQueue{
normal: make(chan snapshotRequest, n),
urgent: make(chan snapshotRequest),
background: make(chan snapshotRequest),
ctx: ctx,
cancel: cancel,
maxOpN: 10000,
logger: l,
}
if sq.logger == nil {
sq.logger = logger.NewStandardLogger(os.Stderr)
}
_ = testhook.Opened(NewAuditor(), sq, nil)
sq.spawnWorkers(w)
return sq
}
type snapshotRequest struct {
frag *fragment
when time.Time
}
// prioritySnapshotQueue gives preference to "immediate" requests, and
// dispreference to "background" requests from ScanHolder. It timestamps
// requests, so it can discard a request if the most recent snapshot is
// newer than the request. The snapshotPending flag in the fragment is
// used to track that a given fragment thinks it has been successfully
// enqueued. Background requests are not considered enqueued, since
// they'll never get processed if there's anything else. In normal workloads,
// immediate/urgent snapshots should be rare, but we'll happily drop
// most requests on the floor; the scanner should pick them up once things
// are quiet.
type prioritySnapshotQueue struct {
logger logger.Logger
urgent chan snapshotRequest
normal chan snapshotRequest
background chan snapshotRequest
ctx context.Context
cancel context.CancelFunc
mu sync.RWMutex
scanWG, workerWG sync.WaitGroup
maxOpN int
observedOpN [16]uint32
stats struct {
enqueued uint32
skipped uint32
}
stopped bool
}
func (sq *prioritySnapshotQueue) spawnWorkers(w int) {
sq.mu.Lock()
defer sq.mu.Unlock()
if sq.ctx.Err() != nil {
sq.logger.Infof("prioritySnapshotQueue worker: already done")
return
}
sq.workerWG.Add(w)
for i := 0; i < w; i++ {
go sq.worker(sq.ctx, sq.urgent, sq.normal, sq.background)
}
}
func (sq *prioritySnapshotQueue) worker(ctx context.Context, urgent, normal, background chan snapshotRequest) {
defer sq.workerWG.Done()
done := ctx.Done()
ok := true
var req snapshotRequest
for ok {
req.frag = nil
select {
case _, ok = <-done:
case req, ok = <-urgent:
default:
select {
case _, ok = <-done:
case req, ok = <-urgent:
case req, ok = <-normal:
default:
select {
case _, ok = <-done:
case req, ok = <-urgent:
case req, ok = <-normal:
case req, ok = <-background:
}
}
}
if req.frag != nil {
sq.process(req)
}
}
}
// process actually runs a fragment. it will do this if either the fragment
// has a pending snapshot, or the force flag is set.
func (sq *prioritySnapshotQueue) process(req snapshotRequest) {
f := req.frag
f.mu.Lock()
defer f.mu.Unlock()
if f.snapshotStamp.Before(req.when) {
f.snapshotErr = f.snapshot()
if f.snapshotErr != nil {
fmt.Printf("ERROR: snapshot error: %v\n", f.snapshotErr)
sq.logger.Errorf("snapshot error: %v", f.snapshotErr)
}
f.snapshotPending = false
f.snapshotCond.Broadcast()
}
}
// Stop shuts down the snapshot queue. It first marks it as done, causing
// the background scanner(s), if any, to shut down, then waits for them, then
// closes and nils the queues. The background scanner has to get stopped
// because otherwise it might try to write to those closed queues.
func (sq *prioritySnapshotQueue) Stop() {
sq.mu.Lock()
defer sq.mu.Unlock()
if sq.stopped {
return
}
sq.stopped = true
sq.cancel()
// scanners need to be done before we close the other channels.
sq.scanWG.Wait()
close(sq.normal)
sq.normal = nil
close(sq.urgent)
sq.urgent = nil
close(sq.background)
sq.background = nil
_ = testhook.Closed(NewAuditor(), sq, nil)
enqueued := atomic.LoadUint32(&sq.stats.enqueued)
skipped := atomic.LoadUint32(&sq.stats.skipped)
if skipped > 0 || enqueued > 1 {
sq.logger.Infof("snapshot queue: enqueued %d, skipped %d\n", sq.stats.enqueued, sq.stats.skipped)
}
}
// Enqueue tries to add a fragment to the queue, if the fragment is not already
// enqueued. You should hold a lock on the fragment when calling this.
func (sq *prioritySnapshotQueue) Enqueue(f *fragment) {
if f.snapshotPending {
return
}
sq.observeOpN(uint32(f.opN))
sq.mu.RLock()
defer sq.mu.RUnlock()
if sq.normal == nil {
sq.logger.Infof("requested snapshot after snapshot queue was closed")
return
}
// we have to set this before enqueing, because it's
// otherwise possible that we're at the head of the queue,
// and the recipient gets the fragment before we execute the
// line after the send.
f.snapshotPending = true
// try to enqueue snapshot
select {
case sq.normal <- snapshotRequest{frag: f, when: time.Now()}:
atomic.AddUint32(&sq.stats.enqueued, 1)
return
default:
atomic.AddUint32(&sq.stats.skipped, 1)
f.snapshotPending = false
return
}
}
// Await returns when f is not pending a snapshot. Call with the fragment lock
// held. Await waits on a condition variable inside f, associated with the
// fragment's lock, so this does not conflict with the lock being used for
// snapshots.
//
// Note that workers don't stop just because the queue's been stopped; only
// the background scanner is stopped. So an Await shouldn't block forever
// even if the queue gets shut down. If you're reading this, possibly that
// analysis is incorrect.
func (sq *prioritySnapshotQueue) Await(f *fragment) (err error) {
for f.snapshotPending {
f.snapshotCond.Wait()
}
err, f.snapshotErr = f.snapshotErr, nil
return err
}
// Immediate forces an immediate snapshot of the given fragment. Call with
// the fragment locked. If the queue is already closing, the fragment does
// not get snapshotted.
func (sq *prioritySnapshotQueue) Immediate(f *fragment) error {
sq.mu.RLock()
// no deferred unlock, because we want to unlock this before calling Await.
// Not because that needs this lock, but because once we're that far, we
// *don't* need this lock anymore so someone else should have it.
if sq.urgent == nil {
sq.mu.RUnlock()
sq.logger.Errorf("requested immediate snapshot after snapshot queue was closed")
return errors.New("requested immediate snapshot after snapshot queue was closed")
}
f.snapshotPending = true
sq.observeOpN(uint32(f.opN))
req := snapshotRequest{frag: f, when: time.Now()}
// if the fragment was already in the work queue, it's *possible*
// that the only available worker just picked it off the queue, and
// is now waiting on getting the fragment's lock, so it can run
// a snapshot. So we let go of the lock on the fragment, send the
// request, then request the fragment lock again, because Await will
// be sleeping on the condition variable associated with the lock,
// which means it needs to hold the lock so it can let it go during
// the wait... No, really, this made sense.
f.mu.Unlock()
sq.urgent <- req
sq.mu.RUnlock()
f.mu.Lock()
return sq.Await(f)
}
// ScanHolder spawns a goroutine which iterates through the holder's
// indexes/fields/views/fragments, looking for fragments which have OpN
// high enough to justify a snapshot but don't seem to have one pending.
// It then dumps these in the low priority background queue.
func (sq *prioritySnapshotQueue) ScanHolder(h *Holder, done chan struct{}) {
sq.mu.Lock()
sq.scanWG.Add(1)
go sq.scanHolderWorker(h, sq.background, done)
sq.mu.Unlock()
}
// observeOpN reports that a given value of opN was "observed", meaning,
// we encountered a fragment which had that value. This happens for every
// enqueue/immediate, including enqueue attempts which fail to actually
// enter the queue, and it also happens for fragments noticed by the background
// scan but which don't have high enough opN to trigger a snapshot.
func (sq *prioritySnapshotQueue) observeOpN(n uint32) {
// aka "log2(n) + 1", or 0 for n==0
pow2 := 32 - bits.LeadingZeros32(n)
// 15 == 16384. Our usual fragment maxOpN is 10k, so most fragments
// should end up in the 8k-16k bucket, rather than the 16k+ bucket,
// unless we've got a lot of ingests with large batches going on,
// in which case the 16k bucket will win.
if pow2 > 15 {
pow2 = 15
}
// store in inverse order so the lowest slot in the array is the
// highest cardinality
atomic.AddUint32(&sq.observedOpN[15-pow2], 1)
}
// computeMaxOpN tries to pick a reasonable new maxOpN for the background
// scan to use. On a quiet system, we want to gradually lower opN, picking
// the fragments with the highest opN values first, because those offer the
// largest benefit. So, whenever we check a fragment in the background, if we
// *don't* snapshot it, we'll "observe" its OpN value, and then we pick a
// value which picks up at least 1/4 of them.
//
// If there's ingest activity, the Immediate and Enqueue operations will
// "observe" the OpN of fragments submitted to them. This can drive OpN back
// up, if those fragments frequently have very high opN values, which reflects
// the fact that we have enough of that activity that we don't need the
// background scanner adding more.
//
// If we have enough ingest activity that the background scanner never actually
// gets to submit work, we'll rarely get here, because the background scanner
// will block until there's no snapshots pending for the normal workload.
// When we do, we'll probably pick a MaxOpN which is dominated by the ingest
// workload's opN values. So for instance, if everything coming in from the
// ingest workload has 10k or more items, because that's the default fragment
// maxOpN, that will probably set the background snapshot queue value to 8k.
func (sq *prioritySnapshotQueue) computeMaxOpN() {
sq.logger.Debugf("observedOpN by power of 2: %d\n", sq.observedOpN[:])
total := uint32(0)
for i := range sq.observedOpN {
total += atomic.LoadUint32(&sq.observedOpN[i])
}
target := (total / 4) + 1
subTotal := uint32(0)
for i := range sq.observedOpN {
v := atomic.LoadUint32(&sq.observedOpN[i])
subTotal += v
if subTotal >= target {
prevMaxOpN := sq.maxOpN
sq.maxOpN = (1 << (15 - uint(i))) / 2
if sq.maxOpN > 0 {
sq.maxOpN--
}
if prevMaxOpN != sq.maxOpN {
sq.logger.Infof("background scan: %d/%d fragments considered have opN %d or higher\n",
subTotal, total, sq.maxOpN)
}
break
}
}
// It's conceptually possible that we'll miss a couple of observations
// here but that's not really important. This is all pretty approximate.
for i := range sq.observedOpN {
atomic.StoreUint32(&sq.observedOpN[i], 0)
}
}
// prioritySnapshotQueueScanner is the data type that implements HolderOperator
// and represents a single scan of a holder, with a given maxOpN.
type prioritySnapshotQueueScanner struct {
HolderFilterAll
HolderProcessNone
sq *prioritySnapshotQueue
holder *Holder
queue chan snapshotRequest
ctx context.Context
maxOpN int
seen, hits, counter int
}
func (s *prioritySnapshotQueueScanner) ProcessFragment(f *fragment) error {
if f == nil {
return nil
}
s.seen++
// we can't defer this reasonably, because otherwise we'll keep
// the fragment locked forever if we end up trying to send it
// to the queue, but the workers are busy on other fragments.
f.mu.Lock()
open := f.open
snapshotPending, opN := f.snapshotPending, f.opN
f.mu.Unlock()
// a pending snapshot is one that is either in the normal or
// immediate queue, or is trying to get into the normal queue
// and about to fail, but either way, it already got observed
// there, so we don't need to observe it here. A closed fragment
// doesn't matter to us -- it should be a transient state that
// happens during a shutdown, or shouldn't happen, but we don't
// care about it.
if snapshotPending || !open {
return nil
}
if opN <= s.maxOpN {
// observe the value but don't do a snapshot
s.sq.observeOpN(uint32(opN))
s.counter++
if s.counter == 1000 {
select {
case <-time.After(1 * time.Second):
case <-s.ctx.Done():
return io.EOF
}
s.counter = 0
}
return nil
}
// we don't observe values when we decide to trigger a snapshot,
// because those values will be changing anyway. we could also
// observe them as zero, but that's also sort of wrong.
s.hits++
select {
case s.queue <- snapshotRequest{frag: f, when: time.Now()}:
s.sq.logger.Debugf("found fragment needing snapshot: %s\n", f.path())
case <-s.ctx.Done():
return io.EOF
}
return nil
}
func contextMergedWithStructChan(ctx context.Context, ch chan struct{}) (context.Context, context.CancelFunc) {
canCancel, cancel := context.WithCancel(ctx)
go func() {
select {
case <-ctx.Done():
cancel()
case <-ch:
cancel()
case <-canCancel.Done():
// don't need to cancel, but do need to exit this
// function
}
}()
return canCancel, cancel
}
// scanHolderWorker is a background task that scans a holder looking for
// fragments which need snapshots taken. It's the cleanup task for snapshots
// that would have been requested by Enqueue, but the queue was full.
func (sq *prioritySnapshotQueue) scanHolderWorker(h *Holder, background chan snapshotRequest, done chan struct{}) {
defer sq.scanWG.Done()
ctx, cancel := contextMergedWithStructChan(sq.ctx, done)
defer cancel()
scanner := &prioritySnapshotQueueScanner{
sq: sq,
holder: h,
queue: background,
ctx: sq.ctx,
maxOpN: sq.maxOpN,
}
for {
err := h.Process(ctx, scanner)
if err != nil {
return
}
if scanner.hits > 0 {
sq.logger.Infof("background scan: %d/%d fragments needed snapshots\n", scanner.hits, scanner.seen)
scanner.hits = 0
} else {
sq.logger.Debugf("background scan: no fragments needed snapshots, waiting\n")
// No reason to be active if we're not finding anything.
select {
case <-time.After(60 * time.Second):
case <-ctx.Done():
return
}
}
scanner.seen = 0
sq.computeMaxOpN()
scanner.maxOpN = sq.maxOpN
}
}

3
stattx.go
View file

@ -12,6 +12,7 @@ import (
"github.com/molecula/featurebase/v3/debugstats"
"github.com/molecula/featurebase/v3/roaring"
txkey "github.com/molecula/featurebase/v3/short_txkey"
"github.com/molecula/featurebase/v3/storage"
"github.com/molecula/featurebase/v3/vprint"
)
@ -56,7 +57,7 @@ func (w *callStats) reset() {
}
func (c *callStats) report() (r string) {
backend := CurrentBackend()
backend := storage.DefaultBackend
r = fmt.Sprintf("callStats: (%v)\n", backend)
c.mu.Lock()
defer c.mu.Unlock()

4
storage/config.go
View file

@ -3,9 +3,7 @@ package storage
// public strings that pilosa/server/config.go can reference
const (
RoaringBackend string = "roaring"
RBFBackend string = "rbf"
BoltBackend string = "bolt"
RBFBackend string = "rbf"
)
// DefaultBackend is set here. pilosa/server/config.go references it

2
test/cluster.go
View file

@ -593,7 +593,7 @@ func prependTestServerOpts(opts []server.CommandOption) []server.CommandOption {
pilosa.OptServerOpenTranslateStore(pilosa.OpenInMemTranslateStore),
pilosa.OptServerNodeDownRetries(5, 100*time.Millisecond),
pilosa.OptServerStorageConfig(&storage.Config{
Backend: pilosa.CurrentBackendOrDefault(),
Backend: storage.DefaultBackend,
FsyncEnabled: false,
}),
),

13
tournament.sh
View file

@ -1,13 +0,0 @@
#!/bin/bash
## tournament.sh runs a sequence of duels between greens and blues.
## Each test run changes the PILOSA_STORAGE_BACKEND and runs either
## one or two backends through the rigors of make testv-race.
## logs are saved to the tourna.log.${i} files.
for i in rbf roaring bolt rbf_roaring roaring_rbf roaring_bolt; do
echo "$(date) starting ${i}, output to tourna.log.${i}"
echo "***=== ${i} ====================*** $(date)" &> tourna.log.${i}
PILOSA_STORAGE_BACKEND=${i} make testv-race 2>&1 > tourna.log.${i}
done

10
tx_test.go
View file

@ -4,13 +4,11 @@ package pilosa_test
import (
"context"
"fmt"
"strings"
"testing"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/http"
"github.com/molecula/featurebase/v3/server"
"github.com/molecula/featurebase/v3/storage"
"github.com/molecula/featurebase/v3/test"
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
)
@ -47,15 +45,7 @@ func queryBalances(m0api *pilosa.API, acctOwnerID uint64, fldAcct0, fldAcct1, in
return
}
func skipForRoaring(t *testing.T) {
src := pilosa.CurrentBackend()
if (storage.DefaultBackend == pilosa.RoaringTxn) || strings.Contains(src, "roaring") {
t.Skip("skip if roaring pseudo-txn involved -- won't show transactional rollback")
}
}
func TestAPI_ImportAtomicRecord(t *testing.T) {
skipForRoaring(t)
c := test.MustRunCluster(t, 1,
[]server.CommandOption{
server.OptCommandServerOptions(

27
txfactory.go
View file

@ -17,8 +17,7 @@ import (
// public strings that pilosa/server/config.go can reference
const (
RoaringTxn string = "roaring"
RBFTxn string = "rbf"
RBFTxn string = "rbf"
)
// DetectMemAccessPastTx true helps us catch places in api and executor
@ -377,9 +376,8 @@ type TxFactory struct {
type txtype int
const (
noneTxn txtype = 0
roaringTxn txtype = 1 // these don't really have any transactions
rbfTxn txtype = 2
noneTxn txtype = 0
rbfTxn txtype = 2
)
// DirectoryName just returns a string version of the transaction type. We
@ -388,8 +386,6 @@ const (
// replaced/removed) during that refactor.
func (ty txtype) DirectoryName() string {
switch ty {
case roaringTxn:
return "roaring"
case rbfTxn:
return "rbf"
}
@ -397,18 +393,12 @@ func (ty txtype) DirectoryName() string {
return ""
}
func (txf *TxFactory) NeedsSnapshot() (b bool) {
return txf.typ == roaringTxn
}
func MustBackendToTxtype(backend string) (typ txtype) {
if strings.Contains(backend, "_") {
panic("blue-green comparisons removed")
}
switch backend {
case RoaringTxn: // "roaring"
return roaringTxn
case RBFTxn: // "rbf"
return rbfTxn
}
@ -839,8 +829,6 @@ func (ty txtype) String() string {
switch ty {
case noneTxn:
return "noneTxn"
case roaringTxn:
return "roaring"
case rbfTxn:
return "rbf"
}
@ -937,25 +925,16 @@ func fileSize(name string) (int64, error) {
var _ = anyGlobalDBWrappersStillOpen // happy linter
func anyGlobalDBWrappersStillOpen() bool {
if globalRoaringReg.Size() != 0 {
return true
}
if globalRbfDBReg.Size() != 0 {
return true
}
return false
}
func (f *TxFactory) hasRoaring() bool {
return f.typ == roaringTxn
}
func (f *TxFactory) hasRBF() bool {
return f.typ == rbfTxn
}
var _ = (&TxFactory{}).hasRoaring // happy linter
func (f *TxFactory) GetDBShardPath(index string, shard uint64, idx *Index, ty txtype, write bool) (shardPath string, err error) {
dbs, err := f.dbPerShard.GetDBShard(index, shard, idx)
if err != nil {

4
txfactory_internal_test.go
View file

@ -8,8 +8,8 @@ import (
func Test_TxFactory_verifyStringConstantsMatch(t *testing.T) {
// txtype.String() method MUST return strings that match
// our const definitions at the top of txfactory.go.
check := []txtype{roaringTxn, rbfTxn}
expect := []string{RoaringTxn, RBFTxn}
check := []txtype{rbfTxn}
expect := []string{RBFTxn}
for i, chk := range check {
obs := chk.String()
if obs != expect[i] {