Merge branch 'fb-1188-ttl' of ssh://github.com/molecula/featurebase into fb-1188-ttl

This commit is contained in:
Hoang Pham 2022-03-03 16:20:11 -06:00
commit 31aa5ceaa4
65 changed files with 1916 additions and 1668 deletions

View file

@ -14,6 +14,7 @@ stages:
- gauntlet
- performance
- post build
- nonblocking
smoke build:
image: golang:$GOVERSION
@ -35,6 +36,15 @@ golangci-lint:
- echo "Checking for issues in new code"
- golangci-lint run
go mod tidy:
stage: lint
image: golang:$GOVERSION
rules:
- if: '$CI_COMMIT_TAG == null && ($CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web")'
script:
- go mod tidy
- git diff --exit-code -- go.mod go.sum
build lattice:
stage: test
image: node:14
@ -84,11 +94,12 @@ run go tests:
- aws
run go tests race:
stage: test
stage: nonblocking # don't let this job block any other jobs because it takes much longer than the other tests.
image: golang:$GOVERSION
rules:
- if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
retry: 1
needs: [] # don't wait to start running this.
script:
- echo "Running featurebase race tests..."
- go test -race -v -timeout=90m ./...
@ -101,7 +112,7 @@ run go tests shardwidth22:
rules:
- if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
script:
- echo "Running featurebase race tests..."
- echo "Running featurebase shardwidth22 tests..."
- go test -timeout=30m -tags=shardwidth22 ./...
tags:
- aws
@ -483,7 +494,7 @@ s3 dump:
perf_able:
stage: performance
rules:
- if: '$CI_PIPELINE_SOURCE == "push"'
- if: '$CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH && $CI_PIPELINE_SOURCE == "push"'
trigger:
include: .gitlab/.perf-able-gitlab-ci.yml
variables:
@ -506,18 +517,21 @@ s3 dump tag:
- aws configure set aws_secret_access_key $AWS_FBCI_SECRET_ACCESS_KEY
- aws configure set region "us-east-2"
- aws configure set aws_profile $PROFILE
- aws s3 cp featurebase_linux_amd64 s3://${LOCATION}/${CI_COMMIT_TAG}/featurebase_linux_amd64
- aws s3 cp roaring-migrate_linux_amd64 s3://${LOCATION}/${CI_COMMIT_TAG}/roaring-migrate_linux_amd64
- aws s3 cp featurebase_linux_arm64 s3://${LOCATION}/${CI_COMMIT_TAG}/featurebase_linux_arm64
- aws s3 cp roaring-migrate_linux_arm64 s3://${LOCATION}/${CI_COMMIT_TAG}/roaring-migrate_linux_arm64
- aws s3 cp featurebase_darwin_amd64 s3://${LOCATION}/${CI_COMMIT_TAG}/featurebase_darwin_amd64
- aws s3 cp roaring-migrate_darwin_amd64 s3://${LOCATION}/${CI_COMMIT_TAG}/roaring-migrate_darwin_amd64
- aws s3 cp featurebase_darwin_arm64 s3://${LOCATION}/${CI_COMMIT_TAG}/featurebase_darwin_arm64
- aws s3 cp roaring-migrate_darwin_arm64 s3://${LOCATION}/${CI_COMMIT_TAG}/roaring-migrate_darwin_arm64
- aws s3 cp NOTICE s3://${LOCATION}/${CI_COMMIT_TAG}/NOTICE
- aws s3 cp install/featurebase.debian.service s3://${LOCATION}/${CI_COMMIT_TAG}/featurebase.debian.service
- aws s3 cp install/featurebase.redhat.service s3://${LOCATION}/${CI_COMMIT_TAG}/featurebase.redhat.service
- aws s3 cp install/featurebase.conf s3://${LOCATION}/${CI_COMMIT_TAG}/featurebase.conf
- |
for goos in "darwin" "linux"; do
for goarch in "amd64" "arm64"; do
dir=featurebase-${CI_COMMIT_TAG}-${goos}-${goarch}
echo "Directory ${dir}"
mkdir $dir
mv featurebase_${goos}_${goarch} ${dir}/featurebase
mv roaring-migrate_${goos}_${goarch} ${dir}/roaring-migrate
cp NOTICE install/featurebase.conf install/featurebase.*.service ${dir}/
tar cvzf ${dir}.tar.gz ${dir}
aws s3 cp ${dir} s3://${LOCATION}/${CI_COMMIT_TAG}/${dir}/ --recursive
aws s3 cp ${dir}.tar.gz s3://${LOCATION}/${CI_COMMIT_TAG}/
done
done
needs:
- job: build for darwin amd64
- job: build for darwin arm64

329
api.go
View file

@ -50,9 +50,6 @@ type API struct {
importWorkerPoolSize int
importWork chan importJob
usageCache *usageCache
schemaDetailsOn bool
Serializer Serializer
}
@ -73,14 +70,6 @@ func OptAPIServer(s *Server) apiOption {
}
}
// Used to configure API option: schemaDetailsOn
func OptAPISchemaDetailsOn(isOn bool) apiOption {
return func(a *API) error {
a.schemaDetailsOn = isOn
return nil
}
}
func OptAPIImportWorkerPoolSize(size int) apiOption {
return func(a *API) error {
a.importWorkerPoolSize = size
@ -938,256 +927,6 @@ func (api *API) PrimaryNode() *topology.Node {
return snap.PrimaryFieldTranslationNode()
}
// Cache of disk usage statistics
type usageCache struct {
data map[string]NodeUsage
refreshInterval time.Duration
lastUpdated time.Time
resetTrigger chan bool
lastCalcDuration time.Duration
waitMultiplier float64
disable bool
muCalculate sync.Mutex
muAssign sync.Mutex
}
var usageCacheMinDuration = 5 * time.Second // If usage takes less than this duration to calculate, don't use the cache.
var usageCacheMinInterval = time.Hour // Refresh interval is forced to be >= this duration.
var usageCacheInitialInterval = time.Hour // Refresh interval starts with this duration.
// NodeUsage represents all usage measurements for one node.
type NodeUsage struct {
Disk DiskUsage `json:"diskUsage"`
Memory MemoryUsage `json:"memoryUsage"`
LastUpdated time.Time `json:"lastUpdated"`
}
// DiskUsage represents the storage space used on disk by one node.
type DiskUsage struct {
Capacity uint64 `json:"capacity,omitempty"`
TotalUse uint64 `json:"totalInUse"`
IndexUsage map[string]IndexUsage `json:"indexes"`
}
// IndexUsage represents the storage space used on disk by one index, on one node.
type IndexUsage struct {
Total uint64 `json:"total"`
IndexKeys uint64 `json:"indexKeys"`
FieldKeysTotal uint64 `json:"fieldKeysTotal"`
Fragments uint64 `json:"fragments"`
Metadata uint64 `json:"metadata"`
Fields map[string]FieldUsage `json:"fields"`
}
// FieldUsage represents the storage space used on disk by one field, on one node
type FieldUsage struct {
Total uint64 `json:"total"`
Fragments uint64 `json:"fragments"`
Keys uint64 `json:"keys"`
Metadata uint64 `json:"metadata"`
}
// MemoryUsage represents the memory used by one node.
type MemoryUsage struct {
Capacity uint64 `json:"capacity"`
TotalUse uint64 `json:"totalInUse"`
}
// Returns disk usage from cache if cache is large. It will recalculate on the spot if the last cacluation was under 5 seconds.
func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.Usage")
defer span.Finish()
if api.usageCache.disable {
resp := make(map[string]NodeUsage)
return resp, nil
}
api.usageCache.muAssign.Lock()
lastCalc := api.usageCache.lastCalcDuration
api.usageCache.muAssign.Unlock()
if lastCalc < usageCacheMinDuration {
err := api.ResetUsageCache()
if err != nil {
api.server.logger.Infof("could not reset usageCache: %s", err)
}
}
api.usageCache.muAssign.Lock()
lastUpdated := api.usageCache.lastUpdated
api.usageCache.muAssign.Unlock()
if lastUpdated == (time.Time{}) {
api.calculateUsage()
}
if !remote {
api.requestUsageOfNodes()
}
return api.usageCache.data, nil
}
// Makes a ui/usage request for each node in cluster to calculates its usage and adds it to the cache
func (api *API) requestUsageOfNodes() {
nodes := api.cluster.Nodes()
for _, node := range nodes {
if node.ID == api.server.nodeID {
continue
}
nodeUsage, err := api.server.defaultClient.GetNodeUsage(context.Background(), &node.URI)
if err != nil {
api.server.logger.Infof("couldn't collect disk usage from %s: %s", node.URI, err)
}
api.usageCache.muAssign.Lock()
api.usageCache.data[node.ID] = nodeUsage[node.ID]
api.usageCache.muAssign.Unlock()
}
}
// Calculates disk usage from scratch if cache has expired for each index and stores the results in the usage cache
func (api *API) calculateUsage() {
api.usageCache.muCalculate.Lock()
defer api.usageCache.muCalculate.Unlock()
api.server.wg.Add(1)
defer api.server.wg.Done()
api.usageCache.muAssign.Lock()
lastUpdated := api.usageCache.lastUpdated
api.usageCache.muAssign.Unlock()
if time.Since(lastUpdated) <= api.usageCache.refreshInterval {
return
}
indexDetails, nodeMetadataBytes, err := api.holder.Txf().IndexUsageDetails(api.isClosing)
if err != nil {
api.server.logger.Infof("couldn't get index usage details: %s", err)
}
if api.isClosing() {
return
}
totalSize := nodeMetadataBytes
for _, s := range indexDetails {
totalSize += s.Total
}
// NOTE: these errors are ignored in api.Info(), but checked here
si := api.server.systemInfo
diskCapacity, err := si.DiskCapacity(api.holder.path)
if err != nil {
api.server.logger.Infof("couldn't read disk capacity: %s", err)
}
memoryCapacity, err := si.MemTotal()
if err != nil {
api.server.logger.Infof("couldn't read memory capacity: %s", err)
}
memoryUse, err := si.MemUsed()
if err != nil {
api.server.logger.Infof("couldn't read memory usage: %s", err)
}
lastUpdated = time.Now()
// Insert into result.
nodeUsage := NodeUsage{
Disk: DiskUsage{
Capacity: diskCapacity,
TotalUse: totalSize,
IndexUsage: indexDetails,
},
Memory: MemoryUsage{
Capacity: memoryCapacity,
TotalUse: memoryUse,
},
LastUpdated: lastUpdated,
}
api.usageCache.muAssign.Lock()
api.usageCache.data = make(map[string]NodeUsage)
api.usageCache.data[api.server.nodeID] = nodeUsage
api.usageCache.lastUpdated = lastUpdated
api.usageCache.muAssign.Unlock()
}
// Periodically calculates disk/memory usage in terms of the duty cycle. The duty cycle represents the percentage of
// time that is spent recalculating this cache. It is specified relatively, rather than by a set interval, because
// scans can take an unpredictably long time.
func (api *API) RefreshUsageCache(dutyCycle float64) {
if dutyCycle == 0 {
api.server.logger.Warnf("usage-duty-cycle set to 0, usage cache and /ui/usage endpoint are disabled")
api.usageCache = &usageCache{
disable: true,
}
return
}
trigger := make(chan bool)
defer close(trigger)
multiplier := 100/dutyCycle - 1
api.usageCache = &usageCache{
data: make(map[string]NodeUsage),
refreshInterval: usageCacheInitialInterval,
resetTrigger: trigger,
lastCalcDuration: 0,
waitMultiplier: multiplier,
}
api.server.logger.Infof("monitoring resource usage with duty cycle %v%%\n", dutyCycle)
for {
start := time.Now()
api.calculateUsage()
api.setRefreshInterval(time.Since(start))
api.server.logger.Infof("updated resource usage cache at %v, took %v, next update in %v\n", api.usageCache.lastUpdated.Format(time.RFC3339), api.usageCache.lastCalcDuration.Truncate(time.Millisecond), api.usageCache.refreshInterval.Truncate(100*time.Millisecond))
select {
case <-trigger:
continue
case <-api.server.closing:
return
case <-time.After(api.usageCache.refreshInterval):
continue
}
}
}
// Refresh interval set in relation to how long the last calculation took.
func (api *API) setRefreshInterval(dur time.Duration) {
refresh := time.Duration(float64(dur) * api.usageCache.waitMultiplier)
if refresh < usageCacheMinInterval {
refresh = usageCacheMinInterval
}
api.usageCache.muAssign.Lock()
api.usageCache.refreshInterval = refresh
api.usageCache.lastCalcDuration = dur
api.usageCache.muAssign.Unlock()
}
// Resets the lastUpdated time and awakens RefreshUsageCache()
func (api *API) ResetUsageCache() error {
if api.usageCache != nil {
api.usageCache.muAssign.Lock()
api.usageCache.lastUpdated = time.Time{}
api.usageCache.muAssign.Unlock()
} else {
return errors.New("invalidating cache: cache not initialized")
}
api.usageCache.resetTrigger <- true
return nil
}
// isClosing returns true if the server is shutting down.
func (api *API) isClosing() bool {
select {
case <-api.server.closing:
return true
default:
return false
}
}
// RecalculateCaches forces all TopN caches to be updated.
// This is done internally within a TopN query, but a user may want to do it ahead of time?
func (api *API) RecalculateCaches(ctx context.Context) error {
@ -1272,38 +1011,6 @@ func (api *API) Schema(ctx context.Context, withViews bool) ([]*IndexInfo, error
return api.holder.limitedSchema()
}
// SchemaDetails returns information about each index in Pilosa including which
// fields they contain. Additional field information such as cardinality unless
// turned off via the schemaDetailsOn cli option.
func (api *API) SchemaDetails(ctx context.Context) ([]*IndexInfo, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.Schema")
defer span.Finish()
schema, err := api.holder.Schema()
if err != nil {
return nil, errors.Wrap(err, "getting schema")
}
if !api.schemaDetailsOn {
return schema, nil
}
for _, index := range schema {
for _, field := range index.Fields {
q := fmt.Sprintf("Count(Distinct(field=%s))", field.Name)
req := QueryRequest{Index: index.Name, Query: q}
resp, err := api.query(ctx, &req)
if err != nil {
return schema, errors.Wrapf(err, "querying cardinality (%s/%s)", index.Name, field.Name)
}
if len(resp.Results) == 0 {
continue
}
if card, ok := resp.Results[0].(uint64); ok {
field.Cardinality = &card
}
}
}
return schema, nil
}
// ApplySchema takes the given schema and applies it across the
// cluster (if remote is false), or just to this node (if remote is
// true). This is designed for the use case of replicating a schema
@ -3252,24 +2959,24 @@ var methodsResizing = map[apiMethod]struct{}{
apiSchema: {},
}
var methodsDegraded = map[apiMethod]struct{}{
apiExportCSV: {},
apiFragmentBlockData: {},
apiFragmentBlocks: {},
apiField: {},
apiIndex: {},
apiQuery: {},
apiRecalculateCaches: {},
apiRemoveNode: {},
apiShardNodes: {},
apiSchema: {},
apiViews: {},
apiStartTransaction: {},
apiFinishTransaction: {},
apiTransactions: {},
apiGetTransaction: {},
apiActiveQueries: {},
}
// var methodsDegraded = map[apiMethod]struct{}{
// apiExportCSV: {},
// apiFragmentBlockData: {},
// apiFragmentBlocks: {},
// apiField: {},
// apiIndex: {},
// apiQuery: {},
// apiRecalculateCaches: {},
// apiRemoveNode: {},
// apiShardNodes: {},
// apiSchema: {},
// apiViews: {},
// apiStartTransaction: {},
// apiFinishTransaction: {},
// apiTransactions: {},
// apiGetTransaction: {},
// apiActiveQueries: {},
// }
var methodsNormal = map[apiMethod]struct{}{
apiCreateField: {},

View file

@ -956,29 +956,6 @@ func TestAPI_IDAlloc(t *testing.T) {
})
}
func TestAPI_SchemaDetailsOff(t *testing.T) {
cluster := test.MustRunCluster(t, 2)
defer cluster.Close()
cmd := cluster.GetNode(0)
err := cmd.API.SetAPIOptions(pilosa.OptAPISchemaDetailsOn(false))
if err != nil {
t.Fatalf("could not toggle schema details to off: %v", err)
}
schema, err := cmd.API.SchemaDetails(context.Background())
if err != nil {
t.Fatalf("getting schema: %v", err)
}
for _, i := range schema {
for _, f := range i.Fields {
if f.Cardinality != nil {
t.Fatalf("expected nil cardinality, got: %v", *f.Cardinality)
}
}
}
}
type mutexCheckIndex struct {
index *pilosa.Index
indexName string

View file

@ -12,7 +12,8 @@ import (
"sync"
"time"
"github.com/molecula/featurebase/v3"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/roaring"
"github.com/pkg/errors"
bolt "go.etcd.io/bbolt"
@ -32,6 +33,8 @@ var (
bucketKeys = []byte("keys")
bucketIDs = []byte("ids")
bucketFree = []byte("free")
freeKey = []byte("free")
)
const (
@ -119,6 +122,8 @@ func (s *TranslateStore) Open() (err error) {
return err
} else if _, err := tx.CreateBucketIfNotExists(bucketIDs); err != nil {
return err
} else if _, err := tx.CreateBucketIfNotExists(bucketFree); err != nil {
return err
}
return nil
}); err != nil {
@ -230,14 +235,26 @@ func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
if idBucket == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketIDs)
}
freeBucket := tx.Bucket(bucketFree)
if freeBucket == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketFree)
}
puts := 0
// we create a freeIDGetter to reduce marshalling
getter := newFreeIDGetter(freeBucket)
defer getter.Close()
for idx, key := range keys {
id, boltKey := findIDByKey(keyBucket, key)
if id != 0 {
result[key] = id
continue
}
id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
// see if we can re-use any IDs first
if id = getter.GetFreeID(); id == 0 {
id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
}
idBytes := idScratch[puts*8 : puts*8+8]
binary.BigEndian.PutUint64(idBytes, id)
puts++
@ -498,6 +515,88 @@ func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
}
}
type boltWrapper struct {
tx *bolt.Tx
db *bolt.DB
}
func (w *boltWrapper) Commit() error {
if w.tx != nil {
return w.tx.Commit()
}
return nil
}
func (w *boltWrapper) Rollback() {
if w.tx != nil {
w.tx.Rollback()
}
}
func (s *TranslateStore) FreeIDs() (*roaring.Bitmap, error) {
result := roaring.NewBitmap()
err := s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketFree)
if bkt == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
}
b := bkt.Get(freeKey)
err := result.UnmarshalBinary(b)
if err != nil {
return err
}
return nil
})
return result, err
}
func (s *TranslateStore) MergeFree(tx *bolt.Tx, newIDs *roaring.Bitmap) error {
bkt := tx.Bucket(bucketFree)
b := bkt.Get(freeKey)
buf := new(bytes.Buffer)
if b != nil { //if existing combine with newIDs
before := roaring.NewBitmap()
err := before.UnmarshalBinary(b)
if err != nil {
return err
}
final := newIDs.Union(before)
_, err = final.WriteTo(buf)
if err != nil {
return err
}
} else {
newIDs.WriteTo(buf)
}
return bkt.Put(freeKey, buf.Bytes())
}
// Delete removes the lookeup pairs in order to make avialble for reuse but doesn't commit the
// transaction for that is tied to the associated rbf transaction being successful
func (s *TranslateStore) Delete(records *roaring.Bitmap) (pilosa.Commitor, error) {
tx, err := s.db.Begin(true)
if err != nil {
return nil, err
}
keyBucket := tx.Bucket(bucketKeys)
idBucket := tx.Bucket(bucketIDs)
ids := records.Slice()
for i := range ids {
id := u64tob(ids[i])
boltKey := idBucket.Get(id)
err = keyBucket.Delete(boltKey)
if err != nil {
tx.Rollback()
return &boltWrapper{}, err
}
err = idBucket.Delete(id)
if err != nil {
tx.Rollback()
return &boltWrapper{}, err
}
}
return &boltWrapper{tx: tx}, s.MergeFree(tx, records)
}
// emptyKey is a sentinel byte slice which stands for "" as a key.
var emptyKey = []byte{
0x00, 0x00, 0x00,
@ -521,6 +620,84 @@ func findIDByKey(bkt *bolt.Bucket, key string) (uint64, []byte) {
return 0, boltKey
}
// freeIDGetter reduces the amount of marshaling required to get multiple ids
type freeIDGetter struct {
freeBucket *bolt.Bucket
b *roaring.Bitmap
changed bool
}
// newFreeIDGetter initializes a new freeIDGetter. If at any point there is a
// failure, it returns an error.
//
// NOTE: For changes to be persisted to the bucket, you must call
// (*freeIDGetter).Close()
func newFreeIDGetter(freeBucket *bolt.Bucket) *freeIDGetter {
g := &freeIDGetter{
freeBucket: freeBucket,
}
// we ignore this value because it's okay if we dont have a bitmap just yet
_ = g.getBitmap()
return g
}
func (g *freeIDGetter) getBitmap() bool {
if g.b == nil {
// get the bitmap from freeBucket
value := g.freeBucket.Get(freeKey)
if value == nil {
return false
}
// turn the value into a bitmap
b := roaring.NewBitmap()
if err := b.UnmarshalBinary(value); err != nil {
return false
}
g.b = b
}
return true
}
// GetFreeID tries to get a free ID from the free id bucket. If at any point it
// fails to do so, it returns a 0. Otherwise, it returns the first free ID in the
// bucket
func (g *freeIDGetter) GetFreeID() (id uint64) {
if !g.getBitmap() {
return 0
}
// get the first free id
id, ok := g.b.Min()
if !ok {
return 0
}
// remove that id from the free id bitmap
if changed, err := g.b.RemoveN(id); changed == 0 || err != nil {
return 0
} else {
g.changed = true
}
return id
}
// Close persists any changes to the bitmap back to the bucket and then nils the
// references for safety.
func (g *freeIDGetter) Close() error {
if g.changed {
// convert bitmap to binary
buf, err := g.b.MarshalBinary()
if err != nil {
return errors.Wrap(err, "closing free ID Getter")
}
// put updated bitmap back into the freeBucket
if err := g.freeBucket.Put(freeKey, buf); err != nil {
return errors.Wrap(err, "closing free ID Getter")
}
}
g.b = nil
g.freeBucket = nil
return nil
}
func findKeyByID(bkt *bolt.Bucket, id uint64) string {
boltKey := bkt.Get(u64tob(id))
if bytes.Equal(boltKey, emptyKey) {

View file

@ -0,0 +1,107 @@
package boltdb
import (
"path/filepath"
"testing"
"github.com/molecula/featurebase/v3/roaring"
bolt "go.etcd.io/bbolt"
)
func TestGetFreeID(t *testing.T) {
boltDir := t.TempDir()
db, err := bolt.Open(filepath.Join(boltDir, "testDB"), 0600, nil)
if err != nil {
t.Fatalf("unexpected error opening test boltdb: %v", err)
}
defer db.Close()
makeTestBucket := func(tx *bolt.Tx, b *roaring.Bitmap) *bolt.Bucket {
if b == nil {
t.Fatalf("unexpected nil bitmap")
}
free, err := tx.CreateBucketIfNotExists(bucketFree)
if err != nil {
t.Fatalf("unexpected error making freeBucket: %v", err)
}
buf, err := b.MarshalBinary()
if err != nil {
t.Fatalf("unexpected error marshaling bitmap (%v) to binary: %v", b, err)
}
if err := free.Put(freeKey, buf); err != nil {
t.Fatalf("unexpected error adding data (%v) to freeBucket: %v", b, err)
}
return free
}
for name, test := range map[string]struct {
bits *roaring.Bitmap
want uint64
}{
"bucket is there, but nobody's home": {
bits: roaring.NewBitmap(),
want: 0,
},
"good bucket": {
bits: roaring.NewBitmap(1, 2, 34, 55, 9000),
want: 1,
},
} {
t.Run(name, func(t *testing.T) {
tx, err := db.Begin(true)
if err != nil {
t.Fatalf("unexpected error starting bolt transaction: %v", err)
}
defer tx.Rollback()
freeBucket := makeTestBucket(tx, test.bits)
getter := newFreeIDGetter(freeBucket)
defer getter.Close()
if got := getter.GetFreeID(); got != test.want {
t.Fatalf("expected %v got %v", test.want, got)
}
})
}
t.Run("CorrectOrdering", func(t *testing.T) {
tx, err := db.Begin(true)
if err != nil {
t.Fatalf("unexpected error starting bolt transaction: %v", err)
}
defer tx.Rollback()
bucket := makeTestBucket(tx, roaring.NewBitmap(1, 34, 2, 55, 9000))
getter := newFreeIDGetter(bucket)
defer getter.Close()
for _, want := range []uint64{1, 2, 34, 55, 9000} {
if got := getter.GetFreeID(); got != want {
t.Fatalf("expected %v got %v", want, got)
}
}
if got := getter.GetFreeID(); got != 0 {
t.Fatalf("expected 0 got %v", got)
}
})
t.Run("NotABitmap", func(t *testing.T) {
tx, err := db.Begin(true)
if err != nil {
t.Fatalf("unexpected error starting bolt transaction: %v", err)
}
defer tx.Rollback()
free, err := tx.CreateBucketIfNotExists(bucketFree)
if err != nil {
t.Fatalf("unexpected error making freeBucket: %v", err)
}
if err := free.Put(freeKey, []byte("this isn't right!")); err != nil {
t.Fatalf("unexpected error adding data to freeBucket: %v", err)
}
getter := newFreeIDGetter(free)
defer getter.Close()
if got := getter.GetFreeID(); got != 0 {
t.Fatalf("expected 0 got %v", got)
}
})
}

View file

@ -10,8 +10,9 @@ import (
"testing"
"time"
"github.com/molecula/featurebase/v3"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/testhook"
"github.com/molecula/featurebase/v3/topology"
)
@ -385,7 +386,53 @@ func MustNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
s.Path = f.Name()
return s
}
func TestTranslateStore_Delete(t *testing.T) {
s := MustOpenNewTranslateStore(t)
defer MustCloseTranslateStore(s)
// Setup initial keys.
ids, err := s.CreateKeys("foo", "bar", "deleteme")
if err != nil {
t.Fatal(err)
}
records := roaring.NewBitmap(ids["deleteme"])
c, err := s.Delete(records)
if err != nil {
t.Fatal(err)
}
if err = c.Commit(); err != nil {
t.Fatal(err)
}
r, e := s.FreeIDs()
if e != nil {
t.Fatal(err)
}
freeids := r.Slice()
if len(freeids) == 0 {
t.Fatalf("expected to have free id")
}
if freeids[0] != ids["deleteme"] {
t.Fatalf("expected [%v] and got %v", ids["deleteme"], freeids[0])
}
records2 := roaring.NewBitmap(ids["foo"])
c, err = s.Delete(records2)
if err != nil {
t.Fatal(err)
}
if err = c.Commit(); err != nil {
t.Fatal(err)
}
r, e = s.FreeIDs()
if e != nil {
t.Fatal(err)
}
freeids = r.Slice()
if len(freeids) != 2 {
t.Fatalf("expected to have 2 free ids")
}
}
func TestTranslateStore_ReadWrite(t *testing.T) {
t.Run("WriteTo_ReadFrom", func(t *testing.T) {
s := MustOpenNewTranslateStore(t)
@ -408,7 +455,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
// Put the contents of the store into a buffer.
buf := bytes.NewBuffer(nil)
expN := int64(32768)
expN := s.Size()
// After this, the buffer should contain batch0.
if n, err := s.WriteTo(buf); err != nil {
@ -458,7 +505,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
func MustOpenNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
s := MustNewTranslateStore(tb)
if err := s.Open(); err != nil {
panic(err)
tb.Fatalf("opening s: %v", err)
}
return s
}

View file

@ -26,6 +26,11 @@ func init() {
var _ Tx = (*catcherTx)(nil)
func (c *catcherTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) {
c.b.RemoveChannel(index, field, view, shard, a, resChan)
return
}
func (c *catcherTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator {
return c.b.NewTxIterator(index, field, view, shard)
}

View file

@ -90,15 +90,9 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.Uint16Var(&srv.Config.Postgres.ConnectionLimit, "postgres.connection-limit", srv.Config.Postgres.ConnectionLimit, "Maximum number of simultaneous postgres connections to allow. (set 0 to disable)")
flags.Uint16Var(&srv.Config.Postgres.SqlVersion, "postgres.sql-version", srv.Config.Postgres.SqlVersion, "Molecula Sql Handling Version (default 1)")
// Disk and Memory usage cache for ui/usage endpoint
flags.Float64Var(&srv.Config.UsageDutyCycle, "usage-duty-cycle", srv.Config.UsageDutyCycle, "Sets the percentage of time that is spent recalculating the disk and memory usage cache. 100.0 for always-running, 0 disables the cache and the /ui/usage endpoint.")
// Future flags.
flags.BoolVar(&srv.Config.Future.Rename, "future.rename", false, "Present application name as FeatureBase. Defaults to false, will default to true in an upcoming release.")
// Toggle /schema/details endpoint.
flags.BoolVar(&srv.Config.SchemaDetailsOn, "schema-details-on", true, "Disable /schema/details endpoint")
// OAuth2.0 identity provider configuration
flags.BoolVar(&srv.Config.Auth.Enable, "auth.enable", false, "Enable AuthN/AuthZ of featurebase, disabled by default.")
flags.StringVar(&srv.Config.Auth.ClientId, "auth.client-id", srv.Config.Auth.ClientId, "Identity Provider's Application/Client ID.")

View file

@ -49,6 +49,21 @@ func TestExecutor_DeleteRecords(t *testing.T) {
})
}
setupBig := func(t *testing.T, r *require.Assertions, c *test.Cluster, Rows uint64) {
t.Helper()
fieldName := "setfield"
c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, fieldName)
rows := make([][2]uint64, ShardWidth*Rows)
for columnID := uint64(0); columnID < ShardWidth; columnID++ {
for rowID := uint64(0); rowID < Rows; rowID++ {
if rowID == 0 || (columnID%rowID+1) != 0 {
rows[rowID] = [2]uint64{rowID, columnID}
}
}
}
c.ImportBits(t, indexName, "setfield", rows)
}
setupKeys := func(t *testing.T, r *require.Assertions, c *test.Cluster) {
t.Helper()
c.CreateField(t, indexName, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "timefield", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"), "0"))
@ -131,6 +146,12 @@ func TestExecutor_DeleteRecords(t *testing.T) {
m = resp.Results[0].(pilosa.ExtractedTable)
after := convertKey(m.Columns)
require.Equal([]string{"B", "C", "D", "two"}, after, "these keyed records after delete")
//validate that column keys got deleted
node := c.GetNode(0)
keys := []string{"A", "one"}
res, err := node.API.FindIndexKeys(context.Background(), indexName, keys...)
require.Nil(err)
require.Empty(res)
})
t.Run("Delete Row", func(t *testing.T) {
setup(t, require, c)
@ -200,8 +221,33 @@ func TestExecutor_DeleteRecords(t *testing.T) {
require.Equal([]uint64{0, 1}, after, "these records should be remaining")
})
})
t.Run("DeleteRecordsBigWithRestart", func(t *testing.T) {
c := test.MustNewCluster(t, 1)
for _, n := range c.Nodes {
n.Config.Cluster.ReplicaN = 1
}
err := c.Start()
defer c.Close()
require.NoError(err, "Start cluster DeleteRecordsBig")
setupBig(t, require, c, 16)
defer tearDown(t, require, c)
node := c.GetNode(0)
resp := c.Query(t, indexName, `Delete(Row(setfield=12))`)
require.NotNil(resp, "Response should not be nil")
require.NotEmpty(resp.Results)
require.Equal(true, resp.Results[0], "Change should have happened")
resp = c.Query(t, indexName, `Count(Row(setfield=12))`)
require.NotNil(resp, "Response should not be nil")
require.NotEmpty(resp.Results)
require.Equal(uint64(0), resp.Results[0], "Should have removed")
err = node.Reopen()
require.NoError(err, "restart cluster DeleteRecordsBig")
err = c.AwaitState(disco.ClusterStateNormal, 10*time.Second)
require.NoError(err, "backToNormal")
})
}
func convert(before []pilosa.ExtractedTableColumn) []uint64 {
result := make([]uint64, 0)
for _, i := range before {

View file

@ -6731,6 +6731,17 @@ func (e *executor) translateCall(c *pql.Call, index string, columnKeys map[strin
}
}
}
// Check if "like" argument is applied to keyed fields.
if _, found := c.Args["like"].(string); found {
fieldName, err := c.FirstStringArg("_field", "field")
if err != nil || fieldName == "" {
return nil, fmt.Errorf("cannot read field name for Rows call")
}
if !idx.Field(fieldName).options.Keys {
return nil, fmt.Errorf("'%s' is not a set/mutex/time field with a string key", fieldName)
}
}
}
// Translate child calls.
@ -8252,70 +8263,128 @@ func (e *executor) executeDeleteRecords(ctx context.Context, qcx *Qcx, index str
return n, nil
}
func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (bool, error) {
func transactExistRow(ctx context.Context, idx *Index, shard uint64, frag *fragment, src *Row) (uint64, error) {
tx := idx.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
rows, err := frag.rows(ctx, tx, 1)
if err != nil {
tx.Rollback()
return 0, err
}
rowID := uint64(len(rows) + 1)
_, err = frag.setRow(tx, src, rowID)
if err != nil {
tx.Rollback()
return 0, err
}
return rowID, tx.Commit()
}
func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (changed bool, err error) {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeDeleteRecordFromShard")
defer span.Finish()
//need to build the bitmap in the call
child := c.Children[0]
row, err := e.executeBitmapCallShard(ctx, qcx, index, child, shard)
if err != nil {
return false, err
src, er := e.executeBitmapCallShard(ctx, qcx, index, child, shard)
if er != nil {
err = er
return
}
if len(row.segments) == 0 { //nothing to remove
if len(src.segments) == 0 { //nothing to remove
return
}
columns := src.segments[0].data //should only be one segment
if columns.Count() == 0 {
return
}
// Fetch index.
idx := e.Holder.Index(index)
if idx == nil {
err = newNotFoundError(ErrIndexNotFound, index)
return
}
return DeleteRowsWithFlow(ctx, src, idx, shard, true)
}
func DeleteRows(ctx context.Context, src *Row, idx *Index, shard uint64) (bool, error) {
return DeleteRowsWithFlow(ctx, src, idx, shard, false)
}
func DeleteRowsWithFlow(ctx context.Context, src *Row, idx *Index, shard uint64, normalFlow bool) (bool, error) {
var existenceFragment *fragment
var deletedRowID uint64
var commitor Commitor = &NopCommitor{}
var err error
if len(src.segments) == 0 { //nothing to remove
return false, nil
}
columns := row.segments[0].data //should only be one segment
columns := src.segments[0].data //should only be one segment
if columns.Count() == 0 {
return false, nil
}
// Fetch index.
idx := e.Holder.Index(index)
if idx == nil {
return false, newNotFoundError(ErrIndexNotFound, index)
}
columnIDs := make([]uint64, 0)
none := make([]uint64, 0) // no bits will be set
tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: idx, Shard: shard})
if err != nil {
return false, err
}
defer finisher(&err)
changed := false
colCounts := make([]int, 0)
toClear := columnIDs[:0]
rowSet := make(map[uint64]struct{})
callback := func(pos uint64) error {
toClear = append(toClear, pos)
rowID := pos / ShardWidth
rowSet[rowID] = struct{}{}
return nil
}
findExisting := roaring.NewBitmapBitmapFilter(columns, callback)
clearFragment := func(frag *fragment) (bool, error) {
// re-zero these
toClear = columnIDs[:0]
rowSet = make(map[uint64]struct{})
err = tx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting)
if idx.Keys() {
//store columns in exits field ToBeDelete row commited
if normalFlow { // normalFlow is the standard path, "not normal" is recoverory
existenceFragment = idx.Holder().fragment(idx.Name(), existenceFieldName, viewStandard, shard)
if existenceFragment == nil {
//no exists field
return false, errors.New("can't bulk delete without existence field")
}
deletedRowID, err = transactExistRow(ctx, idx, shard, existenceFragment, src)
}
commitor, err = deleteKeyTranslation(ctx, idx, shard, columns)
if err != nil {
return false, err
}
colCounts = append(colCounts, len(toClear))
// this will be the remove part
if len(toClear) > 0 {
err = frag.importPositions(tx, none, toClear, rowSet)
if err != nil {
return false, err
}
return true, nil
}
return false, nil
}
writeTx := idx.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
if err != nil {
return false, err
}
defer writeTx.Rollback()
changed := false
defer func() {
//if there is an error on the bit clearing rollback the keys
if err != nil {
changed = false
commitor.Rollback()
return
}
// if there is an error in the key commit, then rollback the delete
// write records before keys to remove possiblity of unmatch keys=records
err = writeTx.Commit()
if err != nil {
changed = false
commitor.Rollback()
return
}
if er := commitor.Commit(); er != nil {
err = er
}
if err != nil {
idx.Holder().Logger.Errorf("problems committing delete in rbf %v shard %v", err, shard)
}
}()
findExisting := roaring.NewBitmapBitmapFilter(columns, func(p uint64) error { return nil })
resChan := make(chan countResults)
clearFragment := func(frag *fragment) (bool, error) {
posChan := make(chan uint64, 8192)
findExisting.SetCallback(func(pos uint64) error {
posChan <- pos
return nil
})
go writeTx.RemoveChannel(frag.index(), frag.field(), frag.view(), frag.shard, posChan, resChan)
err = writeTx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting)
close(posChan)
if err != nil {
return false, err
}
r := <-resChan
return r.changeCount > 0, r.err
}
for _, field := range idx.Fields() {
for _, view := range field.views() {
@ -8330,7 +8399,44 @@ func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, i
if c {
changed = true
}
}
}
close(resChan)
if existenceFragment != nil { //a string keys have been deleted and the deleteRow was created
if normalFlow {
existenceFragment.clearRow(writeTx, deletedRowID)
} else {
// this is if we are recovering from failure and cleaning up
rows, err := existenceFragment.rows(ctx, writeTx, 1)
if err != nil {
return false, err
}
for _, rowId := range rows {
existenceFragment.clearRow(writeTx, rowId)
}
}
}
return changed, nil
}
type Commitor interface {
Rollback()
Commit() error
}
type NopCommitor struct {
}
func (c *NopCommitor) Rollback() {
}
func (c *NopCommitor) Commit() error {
return nil
}
func deleteKeyTranslation(ctx context.Context, idx *Index, shard uint64, records *roaring.Bitmap) (Commitor, error) {
// ShardToShardParition ...
paritionID := topology.ShardToShardPartition(idx.name, shard, idx.holder.partitionN)
return idx.TranslateStore(paritionID).Delete(records)
}

View file

@ -545,3 +545,54 @@ func TestDistinctTimestampUnion(t *testing.T) {
})
}
}
func TestExecutor_DeleteRows(t *testing.T) {
path, _ := testhook.TempDir(t, "pilosa-executor-")
holder := NewHolder(path, mustHolderConfig())
defer holder.Close()
if err := holder.Open(); err != nil {
t.Fatalf("opening holder: %v", err)
}
idx, err := holder.CreateIndex("i", IndexOptions{TrackExistence: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
f, err := idx.CreateField("f", OptFieldTypeDefault())
if err != nil {
t.Fatalf("creating field: %v", err)
}
shard := uint64(0)
tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
defer tx.Rollback()
if _, err = f.SetBit(tx, 1, 1, nil); err != nil {
t.Fatalf("setting bit: %v", err)
}
if err := tx.Commit(); err != nil {
t.Fatalf("failed to commit transaction: %v", err)
}
tx = idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: shard})
defer tx.Rollback()
row, err := f.Row(tx, 1)
if err != nil {
t.Fatalf("failed to read row: %v", err)
}
ctx := context.Background()
changed, err := DeleteRows(ctx, row, idx, shard)
if !changed || err != nil {
t.Fatalf("failed to delete row: %v", err)
}
changed, err = DeleteRows(ctx, row, idx, shard)
if changed {
t.Fatalf("expected delete to not clear bit but it did")
}
}

View file

@ -5452,6 +5452,11 @@ func TestExecutor_Execute_Rows_Keys(t *testing.T) {
t.Fatalf("creating field: %v", err)
}
_, err = c.GetNode(0).API.CreateField(context.Background(), "i", "f_id")
if err != nil {
t.Fatalf("creating field: %v", err)
}
// setup some data. 10 bits in each of shards 0 through 9. starting at
// row/col shardNum and progressing to row/col shardNum+10. Also set the
// previous 2 for each bit if row >0.
@ -5474,8 +5479,9 @@ func TestExecutor_Execute_Rows_Keys(t *testing.T) {
}
tests := []struct {
q string
exp []string
q string
exp []string
expErr string
}{
{
q: `Rows(f)`,
@ -5557,13 +5563,26 @@ func TestExecutor_Execute_Rows_Keys(t *testing.T) {
q: `Rows(f, like="__")`,
exp: []string{"10", "11", "12", "13", "14", "15", "16", "17", "18"},
},
{
q: `Rows(f_id, like=7)`,
expErr: "parsing:",
},
{
q: `Rows(f_id, like="__")`,
expErr: "executing: translating call:",
},
}
for i, test := range tests {
t.Run(fmt.Sprintf("#%d_%s", i, test.q), func(t *testing.T) {
if res, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: test.q}); err != nil {
t.Fatal(err)
if !strings.HasPrefix(err.Error(), test.expErr) {
t.Fatal(err)
}
} else {
if test.expErr != "" {
t.Fatalf("got success, expected error similar to: %+v", test.expErr)
}
rows := res.Results[0].(pilosa.RowIdentifiers)
if !reflect.DeepEqual(rows.Keys, test.exp) {
t.Fatalf("\ngot: %+v\nexp: %+v", rows.Keys, test.exp)

View file

@ -770,50 +770,31 @@ func (f *fragment) setValueBase(txOrig Tx, columnID uint64, bitDepth uint64, val
// sum returns the sum of a given bsiGroup as well as the number of columns involved.
// A bitmap can be passed in to optionally filter the computed columns.
func (f *fragment) sum(tx Tx, filter *Row, bitDepth uint64) (sum int64, count uint64, err error) {
// Compute count based on the existence row.
consider, err := f.row(tx, bsiExistsBit)
if err != nil {
return sum, count, err
} else if filter != nil {
consider = consider.Intersect(filter)
}
count = consider.Count()
// Get negative set
nrow, err := f.row(tx, bsiSignBit)
if err != nil {
return sum, count, err
}
// Filter negative set
nrow = consider.Intersect(nrow)
// Get postive set
prow := consider.Difference(nrow)
// Compute the sum based on the bit count of each row multiplied by the
// place value of each row. For example, 10 bits in the 1's place plus
// 4 bits in the 2's place plus 3 bits in the 4's place equals a total
// sum of 30:
//
// 10*(2^0) + 4*(2^1) + 3*(2^2) = 30
//
// Execute once for positive numbers and once for negative. Subtract the
// negative sum from the positive sum.
for i := uint64(0); i < bitDepth; i++ {
row, err := f.row(tx, uint64(bsiOffsetBit+i))
if err != nil {
return sum, count, err
// If there's a provided filter, but it has no contents for this particular
// shard, we're done and can return early. If there's no provided filter,
// though, we want to run with no-filter, as opposed to an empty filter.
var filterData *roaring.Bitmap
if filter != nil {
for _, seg := range filter.segments {
if seg.shard == f.shard {
filterData = seg.data
break
}
}
psum := int64((1 << i) * row.intersectionCount(prow))
nsum := int64((1 << i) * row.intersectionCount(nrow))
// Squash to reduce the possibility of overflow.
sum += psum - nsum
// if filter is empty, we're done
if filterData == nil {
return 0, 0, nil
}
}
bsiFilt := roaring.NewBitmapBSICountFilter(filterData)
err = tx.ApplyFilter(f.index(), f.field(), f.view(), f.shard, 0, bsiFilt)
if err != nil && err != io.EOF {
return sum, count, errors.Wrap(err, "finding existing positions")
}
return sum, count, nil
c32, sum := bsiFilt.Total()
return sum, uint64(c32), nil
}
// min returns the min of a given bsiGroup as well as the number of columns involved.

View file

@ -1570,6 +1570,7 @@ func TestFragment_WriteTo_ReadFrom(t *testing.T) {
}
// make a read-only Tx after ReadFrom has committed.
tx = idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f1, Shard: f1.shard})
defer tx.Rollback()
// Verify cache is in other fragment.
if n := f1.cache.Len(); n != 1 {

View file

@ -287,6 +287,50 @@ func (h *Holder) IndexesPath() string {
return filepath.Join(h.path, IndexesDir)
}
// processDeleteInflight checks if deletion was in progress when server shutdown
// the _exists field is set to row+1 when delete is started. Upon completion, the row is deleted.
// if _exists>=1, we finish deleting the rows
func (h *Holder) processDeleteInflight() error {
for _, index := range h.Indexes() {
if index.trackExistence {
shards := index.AvailableShards(includeRemote).Slice()
for _, shard := range shards {
inprocessRowIDs := NewRow()
frag := h.fragment(index.name, existenceFieldName, viewStandard, shard)
if frag == nil {
continue
}
tx := index.Txf().NewTx(Txo{Write: !writable, Index: index, Shard: shard})
defer tx.Rollback()
// filter rows based on having _exists>=1, which is used to flag delete in-flight
rows, err := frag.rows(context.Background(), tx, 1)
if err != nil {
return err
}
// check if any rows are found
if len(rows) == 0 {
return nil
}
for _, rowID := range rows {
row, err2 := frag.row(tx, rowID)
if err2 != nil {
return err2
}
inprocessRowIDs = inprocessRowIDs.Union(row)
}
DeleteRows(context.Background(), inprocessRowIDs, index, shard)
}
}
}
return nil
}
// Open initializes the root data directory for the holder.
func (h *Holder) Open() error {
h.opening = true
@ -380,6 +424,9 @@ func (h *Holder) Open() error {
return errors.Wrap(err, "processing foreign index fields")
}
// Check if deletion was in progress when server was shutdown
h.processDeleteInflight()
h.Stats.Open()
h.opened.Close()

View file

@ -2,7 +2,10 @@
package pilosa
import (
"testing"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/testhook"
)
// mustHolderConfig sets up a default holder config for tests.
@ -14,3 +17,74 @@ func mustHolderConfig() *HolderConfig {
cfg.Sharder = disco.InMemSharder
return cfg
}
func TestHolder_ProcessDeleteInflight(t *testing.T) {
path, _ := testhook.TempDir(t, "delete-inflight")
h := NewHolder(path, mustHolderConfig())
defer h.Close()
err := h.Open()
if err != nil {
t.Fatalf("failed to open holder: %v", err)
}
idx, err := h.CreateIndexIfNotExists("i", IndexOptions{TrackExistence: true})
if err != nil {
t.Fatalf("failed to create index: %v", err)
}
f, err := idx.CreateFieldIfNotExists("f", OptFieldTypeDefault())
if err != nil {
t.Fatalf("failed to create field: %v", err)
}
existencefield := idx.existenceFld
shard := uint64(0)
tx := idx.Txf().NewTx(Txo{Write: true, Index: idx, Shard: shard})
defer tx.Rollback()
rowCol := []struct {
row uint64
col uint64
}{
{1, 1},
{1, 2},
{30, 33},
{22, 2},
}
for _, r := range rowCol {
_, err = f.SetBit(tx, r.row, r.col, nil)
if err != nil {
t.Fatalf("failed to set bit: %v", err)
}
_, err = existencefield.SetBit(tx, r.row, r.col, nil)
if err != nil {
t.Fatalf("failed to set bit: %v", err)
}
}
if err = tx.Commit(); err != nil {
t.Fatalf("failed to commit tx: %v", err)
}
err = h.processDeleteInflight()
if err != nil {
t.Fatalf("failed to delete: %v", err)
}
tx = idx.Txf().NewTx(Txo{Write: false, Index: idx, Shard: shard})
defer tx.Rollback()
for _, r := range rowCol {
row, err := f.Row(tx, r.row)
if err != nil {
t.Fatalf("failed to get row: %v", err)
}
existenceRow, err := existencefield.Row(tx, r.row)
if err != nil {
t.Fatalf("failed to get row: %v", err)
}
if len(row.Columns()) != 0 || len(existenceRow.Columns()) != 0 {
t.Fatalf("expected columns for fields to be empty after delete")
}
}
}

View file

@ -454,7 +454,6 @@ func newRouter(handler *Handler) http.Handler {
router.HandleFunc("/version", handler.handleGetVersion).Methods("GET").Name("GetVersion")
// /ui endpoints are for UI use; they may change at any time.
router.HandleFunc("/ui/usage", handler.chkAuthZ(handler.handleGetUsage, authz.Read)).Methods("GET").Name("GetUsage")
router.HandleFunc("/ui/transaction", handler.chkAuthZ(handler.handleGetTransactionList, authz.Read)).Methods("GET").Name("GetTransactionList")
router.HandleFunc("/ui/transaction/", handler.chkAuthZ(handler.handleGetTransactionList, authz.Read)).Methods("GET").Name("GetTransactionList")
router.HandleFunc("/ui/shard-distribution", handler.chkAuthZ(handler.handleGetShardDistribution, authz.Admin)).Methods("GET").Name("GetShardDistribution")
@ -468,6 +467,7 @@ func newRouter(handler *Handler) http.Handler {
router.HandleFunc("/internal/translate/data", handler.chkAuthZ(handler.handlePostTranslateData, authz.Write)).Methods("POST").Name("PostTranslateData")
// other ones
router.HandleFunc("/internal/mem-usage", handler.chkAuthZ(handler.handleGetMemUsage, authz.Read)).Methods("GET").Name("GetUsage")
router.HandleFunc("/internal/fragment/block/data", handler.chkAuthN(handler.handleGetFragmentBlockData)).Methods("GET").Name("GetFragmentBlockData")
router.HandleFunc("/internal/fragment/blocks", handler.chkAuthN(handler.handleGetFragmentBlocks)).Methods("GET").Name("GetFragmentBlocks")
router.HandleFunc("/internal/fragment/data", handler.chkAuthN(handler.handleGetFragmentData)).Methods("GET").Name("GetFragmentData")
@ -928,7 +928,12 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
}
}
// handleGetSchema handles GET /schema/details requests.
// handleGetSchema handles GET /schema/details requests. This is essentially the
// same thing as a GET /schema request, except WithViews is turned on by default.
// Previously, /schema/details returned the cardinality of each field, but this was
// removed for performance reasons. If, at some point in the future, there is a more
// performant way to get the cardinality of a field, that information would be
// included here.
func (h *Handler) handleGetSchemaDetails(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
@ -936,7 +941,7 @@ func (h *Handler) handleGetSchemaDetails(w http.ResponseWriter, r *http.Request)
}
w.Header().Set("Content-Type", "application/json")
schema, err := h.api.SchemaDetails(r.Context())
schema, err := h.api.Schema(r.Context(), true)
if err != nil {
h.logger.Printf("error getting detailed schema: %s", err)
return
@ -989,60 +994,22 @@ func (h *Handler) handlePostSchema(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusNoContent)
}
// handleGetUsage handles GET /ui/usage requests.
func (h *Handler) handleGetUsage(w http.ResponseWriter, r *http.Request) {
// handleGetMemUsage handles GET /internal/mem-usage requests.
func (h *Handler) handleGetMemUsage(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
q := r.URL.Query()
remoteStr := q.Get("remote")
var remote bool
if remoteStr == "true" {
remote = true
}
nodeUsages, err := h.api.Usage(r.Context(), remote)
use, err := GetMemoryUsage()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
}
// if auth is turned on, filter results
if h.auth != nil {
g := r.Context().Value(contextKeyGroupMembership)
if g == nil {
http.Error(w, "Forbidden", http.StatusForbidden)
return
}
if !h.permissions.IsAdmin(g.([]authn.Group)) {
allowed := h.permissions.GetAuthorizedIndexList(g.([]authn.Group), authz.Read)
filteredNodeUsages := map[string]NodeUsage{}
for nodeId, nodeUsage := range nodeUsages {
filteredIndexUsage := NodeUsage{
Disk: DiskUsage{
IndexUsage: map[string]IndexUsage{},
},
}
for index, idxUsage := range nodeUsage.Disk.IndexUsage {
// is it in auth list
for _, authd := range allowed {
if index == authd {
filteredIndexUsage.Disk.IndexUsage[index] = idxUsage
break
}
}
}
filteredNodeUsages[nodeId] = filteredIndexUsage
}
nodeUsages = filteredNodeUsages
}
return
}
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(nodeUsages); err != nil {
h.logger.Errorf("write status response error: %s", err)
if err := json.NewEncoder(w).Encode(use); err != nil {
h.logger.Errorf("write mem usage response error: %s", err)
}
}

View file

@ -773,3 +773,19 @@ func NewTestAuth(t *testing.T) *authn.Auth {
}
return a
}
func TestHandleGetMemUsage(t *testing.T) {
h := Handler{
logger: logger.NewStandardLogger(os.Stdout),
queryLogger: logger.NewStandardLogger(os.Stdout),
}
w := httptest.NewRecorder()
r := httptest.NewRequest("GET", "/whatever", nil)
h.handleGetMemUsage(w, r)
resp := w.Result()
if resp.StatusCode != http.StatusOK {
t.Fatalf("expected %v, got %v", http.StatusOK, resp.StatusCode)
}
}

View file

@ -244,28 +244,6 @@ log-path = "/var/log/molecula/featurebase.log"
# enable-client-verification = true
# ==============================================================================
# Usage Duty Cycle - Featurebase maintains a disk/memory usage cache that is
# calculated periodically in the background and accessed by the UI/usage
# endpoint. Since this disk scan can take a long and unpredictable amount of
# time, its timing behavior is specified in a relative, rather than absolute
# sense. That is, the duty cycle sets the percentage of time that is spent
# recalculating this cache. This setting affects the results received from
# the "/ui/usage" http endpoint, as well as all data file and memory usage
# values and graphs on the webui "tables" page
# Special considerations:
# * If disk usage can be calculated quickly (less than 5 seconds), fresh
# results will be calculated when accessed
# * When disk usage takes longer to calculate, there is a minimum of one
# hour wait between cache recalculations
# Setting this value to 0 will completely disable the calculation of disk usage
#
# usage-duty-cycle = 20
# ==============================================================================
# Use [metric] stanza to define attributes for monitoring.
# [metric]

View file

@ -1,4 +1,4 @@
FROM golang:latest
FROM golang:1.16
WORKDIR /
COPY fakeidp ./

View file

@ -244,28 +244,6 @@
# enable-client-verification = true
# ==============================================================================
# Usage Duty Cycle - Featurebase maintains a disk/memory usage cache that is
# calculated periodically in the background and accessed by the UI/usage
# endpoint. Since this disk scan can take a long and unpredictable amount of
# time, its timing behavior is specified in a relative, rather than absolute
# sense. That is, the duty cycle sets the percentage of time that is spent
# recalculating this cache. This setting affects the results received from
# the "/ui/usage" http endpoint, as well as all data file and memory usage
# values and graphs on the webui "tables" page
# Special considerations:
# * If disk usage can be calculated quickly (less than 5 seconds), fresh
# results will be calculated when accessed
# * When disk usage takes longer to calculate, there is a minimum of one
# hour wait between cache recalculations
# Setting this value to 0 will completely disable the calculation of disk usage
#
# usage-duty-cycle = 20
# ==============================================================================
# Use [metric] stanza to define attributes for monitoring.
# [metric]

View file

@ -1385,38 +1385,6 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pnet.URI, in
return tkresp.Keys, nil
}
// GetNodeUsage retrieves the size-on-disk information for the specified node.
func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]NodeUsage, error) {
u := uri.Path("/ui/usage?remote=true")
req, err := http.NewRequest("GET", u, nil)
if err != nil {
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+Version)
req = AddAuthToken(ctx, req)
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
return nil, err
}
defer resp.Body.Close()
// Read body and unmarshal response.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.Wrap(err, "reading")
}
nodeUsages := make(map[string]NodeUsage) // map of size 1
if err := json.Unmarshal(body, &nodeUsages); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
}
return nodeUsages, nil
}
// GetPastQueries retrieves the query history log for the specified node.
func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error) {
u := uri.Path("/query-history?remote=true")

View file

@ -19,14 +19,12 @@ export const ClusterHealth: FC = () => {
const [cluster, setCluster] = useState<any>();
const [metrics, setMetrics] = useState<any>();
const [info, setInfo] = useState<any>();
const [clusterData, setClusterData] = useState<any>();
const [expanded, setExpanded] = useState<string[]>([]);
const [showMetrics, setShowMetrics] = useState<any>();
const allExpanded = cluster && expanded.length === cluster.nodes.length;
useEffectOnce(() => {
getClusterHealth();
getClusterData();
});
const refreshMetrics = useCallback(() => {
@ -38,15 +36,11 @@ export const ClusterHealth: FC = () => {
useEffect(() => {
const interval = setInterval(() => {
if (!clusterData) {
getClusterData();
}
getClusterHealth();
refreshMetrics();
}, 15000);
return () => clearInterval(interval);
}, [refreshMetrics, cluster, clusterData]);
}, [refreshMetrics, cluster]);
const getClusterHealth = () => {
pilosa.get
@ -76,13 +70,6 @@ export const ClusterHealth: FC = () => {
.catch(() => setMetrics(undefined));
};
const getClusterData = () => {
pilosa.get
.usage()
.then((res) => setClusterData(res.data))
.catch(() => setClusterData(undefined));
};
const toggleAccordion = (nodeId: string) => {
const isExpanded = expanded.includes(nodeId);
if (isExpanded) {
@ -140,7 +127,6 @@ export const ClusterHealth: FC = () => {
key={node.id}
node={node}
info={info}
usage={clusterData ? clusterData[node.id] : undefined}
expanded={expanded.includes(node.id)}
onToggle={() => toggleAccordion(node.id)}
onMetricClick={() => setShowMetrics(node)}

View file

@ -1,4 +1,4 @@
import React, { FC, Fragment, useState } from 'react';
import React, { FC, useState } from 'react';
import Button from '@material-ui/core/Button';
import copy from 'copy-to-clipboard';
import EqualizerIcon from '@material-ui/icons/EqualizerSharp';
@ -11,7 +11,6 @@ import Find from 'lodash/find';
import IconButton from '@material-ui/core/IconButton';
import InfoIcon from '@material-ui/icons/Info';
import Tooltip from '@material-ui/core/Tooltip';
import Typography from '@material-ui/core/Typography';
import { formatBytes } from 'shared/utils/formatBytes';
import { nodeInfo } from './nodeInfo';
import { NODE_STATE } from './nodeStatus';
@ -21,7 +20,6 @@ import css from './Node.module.scss';
type NodeType = {
node: any;
info: any;
usage: any;
expanded: boolean;
onToggle: () => void;
onMetricClick: () => void;
@ -30,24 +28,13 @@ type NodeType = {
export const Node: FC<NodeType> = ({
node,
info,
usage,
expanded,
onToggle,
onMetricClick
onMetricClick,
}) => {
const [copyHost, setCopyHost] = useState<string>('Copy Host');
const [copyID, setCopyID] = useState<string>('Click to Copy');
const { id, isPrimary, state } = node;
const diskTotalInUse = usage?.diskUsage?.totalInUse;
const diskCapacity = usage?.diskUsage?.capacity;
const diskUsagePercentage = diskCapacity
? (diskTotalInUse / diskCapacity) * 100
: undefined;
const memoryTotalInUse = usage?.memoryUsage?.totalInUse;
const memoryCapacity = usage?.memoryUsage?.capacity;
const memoryUsagePercentage = memoryCapacity
? (memoryTotalInUse / memoryCapacity) * 100
: undefined;
const keys = Object.keys(info);
const onCopyHostClick = () => {
@ -103,154 +90,6 @@ export const Node: FC<NodeType> = ({
</span>
</Tooltip>
</div>
<div className={css.nodeUsage}>
<div>
<div className={css.label}>Disk Usage:</div>
<div>
{usage ? (
<Fragment>
<Typography variant="caption">
{formatBytes(diskTotalInUse)}
{diskCapacity
? ` used out of ${formatBytes(diskCapacity)}`
: null}
</Typography>
<div className={css.totalCapacity}>
{diskUsagePercentage ? (
<Tooltip
title={
<Typography variant="caption">
{diskUsagePercentage < 1
? '< 1'
: diskUsagePercentage.toLocaleString(
undefined,
{ maximumFractionDigits: 1 }
)}
% used
</Typography>
}
placement="top"
arrow
>
<div
className={css.totalInUse}
style={{
width: `${
diskUsagePercentage < 1
? 1
: diskUsagePercentage
}%`
}}
/>
</Tooltip>
) : (
<Fragment>
<Tooltip
title={
<Typography variant="caption">
{formatBytes(diskTotalInUse)} used
</Typography>
}
placement="top"
arrow
>
<div
className={css.totalInUse}
style={{ width: '2%' }}
/>
</Tooltip>
<Typography
className={css.unknownCapacity}
variant="caption"
color="textSecondary"
>
Node disk capacity unknown
</Typography>
</Fragment>
)}
</div>
</Fragment>
) : (
<Typography variant="caption" paragraph>
Calculating...
</Typography>
)}
</div>
</div>
<div>
<div className={css.label}>Memory Usage:</div>
<div>
{usage ? (
<Fragment>
<Typography variant="caption">
{formatBytes(memoryTotalInUse)}
{memoryCapacity
? ` used out of ${formatBytes(memoryCapacity)}`
: null}
</Typography>
<div className={css.totalCapacity}>
{memoryUsagePercentage ? (
<Tooltip
title={
<Typography variant="caption">
{memoryUsagePercentage < 1
? '< 1'
: memoryUsagePercentage.toLocaleString(
undefined,
{ maximumFractionDigits: 1 }
)}
% used
</Typography>
}
placement="top"
arrow
>
<div
className={css.totalInUse}
style={{
width: `${
memoryUsagePercentage < 1
? 1
: memoryUsagePercentage
}%`
}}
/>
</Tooltip>
) : (
<Fragment>
<Tooltip
title={
<Typography variant="caption">
{formatBytes(memoryTotalInUse)} used
</Typography>
}
placement="top"
arrow
>
<div
className={css.totalInUse}
style={{ width: '2%' }}
/>
</Tooltip>
<Typography
className={css.unknownCapacity}
variant="caption"
color="textSecondary"
>
Node memory capacity unknown
</Typography>
</Fragment>
)}
</div>
</Fragment>
) : (
<Typography variant="caption" paragraph>
Calculating...
</Typography>
)}
</div>
</div>
</div>
<div className={css.nodeSettings}>
{keys.map((key) => {
const showNode = Find(nodeInfo, (node) => node.name === key);

View file

@ -4,75 +4,38 @@ import Breadcrumbs from '@material-ui/core/Breadcrumbs';
import classNames from 'classnames';
import Fuse from 'fuse.js';
import Highlighter from 'react-highlight-words';
import isEmpty from 'lodash/isEmpty';
import Link from '@material-ui/core/Link';
import map from 'lodash/map';
import moment from 'moment';
import OrderBy from 'lodash/orderBy';
import Reduce from 'lodash/reduce';
import Table from '@material-ui/core/Table';
import TableBody from '@material-ui/core/TableBody';
import TableCell from '@material-ui/core/TableCell';
import TableHead from '@material-ui/core/TableHead';
import TableRow from '@material-ui/core/TableRow';
import TextField from '@material-ui/core/TextField';
import Tooltip from '@material-ui/core/Tooltip';
import Typography from '@material-ui/core/Typography';
import { Block } from 'shared/Block';
import { Pager } from 'shared/Pager';
import { UsageBreakdown } from '../UsageBreakdown';
import css from './MoleculaTable.module.scss';
type MoleculaTableProps = {
table: any;
dataDistribution: any;
lastUpdated: string;
};
export const MoleculaTable: FC<MoleculaTableProps> = ({
table,
dataDistribution,
lastUpdated
lastUpdated,
}) => {
const [page, setPage] = useState<number>(1);
const [resultsPerPage, setResultsPerPage] = useState<number>(10);
const sliceStart = (page - 1) * resultsPerPage;
const [searchText, setSearchText] = useState<string>('');
const [filteredFields, setFiltereedFields] = useState(table.fields);
const [fieldsData, setFieldsData] = useState<{}>({});
const [maxFieldSize, setMaxFieldSize] = useState<number>(0);
const [fieldsData] = useState<{}>({});
const [sort, setSort] = useState<string>('total');
const [sortDir, setSortDir] = useState<'asc' | 'desc'>('desc');
const lastUpdatedMoment = lastUpdated ? moment(lastUpdated).utc() : undefined;
useEffect(() => {
if (dataDistribution && !dataDistribution.uncached) {
const aggregatedFieldsData = Reduce(
dataDistribution.fields,
(result, value) => {
let newResult = {};
const keys = Object.keys(value);
keys.forEach(
(key) =>
(newResult[key] = {
total: result[key].total + value[key].total,
fragments: result[key].fragments + value[key].fragments,
keys: result[key].keys + value[key].keys,
metadata: result[key].metadata + value[key].metadata
})
);
return newResult;
}
);
const sorted = OrderBy(aggregatedFieldsData, ['total'], ['desc']);
if (sorted.length > 0) {
setMaxFieldSize(sorted[0].total);
}
setFieldsData(aggregatedFieldsData);
}
}, [dataDistribution]);
useEffect(() => {
if (searchText.length > 1) {
@ -80,7 +43,7 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
keys: ['name'],
minMatchCharLength: 2,
ignoreLocation: true,
threshold: 0
threshold: 0,
});
const result = fuse.search(searchText);
@ -131,46 +94,6 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
<Typography variant="h5" color="textSecondary">
{table.name}
</Typography>
{lastUpdatedMoment ? (
<div className={css.infoMessage}>
{dataDistribution && dataDistribution.uncached ? (
<Fragment>
Disk usage will be calculated at the next{` `}
<Tooltip
title={
<Fragment>
Disk and memory information shown here are read from a
cache, the behavior of which can be controlled with the{` `}
<code style={{ whiteSpace: 'nowrap' }}>
--usage-duty-cycle
</code>{' '}
command line flag.
</Fragment>
}
placement="top"
arrow
>
<span className={css.infoTooltip}>cache refresh</span>
</Tooltip>
.
</Fragment>
) : (
<Fragment>
Disk usage last updated{' '}
<Tooltip
title={`${lastUpdatedMoment.format('M/D/YYYY hh:mm a')} UTC`}
placement="top"
arrow
>
<span className={css.infoTooltip}>
{lastUpdatedMoment.fromNow()}
</span>
</Tooltip>
.
</Fragment>
)}
</div>
) : null}
<div className={css.layout}>
<div>
<label className={css.label}>keys</label>
@ -180,9 +103,6 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
</code>
</div>
</div>
<div className={css.breakdown}>
<UsageBreakdown data={dataDistribution} />
</div>
</div>
<div>
<div>
@ -211,36 +131,20 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
<TableCell className={css.tableHeader}>
<span
className={classNames(css.sortable, {
[css.currentSort]: sort === 'name'
[css.currentSort]: sort === 'name',
})}
onClick={() => onSortClick('name')}
>
Name{' '}
<ArrowDropDownIcon
className={classNames(css.sortArrow, {
[css.asc]: sortDir === 'asc'
[css.asc]: sortDir === 'asc',
})}
/>
</span>
</TableCell>
<TableCell className={css.tableHeader}>Type</TableCell>
<TableCell className={css.tableHeader}>Cardinality</TableCell>
<TableCell className={css.tableHeader}>Options</TableCell>
<TableCell className={css.tableHeader}>
<span
className={classNames(css.sortable, {
[css.currentSort]: sort === 'total'
})}
onClick={() => onSortClick('total')}
>
Disk Usage{' '}
<ArrowDropDownIcon
className={classNames(css.sortArrow, {
[css.asc]: sortDir === 'asc'
})}
/>
</span>
</TableCell>
</TableRow>
</TableHead>
<TableBody>
@ -266,9 +170,6 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
{type} {showKeys ? (keys ? '(keys)' : '(ID)') : null}
</code>
</TableCell>
<TableCell className={css.tableCell}>
{cardinality ? cardinality.toLocaleString() : '-'}
</TableCell>
<TableCell className={css.tableCell}>
<div className={css.optionsTable}>
{map(rest, (value, key) => {
@ -295,22 +196,6 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
})}
</div>
</TableCell>
<TableCell className={css.tableCell}>
<UsageBreakdown
data={
isEmpty(field)
? field
: dataDistribution
? dataDistribution.uncached
? dataDistribution
: field
: field
}
width={`${(field.total / maxFieldSize) * 150}px`}
showLabel={false}
usageValueSize="small"
/>
</TableCell>
</TableRow>
);
})}

View file

@ -8,42 +8,29 @@ import Tooltip from '@material-ui/core/Tooltip';
import Typography from '@material-ui/core/Typography';
import { Block } from 'shared/Block';
import { SortBy } from 'shared/SortBy';
import { UsageBreakdown } from './UsageBreakdown';
import { useHistory } from 'react-router-dom';
import css from './MoleculaTables.module.scss';
type MoleculaTablesProps = {
tables: any;
dataDistribution: any;
lastUpdated: string;
maxSize: number;
};
export const MoleculaTables: FC<MoleculaTablesProps> = ({
tables,
dataDistribution,
lastUpdated,
maxSize
maxSize,
}) => {
const history = useHistory();
const [sortedTables, setSortedTables] = useState<any>([]);
const lastUpdatedMoment = lastUpdated ? moment(lastUpdated).utc() : undefined;
useEffect(() => {
if (tables && dataDistribution) {
let aggregatedData: any[] = [];
tables.forEach((i) =>
aggregatedData.push({
...dataDistribution[i.name],
...i
})
);
setSortedTables(aggregatedData);
} else if (tables) {
if (tables) {
setSortedTables(tables);
}
}, [tables, dataDistribution]);
}, [tables]);
const handleSortChange = (value: any) => {
const sortDirection = value === 'name' ? 'asc' : 'desc';
@ -96,7 +83,7 @@ export const MoleculaTables: FC<MoleculaTablesProps> = ({
{ label: 'Index Keys Size', value: 'indexKeys' },
{ label: 'Fragment Size', value: 'fragments' },
{ label: 'Field Keys Size', value: 'fieldKeysTotal' },
{ label: 'Metadata Size', value: 'metadata' }
{ label: 'Metadata Size', value: 'metadata' },
]}
defaultValue="name"
onChange={handleSortChange}
@ -111,22 +98,6 @@ export const MoleculaTables: FC<MoleculaTablesProps> = ({
<Card key={name} className={css.tableTile}>
<CardContent>
<div className={css.header}>{name}</div>
<div className={css.section}>
<UsageBreakdown
data={
dataDistribution
? dataDistribution[name]
? dataDistribution[name]
: { uncached: true }
: undefined
}
width={
dataDistribution && dataDistribution[name]
? `${(dataDistribution[name].total / maxSize) * 100}%`
: '0px'
}
/>
</div>
<label className={css.label}>Options</label>
<div className={css.cell}>
<span className={css.label}>keys</span>

View file

@ -1,5 +1,4 @@
import React, { useEffect, useState } from 'react';
import OrderBy from 'lodash/orderBy';
import { MoleculaTable } from './MoleculaTable';
import { MoleculaTables } from './MoleculaTables';
import { pilosa } from 'services/eventServices';
@ -12,9 +11,8 @@ export const MoleculaTablesContainer = () => {
const history = useHistory();
const [tables, setTables] = useState<any>();
const [selectedTable, setSelectedTable] = useState<any>();
const [dataDistribution, setDataDistribution] = useState<any>();
const [maxSize, setMaxSize] = useState<number>(0);
const [lastUpdated, setLastUpdated] = useState<string>('');
const [maxSize] = useState<number>(0);
const [lastUpdated] = useState<string>('');
useEffectOnce(() => {
pilosa.get
@ -26,48 +24,6 @@ export const MoleculaTablesContainer = () => {
.then((res) => setTables(res.data.indexes))
.catch((err) => console.log(err))
);
pilosa.get.usage().then((res) => {
const nodes = Object.keys(res.data);
let data = {};
nodes.forEach((node) => {
const nodeIndexes = res.data[node].diskUsage.indexes;
const indexList = Object.keys(nodeIndexes);
indexList.forEach((i) => {
const nodeData = nodeIndexes[i];
if (data[i]) {
data[i] = {
total: data[i].total + nodeData.total,
fieldKeysTotal: data[i].fieldKeysTotal + nodeData.fieldKeysTotal,
indexKeys: data[i].indexKeys + nodeData.indexKeys,
fragments: data[i].fragments + nodeData.fragments,
metadata: data[i].metadata + nodeData.metadata,
fields: [...data[i].fields, nodeData.fields]
};
} else {
data[i] = {
total: nodeData.total,
fieldKeysTotal: nodeData.fieldKeysTotal,
indexKeys: nodeData.indexKeys,
fragments: nodeData.fragments,
metadata: nodeData.metadata,
fields: [nodeData.fields]
};
}
});
if(!lastUpdated) {
setLastUpdated(res.data[node].lastUpdated);
}
});
const sorted = OrderBy(data, ['total'], ['desc']);
if (sorted.length > 0) {
setMaxSize(sorted[0].total);
}
setDataDistribution(data);
});
});
useEffect(() => {
@ -85,21 +41,10 @@ export const MoleculaTablesContainer = () => {
}, [match, tables, history]);
return selectedTable ? (
<MoleculaTable
table={selectedTable}
dataDistribution={
dataDistribution
? dataDistribution[selectedTable.name]
? dataDistribution[selectedTable.name]
: { uncached: true }
: undefined
}
lastUpdated={lastUpdated}
/>
<MoleculaTable table={selectedTable} lastUpdated={lastUpdated} />
) : (
<MoleculaTables
tables={tables}
dataDistribution={dataDistribution}
lastUpdated={lastUpdated}
maxSize={maxSize}
/>

View file

@ -1,62 +0,0 @@
.label {
font-size: 0.75rem;
color: var(--text-secondary);
margin-bottom: 4px;
font-weight: 400;
}
.usageBreakdown {
display: flex;
align-items: center;
.usageBreakdownLabel {
white-space: nowrap;
margin-right: 8px;
&.smallLabel {
font-size: 12px;
}
}
}
.breakdown {
display: flex;
align-items: center;
height: 13px;
border-radius: 4px;
background: rgba(var(--contrast-rgb), 0.1);
.fieldKeysTotal {
height: 13px;
background: rgba(88, 80, 141, 0.7);
}
.indexKeys {
height: 13px;
background: rgba(255, 99, 97, 0.7);
}
.keys {
height: 13px;
background: rgba(88, 80, 141, 0.7);
}
.fragments {
height: 13px;
background: rgba(255, 166, 0, 0.7);
}
.metadata {
height: 13px;
background: rgba(188, 80, 144, 0.7);
}
.bar:first-child {
border-top-left-radius: 4px;
border-bottom-left-radius: 4px;
}
.bar:last-child {
border-top-right-radius: 4px;
border-bottom-right-radius: 4px;
}
}

View file

@ -1,183 +0,0 @@
import React, { FC, Fragment } from 'react';
import classNames from 'classnames';
import Tooltip from '@material-ui/core/Tooltip';
import Typography from '@material-ui/core/Typography';
import { formatBytes } from 'shared/utils/formatBytes';
import css from './UsageBreakdown.module.scss';
type UsageBreakdownProps = {
data: any;
width?: string;
showLabel?: boolean;
usageValueSize?: 'small' | 'medium';
};
export const UsageBreakdown: FC<UsageBreakdownProps> = ({
data = {},
width,
showLabel = true,
usageValueSize = 'medium'
}) => {
const {
total,
fieldKeysTotal,
indexKeys,
fragments,
metadata,
keys,
uncached
} = data;
const fieldKeysPercentage =
fieldKeysTotal && total ? (fieldKeysTotal / total) * 100 : 0;
const indexKeysPercentage = indexKeys ? (indexKeys / total) * 100 : 0;
const fragmentsPercentage = fragments ? (fragments / total) * 100 : 0;
const metadataPercentage = metadata ? (metadata / total) * 100 : 0;
const keysPercentage = keys && total ? (keys / total) * 100 : 0;
return (
<Fragment>
{showLabel ? <label className={css.label}>Data</label> : null}
<div className={css.usageBreakdown}>
{total ? (
<Fragment>
<span
className={classNames(css.usageBreakdownLabel, {
[css.smallLabel]: usageValueSize === 'small'
})}
>
{formatBytes(total)}
</span>
<div
className={css.breakdown}
style={{ width: width ? width : '100%' }}
>
{fieldKeysTotal ? (
<Tooltip
title={
<Fragment>
<label>Field Keys:</label>
<Typography variant="caption" component="div">
{formatBytes(fieldKeysTotal)} (
{fieldKeysPercentage.toLocaleString(undefined, {
maximumFractionDigits: 1
})}
%)
</Typography>
</Fragment>
}
placement="top"
arrow
>
<div
className={classNames(css.bar, css.fieldKeysTotal)}
style={{ width: `${fieldKeysPercentage}%` }}
/>
</Tooltip>
) : null}
{indexKeys ? (
<Tooltip
title={
<Fragment>
<label>Index Keys:</label>
<Typography variant="caption" component="div">
{formatBytes(indexKeys)} (
{indexKeysPercentage.toLocaleString(undefined, {
maximumFractionDigits: 1
})}
%)
</Typography>
</Fragment>
}
placement="top"
arrow
>
<div
className={classNames(css.bar, css.indexKeys)}
style={{ width: `${indexKeysPercentage}%` }}
/>
</Tooltip>
) : null}
{keys ? (
<Tooltip
title={
<Fragment>
<label>Keys:</label>
<Typography variant="caption" component="div">
{formatBytes(keys)} (
{keysPercentage.toLocaleString(undefined, {
maximumFractionDigits: 1
})}
%)
</Typography>
</Fragment>
}
placement="top"
arrow
>
<div
className={classNames(css.bar, css.keys)}
style={{ width: `${keysPercentage}%` }}
/>
</Tooltip>
) : null}
{fragments ? (
<Tooltip
title={
<Fragment>
<label>Fragments:</label>
<Typography variant="caption" component="div">
{formatBytes(fragments)} (
{fragmentsPercentage.toLocaleString(undefined, {
maximumFractionDigits: 1
})}
%)
</Typography>
</Fragment>
}
placement="top"
arrow
>
<div
className={classNames(css.bar, css.fragments)}
style={{ width: `${fragmentsPercentage}%` }}
/>
</Tooltip>
) : null}
{metadata ? (
<Tooltip
title={
<Fragment>
<label>Metadata:</label>
<Typography variant="caption" component="div">
{formatBytes(metadata)} (
{metadataPercentage.toLocaleString(undefined, {
maximumFractionDigits: 1
})}
%)
</Typography>
</Fragment>
}
placement="top"
arrow
>
<div
className={classNames(css.bar, css.metadata)}
style={{ width: `${metadataPercentage}%` }}
/>
</Tooltip>
) : null}
</div>
</Fragment>
) : uncached ? (
<Typography variant="caption" component="div">
Waiting...
</Typography>
) : (
<Typography variant="caption" component="div">
Calculating...
</Typography>
)}
</div>
</Fragment>
);
};

View file

@ -1 +0,0 @@
export * from './UsageBreakdown';

View file

@ -119,19 +119,7 @@ export const QueryContainer: FC<{}> = () => {
setLoading(false);
}
} else {
let queryArr = query.split(' ');
queryArr.forEach((word, idx) => {
if (word.includes('-')) {
let wordArr = word.split('.');
wordArr.forEach((section, idx) => {
if (section.includes('-') && !word.includes('`')) {
wordArr[idx] = `\`${wordArr[idx]}\``;
}
});
queryArr[idx] = wordArr.join('.');
}
});
querySQL(queryArr.join(' '), handleQueryMessages, handleQueryEnd);
querySQL(query, handleQueryMessages, handleQueryEnd);
}
}
};

View file

@ -42,9 +42,6 @@ export const pilosa = {
metrics() {
return api.get('/metrics.json');
},
usage() {
return api.get('/ui/usage');
},
queryHistory() {
return api.get('/query-history');
},

68
rbf.go
View file

@ -223,6 +223,74 @@ func (tx *RBFTx) Remove(index, field, view string, shard uint64, a ...uint64) (c
// which is expensive in practice and only really useful occasionally.
const sortedParanoia = false
type countResults struct {
changeCount int
err error
}
// RemoveChannel provides a method of streaming in bits or positions and not requiring a large buffer like add and remove
// the bits are input via the posChanel and the results are returned via the retChannel
func (tx *RBFTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) {
name := rbfName(index, field, view, shard)
var lastHi uint64 = math.MaxUint64 // highbits is always less than this starter.
var rc *roaring.Container
var hi uint64
var lo uint16
var err error
changeCount := 0
i := 0
for v := range a {
hi, lo = highbits(v), lowbits(v)
if hi != lastHi {
// either first time through, or changed to a different container.
// do we need put the last updated container now?
if i > 0 {
// not first time through, write what we got.
if rc == nil || (rc.N() == 0) {
err = tx.tx.RemoveContainer(name, lastHi)
if err != nil {
resChan <- countResults{0, errors.Wrap(err, "failed to remove container")}
return
}
} else {
err = tx.tx.PutContainer(name, lastHi, rc)
if err != nil {
resChan <- countResults{0, errors.Wrap(err, "failed to put container")}
return
}
}
}
// get the next container
rc, err = tx.tx.Container(name, hi)
if err != nil {
resChan <- countResults{0, errors.Wrap(err, "failed to retrieve container")}
return
}
} // else same container, keep adding bits to rct.
chng := false
rc, chng = rc.Remove(lo)
if chng {
changeCount++
}
lastHi = hi
i++
}
// write the last updates.
if rc == nil || rc.N() == 0 {
err = tx.tx.RemoveContainer(name, hi)
if err != nil {
resChan <- countResults{0, errors.Wrap(err, "failed to remove container")}
return
}
} else {
err = tx.tx.PutContainer(name, hi, rc)
if err != nil {
resChan <- countResults{0, errors.Wrap(err, "put to remove container")}
return
}
}
resChan <- countResults{changeCount, nil}
}
func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove bool, a ...uint64) (changeCount int, err error) {
if len(a) == 0 {
return 0, nil

View file

@ -474,9 +474,11 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
writeCellN(buf[:], len(group))
offset := dataOffset(len(group))
x := 0
for j, cell := range group {
writeLeafCell(buf[:], j, offset, cell)
offset += align8(cell.Size())
x++
}
if err := c.tx.writePage(buf[:]); err != nil {
@ -614,7 +616,6 @@ func (c *Cursor) deleteLeafCell(key uint64) (err error) {
copy(cells[elem.index:], cells[elem.index+1:])
cells[len(cells)-1] = leafCell{}
cells = cells[:len(cells)-1]
// Write cells to page.
buf := allocPage()
writePageNo(buf[:], elem.pgno)
@ -626,6 +627,7 @@ func (c *Cursor) deleteLeafCell(key uint64) (err error) {
writeLeafCell(buf[:], j, offset, cell)
offset += align8(cell.Size())
}
if err := c.tx.writePage(buf[:]); err != nil {
return err
}

View file

@ -2,6 +2,7 @@
package rbf_test
import (
"bytes"
"io"
"math/bits"
"math/rand"
@ -852,8 +853,10 @@ func TestDumpDot(t *testing.T) {
if err != nil {
t.Fatal(err)
}
rbf.Dumpdot(tx, 0, " ", os.Stdout)
var b bytes.Buffer
rbf.Dumpdot(tx, 0, " ", &b)
}
func TestCursor_UpdateBranchCells(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)

View file

@ -177,7 +177,7 @@ func intoContainer(l leafCell, tx *Tx, replacing *roaring.Container, target []by
case ContainerTypeBitmapPtr:
_, bm, _ := tx.leafCellBitmap(toPgno(cpMaybe))
cloneMaybe := bm
c = roaring.RemakeContainerBitmap(replacing, cloneMaybe)
c = roaring.RemakeContainerBitmapN(replacing, cloneMaybe, int32(l.BitN))
case ContainerTypeBitmap:
c = roaring.RemakeContainerBitmapN(replacing, toArray64(cpMaybe), int32(l.BitN))
case ContainerTypeRLE:
@ -216,9 +216,9 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
case ContainerTypeBitmapPtr:
_, bm, _ := tx.leafCellBitmap(toPgno(cpMaybe))
cloneMaybe := bm
c = roaring.NewContainerBitmap(-1, cloneMaybe)
c = roaring.NewContainerBitmap(l.BitN, cloneMaybe)
case ContainerTypeBitmap:
c = roaring.NewContainerBitmap(-1, toArray64(cpMaybe))
c = roaring.NewContainerBitmap(l.BitN, toArray64(cpMaybe))
case ContainerTypeRLE:
c = roaring.NewContainerRun(toInterval16(cpMaybe))
}

View file

@ -411,17 +411,28 @@ func (db *DB) checkpoint() (err error) {
// Close closes the database.
func (db *DB) Close() (err error) {
// TODO(bbj): Add wait group to hang until last Tx is complete.
// mark db as closed, spawn a thing to wait for existing tx to drain, then
// release the lock so they CAN drain. We do this before getting the
// write lock, so if something else is waiting on rwmu.Lock, and will be
// competing with us, we can ensure that it'll exit out quickly.
db.mu.Lock()
db.opened = false
// wait for transactions to complete
ch := make(chan struct{})
db.afterCurrentTx(func() {
close(ch)
})
db.mu.Unlock()
<-ch
// Wait for writer lock.
db.rwmu.Lock()
defer db.rwmu.Unlock()
// and main DB lock.
db.mu.Lock()
defer db.mu.Unlock()
db.opened = false
// Close mmap handle.
if db.data != nil {
if e := syswrap.Munmap(db.data); e != nil && err == nil {

View file

@ -139,6 +139,90 @@ func TestDB_WAL(t *testing.T) {
t.Fatal(err)
}
})
// initially this is just a cut and paste of the Halt test, except that
// we close the DB while the reads are still running.
t.Run("Close", func(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
}
config := rbfcfg.NewDefaultConfig()
config.MaxWALSize = 16 * rbf.PageSize
config.MaxWALCheckpointSize = 8 * rbf.PageSize
config.MinWALCheckpointSize = 4 * rbf.PageSize
db := MustOpenDB(t, config)
// Continuously run read overlapping transactions.
ctx, cancel := context.WithCancel(context.Background())
g, ctx := errgroup.WithContext(ctx)
for i := 0; i < 10; i++ {
i := i
g.Go(func() error {
time.Sleep(time.Duration(i) * 10 * time.Millisecond) // stagger
for {
if err := ctx.Err(); err != nil {
return nil
}
if err := func() error {
tx, err := db.Begin(false)
if err != nil {
return err
}
// give the db time to close between when we opened and
// when we run the Container call
time.Sleep(10 * time.Millisecond)
_, err = tx.Container("x", 0)
if err != nil {
t.Fatalf("requesting container: %v", err)
}
defer tx.Rollback()
return nil
}(); err != nil {
// it's okay to ErrClosed, because we plan to close
// the database out from under us.
if err != rbf.ErrClosed {
return err
} else {
return nil
}
}
}
})
}
// Generate updates to the DB/WAL.
for i := 0; i < 100; i++ {
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.CreateBitmapIfNotExists("x"); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", uint64(i)); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
time.Sleep(1 * time.Millisecond)
}()
}
// close the db now.
err := db.Close()
if err != nil {
t.Fatalf("closing db: %v", err)
}
// delay a bit to let some readers try to read
time.Sleep(20 * time.Millisecond)
// Stop read transactions & wait.
cancel()
if err := g.Wait(); err != nil {
t.Fatal(err)
}
})
}
func TestDB_Recovery(t *testing.T) {

View file

@ -164,79 +164,6 @@ func GenerateValues(rand *rand.Rand, n int) []uint64 {
return a
}
var _ = ToRows
// ToRows returns a sorted list of rows from a set of values.
func ToRows(values []uint64) []*Row {
m := make(map[uint64][]uint64)
for _, v := range values {
id := v / rbf.ShardWidth
m[id] = append(m[id], v&rbf.RowValueMask)
}
a := make([]*Row, 0, len(m))
for id, values := range m {
a = append(a, &Row{ID: id, Values: values})
}
sort.Slice(a, func(i, j int) bool { return a[i].ID < a[j].ID })
return a
}
var _ = Row{}
type Row struct {
ID uint64
Values []uint64
}
func (r *Row) Bitmap() []uint64 {
a := make([]uint64, rbf.ShardWidth/64)
for _, v := range r.Values {
a[v/64] |= 1 << (v % 64)
}
return a
}
// Union returns the union of r and other's values.
func (r *Row) Union(other *Row) []uint64 {
m := make(map[uint64]struct{})
for _, v := range r.Values {
m[v] = struct{}{}
}
for _, v := range other.Values {
m[v] = struct{}{}
}
a := make([]uint64, 0, len(m))
for v := range m {
a = append(a, v)
}
sort.Slice(a, func(i, j int) bool { return a[i] < a[j] })
return a
}
// Intersect returns the intersection of r & other's values.
func (r *Row) Intersect(other *Row) []uint64 {
m := make(map[uint64]struct{})
for _, v := range r.Values {
m[v] = struct{}{}
}
a := make([]uint64, 0)
used := make(map[uint64]struct{})
for _, v := range other.Values {
if _, ok := used[v]; ok {
continue
}
if _, ok := m[v]; ok {
used[v] = struct{}{}
a = append(a, v)
}
}
sort.Slice(a, func(i, j int) bool { return a[i] < a[j] })
return a
}
// QuickCheck executes fn multiple times with a different PRNG.
func QuickCheck(t *testing.T, fn func(t *testing.T, rand *rand.Rand)) {
for i := 0; i < *quickCheckN; i++ {

View file

@ -201,6 +201,29 @@ func (tx *Tx) BitmapNames() ([]string, error) {
return a, nil
}
// BitmapExist returns true if bitmap exists.
func (tx *Tx) BitmapExists(name string) (bool, error) {
tx.mu.Lock()
defer tx.mu.Unlock()
return tx.bitmapExists(name)
}
func (tx *Tx) bitmapExists(name string) (bool, error) {
if tx.db == nil {
return false, ErrTxClosed
} else if name == "" {
return false, ErrBitmapNameRequired
}
// Read root records and find entry for bitmap.
records, err := tx.RootRecords()
if err != nil {
return false, err
}
_, ok := records.Get(name)
return ok, nil
}
// CreateBitmap creates a new empty bitmap with the given name.
// Returns an error if the bitmap already exists.
func (tx *Tx) CreateBitmap(name string) error {
@ -561,6 +584,31 @@ func (tx *Tx) Contains(name string, v uint64) (bool, error) {
return c.Contains(v)
}
// Depth returns the depth of the b-tree for a bitmap.
func (tx *Tx) Depth(name string) (int, error) {
tx.mu.RLock()
defer tx.mu.RUnlock()
if tx.db == nil {
return 0, ErrTxClosed
} else if name == "" {
return 0, ErrBitmapNameRequired
}
c, err := tx.cursor(name)
if err == ErrBitmapNotFound {
return 0, nil
} else if err != nil {
return 0, err
}
defer c.Close()
if err := c.First(); err != nil {
return 0, err
}
return c.stack.top + 1, nil
}
// Cursor returns an instance of a cursor this bitmap.
func (tx *Tx) Cursor(name string) (*Cursor, error) {
tx.mu.RLock()
@ -669,6 +717,7 @@ func (tx *Tx) container(name string, key uint64) (*roaring.Container, error) {
func (tx *Tx) PutContainer(name string, key uint64, ct *roaring.Container) error {
tx.mu.Lock()
defer tx.mu.Unlock()
return tx.putContainer(name, key, ct)
}
@ -725,7 +774,6 @@ func (tx *Tx) removeContainer(name string, key uint64) error {
if exact, err := c.Seek(key); err != nil || !exact {
return err
}
return c.deleteLeafCell(key)
}
@ -1125,7 +1173,7 @@ func (tx *Tx) readPage(pgno uint32) (_ []byte, isHeap bool, err error) {
// Verify page number requested is within current size of database.
pageN := readMetaPageN(tx.meta[:])
if pgno > pageN {
if pgno >= pageN {
return nil, false, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN-1)
}
@ -1210,6 +1258,22 @@ func (tx *Tx) ContainerIterator(name string, key uint64) (citer roaring.Containe
return &containerIterator{cursor: c}, exact, nil
}
// Shared pool for in-memory database pages.
// These are used before being flushed to disk.
var containerFilterPool = &sync.Pool{}
func getContainerFilter(c *Cursor, filter roaring.BitmapFilter, tx *Tx) *containerFilter {
existing := containerFilterPool.Get()
if existing == nil {
return &containerFilter{cursor: c, filter: filter, tx: tx}
}
f := existing.(*containerFilter)
f.cursor = c
f.filter = filter
f.tx = tx
return f
}
func (tx *Tx) ApplyFilter(name string, key uint64, filter roaring.BitmapFilter) (err error) {
tx.mu.RLock()
defer tx.mu.RUnlock()
@ -1225,7 +1289,7 @@ func (tx *Tx) ApplyFilter(name string, key uint64, filter roaring.BitmapFilter)
if err != nil {
return err
}
f := containerFilter{cursor: c, filter: filter, tx: tx}
f := getContainerFilter(c, filter, tx)
defer f.Close()
return f.Apply()
}
@ -1567,6 +1631,8 @@ type containerFilter struct {
func (s *containerFilter) Close() {
s.cursor.Close()
s.cursor = nil
containerFilterPool.Put(s)
}
func (s *containerFilter) Apply() (err error) {
@ -1932,6 +1998,7 @@ func (tx *Tx) Pages(pgnos []uint32) ([]Page, error) {
// PageInfos returns meta data about all pages in the database.
func (tx *Tx) PageInfos() ([]PageInfo, error) {
var errorList ErrorList
infos := make([]PageInfo, tx.PageN())
// Read meta page info.

View file

@ -2,6 +2,7 @@
package rbf_test
import (
"bytes"
"encoding/binary"
"fmt"
"math/rand"
@ -443,7 +444,7 @@ func TestTx_DeallocateToFreeList(t *testing.T) {
}
}
func TestTx_Remove(t *testing.T) {
func TestTx_RemoveContainer(t *testing.T) {
t.Parallel()
db := MustOpenDB(t)
@ -541,6 +542,309 @@ func TestTx_AddRemove_Quick(t *testing.T) {
})
}
func TestTx_Remove(t *testing.T) {
t.Run("FullContiguous", func(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
}
for _, bitN := range []uint64{1000, 100000, 2000000} {
t.Run(fmt.Sprint(bitN), func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
// Add bits
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
for i := uint64(0); i < bitN; i++ {
if _, err := tx.Add("x", i); err != nil {
t.Fatalf("Add(%d) err=%q", i, err)
}
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
// Remove bits
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
for i := uint64(0); i < bitN; i++ {
if _, err := tx.Remove("x", i); err != nil {
t.Fatalf("Remove(%d) err=%q", i, err)
}
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
// Verify that all bits have been removed.
tx := MustBegin(t, db, false)
defer tx.Rollback()
if n, err := tx.Count("x"); err != nil {
t.Fatal(err)
} else if got, want := n, uint64(0); got != want {
t.Fatalf("Count=%d, want %d", got, want)
}
})
}
})
t.Run("PartialContiguous", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
// Add bits
const bitN = 100000
const multiplier = 7 // space out bits so we span more containers
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
for i := uint64(0); i < bitN; i++ {
if _, err := tx.Add("x", i*multiplier); err != nil {
t.Fatalf("Add(%d) err=%q", i, err)
}
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
// Remove some bits in small contiguous chunks.
var deleteN int
for i := uint64(bitN / 2); i < bitN; {
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
for j := uint64(0); j < 100; i, j = i+1, j+1 {
if n, err := tx.Remove("x", i*multiplier); err != nil || n != 1 {
t.Fatalf("Remove(%d)=(%v,%q)", i, n, err)
}
deleteN++
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
}
// Verify that we have the correct count afterward.
tx := MustBegin(t, db, false)
defer tx.Rollback()
if n, err := tx.Count("x"); err != nil {
t.Fatal(err)
} else if got, want := n, uint64(bitN-deleteN); got != want {
t.Fatalf("Count=%d, want %d", got, want)
}
})
t.Run("PartialNonContiguous", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
// Add bits
const bitN = 100000
const multiplier = 7 // space out bits
bits := make([]uint64, 0, bitN)
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
for i := uint64(0); i < bitN; i++ {
if _, err := tx.Add("x", i*multiplier); err != nil {
t.Fatalf("Add(%d) err=%q", i, err)
}
bits = append(bits, i*multiplier)
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
// Remove some bits in small contiguous chunks.
var deleteN int
perm := rand.Perm(len(bits))
for i := uint64(bitN / 2); i < bitN; {
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
for j := uint64(0); j < 100; i, j = i+1, j+1 {
value := bits[perm[i]]
if n, err := tx.Remove("x", value); err != nil || n != 1 {
t.Fatalf("Remove(%d)=(%v,%q)", value, n, err)
}
deleteN++
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
}
// Verify that we have the correct count afterward.
tx := MustBegin(t, db, false)
defer tx.Rollback()
if n, err := tx.Count("x"); err != nil {
t.Fatal(err)
} else if got, want := n, uint64(bitN-deleteN); got != want {
t.Fatalf("Count=%d, want %d", got, want)
}
})
t.Run("DeleteEmptyBitmap", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
// Create bitmap.
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
// Remove bitmap.
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.DeleteBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
// Ensure bitmap no longer exists.
tx := MustBegin(t, db, false)
defer tx.Rollback()
if exists, err := tx.BitmapExists("x"); err != nil {
t.Fatal(err)
} else if exists {
t.Fatal("expected bitmap to be removed")
}
})
t.Run("WithTreeDepth", func(t *testing.T) {
for depth := 1; depth <= 3; depth++ {
t.Run(fmt.Sprint(depth), func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
// Create bitmap & insert until we hit a tree depth.
var bitN int
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
for i := uint64(0); ; i++ {
if _, err := tx.Add("x", i<<16); err != nil {
t.Fatalf("Add(%d) err=%q", i<<16, err)
}
bitN++
if d, err := tx.Depth("x"); err != nil {
t.Fatal(err)
} else if d == depth {
break
}
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
// Remove all bits in reverse order.
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
for i := bitN - 1; i >= 0; i-- {
if n, err := tx.Remove("x", uint64(i)<<16); err != nil || n != 1 {
t.Fatalf("Remove(%d)=(%v,%q)", uint64(i)<<16, n, err)
}
}
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
// Ensure bitmap no longer exists.
tx := MustBegin(t, db, false)
defer tx.Rollback()
for i := uint64(0); i < uint64(bitN); i++ {
if ok, err := tx.Contains("x", i<<16); err != nil || ok {
t.Fatalf("Contains(%d)=(%v,%q)", i<<16, ok, err)
}
}
})
}
})
t.Run("RollbackAfterDelete", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}()
// Add bits
const bitN = 1000
for i := uint64(0); i < bitN; i++ {
func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
if _, err := tx.Add("x", i<<16); err != nil {
t.Fatalf("Add(%d) err=%q", i<<16, err)
}
// Only commit every other bit.
if i%2 == 1 {
if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}
}()
}
// Verify that we have the correct count afterward.
tx := MustBegin(t, db, false)
defer tx.Rollback()
if n, err := tx.Count("x"); err != nil {
t.Fatal(err)
} else if got, want := n, uint64(bitN/2); got != want {
t.Fatalf("Count=%d, want %d", got, want)
}
})
}
func TestTx_Multiple_CreateBitmap(t *testing.T) {
rand := rand.New(rand.NewSource(0))
db := MustOpenDB(t)
@ -742,7 +1046,11 @@ func TestTx_DeleteBitmapsWithPrefix(t *testing.T) {
t.Fatal(err)
}
}
checkInfos := func() {
var b bytes.Buffer
pBuf := func(msg string, args ...interface{}) (int, error) {
return fmt.Fprintf(&b, msg, args...)
}
checkInfos := func(pf func(string, ...interface{}) (int, error)) {
tx := MustBegin(t, db, false)
defer tx.Rollback()
infos, err := tx.PageInfos()
@ -750,34 +1058,34 @@ func TestTx_DeleteBitmapsWithPrefix(t *testing.T) {
for pgno, info := range infos {
switch info := info.(type) {
case *rbf.MetaPageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "meta")
fmt.Printf("pageN=%d,walid=%d,rootrec=%d,freelist=%d\n", info.PageN, info.WALID, info.RootRecordPageNo, info.FreelistPageNo)
pf("%-8d ", pgno)
pf("%-10s ", "meta")
pf("pageN=%d,walid=%d,rootrec=%d,freelist=%d\n", info.PageN, info.WALID, info.RootRecordPageNo, info.FreelistPageNo)
case *rbf.RootRecordPageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "rootrec")
fmt.Printf("next=%d\n", info.Next)
pf("%-8d ", pgno)
pf("%-10s ", "rootrec")
pf("next=%d\n", info.Next)
case *rbf.LeafPageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "leaf")
fmt.Printf("flags=x%x,celln=%d\n", info.Flags, info.CellN)
pf("%-8d ", pgno)
pf("%-10s ", "leaf")
pf("flags=x%x,celln=%d\n", info.Flags, info.CellN)
case *rbf.BranchPageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "branch")
fmt.Printf("flags=x%x,celln=%d\n", info.Flags, info.CellN)
pf("%-8d ", pgno)
pf("%-10s ", "branch")
pf("flags=x%x,celln=%d\n", info.Flags, info.CellN)
case *rbf.BitmapPageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "bitmap")
fmt.Printf("-\n")
pf("%-8d ", pgno)
pf("%-10s ", "bitmap")
pf("-\n")
case *rbf.FreePageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "free")
fmt.Printf("-\n")
pf("%-8d ", pgno)
pf("%-10s ", "free")
pf("-\n")
default:
t.Fatal(fmt.Sprintf("unexpected page info type %T", info))
@ -806,19 +1114,19 @@ func TestTx_DeleteBitmapsWithPrefix(t *testing.T) {
ifError(tx.Commit())
}
checkInfos()
checkInfos(pBuf)
populate()
checkInfos()
checkInfos(pBuf)
ifError(db.Check())
tx := MustBegin(t, db, true)
tx.DeleteBitmapsWithPrefix(prefix)
ifError(tx.Commit())
ifError(db.Check())
checkInfos()
checkInfos(pBuf)
populate()
ifError(db.Check())
checkInfos()
checkInfos(pBuf)
}

View file

@ -618,7 +618,7 @@ func (c *Container) setBitmap(bitmap []uint64) {
}
}
if len(bitmap) != 1024 {
panic("illegal bitmap length")
panic(fmt.Sprintf("illegal bitmap length %v", len(bitmap)))
}
c.pointer, c.len, c.cap = (*uint16)(unsafe.Pointer(&bitmap[0])), bitmapN, bitmapN
c.flags &^= flagPristine

View file

@ -584,6 +584,10 @@ type BitmapBitmapFilter struct {
callback func(uint64) error
}
func (b *BitmapBitmapFilter) SetCallback(cb func(uint64) error) {
b.callback = cb
}
func (b *BitmapBitmapFilter) ConsiderKey(key FilterKey, n int32) FilterResult {
pos := key & keyMask
if b.containers[pos] == nil || n == 0 {
@ -875,3 +879,136 @@ func ApplyFilterToIterator(filter BitmapFilter, iter ContainerIterator) error {
}
return nil
}
// BitmapBSICountFilter gives counts of values in each value-holding row
// of a BSI field, constrained by a filter. The first row of the data is
// taken to be an existence bit, which is intersected into the filter to
// constrain it, and the second is used as a sign bit. The rows after that
// are treated as value rows, and their counts of bits, overlapping with
// positive and negative bits in the sign rows, are returned to a callback
// function.
//
// The total counts of positions evaluated are returned with a row count
// of ^uint64(0) prior to row counts.
type BitmapBSICountFilter struct {
containers []*Container
positive []*Container
negative []*Container
nextOffsets []uint64
count int32
psum, nsum uint64
}
func (b *BitmapBSICountFilter) Total() (count int32, total int64) {
return b.count, int64(b.psum) - int64(b.nsum)
}
func (b *BitmapBSICountFilter) ConsiderKey(key FilterKey, n int32) FilterResult {
pos := key & keyMask
if b.containers[pos] == nil || n == 0 {
return key.RejectUntilOffset(b.nextOffsets[pos])
}
return key.NeedData()
}
func (b *BitmapBSICountFilter) ConsiderData(key FilterKey, data *Container) FilterResult {
pos := key & keyMask
filter := b.containers[pos]
if filter == nil {
key.RejectUntilOffset(b.nextOffsets[pos])
}
row := uint64(key >> rowExponent) // row count within the fragment
// How do we translate the filter and existence bit into actionable things?
// Assume the sign row is empty. We want positive values for anything in
// the intersection of the filter and the positive bits. If the sign row
// isn't empty, we want positive values for that intersection, less the
// sign row, and negative for the intersection of the filter/positive and
// the sign bits. So we can just stash the intermediate filter+existence
// as positive, then split it up if we have sign bits, which we often don't.
setup := false
switch row {
case 0: // existence bit
b.positive[pos] = intersect(b.containers[pos], data)
if b.positive[pos] == data {
b.positive[pos] = b.positive[pos].Clone()
}
b.count += int32(b.positive[pos].N())
setup = true
case 1: // sign bit
// split into negative/positive components. doesn't affect total
// count.
b.negative[pos] = intersect(b.positive[pos], data)
if b.negative[pos] == data {
b.negative[pos] = b.negative[pos].Clone()
}
b.positive[pos] = difference(b.positive[pos], data)
setup = true
}
// if we were doing setup (first two rows), we're done
if setup {
return key.MatchOneUntilOffset(b.nextOffsets[pos])
}
// helpful reminder: a nil container is a valid empty container, and
// intersectionCount knows this.
pcount := intersectionCount(b.positive[pos], data)
ncount := intersectionCount(b.negative[pos], data)
b.psum += (uint64(pcount) << (row - 2))
b.nsum += (uint64(ncount) << (row - 2))
return key.MatchOneUntilOffset(b.nextOffsets[pos])
}
// NewBitmapBSICountFilter creates a BitmapBSICountFilter, used for tasks
// like computing the sum of a BSI field matching a given filter.
//
// The input filter is assumed to represent one "row" of a shard's data,
// which is to say, a range of up to rowWidth consecutive containers starting
// at some multiple of rowWidth. We coerce that to the 0..rowWidth range
// because offset-within-row is what we care about.
func NewBitmapBSICountFilter(filter *Bitmap) *BitmapBSICountFilter {
containers := make([]*Container, rowWidth*3)
b := &BitmapBSICountFilter{
containers: containers[:rowWidth],
positive: containers[rowWidth : rowWidth*2],
negative: containers[rowWidth*2 : rowWidth*3],
nextOffsets: make([]uint64, rowWidth),
}
if filter == nil {
for i := range b.containers {
b.containers[i] = NewContainerRun([]Interval16{{Start: 0, Last: 65535}})
b.nextOffsets[i] = uint64(i+1) % rowWidth
}
return b
}
count := 0
iter, _ := filter.Containers.Iterator(0)
last := uint64(0)
for iter.Next() {
k, v := iter.Value()
// Coerce container key into the 0-rowWidth range we'll be
// using to compare against containers within each row.
k = k & keyMask
b.containers[k] = v
last = k
count++
}
// if there's only one container, we need to populate everything with
// its position.
if count == 1 {
for i := range b.containers {
b.nextOffsets[i] = last
}
} else {
// Point each container at the offset of the next valid container.
// With sparse bitmaps this will potentially make skipping faster.
for i := range b.containers {
if b.containers[i] != nil {
for int(last) != i {
b.nextOffsets[last] = uint64(i)
last = (last + 1) % rowWidth
}
}
}
}
return b
}

View file

@ -675,6 +675,47 @@ func (b *Bitmap) Intersect(other *Bitmap) *Bitmap {
return output
}
func (b *Bitmap) Hash(hash uint64) uint64 {
const (
offset = 14695981039346656037
prime = 1099511628211
)
if hash == 0 {
hash = uint64(offset)
}
it, _ := b.Containers.Iterator(0)
for it.Next() {
ki, _ := it.Value()
hash ^= uint64(ki)
hash *= prime
}
it, _ = b.Containers.Iterator(0)
for it.Next() {
_, ci := it.Value()
hash ^= 0
hash *= prime
if ci.N() > 0 {
var bytes []byte
switch ci.typ() {
case ContainerArray:
bytes = fromArray16(ci.array())
case ContainerBitmap:
bytes = fromArray64(ci.bitmap())
case ContainerRun:
bytes = fromInterval16(ci.runs())
}
for _, b := range bytes {
hash ^= uint64(b)
hash *= prime
}
}
}
return hash
}
type mutableContainersIterator struct {
c Containers
@ -7488,3 +7529,13 @@ func (c *Container) Slice() (r []uint16) {
}
return r
}
func fromArray16(a []uint16) []byte {
return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*2 : len(a)*2]
}
func fromArray64(a []uint64) []byte {
return (*[8192]byte)(unsafe.Pointer(&a[0]))[:8192:8192]
}
func fromInterval16(a []Interval16) []byte {
return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*4 : len(a)*4]
}

View file

@ -4825,3 +4825,26 @@ func TestVariousBitmap(t *testing.T) {
t.Fatal("nil AddN should be 0")
}
}
func TestBitmapHash(t *testing.T) {
a, b := NewContainerBitmapN(getFullBitmap(), MaxContainerVal+1), NewContainerBitmapN(getFullBitmap(), MaxContainerVal+1)
arr := NewContainerArray([]uint16{1, 2, 3, 5, 8})
run := NewContainerRun([]Interval16{{Start: 0, Last: 32}})
ba := NewBitmap()
bb := NewBitmap()
ba.Containers.Put(1, arr)
ba.Containers.Put(2, run)
ba.Containers.Put(101, a)
ba.Containers.Put(102, a)
bb.Containers.Put(1, arr)
bb.Containers.Put(2, run)
bb.Containers.Put(101, b)
bb.Containers.Put(102, b)
if ba.Hash(0) != bb.Hash(0) {
t.Fatal("hash should be equal")
}
bb.Containers.Put(103, b)
if ba.Hash(0) == bb.Hash(0) {
t.Fatal("hash should be different")
}
}

9
row.go
View file

@ -122,6 +122,15 @@ func (r *Row) ToTable() (*pb.TableResponse, error) {
return pb.RowsToTable(r, n)
}
// Hash calculate checksum code be useful in block hash join
func (r *Row) Hash() uint64 {
hash := uint64(0)
for i := range r.segments {
hash = r.segments[i].data.Hash(hash)
}
return hash
}
// ToRows implements the ToRowser interface.
func (r *Row) ToRows(callback func(*pb.RowResponse) error) error {
if len(r.Keys) > 0 {

View file

@ -44,6 +44,7 @@ var _ broadcaster = &Server{}
type Server struct { // nolint: maligned
// Close management.
wg sync.WaitGroup
muWG sync.Mutex
closing chan struct{}
// Internal
@ -100,6 +101,26 @@ func (s *Server) Holder() *Holder {
return s.holder
}
// addToWaitGroup adds to the server WaitGroup but makes sure the server isn't
// closing, and that the WaitGroup is not already waiting before it adds
func (s *Server) addToWaitGroup(delta int) bool {
select {
case <-s.closing:
return false
default:
s.muWG.Lock()
defer s.muWG.Unlock()
select {
case <-s.closing:
// if we're closing after having gotten the lock, stop!!
return false
default:
s.wg.Add(delta)
return true
}
}
}
// ServerOption is a functional option type for pilosa.Server
type ServerOption func(s *Server) error
@ -601,7 +622,10 @@ func (s *Server) Open() error {
// Start background process listening for translation
// sync resets.
s.wg.Add(1)
if ok := s.addToWaitGroup(1); !ok {
return fmt.Errorf("closing server while opening server is NOT allowed")
}
go func() { defer s.wg.Done(); s.monitorResetTranslationSync() }()
go func() { _ = s.translationSyncer.Reset() }()
@ -628,7 +652,10 @@ func (s *Server) Open() error {
return errors.Wrap(err, "setting nodeState")
}
s.wg.Add(4)
if ok := s.addToWaitGroup(4); !ok {
return fmt.Errorf("closing server while opening server is NOT allowed")
}
go func() { defer s.wg.Done(); s.monitorAntiEntropy() }()
go func() { defer s.wg.Done(); s.monitorRuntime() }()
go func() { defer s.wg.Done(); s.monitorDiagnostics() }()
@ -643,14 +670,18 @@ func (s *Server) Open() error {
return toSend
}()
s.wg.Add(1)
if ok := s.addToWaitGroup(1); !ok {
return fmt.Errorf("closing server while opening server is NOT allowed")
}
go func() {
defer s.wg.Done()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
s.wg.Add(1)
if ok := s.addToWaitGroup(1); !ok {
// the server is closing, stop!!
return
}
go func() {
defer s.wg.Done()
defer cancel()
@ -728,11 +759,15 @@ func (s *Server) Close() error {
case <-s.closing:
return nil
default:
errE := s.executor.Close()
// get the muWG lock so that noone adds to the WaitGroup while it Waits
s.muWG.Lock()
defer s.muWG.Unlock()
// Notify goroutines to stop.
close(s.closing)
s.wg.Wait()
errE := s.executor.Close()
var errh, errd error
var errhs error
var errc error
@ -788,8 +823,11 @@ func (s *Server) monitorResetTranslationSync() {
case <-s.closing:
return
case <-s.resetTranslationSyncCh:
if ok := s.addToWaitGroup(1); !ok {
// the server is closing!!! stop!!
return
}
s.logger.Infof("holder translation sync beginning")
s.wg.Add(1)
go func() {
// Obtaining this lock ensures that there is only
// one instance of resetTranslationSync() running

View file

@ -214,9 +214,6 @@ type Config struct {
// LookupDBDSN is an external database to connect to for `ExternalLookup` queries.
LookupDBDSN string `toml:"lookup-db-dsn"`
// The percentage of time spent recalculating the disk and memory usage cache.
UsageDutyCycle float64 `toml:"usage-duty-cycle"`
// Future flags are used to represent features or functionality which is not
// yet the default behavior, but will be in a future release.
Future struct {
@ -225,9 +222,6 @@ type Config struct {
Rename bool `toml:"rename"`
} `toml:"future"`
// Toggles /schema/details endpoint. If off, it returns empty.
SchemaDetailsOn bool `toml:"schema-details-on"`
Auth Auth
}
@ -390,15 +384,9 @@ func NewConfig() *Config {
c.Etcd.PeerCertFile = ""
c.Etcd.PeerKeyFile = ""
// Disk and Memory Usage
c.UsageDutyCycle = 20.0
// Future flags.
c.Future.Rename = false
// Schema Details Toggle
c.SchemaDetailsOn = true
return c
}

View file

@ -1007,6 +1007,10 @@ func TestQuerySQLWithError(t *testing.T) {
sql: "select _id, age, field_not_found from grouper",
err: pilosa.ErrFieldNotFound,
},
{
sql: "select age, color, count(*) from grouper group by field_not_found, age, color",
err: pilosa.ErrFieldNotFound,
},
}
for i, test := range tests {

View file

@ -302,8 +302,7 @@ func TestHandler_Endpoints(t *testing.T) {
}
var bodySchema pilosa.Schema
if err := json.Unmarshal(w.Body.Bytes(),
&bodySchema); err != nil {
if err := json.Unmarshal(w.Body.Bytes(), &bodySchema); err != nil {
t.Fatalf("unexpected unmarshalling error: %v", err)
}
// DO NOT COMPARE `CreatedAt` - reset to 0
@ -316,9 +315,8 @@ func TestHandler_Endpoints(t *testing.T) {
//
var targetSchema pilosa.Schema
target := fmt.Sprintf(`{"indexes":[{"name":"i0","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false},"cardinality":0},{"name":"f1","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false},"cardinality":1,"views":[{"name":"standard"}]}],"shardWidth":%[1]d},{"name":"i1","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false},"cardinality":1,"views":[{"name":"standard"}]}],"shardWidth":%[1]d},{"name":"i2","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":1000,"keys":false},"cardinality":1,"views":[{"name":"standard"}]},{"name":"f1","options":{"type":"int","base":0,"bitDepth":0,"min":-100,"max":100,"keys":false,"foreignIndex":""},"cardinality":4,"views":[{"name":"bsig_f1"}]},{"name":"f2","options":{"type":"decimal","base":0,"scale":1,"bitDepth":0,"min":-10,"max":10,"keys":false},"cardinality":5,"views":[{"name":"bsig_f2"}]},{"name":"f3","options":{"type":"time","timeQuantum":"YMDH","keys":false,"noStandardView":false},"cardinality":1,"views":[{"name":"standard"}]},{"name":"f4","options":{"type":"mutex","cacheType":"ranked","cacheSize":5000,"keys":false},"cardinality":1,"views":[{"name":"standard"}]},{"name":"f5","options":{"type":"bool"},"cardinality":1,"views":[{"name":"standard"}]}],"shardWidth":%[1]d}]}`, pilosa.ShardWidth)
if err := json.Unmarshal([]byte(target),
&targetSchema); err != nil {
target := fmt.Sprintf(`{"indexes":[{"name":"i0","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false}},{"name":"f1","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false},"views":[{"name":"standard"}]}],"shardWidth":%[1]d},{"name":"i1","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false},"views":[{"name":"standard"}]}],"shardWidth":%[1]d},{"name":"i2","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":1000,"keys":false},"views":[{"name":"standard"}]},{"name":"f1","options":{"type":"int","base":0,"bitDepth":0,"min":-100,"max":100,"keys":false,"foreignIndex":""},"views":[{"name":"bsig_f1"}]},{"name":"f2","options":{"type":"decimal","base":0,"scale":1,"bitDepth":0,"min":-10,"max":10,"keys":false},"views":[{"name":"bsig_f2"}]},{"name":"f3","options":{"type":"time","timeQuantum":"YMDH","keys":false,"noStandardView":false},"views":[{"name":"standard"}]},{"name":"f4","options":{"type":"mutex","cacheType":"ranked","cacheSize":5000,"keys":false},"views":[{"name":"standard"}]},{"name":"f5","options":{"type":"bool"},"views":[{"name":"standard"}]}],"shardWidth":%[1]d}]}`, pilosa.ShardWidth)
if err := json.Unmarshal([]byte(target), &targetSchema); err != nil {
t.Fatalf("unexpected unmarshalling error: %v", err)
}
@ -327,38 +325,6 @@ func TestHandler_Endpoints(t *testing.T) {
}
})
t.Run("SchemaDetailsOff", func(t *testing.T) {
err := cmd.API.SetAPIOptions(pilosa.OptAPISchemaDetailsOn(false))
if err != nil {
t.Fatalf("setting schema details option")
}
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/schema/details", nil))
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
var bodySchema pilosa.Schema
if err := json.Unmarshal(w.Body.Bytes(),
&bodySchema); err != nil {
t.Fatalf("unexpected unmarshalling error: %v", err)
}
for _, i := range bodySchema.Indexes {
for _, f := range i.Fields {
if f.Cardinality != nil {
t.Fatalf("expected nil cardinality, got: %v", *f.Cardinality)
}
}
}
err = cmd.API.SetAPIOptions(pilosa.OptAPISchemaDetailsOn(true))
if err != nil {
t.Fatalf("could not toggle schema details to on: %v", err)
}
})
t.Run("Import", func(t *testing.T) {
indexInfo, err := cmd.API.Schema(context.Background(), false)
if err != nil {
@ -517,48 +483,6 @@ func TestHandler_Endpoints(t *testing.T) {
}
})
// UI/usage returns disk and memory usage from a precalculated cache.
// Since the cache calculates the cache on server startup, and tests create indexes thereafter
// the cache initially has 0 indexes when the test suite is ran. Therefore, this test first
// resets the cache.
t.Run("UI/usage", func(t *testing.T) {
if cmd.API.ResetUsageCache() != nil {
t.Fatal(err)
}
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/ui/usage", nil))
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
nodeUsages := make(map[string]pilosa.NodeUsage)
if err := json.Unmarshal(w.Body.Bytes(), &nodeUsages); err != nil {
t.Fatalf("unmarshal")
}
for _, nodeUsage := range nodeUsages {
if nodeUsage.Disk.TotalUse < 1 {
t.Fatalf("expected some disk use, got %d", nodeUsage.Disk.TotalUse)
}
if nodeUsage.Disk.Capacity < 1 {
t.Fatalf("expected some disk capacity, got %d", nodeUsage.Disk.Capacity)
}
if nodeUsage.Memory.TotalUse < 1 {
t.Fatalf("expected some memory use, got %d", nodeUsage.Memory.TotalUse)
}
if nodeUsage.Memory.Capacity < 1 {
t.Fatalf("expected some memory capacity, got %d", nodeUsage.Memory.Capacity)
}
numIndexes := len(nodeUsage.Disk.IndexUsage)
if numIndexes != 3 {
t.Fatalf("wrong length index usage list: expected %d, got %d", 3, numIndexes)
}
numFields := len(nodeUsage.Disk.IndexUsage["i1"].Fields)
if numFields != len(i1.Fields()) {
t.Fatalf("wrong length field usage list: expected %d, got %d", len(i1.Fields()), numFields)
}
}
})
t.Run("UI/shard-distribution", func(t *testing.T) {
// This tests the response structure, not the cluster behavior.
w := httptest.NewRecorder()

View file

@ -271,8 +271,6 @@ func (m *Command) Start() (err error) {
}
}
go m.API.RefreshUsageCache(m.Config.UsageDutyCycle)
_ = testhook.Opened(pilosa.NewAuditor(), m, nil)
close(m.Started)
return nil
@ -511,7 +509,6 @@ func (m *Command) SetupServer() error {
m.API, err = pilosa.NewAPI(
pilosa.OptAPIServer(m.Server),
pilosa.OptAPIImportWorkerPoolSize(m.Config.ImportWorkerPoolSize),
pilosa.OptAPISchemaDetailsOn(m.Config.SchemaDetailsOn),
)
if err != nil {
return errors.Wrap(err, "new api")
@ -519,10 +516,6 @@ func (m *Command) SetupServer() error {
// Tell server about its new API, which its client will need.
m.Server.SetAPI(m.API)
if err != nil {
return errors.Wrap(err, "new grpc server")
}
var p authz.GroupPermissions
if m.Config.Auth.Enable {
m.Config.MustValidateAuth()

View file

@ -35,3 +35,34 @@ func TestMonitorAntiEntropyZero(t *testing.T) {
t.Fatalf("monitorAntiEntropy should have returned immediately with duration 0")
}
}
func TestAddToWaitGroup(t *testing.T) {
// if this test times out / panics we have a problem, otherwise we're fine
td := t.TempDir()
cfg := &storage.Config{FsyncEnabled: false, Backend: storage.DefaultBackend}
s, err := NewServer(OptServerDataDir(td), OptServerStorageConfig(cfg))
if err != nil {
t.Fatalf("making new server: %v", err)
}
oks := make(chan bool, 10)
for i := 0; i < 10; i++ {
go func() {
oks <- s.addToWaitGroup(1)
time.Sleep(10 * time.Millisecond)
defer s.wg.Done()
}()
}
for i := 0; i < 10; i++ {
ok := <-oks
if !ok {
t.Fatalf("unexpected close during WaitGroup add")
}
}
s.Close()
if ok := s.addToWaitGroup(1); ok {
t.Fatalf("shouldn't be able to add while server is closing")
}
}

View file

@ -3,10 +3,13 @@ package sql_test
import (
"context"
"math"
"testing"
"github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/sql"
"github.com/molecula/featurebase/v3/test"
"vitess.io/vitess/go/vt/sqlparser"
)
func TestHandler(t *testing.T) {
@ -28,3 +31,52 @@ func TestHandler(t *testing.T) {
}
}
func TestSelectHandler_MapSelect(t *testing.T) {
cluster := test.MustRunCluster(t, 1)
defer cluster.Close()
api := cluster.GetNode(0).API
if _, err := api.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if _, err = api.CreateField(context.Background(), "i", "bytes", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
} else if _, err = api.CreateField(context.Background(), "i", "duration_time", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)); err != nil {
t.Fatal(err)
} else if _, err = api.CreateField(context.Background(), "i", "timestamp", pilosa.OptFieldTypeTimestamp(pilosa.DefaultEpoch, pilosa.TimeUnitSeconds)); err != nil {
t.Fatal(err)
}
for _, tt := range []struct {
name string
input string
output string
}{
{
name: "WhereTimestamp",
input: `SELECT * FROM i WHERE timestamp>"2000-01-01T00:00:00Z"`,
output: `Extract(Row(timestamp>"2000-01-01T00:00:00Z"),Rows(bytes),Rows(duration_time),Rows(timestamp))`,
},
{
name: "WhereTimestampWithSpaces",
input: `SELECT * FROM i WHERE timestamp > "2000-01-01T00:00:00Z"`,
output: `Extract(Row(timestamp>"2000-01-01T00:00:00Z"),Rows(bytes),Rows(duration_time),Rows(timestamp))`,
},
} {
t.Run(tt.name, func(t *testing.T) {
query, err := sql.NewMapper().MapSQL(tt.input)
if err != nil {
t.Fatal(err)
}
h := sql.NewSelectHandler(api)
mr, err := h.MapSelect(context.Background(), query.Statement.(*sqlparser.Select), query.Mask)
if err != nil {
t.Fatal(err)
} else if got, want := mr.Query, tt.output; got != want {
t.Fatalf("unexpected pql\npql: %s\nwant: %s", got, want)
}
})
}
}

View file

@ -15,32 +15,32 @@ const timeFormat = "2006-01-02T15:04"
// LT creates a less than query.
func LT(fieldName string, value interface{}) string {
return fmt.Sprintf("Row(%s<%s)", fieldName, intOrFloat(value))
return fmt.Sprintf("Row(%s<%s)", fieldName, formatValue(value))
}
// LTE creates a less than or equal query.
func LTE(fieldName string, value interface{}) string {
return fmt.Sprintf("Row(%s<=%s)", fieldName, intOrFloat(value))
return fmt.Sprintf("Row(%s<=%s)", fieldName, formatValue(value))
}
// GT creates a greater than query.
func GT(fieldName string, value interface{}) string {
return fmt.Sprintf("Row(%s>%s)", fieldName, intOrFloat(value))
return fmt.Sprintf("Row(%s>%s)", fieldName, formatValue(value))
}
// GTE creates a greater than or equal query.
func GTE(fieldName string, value interface{}) string {
return fmt.Sprintf("Row(%s>=%s)", fieldName, intOrFloat(value))
return fmt.Sprintf("Row(%s>=%s)", fieldName, formatValue(value))
}
// Equals creates an equals query.
func Equals(fieldName string, value interface{}) string {
return fmt.Sprintf("Row(%s=%s)", fieldName, intOrFloat(value))
return fmt.Sprintf("Row(%s=%s)", fieldName, formatValue(value))
}
// NotEquals creates a not equals query.
func NotEquals(fieldName string, value interface{}) string {
return fmt.Sprintf("Row(%s!=%s)", fieldName, intOrFloat(value))
return fmt.Sprintf("Row(%s!=%s)", fieldName, formatValue(value))
}
// NotNull creates a not equal to null query.
@ -94,12 +94,18 @@ func Like(fieldName string, pattern string) string {
// Between creates a between query.
func Between(fieldName string, a interface{}, b interface{}) string {
return fmt.Sprintf("Row(%s >< [%s,%s])", fieldName, intOrFloat(a), intOrFloat(b))
return fmt.Sprintf("Row(%s >< [%s,%s])", fieldName, formatValue(a), formatValue(b))
}
// Distinct creates a Distinct query.
func Distinct(indexName, fieldName string) string {
return fmt.Sprintf("Distinct(Row(%s!=null),index='%s',field='%s')", fieldName, indexName, fieldName)
func Distinct(indexName, fieldName, rowCall string) string {
var b strings.Builder
fmt.Fprintf(&b, `Distinct(`)
if rowCall != "" {
fmt.Fprintf(&b, `%s, `, rowCall)
}
fmt.Fprintf(&b, `index='%s',field='%s')`, indexName, fieldName)
return b.String()
}
// RowDistinct creates a Distinct query with the given row filter.
@ -269,8 +275,10 @@ func formatIDKey(idKey interface{}) (string, error) {
}
}
func intOrFloat(value interface{}) string {
func formatValue(value interface{}) string {
switch value.(type) {
case string:
return fmt.Sprintf("%q", value)
case float64, float32:
// In order to test expected values, we set the precision
// to 8. TODO: It's likely we'll need to address this

View file

@ -347,6 +347,34 @@ func AssignHeaders(rowser pproto.ToRowser, headers ...Column) pproto.ToRowser {
return &assignHeadersRowser{rowser, headers}
}
type staticHeaderRowser struct {
rowser pproto.ToRowser
cols []Column
}
func (a *staticHeaderRowser) ToRows(fn func(*pproto.RowResponse) error) error {
return a.rowser.ToRows(func(row *pproto.RowResponse) error {
var out pproto.RowResponse
headers := make([]*pproto.ColumnInfo, len(row.Headers))
for i := range row.Headers {
header := row.Headers[i]
header.Name = a.cols[i].Name()
headers[i] = header
}
out.Headers = headers
out.Columns = row.Columns
return fn(&out)
})
}
// StaticHeaders assigns fixed cols to a ToRowser.
func StaticHeaders(rowser pproto.ToRowser, cols ...Column) pproto.ToRowser {
return &staticHeaderRowser{rowser, cols}
}
var (
ErrIncompleteHeaders = errors.New("incomplete header assignment")
ErrFieldNotInHeaders = errors.New("field not found in source header")

View file

@ -29,6 +29,17 @@ func newRouter() *router {
handlerSelectFieldsFromTableWhere{},
)
////
selectRouter.addFilter(
NewQueryMask(
SelectPartDistinct|SelectPartField,
FromPartTable,
WherePartFieldCondition|WherePartMultiFieldCondition,
0,
0,
),
[]QueryMask{},
handlerSelectDistinctFromTable{},
)
selectRouter.addRoute("select distinct fld from tbl", handlerSelectDistinctFromTable{})
////
selectRouter.addFilter(
@ -58,7 +69,7 @@ func newRouter() *router {
groupByOptional := NewQueryMask(
SelectPartField|SelectPartFields|SelectPartCountStar|SelectPartSumField,
FromPartTable,
WherePartFieldCondition, // TODO: this can probably handle fields as well
WherePartFieldCondition|WherePartMultiFieldCondition,
GroupByPartField|GroupByPartFields,
HavingPartCondition,
)

View file

@ -34,14 +34,14 @@ func (s *SelectHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.T
if !ok {
return nil, fmt.Errorf("statement is not type select: %T", mapped.Statement)
}
mr, err := s.mapSelect(ctx, stmt, mapped.Mask)
mr, err := s.MapSelect(ctx, stmt, mapped.Mask)
if err != nil {
return nil, errors.Wrap(err, "mapping select")
}
return s.execMappingResult(ctx, mr, mapped.SQL)
}
func (s *SelectHandler) mapSelect(ctx context.Context, selectStmt *sqlparser.Select, qm QueryMask) (*MappingResult, error) {
func (s *SelectHandler) MapSelect(ctx context.Context, selectStmt *sqlparser.Select, qm QueryMask) (*MappingResult, error) {
// Get the handler for this query mask.
hndlr := s.router.handler(qm)
if hndlr == nil {
@ -305,6 +305,15 @@ func (h handlerSelectDistinctFromTable) Apply(stmt *sqlparser.Select, qm QueryMa
return nil, errors.New("distinct requires a valid field column")
}
var wherePQL string
if stmt.Where != nil {
if wherePQL, err = extractWhere(index, stmt.Where.Expr); err != nil {
return nil, err
}
} else {
wherePQL = All()
}
limit, offset, hasLimit, hasOffset, err := extractLimitOffset(stmt)
if err != nil {
return nil, errors.Wrap(err, "extracting limit")
@ -315,22 +324,8 @@ func (h handlerSelectDistinctFromTable) Apply(stmt *sqlparser.Select, qm QueryMa
return nil, errors.Wrap(err, "extracting order by")
}
// Determine the type of the field needing distinct.
// If the pilosa field is type int, handle it as a Distinct() query.
// Otherwise, use Rows()
// TODO: ensure this works for all field types (bool, time, etc).
var qo string
if fieldCol.Field.Type() == pilosa.FieldTypeInt || fieldCol.Field.Type() == pilosa.FieldTypeTimestamp {
qo = Distinct(fieldCol.Field.Index(), fieldCol.Field.Name())
} else {
if !qm.HasOrderBy() && limit > 0 {
if qo, err = RowsLimit(fieldCol.Field.Name(), int64(limit)); err != nil {
return nil, errors.Wrap(err, "creating Rows query")
}
} else {
qo = Rows(fieldCol.Field.Name())
}
}
// We use a Distinct call instead of Rows as it supports filtering.
qo := Distinct(fieldCol.Field.Index(), fieldCol.Field.Name(), wherePQL)
mr := &MappingResult{
IndexName: indexName,
@ -340,7 +335,7 @@ func (h handlerSelectDistinctFromTable) Apply(stmt *sqlparser.Select, qm QueryMa
// Assign headers to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return AssignHeaders(result, selectFields...)
return StaticHeaders(result, selectFields...)
})
if qm.HasOrderBy() {
@ -598,8 +593,7 @@ func (h handlerSelectGroupBy) Apply(stmt *sqlparser.Select, qm QueryMask, indexF
rowsQueries := []string{}
for _, fieldName := range groupByFieldNames {
field := index.Field(fieldName)
rowsQueries = append(rowsQueries, Rows(field.Name()))
rowsQueries = append(rowsQueries, Rows(fieldName))
}
var wherePQL string
@ -796,7 +790,7 @@ func (h handlerSelectJoin) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc
// Build the Distinct() portion of the query on the secondary.
var distinctQry string
if secondaryWhere == "" {
distinctQry = Distinct(secondaryField.Index(), secondaryField.Name())
distinctQry = Distinct(secondaryField.Index(), secondaryField.Name(), "")
} else {
distinctQry = RowDistinct(secondaryField.Index(), secondaryField.Name(), secondaryWhere)
}

View file

@ -81,11 +81,6 @@ func TestStatsCount_TopN(t *testing.T) {
defer c.Close()
hldr := test.Holder{Holder: c.GetNode(0).Server.Holder()}
hldr.SetBit("d", "f", 0, 0)
hldr.SetBit("d", "f", 0, 1)
hldr.SetBit("d", "f", 0, pilosa.ShardWidth)
hldr.SetBit("d", "f", 0, pilosa.ShardWidth+2)
// Execute query.
called := false
hldr.Holder.Stats = &MockStats{
@ -101,6 +96,12 @@ func TestStatsCount_TopN(t *testing.T) {
called = true
},
}
hldr.SetBit("d", "f", 0, 0)
hldr.SetBit("d", "f", 0, 1)
hldr.SetBit("d", "f", 0, pilosa.ShardWidth)
hldr.SetBit("d", "f", 0, pilosa.ShardWidth+2)
if _, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "d", Query: `TopN(field=f, n=2)`}); err != nil {
t.Fatal(err)
}

View file

@ -159,6 +159,7 @@ const (
kOffsetRange
kLast // mark the end, always keep this last. The following aren't tracked atm:
kType
kRemoveChannel
)
func (k kall) String() string {
@ -205,6 +206,8 @@ func (k kall) String() string {
return "kLast"
case kType:
return "kType"
case kRemoveChannel:
return "kRemoveChannel"
}
vprint.PanicOn(fmt.Sprintf("unknown kall '%v'", int(k)))
return ""
@ -221,6 +224,15 @@ func (c *statTx) NewTxIterator(index, field, view string, shard uint64) *roaring
}()
return c.b.NewTxIterator(index, field, view, shard)
}
func (c *statTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) {
me := kRemoveChannel
t0 := time.Now()
defer func() {
c.stats.add(me, time.Since(t0))
}()
c.b.RemoveChannel(index, field, view, shard, a, resChan)
return
}
func (c *statTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) {
me := kImportRoaringBits

View file

@ -11,6 +11,7 @@ import (
"sync"
"github.com/molecula/featurebase/v3/ingest"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/topology"
"github.com/pkg/errors"
)
@ -84,6 +85,8 @@ type TranslateStore interface { // TODO: refactor this interface; readonly shoul
// It should read from the reader and replace the data store with
// the read payload.
ReadFrom(io.Reader) (int64, error)
Delete(records *roaring.Bitmap) (Commitor, error)
}
// This implements ingest's key translator interface, which differs
@ -420,6 +423,16 @@ func (s *InMemTranslateStore) SetReadOnly(v bool) {
defer s.mu.Unlock()
s.readOnly = v
}
func (s *InMemTranslateStore) Delete(records *roaring.Bitmap) (Commitor, error) {
s.mu.Lock()
defer s.mu.Unlock()
for _, id := range records.Slice() {
key := s.keysByID[id]
delete(s.keysByID, id)
delete(s.idsByKey, key)
}
return &NopCommitor{}, nil
}
// FindKeys looks up the ID for each key.
// Keys are not created if they do not exist.

1
tx.go
View file

@ -133,6 +133,7 @@ type Tx interface {
GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error)
GetFieldSizeBytes(index, field string) (uint64, error)
RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults)
}
// GenericApplyFilter implements ApplyFilter in terms of tx.ContainerIterator,

View file

@ -242,5 +242,4 @@ func TestAPI_ImportAtomicRecord(t *testing.T) {
if iraBit {
PanicOn("IRA bit should have been cleared")
}
}

View file

@ -4,7 +4,6 @@ package pilosa
import (
"fmt"
"os"
"path"
"strings"
"sync"
@ -471,192 +470,6 @@ func (f *TxFactory) DeleteFragmentFromStore(
return f.dbPerShard.DeleteFragment(index, field, view, shard, frag)
}
// IndexUsageDetails computes the sum of filesizes used by the node, broken down
// by index, field, fragments and keys.
func (f *TxFactory) IndexUsageDetails(isClosing func() bool) (map[string]IndexUsage, uint64, error) {
indexUsage := make(map[string]IndexUsage)
holderPath, err := expandDirName(f.holder.path)
if err != nil {
return indexUsage, 0, errors.Wrap(err, "expanding data directory")
}
indexesPath, err := expandDirName(f.holder.IndexesPath())
if err != nil {
return indexUsage, 0, errors.Wrap(err, "expanding indexes directory")
}
idxs := f.holder.Indexes()
qcx := f.NewQcx()
defer qcx.Abort()
for _, idx := range idxs {
index := idx.name
indexPath := path.Join(indexesPath, index)
// field usage
fieldUsages := make(map[string]FieldUsage)
fragmentsTotal := uint64(0)
fieldKeysTotal := uint64(0)
fieldMetaBytesTotal := uint64(0)
fieldsTotal := uint64(0)
flds := idx.Fields()
for _, fld := range flds {
field := fld.Name()
if field == "_keys" {
continue
}
fUsage, err := f.fieldUsage(indexPath, fld)
if err != nil {
return indexUsage, 0, errors.Wrapf(err, "getting disk usage for index (%s)", index)
}
// non-roaring field usage
fragmentUsage := uint64(0)
for _, shard := range fld.AvailableShards(true).Slice() {
if isClosing() {
return nil, 0, nil
}
if err := func() error {
tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard})
if err != nil {
return errors.Wrap(err, "qcx.GetTx")
}
defer finisher(nil)
fieldBytes, err := tx.GetFieldSizeBytes(index, field)
if err != nil {
return errors.Wrapf(err, "getting disk usage for non-roaring fragments (%s)", field)
}
fragmentUsage += fieldBytes
return nil
}(); err != nil {
return indexUsage, 0, err
}
}
// add non-roaring to roaring
fUsage.Fragments += fragmentUsage
fUsage.Total += fragmentUsage
// add to running total
fieldMetaBytesTotal += fUsage.Metadata
fieldKeysTotal += fUsage.Keys
fragmentsTotal += fUsage.Fragments
fieldsTotal += fUsage.Total
fieldUsages[field] = fUsage
}
// index metadata
indexMetaBytes, err := directoryUsage(indexPath, false)
if err != nil {
return indexUsage, 0, errors.Wrapf(err, "getting disk usage for index metadata (%s)", index)
}
// index keys usage
indexKeysBytes := uint64(0)
if idx.keys {
keysPath := path.Join(indexPath, translateStoreDir)
indexKeysBytes, _ = directoryUsage(keysPath, true) // if directory doesn't exist, size = 0
}
indexUsage[index] = IndexUsage{
Total: indexMetaBytes + indexKeysBytes + fieldsTotal,
Metadata: indexMetaBytes + fieldMetaBytesTotal,
IndexKeys: indexKeysBytes,
FieldKeysTotal: fieldKeysTotal,
Fragments: fragmentsTotal,
Fields: fieldUsages,
}
}
// node metadata, e.g. id allocator
nodeMetaBytes, err := directoryUsage(holderPath, false)
if err != nil {
return indexUsage, 0, errors.Wrapf(err, "getting disk usage for node metadata")
}
return indexUsage, nodeMetaBytes, nil
}
// fieldUsage computes the sum of filesizes used by a field in
// the filesystem tree (roaring storage), broken down by keys and fragments.
func (f *TxFactory) fieldUsage(indexPath string, fld *Field) (FieldUsage, error) {
fieldUsage := FieldUsage{}
field := fld.name
// row keys
keysBytes := int64(0)
var err error
keysBytes, err = fileSize(fld.TranslateStorePath())
if err != nil {
// if file doesn't exist, size = 0
keysBytes = 0
}
// field metadata
fieldPath := path.Join(indexPath, FieldsDir, field)
metaBytes, err := directoryUsage(fieldPath, false) // this includes keys
if err != nil {
return fieldUsage, errors.Wrapf(err, "getting disk usage for field meta (%s)", field)
}
// fragment data
viewsPath := path.Join(fieldPath, "views")
fragmentBytes := uint64(0)
if dirExists(viewsPath) {
fragmentBytes, err = directoryUsage(viewsPath, true)
if err != nil {
return fieldUsage, errors.Wrapf(err, "getting disk usage for field fragments (%s)", field)
}
}
fieldUsage = FieldUsage{
Total: metaBytes + fragmentBytes, // metaBytes includes keys
Metadata: metaBytes - uint64(keysBytes),
Fragments: fragmentBytes,
Keys: uint64(keysBytes),
}
return fieldUsage, nil
}
// NOTE: Go 1.16 introduced a new Readdir() method that is supposed to be more performant.
// Not yet upgraded b/c new method is not compatible with older versions of Go.
func directoryUsage(fname string, recursive bool) (uint64, error) {
if !dirExists(fname) {
return 0, errors.Errorf("directory does not exist (%s)", fname)
}
var size uint64
dir, err := os.Open(fname)
if err != nil {
return 0, errors.Wrap(err, "opening data subdirectory")
}
defer dir.Close()
files, err := dir.Readdir(-1)
if err != nil {
return 0, errors.Wrap(err, "reading data subdirectory")
}
for _, file := range files {
if recursive && file.IsDir() {
sz, err := directoryUsage(path.Join(fname, file.Name()), true)
if err != nil {
return 0, err
}
size += sz
} else {
size += uint64(file.Size()) // NOTE this cast is safe for regular files, not necessarily others
}
}
return size, nil
}
// CloseIndex is a no-op. This seems to be in place for debugging purposes.
func (f *TxFactory) CloseIndex(idx *Index) error {
return nil

17
util.go
View file

@ -4,8 +4,11 @@ package pilosa
// util.go: a place for generic, reusable utilities.
import (
"fmt"
"reflect"
"time"
"github.com/shirou/gopsutil/v3/mem"
)
// LeftShifted16MaxContainerKey is 0xffffffffffff0000. It is similar
@ -54,3 +57,17 @@ func GetLoopProgress(start time.Time, now time.Time, iteration uint, total uint)
func FormatTimestampNano(value, base int64, timeUnit string) string {
return time.Unix(0, (value+base)*TimeUnitNanos(timeUnit)).UTC().Format(time.RFC3339Nano)
}
type MemoryUsage struct {
Capacity uint64 `json:"capacity"`
TotalUse uint64 `json:"totalUsed"`
}
// GetMemoryUsage gets the memory usage
func GetMemoryUsage() (MemoryUsage, error) {
usage, err := mem.VirtualMemory()
if usage == nil || err != nil {
return MemoryUsage{}, fmt.Errorf("reading virtual memory: %v", err)
}
return MemoryUsage{Capacity: usage.Total, TotalUse: usage.Used}, nil
}

View file

@ -90,3 +90,9 @@ func TestFormatTimestampNano(t *testing.T) {
t.Fatal("Timestamp not formatted properly")
}
}
func TestGetMemoryUsage(t *testing.T) {
if _, err := GetMemoryUsage(); err != nil {
t.Fatalf("unexpected error getting memory usage: %v", err)
}
}