scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-07 19:37:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'master' into sup112

Browse source
This commit is contained in:
reesporte 2021-12-06 10:43:52 -06:00
commit 4c9029e0b6
39 changed files with 873 additions and 430 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
auth/auth.go Normal file
View file

@ -0,0 +1,38 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package auth
type Auth struct {
// Enable AuthZ/AuthN for featurebase server
Enable bool `toml:"enable"`
// Application/Client ID
ClientId string `toml:"client-id"`
// Client Secret
ClientSecret string `toml:"client-secret"`
// Authorize URL
AuthorizeURL string `toml:"authorize-url"`
// Token URL
TokenURL string `toml:"token-url"`
// Group Endpoint URL
GroupEndpointURL string `toml:"group-endpoint-url"`
// Scope URL
ScopeURL string `toml:"scope-url"`
}

74
catcher.go
View file

@ -17,7 +17,7 @@ package pilosa
import (
"github.com/molecula/featurebase/v2/roaring"
txkey "github.com/molecula/featurebase/v2/short_txkey"
. "github.com/molecula/featurebase/v2/vprint"
"github.com/molecula/featurebase/v2/vprint"
)
// catcher is useful to report error locations with a
@ -46,8 +46,8 @@ func (c *catcherTx) NewTxIterator(index, field, view string, shard uint64) *roar
func (c *catcherTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) {
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see ImportRoaringBits() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see ImportRoaringBits() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.ImportRoaringBits(index, field, view, shard, rit, clear, log, rowSize)
@ -56,8 +56,8 @@ func (c *catcherTx) ImportRoaringBits(index, field, view string, shard uint64, r
func (c *catcherTx) Rollback() {
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Rollback() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Rollback() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
c.b.Rollback()
@ -67,8 +67,8 @@ func (c *catcherTx) Commit() error {
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Commit() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Commit() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Commit()
@ -78,8 +78,8 @@ func (c *catcherTx) RoaringBitmap(index, field, view string, shard uint64) (*roa
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see RoaringBitmap() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see RoaringBitmap() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.RoaringBitmap(index, field, view, shard)
@ -89,8 +89,8 @@ func (c *catcherTx) Container(index, field, view string, shard uint64, key uint6
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Container() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Container() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Container(index, field, view, shard, key)
@ -100,8 +100,8 @@ func (c *catcherTx) PutContainer(index, field, view string, shard uint64, key ui
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see PutContainer() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see PutContainer() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.PutContainer(index, field, view, shard, key, rc)
@ -111,8 +111,8 @@ func (c *catcherTx) RemoveContainer(index, field, view string, shard uint64, key
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see RemoveContainer() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see RemoveContainer() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.RemoveContainer(index, field, view, shard, key)
@ -122,8 +122,8 @@ func (c *catcherTx) Add(index, field, view string, shard uint64, a ...uint64) (c
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Add() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Add() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Add(index, field, view, shard, a...)
@ -133,8 +133,8 @@ func (c *catcherTx) Remove(index, field, view string, shard uint64, a ...uint64)
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Remove() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Remove() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Remove(index, field, view, shard, a...)
@ -144,8 +144,8 @@ func (c *catcherTx) Contains(index, field, view string, shard uint64, key uint64
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Contains() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Contains() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Contains(index, field, view, shard, key)
@ -155,8 +155,8 @@ func (c *catcherTx) ContainerIterator(index, field, view string, shard uint64, f
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see ContainerIterator() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see ContainerIterator() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.ContainerIterator(index, field, view, shard, firstRoaringContainerKey)
@ -166,8 +166,8 @@ func (c *catcherTx) ForEach(index, field, view string, shard uint64, fn func(i u
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see ForEach() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see ForEach() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.ForEach(index, field, view, shard, fn)
@ -177,8 +177,8 @@ func (c *catcherTx) ForEachRange(index, field, view string, shard uint64, start,
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see ForEachRange() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see ForEachRange() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.ForEachRange(index, field, view, shard, start, end, fn)
@ -188,8 +188,8 @@ func (c *catcherTx) Count(index, field, view string, shard uint64) (uint64, erro
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Count() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Count() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Count(index, field, view, shard)
@ -199,8 +199,8 @@ func (c *catcherTx) Max(index, field, view string, shard uint64) (uint64, error)
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Max() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Max() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Max(index, field, view, shard)
@ -210,8 +210,8 @@ func (c *catcherTx) Min(index, field, view string, shard uint64) (uint64, bool,
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Min() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Min() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Min(index, field, view, shard)
@ -221,8 +221,8 @@ func (c *catcherTx) CountRange(index, field, view string, shard uint64, start, e
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see CountRange() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see CountRange() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.CountRange(index, field, view, shard, start, end)
@ -232,8 +232,8 @@ func (c *catcherTx) OffsetRange(index, field, view string, shard, offset, start,
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see OffsetRange() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see OffsetRange() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.OffsetRange(index, field, view, shard, offset, start, end)

1
client/client_it_test.go
View file

@ -592,7 +592,6 @@ func TestClientAgainstCluster(t *testing.T) {
target := []uint64{100}
require.Equalf(t, target, resp.Result().Row().Columns, "Row Result Columns")
})
t.Run("StoreQuery", func(t *testing.T) {
schema := NewSchema()
testIndexStore := schema.Index("test-index-store")

14
client/csv/csv.go
View file

@ -56,7 +56,7 @@ func ColumnUnmarshallerWithTimestamp(format Format, timestampFormat string) Reco
column := client.Column{}
parts := strings.Split(text, ",")
if len(parts) < 2 {
return nil, errors.New("Invalid CSV line")
return nil, errors.New("invalid CSV line")
}
hasRowKey := format == RowKeyColumnID || format == RowKeyColumnKey
@ -67,7 +67,7 @@ func ColumnUnmarshallerWithTimestamp(format Format, timestampFormat string) Reco
} else {
column.RowID, err = strconv.ParseUint(parts[0], 10, 64)
if err != nil {
return nil, errors.New("Invalid row ID")
return nil, errors.New("invalid row ID")
}
}
@ -76,7 +76,7 @@ func ColumnUnmarshallerWithTimestamp(format Format, timestampFormat string) Reco
} else {
column.ColumnID, err = strconv.ParseUint(parts[1], 10, 64)
if err != nil {
return nil, errors.New("Invalid column ID")
return nil, errors.New("invalid column ID")
}
}
@ -166,17 +166,17 @@ func FieldValueUnmarshaller(format Format) RecordUnmarshaller {
return func(text string) (client.Record, error) {
parts := strings.Split(text, ",")
if len(parts) < 2 {
return nil, errors.New("Invalid CSV")
return nil, errors.New("invalid CSV")
}
value, err := strconv.ParseInt(parts[1], 10, 64)
if err != nil {
return nil, errors.New("Invalid value")
return nil, errors.New("invalid value")
}
switch format {
case ColumnID:
columnID, err := strconv.ParseUint(parts[0], 10, 64)
if err != nil {
return nil, errors.New("Invalid column ID at line: %d")
return nil, errors.New("invalid column ID at line: %d")
}
return client.FieldValue{
ColumnID: uint64(columnID),
@ -188,7 +188,7 @@ func FieldValueUnmarshaller(format Format) RecordUnmarshaller {
Value: value,
}, nil
default:
return nil, fmt.Errorf("Invalid format: %d", format)
return nil, fmt.Errorf("invalid format: %d", format)
}
}
}

24
cmd/badloader/badloader.go
View file

@ -20,16 +20,15 @@ import (
"context"
"time"
//"fmt"
"fmt"
"io"
"io/ioutil"
gohttp "net/http"
"github.com/molecula/featurebase/v2"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/http"
pnet "github.com/molecula/featurebase/v2/net"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
"os"
"strconv"
@ -38,7 +37,6 @@ import (
func UploadTar(srcFile string, client *http.InternalClient) error {
t0 := time.Now()
f, err := os.Open(srcFile)
if err != nil {
return (err)
@ -65,7 +63,7 @@ func UploadTar(srcFile string, client *http.InternalClient) error {
header, err := tarReader.Next()
if err == io.EOF {
if header != nil {
PanicOn("header should not be nil on err io.EOF")
vprint.PanicOn("header should not be nil on err io.EOF")
}
//submit any stuff we have left
if len(viewData) > 0 {
@ -75,13 +73,13 @@ func UploadTar(srcFile string, client *http.InternalClient) error {
// Submit(lastIndex, lastField, lastShard, request)
uri := GetImportRoaringURI(lastIndex, lastShard)
err := client.ImportRoaring(context.Background(), uri, lastIndex, lastField, lastShard, false, request)
PanicOn(err)
vprint.PanicOn(err)
}
return nil
}
n++
if n%500 == 0 {
VV("n = %v, progress, elapsed '%v'", n, time.Since(t0))
vprint.VV("n = %v, progress, elapsed '%v'", n, time.Since(t0))
}
parts := strings.Split(header.Name, "/")
//vv("parts = '%#v'", parts)
@ -100,7 +98,7 @@ func UploadTar(srcFile string, client *http.InternalClient) error {
}
//vv("about to submit lastIndex='%v' lastShard='%v'", lastIndex, lastShard)
uri := GetImportRoaringURI(lastIndex, lastShard)
PanicOn(client.ImportRoaring(context.Background(), uri, lastIndex, lastField, lastShard, false, request))
vprint.PanicOn(client.ImportRoaring(context.Background(), uri, lastIndex, lastField, lastShard, false, request))
viewData = make(map[string][]byte)
//vv("done with submit lastIndex='%v' lastShard='%v'; took='%v'", lastIndex, lastShard, time.Since(t0))
@ -111,7 +109,7 @@ func UploadTar(srcFile string, client *http.InternalClient) error {
return err
}
if _, already := viewData[view]; already {
PanicOn(fmt.Sprintf("view '%v' already present!", view))
vprint.PanicOn(fmt.Sprintf("view '%v' already present!", view))
}
viewData[view] = roaringData
lastIndex = index
@ -130,12 +128,12 @@ func main() {
host := "127.0.0.1:10101"
h := &gohttp.Client{}
c, err := http.NewInternalClient(host, h)
PanicOn(err)
vprint.PanicOn(err)
tarSrcPath := "q2.tar.gz"
t0 := time.Now()
PanicOn(UploadTar(tarSrcPath, c))
VV("total elapsed '%v'", time.Since(t0))
vprint.PanicOn(UploadTar(tarSrcPath, c))
vprint.VV("total elapsed '%v'", time.Since(t0))
}
var globURI *pnet.URI
@ -143,7 +141,7 @@ var globURI *pnet.URI
func init() {
var err error
globURI, err = pnet.NewURIFromHostPort("127.0.0.1", 10101)
PanicOn(err)
vprint.PanicOn(err)
}
// get correct node to go to.

7
cmd/random-query/main.go
View file

@ -35,7 +35,6 @@ import (
"github.com/molecula/featurebase/v2/pb"
"github.com/molecula/featurebase/v2/pql"
"github.com/molecula/featurebase/v2/vprint"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/pkg/errors"
vegeta "github.com/tsenart/vegeta/v12/lib"
)
@ -332,7 +331,7 @@ func (cfg *RandomQueryConfig) Setup(api API) (err error) {
pql := fmt.Sprintf("Rows(%v)", fld.Name)
res, err := api.Query(ctx, ii.Name, &pilosa.QueryRequest{Index: ii.Name, Query: pql})
PanicOn(err)
vprint.PanicOn(err)
switch x := res.Results[0].(type) {
case *pilosa.RowIdentifiers:
cfg.AddResponse(ii.Name, fld.Name, x, fld.Options.Type == "time", fld.Options.Type == "set")
@ -345,7 +344,7 @@ func (cfg *RandomQueryConfig) Setup(api API) (err error) {
case "decimal":
cfg.AddIntField(ii.Name, fld.Name, fld.Options.Min, fld.Options.Max, fld.Options.Scale, fld.Options.Type == "decimal")
default:
AlwaysPrintf("ignoring field %q: unhandled type %q\n", fld.Name, fld.Options.Type)
vprint.AlwaysPrintf("ignoring field %q: unhandled type %q\n", fld.Name, fld.Options.Type)
}
}
}
@ -495,7 +494,7 @@ func (cfg *RandomQueryConfig) GenQuery(index string) (pql string, err error) {
key = fmt.Sprintf(`%v="%v"`, fr.Field, fr.RowKey)
c := strings.LastIndex(fr.RowKey, "_")
n, err := strconv.Atoi(fr.RowKey[c+1:])
PanicOn(err)
vprint.PanicOn(err)
n += 1
fr.RowKey = fmt.Sprintf("%v%v", fr.RowKey[:c+1], n)
} else {

4
cmd/restore.go
View file

@ -35,8 +35,8 @@ The Restore command will take a backup archive and restore it to a new, clean cl
},
}
flags := restoreCmd.Flags()
flags.StringVarP(&cmd.Path, "source", "s", "", "pilosa backup file; specify '-' to restore from stdin tar stream")
flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of Pilosa.")
flags.StringVarP(&cmd.Path, "source", "s", "", "backup file; specify '-' to restore from stdin tar stream")
flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.")
flags.IntVar(&cmd.Concurrency, "concurrency", 1, "number of concurrent uploads")
ctl.SetTLSConfig(
flags, "",

30
cmd/slurp/slurp.go
View file

@ -30,10 +30,10 @@ import (
"strings"
"time"
"github.com/molecula/featurebase/v2"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/http"
pnet "github.com/molecula/featurebase/v2/net"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
)
// slurp: slurp is a load-tester for importing bulk data.
@ -60,7 +60,7 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
field := parts[2]
view := parts[4]
shard, err := strconv.ParseUint(parts[6], 10, 64)
PanicOn(err)
vprint.PanicOn(err)
if index != r.lastIndex || field != r.lastField || shard != r.lastShard {
err := r.Upload()
if err != nil {
@ -84,7 +84,7 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
if err != nil {
return err
}
VV("Finished import %v", time.Since(r.start))
vprint.VV("Finished import %v", time.Since(r.start))
if r.profile != "" {
stopProfile(r.host, r.profile)
@ -104,21 +104,21 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error {
}
byteData, err := ioutil.ReadAll(tr)
PanicOn(err)
vprint.PanicOn(err)
br := bytes.NewReader(byteData)
err = r.client.ImportFieldKeys(context.Background(), uri, index, fieldName, false, br)
if err != nil {
return err
}
default:
VV("%v", h.Name)
vprint.VV("%v", h.Name)
index := parts[1]
partition, err := strconv.ParseUint(v, 10, 64)
if err != nil {
return err
}
byteData, err := ioutil.ReadAll(tr)
PanicOn(err)
vprint.PanicOn(err)
br := bytes.NewReader(byteData)
err = r.client.ImportIndexKeys(context.Background(), uri, index, int(partition), false, br)
@ -176,10 +176,10 @@ func UploadTar(srcFile string, client *http.InternalClient, profile, host string
break
}
if err != nil {
PanicOn(err)
vprint.PanicOn(err)
}
err = runner.NewHeader(header, tarReader)
PanicOn(err)
vprint.PanicOn(err)
}
return nil
}
@ -194,7 +194,7 @@ func main() {
flag.Parse()
uri, err := pnet.NewURIFromAddress(host)
PanicOn(err)
vprint.PanicOn(err)
globURI = uri
h := &gohttp.Client{}
@ -202,12 +202,12 @@ func main() {
startProfile(host)
}
c, err := http.NewInternalClient(host, h)
PanicOn(err)
vprint.PanicOn(err)
t0 := time.Now()
println("uploading", tarSrcPath)
PanicOn(UploadTar(tarSrcPath, c, profile, host))
VV("total elapsed '%v'", time.Since(t0))
vprint.PanicOn(UploadTar(tarSrcPath, c, profile, host))
vprint.VV("total elapsed '%v'", time.Since(t0))
}
func startProfile(host string) {
@ -248,10 +248,10 @@ func stopProfile(host, outfile string) {
}
fd, err := os.Create(outfile)
PanicOn(err)
vprint.PanicOn(err)
defer fd.Close()
_, err = io.Copy(fd, resp.Body)
PanicOn(err)
vprint.PanicOn(err)
}

2
ctl/backup.go
View file

@ -103,7 +103,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
}
}
if len(indexes) <= 0 {
return fmt.Errorf("Index not found to back up")
return fmt.Errorf("index not found to back up")
}
}

2
ctl/restore.go
View file

@ -153,7 +153,7 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.
//NOTE SHOULD ONLY BE ONE
for _, index := range schema.Indexes {
if exists(index.Name) {
return fmt.Errorf("Index Exists %v", index.Name)
return fmt.Errorf("index Exists %v", index.Name)
}
logger.Printf("Create INDEX %v", index.Name)
err = cmd.client.CreateIndex(ctx, index.Name, index.Options)

10
ctl/server.go
View file

@ -121,4 +121,14 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
// 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.")
flags.StringVar(&srv.Config.Auth.ClientSecret, "auth.client-secret", srv.Config.Auth.ClientSecret, "Identity Provider's Client Secret.")
flags.StringVar(&srv.Config.Auth.AuthorizeURL, "auth.authorize-url", srv.Config.Auth.AuthorizeURL, "Identity Provider's Authorize URL.")
flags.StringVar(&srv.Config.Auth.TokenURL, "auth.token-url", srv.Config.Auth.TokenURL, "Identity Provider's Token URL.")
flags.StringVar(&srv.Config.Auth.GroupEndpointURL, "auth.group-endpoint-url", srv.Config.Auth.GroupEndpointURL, "Identity Provider's Group endpoint URL.")
flags.StringVar(&srv.Config.Auth.ScopeURL, "auth.scope-url", srv.Config.Auth.ScopeURL, "Identity Provider's Scope URL.")
}

16
dbshard.go
View file

@ -28,7 +28,7 @@ import (
"github.com/molecula/featurebase/v2/storage"
"github.com/pkg/errors"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
)
var _ = sort.Sort
@ -279,7 +279,7 @@ func (per *DBPerShard) LoadExistingDBs() (err error) {
func (txf *TxFactory) NewDBPerShard(typ txtype, holderDir string, holder *Holder) (d *DBPerShard) {
if holder.cfg == nil || holder.cfg.RBFConfig == nil || holder.cfg.StorageConfig == nil {
PanicOn("must have holder.cfg.RBFConfig and holder.cfg.StorageConfig set here")
vprint.PanicOn("must have holder.cfg.RBFConfig and holder.cfg.StorageConfig set here")
}
hasRoaring := false
@ -422,7 +422,7 @@ func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *In
if dbs != nil && dbs.closed {
// roaring txn are nil/fake anyway. Don't freak out.
if per.typ != roaringTxn {
PanicOn(fmt.Sprintf("cannot retain closed dbs across holder ReOpen dbs='%p'; per.typ='%v'", dbs, per.typ))
vprint.PanicOn(fmt.Sprintf("cannot retain closed dbs across holder ReOpen dbs='%p'; per.typ='%v'", dbs, per.typ))
}
}
if !ok {
@ -449,11 +449,11 @@ func (per *DBPerShard) unprotectedGetDBShard(index string, shard uint64, idx *In
registry = globalRbfDBReg
registry.(*rbfDBRegistrar).SetRBFConfig(per.RBFConfig)
default:
PanicOn(fmt.Sprintf("unknown txtyp: '%v'", dbs.typ))
vprint.PanicOn(fmt.Sprintf("unknown txtyp: '%v'", dbs.typ))
}
path := dbs.pathForType(dbs.typ)
w, err := registry.OpenDBWrapper(path, DetectMemAccessPastTx, per.StorageConfig)
PanicOn(err)
vprint.PanicOn(err)
h := idx.Holder()
w.SetHolder(h)
dbs.Open = true
@ -470,7 +470,7 @@ func (per *DBPerShard) Close() (err error) {
for _, dbi := range per.dbh.Index {
for _, dbs := range dbi.Shard {
err = dbs.Close()
PanicOn(err)
vprint.PanicOn(err)
}
}
return
@ -546,7 +546,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
ignoreEmpty := false
includeRoot := true
dbf, err := listDirUnderDir(path, includeRoot, ignoreEmpty)
PanicOn(err)
vprint.PanicOn(err)
for _, nm := range dbf {
base := filepath.Base(nm)
@ -561,7 +561,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
// Parse filename into integer.
shard, err := strconv.ParseUint(base[lenOfShardPrefix:], 10, 64)
if err != nil {
PanicOn(err)
vprint.PanicOn(err)
continue
}

6
etcd/embed.go
View file

@ -555,7 +555,7 @@ func (e *Etcd) deleteNodeData(key []byte, revision int64) error {
e.knownNodes[peerID].resizeState = ""
e.nodeStatesDirty = true
default:
return fmt.Errorf("node watch: invalid prefix %q\n", prefix)
return fmt.Errorf("node watch: invalid prefix %q", prefix)
}
return nil
}
@ -586,7 +586,7 @@ func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error)
var newNode topology.Node
err := json.Unmarshal(value, &newNode)
if err != nil {
return fmt.Errorf("json unmarshal of node metadata: %v\n", err)
return fmt.Errorf("json unmarshal of node metadata: %v", err)
}
e.knownNodes[peerID].topologyNode = &newNode
// This saves us one remake of the node later, probably.
@ -599,7 +599,7 @@ func (e *Etcd) putNodeData(key []byte, value []byte, revision int64) (err error)
e.knownNodes[peerID].resizeState = string(value)
e.nodeStatesDirty = true
default:
return fmt.Errorf("node watch: invalid prefix %q\n", prefix)
return fmt.Errorf("node watch: invalid prefix %q", prefix)
}
return nil
}

2
executor.go
View file

@ -2098,7 +2098,7 @@ func (e *executor) executeTopKShard(ctx context.Context, qcx *Qcx, index string,
return e.executeTopKShardTime(ctx, tx, filterBitmap, index, fieldName, shard, fromTime, toTime)
}
fallthrough
case FieldTypeSet:
case FieldTypeSet, FieldTypeMutex:
return e.executeTopKShardSet(ctx, tx, filterBitmap, index, fieldName, shard)
default:
return nil, errors.Errorf("field type %q is not yet supported by TopK", ftype)

69
executor_test.go
View file

@ -1688,34 +1688,63 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
}
func TestExecutor_Execute_TopK_Set(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
// Load some test data into a set field.
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, "f")
c.ImportBits(t, "i", "f", [][2]uint64{
func TestExecutor_ExecuteTopK(t *testing.T) {
baseBits := [][2]uint64{
{0, 0},
{0, 1},
{0, ShardWidth + 2},
{10, 2},
{10, ShardWidth},
{10, 2 * ShardWidth},
{10, ShardWidth + 1},
{20, ShardWidth},
})
// Execute query.
if result, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `TopK(f, k=2)`}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result.Results, []interface{}{&pilosa.PairsField{
Pairs: []pilosa.Pair{
{ID: 10, Count: 4},
{ID: 0, Count: 3},
}
tests := []struct {
fieldName string
fieldOptions []pilosa.FieldOption
bits [][2]uint64
query string
result []pilosa.Pair
}{
{
fieldName: "f",
bits: append(baseBits, [2]uint64{0, 1}),
query: "TopK(f, k=2)",
result: []pilosa.Pair{
{ID: 10, Count: 4},
{ID: 0, Count: 3},
},
},
Field: "f",
}}) {
t.Fatalf("unexpected result: %s", spew.Sdump(result))
{
fieldName: "fmutex",
fieldOptions: []pilosa.FieldOption{pilosa.OptFieldTypeMutex(pilosa.CacheTypeRanked, 10)},
bits: baseBits,
query: "TopK(f, k=2)",
result: []pilosa.Pair{
{ID: 10, Count: 3},
{ID: 0, Count: 2},
},
},
}
c := test.MustRunCluster(t, 3)
defer c.Close()
for _, tst := range tests {
t.Run(tst.fieldName, func(t *testing.T) {
pilosa.OptFieldTypeMutex(pilosa.CacheTypeRanked, 10)
c.CreateField(t, "i", pilosa.IndexOptions{TrackExistence: true}, tst.fieldName)
c.ImportBits(t, "i", tst.fieldName, tst.bits)
if result, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: tst.query}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result.Results, []interface{}{&pilosa.PairsField{
Pairs: []pilosa.Pair{
{ID: 10, Count: 4},
{ID: 0, Count: 3},
},
Field: "f",
}}) {
t.Fatalf("unexpected result: %s", spew.Sdump(result))
}
})
}
}

12
fragment.go
View file

@ -51,7 +51,7 @@ import (
"github.com/molecula/featurebase/v2/testhook"
"github.com/molecula/featurebase/v2/topology"
"github.com/molecula/featurebase/v2/tracing"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
"github.com/pkg/errors"
)
@ -204,7 +204,7 @@ func newFragment(holder *Holder, spec fragSpec, shard uint64, flags byte) *fragm
idx := holder.Index(spec.index.name)
if idx == nil {
PanicOn(fmt.Sprintf("got nil idx back for '%v' from holder!", spec.index))
vprint.PanicOn(fmt.Sprintf("got nil idx back for '%v' from holder!", spec.index))
}
f := &fragment{
@ -615,7 +615,7 @@ func (f *fragment) row(tx Tx, rowID uint64) (*Row, error) {
func (f *fragment) mustRow(tx Tx, rowID uint64) *Row {
row, err := f.row(tx, rowID)
if err != nil {
PanicOn(err)
vprint.PanicOn(err)
}
return row
}
@ -1072,7 +1072,7 @@ func (f *fragment) setValueBase(txOrig Tx, columnID uint64, bitDepth uint64, val
tx = f.idx.holder.txf.NewTx(Txo{Write: writable, Index: f.idx, Fragment: f, Shard: f.shard})
defer func() {
if err == nil {
PanicOn(tx.Commit())
vprint.PanicOn(tx.Commit())
} else {
tx.Rollback()
}
@ -1975,7 +1975,7 @@ func (f *fragment) Blocks() ([]FragmentBlock, error) {
idx := f.holder.Index(f.index())
if idx == nil {
err := fmt.Errorf("index() was nil in fragment.Blocks(): f.index()='%v'", f.index())
PanicOn(err)
vprint.PanicOn(err)
return nil, err
}
tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard})
@ -2350,7 +2350,7 @@ func (p *parallelSlices) fullPrune() {
return
}
if len(p.rows) != len(p.cols) {
PanicOn("parallelSlices must have same length for rows and columns")
vprint.PanicOn("parallelSlices must have same length for rows and columns")
}
unsorted := p.prune()
if unsorted {

4
holder.go
View file

@ -36,7 +36,7 @@ import (
"github.com/molecula/featurebase/v2/storage"
"github.com/molecula/featurebase/v2/testhook"
"github.com/molecula/featurebase/v2/topology"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -300,7 +300,7 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
storage.SetRowCacheOn(cfg.RowcacheOn)
txf, err := NewTxFactory(cfg.StorageConfig.Backend, h.IndexesPath(), h)
PanicOn(err)
vprint.PanicOn(err)
h.txf = txf
_ = testhook.Created(h.Auditor, h, nil)

3
http/client.go
View file

@ -667,6 +667,9 @@ func (c *InternalClient) importHelper(ctx context.Context, req pilosa.Message, p
// request over the wire, even though we still have to go through
// the http interface.
nodes, err = c.Nodes(ctx)
if err != nil {
return errors.Wrap(err, "getting nodes")
}
}
// "us" is a usable local node if any, "them" is every node that we need

214
http/handler.go
View file

@ -3091,7 +3091,7 @@ func (h *Handler) handlePostTranslateIndexDB(w http.ResponseWriter, r *http.Requ
resp.write(w, err)
}
func (h *Handler) handleFindIndexKeys(w http.ResponseWriter, r *http.Request) {
func (h *Handler) handleFindOrCreateKeys(w http.ResponseWriter, r *http.Request, requireField bool, create bool) {
// Verify input and output types
if r.Header.Get("Content-Type") != "application/json" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
@ -3101,178 +3101,76 @@ func (h *Handler) handleFindIndexKeys(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName, ok := mux.Vars(r)["index"]
if !ok {
http.Error(w, "index name is required", http.StatusBadRequest)
return
}
bd, err := readBody(r)
if err != nil {
http.Error(w, "failed to read body", http.StatusBadRequest)
return
}
var indexName, fieldName string
var keys []string
err = json.Unmarshal(bd, &keys)
if err != nil {
http.Error(w, "failed to decode request", http.StatusBadRequest)
return
}
err := func() error {
var ok bool
indexName, ok = mux.Vars(r)["index"]
if !ok {
return errors.New("index name is required")
}
translations, err := h.api.FindIndexKeys(r.Context(), indexName, keys...)
if err != nil {
http.Error(w, "translating keys", http.StatusBadRequest)
return
}
if requireField {
fieldName, ok = mux.Vars(r)["field"]
if !ok {
return errors.New("field name is required")
}
}
err = json.NewEncoder(w).Encode(translations)
bd, err := readBody(r)
if err != nil {
return fmt.Errorf("failed to read body: %v", err)
}
err = json.Unmarshal(bd, &keys)
if err != nil {
return fmt.Errorf("failed to decode request: %v", err)
}
return nil
}()
if err != nil {
http.Error(w, "encoding result", http.StatusBadRequest)
http.Error(w, err.Error(), http.StatusBadRequest)
}
var translations map[string]uint64
switch {
case requireField && create:
translations, err = h.api.CreateFieldKeys(r.Context(), indexName, fieldName, keys...)
case requireField && !create:
translations, err = h.api.FindFieldKeys(r.Context(), indexName, fieldName, keys...)
case !requireField && create:
translations, err = h.api.CreateIndexKeys(r.Context(), indexName, keys...)
case !requireField && !create:
translations, err = h.api.FindIndexKeys(r.Context(), indexName, keys...)
}
if err != nil {
http.Error(w, fmt.Sprintf("translating keys: %v", err), http.StatusInternalServerError)
return
}
data, err := json.Marshal(translations)
if err != nil {
http.Error(w, fmt.Sprintf("encoding response: %v", err), http.StatusInternalServerError)
}
_, err = w.Write(data)
if err != nil {
h.logger.Printf("writing CreateFieldKeys response: %v", err)
}
}
func (h *Handler) handleFindIndexKeys(w http.ResponseWriter, r *http.Request) {
h.handleFindOrCreateKeys(w, r, false, false)
}
func (h *Handler) handleFindFieldKeys(w http.ResponseWriter, r *http.Request) {
// Verify input and output types
if r.Header.Get("Content-Type") != "application/json" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
}
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName, ok := mux.Vars(r)["index"]
if !ok {
http.Error(w, "index name is required", http.StatusBadRequest)
return
}
fieldName, ok := mux.Vars(r)["field"]
if !ok {
http.Error(w, "field name is required", http.StatusBadRequest)
return
}
bd, err := readBody(r)
if err != nil {
http.Error(w, "failed to read body", http.StatusBadRequest)
return
}
var keys []string
err = json.Unmarshal(bd, &keys)
if err != nil {
http.Error(w, "failed to decode request", http.StatusBadRequest)
return
}
translations, err := h.api.FindFieldKeys(r.Context(), indexName, fieldName, keys...)
if err != nil {
http.Error(w, "translating keys", http.StatusBadRequest)
return
}
err = json.NewEncoder(w).Encode(translations)
if err != nil {
http.Error(w, "encoding result", http.StatusBadRequest)
return
}
h.handleFindOrCreateKeys(w, r, true, false)
}
func (h *Handler) handleCreateIndexKeys(w http.ResponseWriter, r *http.Request) {
// Verify input and output types
if r.Header.Get("Content-Type") != "application/json" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
}
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName, ok := mux.Vars(r)["index"]
if !ok {
http.Error(w, "index name is required", http.StatusBadRequest)
return
}
bd, err := readBody(r)
if err != nil {
http.Error(w, "failed to read body", http.StatusBadRequest)
return
}
var keys []string
err = json.Unmarshal(bd, &keys)
if err != nil {
http.Error(w, "failed to decode request", http.StatusBadRequest)
return
}
translations, err := h.api.CreateIndexKeys(r.Context(), indexName, keys...)
if err != nil {
http.Error(w, "translating keys", http.StatusBadRequest)
return
}
err = json.NewEncoder(w).Encode(translations)
if err != nil {
http.Error(w, "encoding result", http.StatusBadRequest)
return
}
h.handleFindOrCreateKeys(w, r, false, true)
}
func (h *Handler) handleCreateFieldKeys(w http.ResponseWriter, r *http.Request) {
// Verify input and output types
if r.Header.Get("Content-Type") != "application/json" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
}
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName, ok := mux.Vars(r)["index"]
if !ok {
http.Error(w, "index name is required", http.StatusBadRequest)
return
}
fieldName, ok := mux.Vars(r)["field"]
if !ok {
http.Error(w, "field name is required", http.StatusBadRequest)
return
}
bd, err := readBody(r)
if err != nil {
http.Error(w, "failed to read body", http.StatusBadRequest)
return
}
var keys []string
err = json.Unmarshal(bd, &keys)
if err != nil {
http.Error(w, "failed to decode request", http.StatusBadRequest)
return
}
translations, err := h.api.CreateFieldKeys(r.Context(), indexName, fieldName, keys...)
if err != nil {
http.Error(w, "translating keys", http.StatusBadRequest)
return
}
err = json.NewEncoder(w).Encode(translations)
if err != nil {
http.Error(w, "encoding result", http.StatusBadRequest)
return
}
h.handleFindOrCreateKeys(w, r, true, true)
}
func (h *Handler) handleMatchField(w http.ResponseWriter, r *http.Request) {

112
http/handler_test.go
View file

@ -19,6 +19,7 @@ import (
"fmt"
"net"
gohttp "net/http"
"strings"
"testing"
pilosa "github.com/molecula/featurebase/v2"
@ -167,6 +168,115 @@ func TestIngestSchemaHandler(t *testing.T) {
schemaURL := fmt.Sprintf("%s/internal/schema", m.URL())
resp := test.Do(t, "POST", schemaURL, string(schema))
if resp.StatusCode != gohttp.StatusOK {
t.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
t.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
// now, try again, expecting a failure:
resp = test.Do(t, "POST", schemaURL, string(schema))
if resp.StatusCode != gohttp.StatusConflict {
t.Errorf("invalid status: expected 409, got %d, body=%s", resp.StatusCode, resp.Body)
}
}
func TestTranslationHandlers(t *testing.T) {
// reusable data for the tests
nameBytes, err := json.Marshal([]string{"a", "b", "c"})
if err != nil {
t.Fatalf("marshalling json: %v", err)
}
names := string(nameBytes)
c := test.MustRunCluster(t, 1)
defer c.Close()
schema := `
{
"index-name": "example",
"primary-key-type": "string",
"index-action": "create",
"fields": [
{
"field-name": "stringset",
"field-type": "string",
"field-options": {
"cache-type": "ranked",
"cache-size": 100000
}
}
]
}
`
m := c.GetPrimary()
schemaURL := fmt.Sprintf("%s/internal/schema", m.URL())
resp := test.Do(t, "POST", schemaURL, string(schema))
if resp.StatusCode != gohttp.StatusOK {
t.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
baseURLs := []string{
fmt.Sprintf("%s/internal/translate/index/example/", m.URL()),
fmt.Sprintf("%s/internal/translate/field/example/stringset/", m.URL()),
fmt.Sprintf("%s/internal/translate/field/example/nonexistent/", m.URL()),
}
for _, url := range baseURLs {
expectFailure := strings.HasSuffix(url, "/nonexistent/")
createURL := url + "keys/create"
findURL := url + "keys/find"
var results map[string]uint64
if expectFailure {
resp := test.Do(t, "POST", findURL, names)
if resp.StatusCode != gohttp.StatusInternalServerError {
t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
resp = test.Do(t, "POST", createURL, names)
if resp.StatusCode != gohttp.StatusInternalServerError {
t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
continue
}
// try to find them when they don't exist
resp := test.Do(t, "POST", findURL, names)
if resp.StatusCode != gohttp.StatusOK {
t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
err := json.Unmarshal([]byte(resp.Body), &results)
if err != nil {
t.Fatalf("unmarshalling result: %v", err)
}
if len(results) != 0 {
t.Fatalf("finding keys before any were set: expected no results, got %d (%q)", len(results), results)
}
// try to create them, but malformed, so we expect an error
resp = test.Do(t, "POST", createURL, names[:6])
if resp.StatusCode != gohttp.StatusBadRequest {
t.Fatalf("invalid status: expected 400, got %d, body=%s", resp.StatusCode, resp.Body)
}
// try to create them
resp = test.Do(t, "POST", createURL, names)
if resp.StatusCode != gohttp.StatusOK {
t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
err = json.Unmarshal([]byte(resp.Body), &results)
if err != nil {
t.Fatalf("unmarshalling result: %v", err)
}
if len(results) != 3 {
t.Fatalf("finding keys before any were set: expected 3 results, got %d (%q)", len(results), results)
}
// try to find them now that they exist
resp = test.Do(t, "POST", findURL, names)
if resp.StatusCode != gohttp.StatusOK {
t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
}
err = json.Unmarshal([]byte(resp.Body), &results)
if err != nil {
t.Fatalf("unmarshalling result: %v", err)
}
if len(results) != 3 {
t.Fatalf("finding keys before any were set: expected 3 results, got %d (%q)", len(results), results)
}
}
}

10
install/featurebase.conf
View file

@ -371,3 +371,13 @@ log-path = "/var/log/molecula/featurebase.log"
# ==============================================================================
# Enable/Disable AuthN/AuthZ for featurebase
# Can choose identity provider, pass authorize and user-info endpoints, and client id
# [auth]
# enable = false
# client-id = ""
# client-secret = ""
# authorize-url = ""
# token-url = ""
# group-endpoint-url = ""
# scope-url = ""

7
internal/clustertests/pause_node_test.go
View file

@ -31,7 +31,6 @@ import (
boltdb "github.com/molecula/featurebase/v2/boltdb"
"github.com/molecula/featurebase/v2/disco"
"github.com/molecula/featurebase/v2/http"
picli "github.com/molecula/featurebase/v2/http"
"github.com/molecula/featurebase/v2/net"
"github.com/molecula/featurebase/v2/topology"
"github.com/pkg/errors"
@ -81,7 +80,7 @@ func getAddress(node string) string {
func getClients(addrs []string) ([]*http.InternalClient, error) {
clients := make([]*http.InternalClient, 0, len(addrs))
for _, addr := range addrs {
c, err := picli.NewInternalClient(addr, picli.GetHTTPClient(nil))
c, err := http.NewInternalClient(addr, http.GetHTTPClient(nil))
if err != nil {
return nil, err
}
@ -102,7 +101,7 @@ func getURIsFromAddresses(addrs []string) ([]*net.URI, error) {
return uris, nil
}
func readIndexTranslateData(ctx context.Context, client *picli.InternalClient, dirPath, index string, partition int) error {
func readIndexTranslateData(ctx context.Context, client *http.InternalClient, dirPath, index string, partition int) error {
// read translateStore contents from endpoint
r, err := client.IndexTranslateDataReader(ctx, index, partition)
if err != nil {
@ -186,7 +185,7 @@ var errOpRetriable = errors.New("If operation failed on this error, it can be re
func verifyNodeHasGivenKeys(ctx context.Context, node, index, dirPath string, keys []string) error {
// get client that's connected to node
address := getAddress(node)
client, err := picli.NewInternalClient(address, picli.GetHTTPClient(nil))
client, err := http.NewInternalClient(address, http.GetHTTPClient(nil))
if err != nil {
return err
}

22
lru/lru.go
View file

@ -82,16 +82,6 @@ func (c *Cache) Get(key Key) (value interface{}, ok bool) {
return nil, false
}
// remove removes the provided key from the cache.
func (c *Cache) remove(key Key) { // nolint: staticcheck,unused
if c.cache == nil {
return
}
if ele, hit := c.cache[key]; hit {
c.removeElement(ele)
}
}
// removeOldest removes the oldest item from the cache.
func (c *Cache) removeOldest() {
if c.cache == nil {
@ -119,15 +109,3 @@ func (c *Cache) Len() int {
}
return c.ll.Len()
}
// clear purges all stored items from the cache.
func (c *Cache) clear() { // nolint: staticcheck,unused
if c.OnEvicted != nil {
for _, e := range c.cache {
kv := e.Value.(*entry)
c.OnEvicted(kv.key, kv.value)
}
}
c.ll = nil
c.cache = nil
}

22
pprof.go
View file

@ -24,7 +24,7 @@ import (
_ "net/http/pprof" // Imported for its side-effect of registering pprof endpoints with the server.
"github.com/molecula/featurebase/v2/storage"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
)
// CPUProfileForDur (where "Dur" is short for "Duration"), is used for
@ -38,18 +38,18 @@ func CPUProfileForDur(dur time.Duration, outpath string) {
}
path := outpath + "." + backend
f, err := os.Create(path)
PanicOn(err)
vprint.PanicOn(err)
if dur == 0 {
dur = time.Minute
}
AlwaysPrintf("starting cpu profile for dur '%v', output to '%v'", dur, path)
vprint.AlwaysPrintf("starting cpu profile for dur '%v', output to '%v'", dur, path)
_ = pprof.StartCPUProfile(f)
go func() {
<-time.After(dur)
pprof.StopCPUProfile()
f.Close()
AlwaysPrintf("stopping cpu profile after dur '%v', output: '%v'", dur, path)
vprint.AlwaysPrintf("stopping cpu profile after dur '%v', output: '%v'", dur, path)
}()
}
@ -64,20 +64,20 @@ func MemProfileForDur(dur time.Duration, outpath string) {
}
path := outpath + "." + backend
f, err := os.Create(path)
PanicOn(err)
vprint.PanicOn(err)
if dur == 0 {
dur = time.Minute
}
AlwaysPrintf("will write memory profile after dur '%v', output to '%v'", dur, path)
vprint.AlwaysPrintf("will write memory profile after dur '%v', output to '%v'", dur, path)
go func() {
<-time.After(dur)
runtime.GC() // get up-to-date statistics
if err := pprof.WriteHeapProfile(f); err != nil {
PanicOn(fmt.Sprintf("could not write memory profile: %v", err))
vprint.PanicOn(fmt.Sprintf("could not write memory profile: %v", err))
}
f.Close()
AlwaysPrintf("wrote memory profile after dur '%v', output: '%v'", dur, path)
vprint.AlwaysPrintf("wrote memory profile after dur '%v', output: '%v'", dur, path)
}()
}
@ -92,7 +92,7 @@ var _ = pprofProfile{}
func newPprof() (pp *pprofProfile) {
pp = &pprofProfile{}
f, err := os.Create("cpu.manual.pprof")
PanicOn(err)
vprint.PanicOn(err)
pp.fdCpu = f
_ = pprof.StartCPUProfile(pp.fdCpu)
@ -105,11 +105,11 @@ func (pp *pprofProfile) Close() {
pp.fdCpu.Close()
f, err := os.Create("mem.manual.pprof")
PanicOn(err)
vprint.PanicOn(err)
runtime.GC() // get up-to-date statistics
if err := pprof.WriteHeapProfile(f); err != nil {
PanicOn(fmt.Sprintf("could not write memory profile: %v", err))
vprint.PanicOn(fmt.Sprintf("could not write memory profile: %v", err))
}
f.Close()
}

2
pql/ast.go
View file

@ -586,7 +586,7 @@ func (c *Call) CheckCallInfo() error {
case string, int64:
continue
default:
return fmt.Errorf("'%s': arg '%s' needed a string or integer value, got %T.",
return fmt.Errorf("'%s': arg '%s' needed a string or integer value, got %T",
c.String(), k, v)
}
}

4
rbf.go
View file

@ -28,7 +28,7 @@ import (
txkey "github.com/molecula/featurebase/v2/short_txkey"
"github.com/molecula/featurebase/v2/storage"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
"github.com/pkg/errors"
)
@ -411,7 +411,7 @@ func (tx *RBFTx) ImportRoaringBits(index, field, view string, shard uint64, rit
func (tx *RBFTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator {
b, err := tx.RoaringBitmap(index, field, view, shard)
PanicOn(err)
vprint.PanicOn(err)
return b.Iterator()
}

30
rbf/rbf.go
View file

@ -30,7 +30,7 @@ import (
"github.com/benbjohnson/immutable"
"github.com/molecula/featurebase/v2/roaring"
"github.com/molecula/featurebase/v2/shardwidth"
. "github.com/molecula/featurebase/v2/vprint"
"github.com/molecula/featurebase/v2/vprint"
)
const (
@ -356,7 +356,7 @@ func (c *leafCell) Bitmap(tx *Tx) []uint64 {
_, bm, _ := tx.leafCellBitmap(toPgno(c.Data))
return bm
default:
PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
vprint.PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
}
return nil
}
@ -383,7 +383,7 @@ func (c *leafCell) Values(tx *Tx) []uint16 {
case ContainerTypeNone:
return []uint16{}
default:
PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
vprint.PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
}
return nil
}
@ -411,7 +411,7 @@ func (c *leafCell) firstValue(tx *Tx) uint16 {
return r[0].Start
case ContainerTypeBitmapPtr:
_, slc, err := tx.leafCellBitmap(toPgno(c.Data))
PanicOn(err)
vprint.PanicOn(err)
for i, v := range slc {
for j := uint(0); j < 64; j++ {
if v&(1<<j) != 0 {
@ -419,9 +419,9 @@ func (c *leafCell) firstValue(tx *Tx) uint16 {
}
}
}
PanicOn(fmt.Errorf("rbf.leafCell.firstValue(): no values set in bitmap container: key=%d", c.Key))
vprint.PanicOn(fmt.Errorf("rbf.leafCell.firstValue(): no values set in bitmap container: key=%d", c.Key))
default:
PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
vprint.PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
}
return 0
@ -436,7 +436,7 @@ func (c *leafCell) lastValueFromBitmap(a []uint64) uint16 {
}
}
}
PanicOn(fmt.Errorf("rbf.leafCell.lastValueFromBitmap(): no values set in bitmap container: key=%d", c.Key))
vprint.PanicOn(fmt.Errorf("rbf.leafCell.lastValueFromBitmap(): no values set in bitmap container: key=%d", c.Key))
return 0
}
@ -456,10 +456,10 @@ func (c *leafCell) lastValue(tx *Tx) uint16 {
case ContainerTypeBitmapPtr:
_, a, err := tx.leafCellBitmap(toPgno(c.Data))
PanicOn(err)
vprint.PanicOn(err)
return c.lastValueFromBitmap(a)
default:
PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
vprint.PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
}
return 0
}
@ -483,10 +483,10 @@ func (c *leafCell) countRange(tx *Tx, start, end int32) (n int) {
return int(roaring.BitmapCountRange(toArray64(c.Data), start, end))
case ContainerTypeBitmapPtr:
_, a, err := tx.leafCellBitmap(toPgno(c.Data))
PanicOn(err)
vprint.PanicOn(err)
return int(roaring.BitmapCountRange(a, start, end))
default:
PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
vprint.PanicOn(fmt.Errorf("invalid container type: %d", c.Type))
}
return
}
@ -567,7 +567,7 @@ func readLeafCellBytesAtOffset(page []byte, offset int) []byte {
case ContainerTypeBitmapPtr:
return buf[:leafCellHeaderSize+4]
default:
PanicOn(fmt.Errorf("invalid cell type: %d", typ))
vprint.PanicOn(fmt.Errorf("invalid cell type: %d", typ))
}
return nil
}
@ -723,13 +723,13 @@ func Walk(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) {
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
page, _, err := tx.readPage(pgno)
if err != nil {
PanicOn(err)
vprint.PanicOn(err)
}
// Read all records on the page.
a, err := readRootRecords(page)
if err != nil {
PanicOn(err)
vprint.PanicOn(err)
}
v(pgno, a)
// Read next overflow page number.
@ -739,7 +739,7 @@ func Walk(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) {
func assert(condition bool) {
if !condition {
PanicOn(fmt.Errorf("assertion failed"))
vprint.PanicOn(fmt.Errorf("assertion failed"))
}
}

34
rbf/tx.go
View file

@ -25,7 +25,7 @@ import (
"github.com/benbjohnson/immutable"
"github.com/molecula/featurebase/v2/roaring"
txkey "github.com/molecula/featurebase/v2/short_txkey"
. "github.com/molecula/featurebase/v2/vprint"
"github.com/molecula/featurebase/v2/vprint"
)
var _ = txkey.ToString
@ -155,7 +155,7 @@ func (tx *Tx) Rollback() {
// Disconnect transaction from DB.
tx.db.mu.Lock()
defer tx.db.mu.Unlock()
PanicOn(tx.db.removeTx(tx))
vprint.PanicOn(tx.db.removeTx(tx))
}
// Root returns the root page number for a bitmap. Returns 0 if the bitmap does not exist.
@ -333,7 +333,6 @@ func (tx *Tx) DeleteBitmapsWithPrefix(prefix string) error {
if !strings.HasPrefix(name.(string), prefix) {
continue
}
// Deallocate all pages in the tree.
if err := tx.deallocateTree(pgno.(uint32)); err != nil {
return err
@ -964,7 +963,7 @@ func (tx *Tx) freelistCleanup(outErr *error) {
}
return
} else if !changed {
PanicOn(fmt.Sprintf("rbf.Tx.freePgno(): double free: %d", tx.pendingFreelistAdds))
vprint.PanicOn(fmt.Sprintf("rbf.Tx.freePgno(): double free: %d", tx.pendingFreelistAdds))
}
}
}
@ -990,7 +989,7 @@ func (tx *Tx) allocatePgno() (_ uint32, outErr error) {
if changed, err := c.Remove(uint64(pgno)); err != nil {
return 0, err
} else if !changed {
PanicOn(fmt.Sprintf("tx.Tx.allocatePgno(): double alloc: %d", pgno))
vprint.PanicOn(fmt.Sprintf("tx.Tx.allocatePgno(): double alloc: %d", pgno))
}
return pgno, nil
}
@ -1042,7 +1041,7 @@ func (tx *Tx) freePgno(pgno uint32) (outErr error) {
if changed, err := c.Add(uint64(pgno)); err != nil {
return err
} else if !changed {
PanicOn(fmt.Sprintf("rbf.Tx.freePgno(): double free: %d", pgno))
vprint.PanicOn(fmt.Sprintf("rbf.Tx.freePgno(): double free: %d", pgno))
}
return nil
}
@ -1065,6 +1064,13 @@ func (tx *Tx) deallocateTree(pgno uint32) error {
return tx.freePgno(pgno)
case PageTypeLeaf:
for i, n := 0, readCellN(page); i < n; i++ {
if cell := readLeafCell(page, i); cell.Type == ContainerTypeBitmapPtr {
if err := tx.freePgno(toPgno(cell.Data)); err != nil {
return err
}
}
}
return tx.freePgno(pgno)
default:
return fmt.Errorf("rbf.Tx.deallocateTree(): invalid page type: pgno=%d type=%d", pgno, typ)
@ -1282,7 +1288,7 @@ func (tx *Tx) ForEachRange(name string, start, end uint64, fn func(uint64) error
}
}
default:
PanicOn(fmt.Sprintf("invalid container type: %d", cell.Type))
vprint.PanicOn(fmt.Sprintf("invalid container type: %d", cell.Type))
}
}
}
@ -1458,11 +1464,11 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) {
func (tx *Tx) OffsetRange(name string, offset, start, endx uint64) (*roaring.Bitmap, error) {
if lowbits(offset) != 0 {
PanicOn("offset must not contain low bits")
vprint.PanicOn("offset must not contain low bits")
} else if lowbits(start) != 0 {
PanicOn("range start must not contain low bits")
vprint.PanicOn("range start must not contain low bits")
} else if lowbits(endx) != 0 {
PanicOn("range endx must not contain low bits")
vprint.PanicOn("range endx must not contain low bits")
}
tx.mu.RLock()
@ -1595,7 +1601,7 @@ func (si *emptyContainerIterator) Next() bool {
return false
}
func (si *emptyContainerIterator) Value() (uint64, *roaring.Container) {
PanicOn("emptyContainerIterator never has any Values")
vprint.PanicOn("emptyContainerIterator never has any Values")
return 0, nil
}
@ -1727,7 +1733,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
err = tx.putContainerWithCursor(cur, itrKey, newC)
if err != nil {
PanicOn(err)
vprint.PanicOn(err)
return
}
continue
@ -1869,7 +1875,7 @@ func (tx *Tx) Pages(pgnos []uint32) ([]Page, error) {
pages = append(pages, &FreePage{FreePageInfo: info})
default:
PanicOn(fmt.Sprintf("invalid page info type %T", info))
vprint.PanicOn(fmt.Sprintf("invalid page info type %T", info))
}
}
@ -1989,7 +1995,7 @@ func (tx *Tx) walkPageInfo(infos []PageInfo, root uint32, name string) error {
Tree: name,
}
default:
PanicOn(fmt.Sprintf("unexpected page type %d for page %d", typ, pgno))
vprint.PanicOn(fmt.Sprintf("unexpected page type %d for page %d", typ, pgno))
}
return nil

99
rbf/tx_test.go
View file

@ -684,3 +684,102 @@ func TestTx_CreateBitmap(t *testing.T) {
}
})
}
func TestTx_DeleteBitmapsWithPrefix(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
prefix := "abc"
bitmapSize := 10000
// var err error
// create a interleaved set up array and bitmap containers
bits := make([]uint64, bitmapSize)
x := uint64(1)
for i := 0; i < len(bits); i++ {
bits[i] = x
x = x + 2
}
ifError := func(err error) {
if err != nil {
t.Fatal(err)
}
}
checkInfos := func() {
tx := MustBegin(t, db, false)
defer tx.Rollback()
infos, err := tx.PageInfos()
ifError(err)
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)
case *rbf.RootRecordPageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "rootrec")
fmt.Printf("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)
case *rbf.BranchPageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "branch")
fmt.Printf("flags=x%x,celln=%d\n", info.Flags, info.CellN)
case *rbf.BitmapPageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "bitmap")
fmt.Printf("-\n")
case *rbf.FreePageInfo:
fmt.Printf("%-8d ", pgno)
fmt.Printf("%-10s ", "free")
fmt.Printf("-\n")
default:
t.Fatal(fmt.Sprintf("unexpected page info type %T", info))
}
}
}
populate := func() {
tx := MustBegin(t, db, true)
defer tx.Rollback()
for i := uint64(0); i < 16; i++ {
bm := roaring.NewBitmap()
if i%3 == 0 {
bm.Put(i, roaring.NewContainerBitmap(6144, bits))
if _, err := tx.AddRoaring(prefix, bm); err != nil {
panic(err)
}
} else {
bm.Put(i, roaring.NewContainerArray([]uint16{2, 4, 5, 7}))
if _, err := tx.AddRoaring(prefix, bm); err != nil {
panic(err)
}
}
}
ifError(tx.Commit())
}
checkInfos()
populate()
checkInfos()
ifError(db.Check())
tx := MustBegin(t, db, true)
tx.DeleteBitmapsWithPrefix(prefix)
ifError(tx.Commit())
ifError(db.Check())
checkInfos()
populate()
ifError(db.Check())
checkInfos()
}

10
rbf/util.go
View file

@ -18,7 +18,7 @@ import (
"strings"
txkey "github.com/molecula/featurebase/v2/short_txkey"
. "github.com/molecula/featurebase/v2/vprint"
"github.com/molecula/featurebase/v2/vprint"
)
// we don't currently use dumpAllPages but it's tricky enough to get right
@ -60,9 +60,9 @@ func (tx *Tx) dumpAllPages(showLeaves bool) error {
fmt.Printf("next=%d\n", info.Next)
page, _, err := tx.readPage(uint32(pgno))
PanicOn(err)
vprint.PanicOn(err)
rootRecords, err := readRootRecords(page)
PanicOn(err)
vprint.PanicOn(err)
for k, rr := range rootRecords {
fmt.Printf(" [%02v] Name:'%v' pgno:%v\n", k, prefixToString(rr.Name), rr.Pgno)
}
@ -77,7 +77,7 @@ func (tx *Tx) dumpAllPages(showLeaves bool) error {
fmt.Printf("flags=x%x,celln=%d\n", info.Flags, info.CellN)
page, _, err := tx.readPage(uint32(pgno))
PanicOn(err)
vprint.PanicOn(err)
var leafCells [PageSize / 8]leafCell
cells := readLeafCells(page, leafCells[:])
@ -92,7 +92,7 @@ func (tx *Tx) dumpAllPages(showLeaves bool) error {
fmt.Printf("flags=x%x,celln=%d\n", info.Flags, info.CellN)
page, _, err := tx.readPage(uint32(pgno))
PanicOn(err)
vprint.PanicOn(err)
cells := readBranchCells(page)
for i, cell := range cells {

4
rrtx.go
View file

@ -28,7 +28,7 @@ import (
txkey "github.com/molecula/featurebase/v2/short_txkey"
"github.com/molecula/featurebase/v2/storage"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
"github.com/pkg/errors"
)
@ -97,7 +97,7 @@ func roaringMapOfShards(optionalViewPath string) (shardMap map[uint64]bool, err
// the transaction Commits or Rollsback.
func (tx *RoaringTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator {
b, err := tx.bitmap(index, field, view, shard)
PanicOn(err)
vprint.PanicOn(err)
return b.Iterator()
}

49
server/config.go
View file

@ -19,11 +19,13 @@ import (
"fmt"
"log"
"net"
"net/url"
"runtime"
"strconv"
"strings"
"time"
"github.com/molecula/featurebase/v2/auth"
petcd "github.com/molecula/featurebase/v2/etcd"
rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg"
"github.com/molecula/featurebase/v2/storage"
@ -240,6 +242,9 @@ type Config struct {
// Toggles /schema/details endpoint. If off, it returns empty.
SchemaDetailsOn bool `toml:"schema-details-on"`
// Enable AuthZ/AuthN
Auth auth.Auth `toml:"auth"`
}
// Namespace returns the namespace to use based on the Future flag.
@ -273,6 +278,7 @@ func (c *Config) validate() error {
"Etcd.ClusterURL", c.Etcd.ClusterURL,
"Postgres.Bind", c.Postgres.Bind,
}
ports := make(map[int]bool)
n := len(hostPort)
for i := 0; i < n; i += 2 {
@ -602,3 +608,46 @@ func lookupAddr(ctx context.Context, resolver *net.Resolver, host string) (strin
// No IPv4 address, return the first resolved address instead.
return addrs[0].String(), nil
}
func (c *Config) ValidateAuth() ([]error, error) {
if !c.Auth.Enable {
return []error{}, nil
}
authConfig := map[string]string{
"ClientId": c.Auth.ClientId,
"ClientSecret": c.Auth.ClientSecret,
"AuthorizeURL": c.Auth.AuthorizeURL,
"TokenURL": c.Auth.TokenURL,
"GroupEndpointURL": c.Auth.GroupEndpointURL,
"ScopeURL": c.Auth.ScopeURL,
}
errors := make([]error, 0)
for name, value := range authConfig {
if value == "" {
errors = append(errors, fmt.Errorf("Empty string for auth config %s", name))
continue
}
if strings.Contains(name, "URL") {
_, err := url.ParseRequestURI(value)
if err != nil {
errors = append(errors, fmt.Errorf("Invalid URL for auth config %s: %s", name, err))
continue
}
}
}
if len(errors) > 0 {
return errors, fmt.Errorf("there were errors validating config")
}
return errors, nil
}
func (c *Config) MustValidateAuth() {
if errors, err := c.ValidateAuth(); err != nil {
for _, e := range errors {
log.Println(e)
}
log.Fatal(err)
}
}

212
server/config_internal_test.go
View file

@ -21,6 +21,8 @@ import (
"os"
"strings"
"testing"
"github.com/molecula/featurebase/v2/auth"
)
type addrs struct{ bind, advertise string }
@ -288,3 +290,213 @@ func TestConfig_validateAddrsGRPC(t *testing.T) {
})
}
}
func TestConfig_validateAuth(t *testing.T) {
errorMesgEmpty := "Empty string"
errorMesgURL := "Invalid URL"
validTestURL := "https://url.com/"
validClientID := "clientid"
validClientSecret := "clientSecret"
notValidURL := "not-a-url"
emptyString := ""
enable := true
disable := false
tests := []struct {
expErrs []string
input auth.Auth
}{
{
// Auth enabled, all configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: emptyString,
ClientSecret: emptyString,
AuthorizeURL: emptyString,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: emptyString,
AuthorizeURL: emptyString,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: emptyString,
ClientSecret: validClientSecret,
AuthorizeURL: emptyString,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: emptyString,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: validTestURL,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some configs are set to empty string
[]string{
errorMesgEmpty,
errorMesgEmpty,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: validTestURL,
TokenURL: validTestURL,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
{
// Auth enabled, some strings are set to invalid URL
[]string{
errorMesgURL,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: notValidURL,
TokenURL: validTestURL,
GroupEndpointURL: validTestURL,
ScopeURL: validTestURL,
},
},
{
// Auth enabled, some strings are set to invalid URL
[]string{
errorMesgURL,
errorMesgURL,
},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: validTestURL,
TokenURL: notValidURL,
GroupEndpointURL: notValidURL,
ScopeURL: validTestURL,
},
},
{
// Auth enabled, all configs are set properly
[]string{},
auth.Auth{
Enable: enable,
ClientId: validClientID,
ClientSecret: validClientSecret,
AuthorizeURL: validTestURL,
TokenURL: validTestURL,
GroupEndpointURL: validTestURL,
ScopeURL: validTestURL,
},
},
{
// Auth disabled, all configs are set to empty string
[]string{},
auth.Auth{
Enable: disable,
ClientId: emptyString,
ClientSecret: emptyString,
AuthorizeURL: emptyString,
TokenURL: emptyString,
GroupEndpointURL: emptyString,
ScopeURL: emptyString,
},
},
}
for i, test := range tests {
t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
c := NewConfig()
c.Auth = test.input
errors, err := c.ValidateAuth()
if len(test.expErrs) > 0 {
if err == nil {
t.Fatal("expected errors, but none were found")
}
}
if len(errors) != len(test.expErrs) {
fmt.Printf("%+v\n", errors)
t.Fatalf("expected %v errors but got %v", len(test.expErrs), len(errors))
}
for i, e := range errors {
if !strings.Contains(e.Error(), test.expErrs[i]) {
t.Errorf("expected error to contain %s, but got %s", test.expErrs[i], e.Error())
}
}
})
}
}

4
server/server.go
View file

@ -234,6 +234,10 @@ func (m *Command) Start() (err error) {
return errors.Wrap(err, "setting resource limits")
}
if m.Config.Auth.Enable == true {
m.Config.MustValidateAuth()
}
// Initialize server.
if err = m.Server.Open(); err != nil {
return errors.Wrap(err, "opening server")

76
stattx.go
View file

@ -25,7 +25,7 @@ import (
"github.com/molecula/featurebase/v2/debugstats"
"github.com/molecula/featurebase/v2/roaring"
txkey "github.com/molecula/featurebase/v2/short_txkey"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
)
// statTx is useful to profile on a
@ -218,7 +218,7 @@ func (k kall) String() string {
case kType:
return "kType"
}
PanicOn(fmt.Sprintf("unknown kall '%v'", int(k)))
vprint.PanicOn(fmt.Sprintf("unknown kall '%v'", int(k)))
return ""
}
@ -243,8 +243,8 @@ func (c *statTx) ImportRoaringBits(index, field, view string, shard uint64, rit
}()
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see ImportRoaringBits() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see ImportRoaringBits() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.ImportRoaringBits(index, field, view, shard, rit, clear, log, rowSize)
@ -259,8 +259,8 @@ func (c *statTx) Rollback() {
}()
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Rollback() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Rollback() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
c.b.Rollback()
@ -276,8 +276,8 @@ func (c *statTx) Commit() error {
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Commit() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Commit() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Commit()
@ -293,8 +293,8 @@ func (c *statTx) RoaringBitmap(index, field, view string, shard uint64) (*roarin
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see RoaringBitmap() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see RoaringBitmap() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.RoaringBitmap(index, field, view, shard)
@ -310,8 +310,8 @@ func (c *statTx) Container(index, field, view string, shard uint64, key uint64)
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Container() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Container() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Container(index, field, view, shard, key)
@ -327,8 +327,8 @@ func (c *statTx) PutContainer(index, field, view string, shard uint64, key uint6
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see PutContainer() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see PutContainer() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.PutContainer(index, field, view, shard, key, rc)
@ -344,8 +344,8 @@ func (c *statTx) RemoveContainer(index, field, view string, shard uint64, key ui
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see RemoveContainer() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see RemoveContainer() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.RemoveContainer(index, field, view, shard, key)
@ -361,8 +361,8 @@ func (c *statTx) Add(index, field, view string, shard uint64, a ...uint64) (chan
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Add() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Add() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Add(index, field, view, shard, a...)
@ -378,8 +378,8 @@ func (c *statTx) Remove(index, field, view string, shard uint64, a ...uint64) (c
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Remove() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Remove() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Remove(index, field, view, shard, a...)
@ -395,8 +395,8 @@ func (c *statTx) Contains(index, field, view string, shard uint64, key uint64) (
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Contains() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Contains() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Contains(index, field, view, shard, key)
@ -412,8 +412,8 @@ func (c *statTx) ContainerIterator(index, field, view string, shard uint64, firs
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see ContainerIterator() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see ContainerIterator() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.ContainerIterator(index, field, view, shard, firstRoaringContainerKey)
@ -433,8 +433,8 @@ func (c *statTx) ForEach(index, field, view string, shard uint64, fn func(i uint
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see ForEach() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see ForEach() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.ForEach(index, field, view, shard, fn)
@ -450,8 +450,8 @@ func (c *statTx) ForEachRange(index, field, view string, shard uint64, start, en
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see ForEachRange() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see ForEachRange() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.ForEachRange(index, field, view, shard, start, end, fn)
@ -467,8 +467,8 @@ func (c *statTx) Count(index, field, view string, shard uint64) (uint64, error)
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Count() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Count() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Count(index, field, view, shard)
@ -484,8 +484,8 @@ func (c *statTx) Max(index, field, view string, shard uint64) (uint64, error) {
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Max() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Max() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Max(index, field, view, shard)
@ -501,8 +501,8 @@ func (c *statTx) Min(index, field, view string, shard uint64) (uint64, bool, err
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see Min() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see Min() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.Min(index, field, view, shard)
@ -518,8 +518,8 @@ func (c *statTx) CountRange(index, field, view string, shard uint64, start, end
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see CountRange() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see CountRange() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.CountRange(index, field, view, shard, start, end)
@ -534,8 +534,8 @@ func (c *statTx) OffsetRange(index, field, view string, shard, offset, start, en
defer func() {
if r := recover(); r != nil {
AlwaysPrintf("see OffsetRange() PanicOn '%v' at '%v'", r, Stack())
PanicOn(r)
vprint.AlwaysPrintf("see OffsetRange() PanicOn '%v' at '%v'", r, vprint.Stack())
vprint.PanicOn(r)
}
}()
return c.b.OffsetRange(index, field, view, shard, offset, start, end)

8
test/holder.go
View file

@ -19,10 +19,10 @@ import (
"testing"
"time"
"github.com/molecula/featurebase/v2"
pilosa "github.com/molecula/featurebase/v2"
"github.com/molecula/featurebase/v2/pql"
"github.com/molecula/featurebase/v2/testhook"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
"github.com/pkg/errors"
)
@ -155,7 +155,7 @@ func (h *Holder) SetBitTime(index, field string, rowID, columnID uint64, t *time
if err != nil {
panic(err)
}
PanicOn(tx.Commit())
vprint.PanicOn(tx.Commit())
}
// ClearBit clears a bit on the given field.
@ -174,7 +174,7 @@ func (h *Holder) ClearBit(index, field string, rowID, columnID uint64) {
if err != nil {
panic(err)
}
PanicOn(tx.Commit())
vprint.PanicOn(tx.Commit())
}
// MustSetBits sets columns on a row. Panic on error.

2
testhook/auditor.go
View file

@ -14,6 +14,8 @@
package testhook
//TODO: Check() and FinalCheck() should return error as the last argument
import (
"fmt"
"reflect"

50
txfactory.go
View file

@ -24,7 +24,7 @@ import (
"sync"
"github.com/molecula/featurebase/v2/testhook"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
"github.com/pkg/errors"
)
@ -174,7 +174,7 @@ func (q *Qcx) Reset() {
q.mu.Lock()
defer q.mu.Unlock()
if !q.done {
PanicOn("must call Qcx.Abort() or Qcx.Finish() before calling Reset().")
vprint.PanicOn("must call Qcx.Abort() or Qcx.Finish() before calling Reset().")
}
q.unprotected_reset()
}
@ -272,16 +272,16 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) {
// verify that shard and index match!
ro := qcx.RequiredTxo
if o.Shard != ro.Shard {
PanicOn(fmt.Sprintf("shard mismatch: o.Shard = %v while qcx.RequiredTxo.Shard = %v", o.Shard, ro.Shard))
vprint.PanicOn(fmt.Sprintf("shard mismatch: o.Shard = %v while qcx.RequiredTxo.Shard = %v", o.Shard, ro.Shard))
}
if o.Index == nil {
PanicOn("o.Index annot be nil")
vprint.PanicOn("o.Index annot be nil")
}
if ro.Index == nil {
PanicOn("ro.Index annot be nil")
vprint.PanicOn("ro.Index annot be nil")
}
if o.Index.name != ro.Index.name {
PanicOn(fmt.Sprintf("index mismatch: o.Index = %v while qcx.RequiredTxo.Index = %v", o.Index.name, ro.Index.name))
vprint.PanicOn(fmt.Sprintf("index mismatch: o.Index = %v while qcx.RequiredTxo.Index = %v", o.Index.name, ro.Index.name))
}
return *qcx.RequiredForAtomicWriteTx, NoopFinisher, nil
}
@ -312,7 +312,7 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) {
// so defer finisher(nil) means always Commit writes, ignoring
// the enclosing functions return status.
if perr == nil || *perr == nil {
PanicOn(tx.Commit())
vprint.PanicOn(tx.Commit())
} else {
tx.Rollback()
}
@ -331,7 +331,7 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) {
// to this shard/index will re-use it.
func (qcx *Qcx) StartAtomicWriteTx(o Txo) {
if !o.Write {
PanicOn("must have o.Write true")
vprint.PanicOn("must have o.Write true")
}
qcx.mu.Lock()
defer qcx.mu.Unlock()
@ -349,16 +349,16 @@ func (qcx *Qcx) StartAtomicWriteTx(o Txo) {
// verify that shard and index match!
ro := qcx.RequiredTxo
if o.Shard != ro.Shard {
PanicOn(fmt.Sprintf("shard mismatch: o.Shard = %v while qcx.RequiredTxo.Shard = %v", o.Shard, ro.Shard))
vprint.PanicOn(fmt.Sprintf("shard mismatch: o.Shard = %v while qcx.RequiredTxo.Shard = %v", o.Shard, ro.Shard))
}
if o.Index == nil {
PanicOn("o.Index annot be nil")
vprint.PanicOn("o.Index annot be nil")
}
if ro.Index == nil {
PanicOn("ro.Index annot be nil")
vprint.PanicOn("ro.Index annot be nil")
}
if o.Index.name != ro.Index.name {
PanicOn(fmt.Sprintf("index mismatch: o.Index = %v while qcx.RequiredTxo.Index = %v", o.Index.name, ro.Index.name))
vprint.PanicOn(fmt.Sprintf("index mismatch: o.Index = %v while qcx.RequiredTxo.Index = %v", o.Index.name, ro.Index.name))
}
}
@ -401,7 +401,7 @@ func (ty txtype) DirectoryName() string {
case rbfTxn:
return "rbf"
}
PanicOn(fmt.Sprintf("unkown txtype %v", int(ty)))
vprint.PanicOn(fmt.Sprintf("unkown txtype %v", int(ty)))
return ""
}
@ -714,7 +714,7 @@ type grpkey struct {
func mustHaveIndexShard(o *Txo) {
if o.Index == nil || o.Index.name == "" {
PanicOn("index must be set on Txo")
vprint.PanicOn("index must be set on Txo")
}
}
@ -754,10 +754,10 @@ func (g *TxGroup) AddTx(tx Tx, o Txo) {
g.mu.Lock()
defer g.mu.Unlock()
if g.finished {
PanicOn("in TxGroup.Finish(): TxGroup already finished")
vprint.PanicOn("in TxGroup.Finish(): TxGroup already finished")
}
if NilInside(tx) {
PanicOn("Cannot add nil Tx to TxGroup")
vprint.PanicOn("Cannot add nil Tx to TxGroup")
}
g.reads = append(g.reads, tx)
@ -765,7 +765,7 @@ func (g *TxGroup) AddTx(tx Tx, o Txo) {
key := grpkey{index: o.Index.name, shard: o.Shard}
prior, ok := g.all[key]
if ok {
PanicOn(fmt.Sprintf("already have Tx in group for this, we should have re-used it! prior is '%v'; tx='%v'", prior, tx))
vprint.PanicOn(fmt.Sprintf("already have Tx in group for this, we should have re-used it! prior is '%v'; tx='%v'", prior, tx))
}
g.all[key] = tx
}
@ -777,7 +777,7 @@ func (g *TxGroup) FinishGroup() (err error) {
g.mu.Lock()
defer g.mu.Unlock()
if g.finished {
PanicOn("in TxGroup.Finish(): TxGroup already finished")
vprint.PanicOn("in TxGroup.Finish(): TxGroup already finished")
}
g.finished = true
for _, r := range g.reads {
@ -817,27 +817,27 @@ func (f *TxFactory) NewTx(o Txo) (txn Tx) {
if o.Fragment != nil {
if o.Fragment.index() != indexName {
PanicOn(fmt.Sprintf("inconsistent NewTx request: o.Fragment.index='%v' but indexName='%v'", o.Fragment.index(), indexName))
vprint.PanicOn(fmt.Sprintf("inconsistent NewTx request: o.Fragment.index='%v' but indexName='%v'", o.Fragment.index(), indexName))
}
if o.Fragment.shard != o.Shard {
PanicOn(fmt.Sprintf("inconsistent NewTx request: o.Fragment.shard='%v' but o.Shard='%v'", o.Fragment.shard, o.Shard))
vprint.PanicOn(fmt.Sprintf("inconsistent NewTx request: o.Fragment.shard='%v' but o.Shard='%v'", o.Fragment.shard, o.Shard))
}
}
// look up in the collection of open databases, and get our
// per-shard database. Opens a new one if needed.
dbs, err := f.dbPerShard.GetDBShard(indexName, o.Shard, o.Index)
PanicOn(err)
vprint.PanicOn(err)
if dbs.Shard != o.Shard {
PanicOn(fmt.Sprintf("asked for o.Shard=%v but got dbs.Shard=%v", int(o.Shard), int(dbs.Shard)))
vprint.PanicOn(fmt.Sprintf("asked for o.Shard=%v but got dbs.Shard=%v", int(o.Shard), int(dbs.Shard)))
}
//vv("got dbs='%p' for o.Index='%v'; shard='%v'; dbs.typ='%#v'; dbs.W='%#v'", dbs, o.Index.name, o.Shard, dbs.typ, dbs.W)
o.dbs = dbs
tx, err := dbs.NewTx(o.Write, indexName, o)
if err != nil {
PanicOn(errors.Wrap(err, "dbs.NewTx transaction errored"))
vprint.PanicOn(errors.Wrap(err, "dbs.NewTx transaction errored"))
}
return tx
}
@ -852,7 +852,7 @@ func (ty txtype) String() string {
case rbfTxn:
return "rbf"
}
PanicOn(fmt.Sprintf("unhandled ty '%v' in txtype.String()", int(ty)))
vprint.PanicOn(fmt.Sprintf("unhandled ty '%v' in txtype.String()", int(ty)))
return ""
}
@ -905,7 +905,7 @@ func listFilesUnderDir(root string, includeRoot bool, requiredSuffix string, ign
// ignore
} else {
if info == nil {
PanicOn(fmt.Sprintf("info was nil for path = '%v'", path))
vprint.PanicOn(fmt.Sprintf("info was nil for path = '%v'", path))
}
if info.IsDir() {
// skip directories.

14
view.go
View file

@ -30,7 +30,7 @@ import (
"github.com/molecula/featurebase/v2/roaring"
"github.com/molecula/featurebase/v2/stats"
"github.com/molecula/featurebase/v2/testhook"
. "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck
"github.com/molecula/featurebase/v2/vprint"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -70,7 +70,7 @@ type view struct {
// newView returns a new instance of View.
func newView(holder *Holder, path, index, field, name string, fieldOptions FieldOptions) *view {
PanicOn(ValidateName(name))
vprint.PanicOn(ValidateName(name))
return &view{
path: path,
@ -116,7 +116,7 @@ func (v *view) addKnownShard(shard uint64) {
atomic.StoreUint32(&v.knownShardsCopied, 0)
}
_, err := v.knownShards.Add(shard)
PanicOn(err)
vprint.PanicOn(err)
}
// removeKnownShard removes a known shard from v. See the notes on addKnownShard.
@ -514,7 +514,7 @@ func (v *view) setBit(txOrig Tx, rowID, columnID uint64) (changed bool, err erro
tx = v.idx.holder.txf.NewTx(Txo{Write: writable, Index: v.idx, Fragment: frag, Shard: shard})
defer func() {
if err == nil {
PanicOn(tx.Commit())
vprint.PanicOn(tx.Commit())
} else {
tx.Rollback()
}
@ -536,7 +536,7 @@ func (v *view) clearBit(txOrig Tx, rowID, columnID uint64) (changed bool, err er
tx = v.idx.holder.txf.NewTx(Txo{Write: writable, Index: v.idx, Fragment: frag, Shard: shard})
defer func() {
if err == nil {
PanicOn(tx.Commit())
vprint.PanicOn(tx.Commit())
} else {
tx.Rollback()
}
@ -576,7 +576,7 @@ func (v *view) setValue(txOrig Tx, columnID uint64, bitDepth uint64, value int64
tx = v.idx.holder.txf.NewTx(Txo{Write: writable, Index: v.idx, Fragment: frag, Shard: shard})
defer func() {
if err == nil {
PanicOn(tx.Commit())
vprint.PanicOn(tx.Commit())
} else {
tx.Rollback()
}
@ -599,7 +599,7 @@ func (v *view) clearValue(txOrig Tx, columnID uint64, bitDepth uint64, value int
tx = v.idx.holder.txf.NewTx(Txo{Write: writable, Index: v.idx, Fragment: frag, Shard: shard})
defer func() {
if err == nil {
PanicOn(tx.Commit())
vprint.PanicOn(tx.Commit())
} else {
tx.Rollback()
}