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 metric-typos

Browse source
This commit is contained in:
Travis Turner 2020-09-14 17:20:49 -05:00 • committed by GitHub
commit d2c6a8ddde
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
28 changed files with 261 additions and 103 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.circleci/config.yml
View file

@ -186,7 +186,7 @@ workflows:
matrix:
parameters:
golang_version: ["1.14", "1.13"]
resource_class: large
resource_class: xlarge
requires:
- setup
filters:

16
badger.go
View file

@ -350,7 +350,8 @@ func (r *badgerRegistrar) OpenDBWrapper(bpath string, doAllocZero bool) (DBWrapp
}
func (w *BadgerDBWrapper) DeleteDBPath(dbs *DBShard) error {
panic("TODO")
path := dbs.pathForType(badgerTxn)
return os.RemoveAll(path)
}
// DeleteIndex deletes all the containers associated with
@ -1627,7 +1628,7 @@ func (tx *BadgerTx) ImportRoaringBits(index, field, view string, shard uint64, i
return
}
func (tx *BadgerTx) toContainer(typ byte, v []byte) (r *roaring.Container) {
func (tx *BadgerTx) toContainer(typ byte, v []byte) (c *roaring.Container) {
if len(v) == 0 {
return nil
@ -1668,29 +1669,28 @@ func (tx *BadgerTx) toContainer(typ byte, v []byte) (r *roaring.Container) {
switch typ {
case roaring.ContainerArray:
c := roaring.NewContainerArray(toArray16(w))
c = roaring.NewContainerArray(toArray16(w))
if tx.doAllocZero {
// tx.acMu was acquired above, and Unlock deferred.
tx.ourContainers = append(tx.ourContainers, c)
}
return c
case roaring.ContainerBitmap:
c := roaring.NewContainerBitmap(-1, toArray64(w))
c = roaring.NewContainerBitmap(-1, toArray64(w))
if tx.doAllocZero {
// tx.acMu was acquired above, and Unlock deferred.
tx.ourContainers = append(tx.ourContainers, c)
}
return c
case roaring.ContainerRun:
c := roaring.NewContainerRun(toInterval16(w))
c = roaring.NewContainerRun(toInterval16(w))
if tx.doAllocZero {
// tx.acMu was acquired above, and Unlock deferred.
tx.ourContainers = append(tx.ourContainers, c)
}
return c
default:
panic(fmt.Sprintf("unknown container: %v", typ))
}
c.SetMapped(true)
return c
}
// StringifiedBadgerKeys returns a string with all the container

45
dbshard.go
View file

@ -63,7 +63,8 @@ type DBRegistry interface {
}
type DBShard struct {
Path string
HolderPath string
Index string
Shard uint64
Open bool
@ -118,8 +119,8 @@ func (dbs *DBShard) Close() (err error) {
return
}
func (dbs *DBShard) String() string {
return dbs.Path
func (dbs *DBShard) HolderString() string {
return dbs.HolderPath
}
// Cleanup must be called at every commit/rollback of a Tx, in
@ -234,7 +235,7 @@ func (per *DBPerShard) HasData(which int) (hasData bool, err error) {
func (per *DBPerShard) ListOpenString() (r string) {
for v := range per.Flatmap {
r += v.Path + " -> " + v.W[per.useOpenList].OpenListString() + "\n"
r += v.HolderPath + " -> " + v.W[per.useOpenList].OpenListString() + "\n"
}
return
}
@ -282,9 +283,17 @@ func (per *DBPerShard) DeleteIndex(index string) (err error) {
return nil
}
for _, dbs := range dbi.Shard {
err := dbs.Close()
panicOn(err)
panicOn(os.RemoveAll(dbs.Path))
err = dbs.Close()
if err != nil {
return errors.Wrap(err, "DBPerShard.DeleteIndex dbs.Close()")
}
for _, ty := range per.types {
path := dbs.pathForType(ty)
err = os.RemoveAll(path)
if err != nil {
return errors.Wrap(err, fmt.Sprintf("DBPerShard.DeleteIndex os.RemoveAll('%v')", path))
}
}
}
return
}
@ -362,8 +371,9 @@ func (per *DBPerShard) DumpAll() {
}
}
func (per *DBPerShard) Path(index string, shard uint64) string {
return per.Dir + sep + index + sep + fmt.Sprintf("%04v", shard)
func (dbs *DBShard) pathForType(ty txtype) string {
// top level paths will end in "@@"
return dbs.HolderPath + sep + dbs.Index + ".index.txstores@@@" + sep + "store" + ty.FileSuffix() + "@" + sep + fmt.Sprintf("shard.%04v", dbs.Shard)
}
func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs *DBShard, err error) {
@ -388,11 +398,12 @@ func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs *
ParentDBIndex: dbi,
Index: index,
Shard: shard,
Path: per.Path(index, shard),
idx: idx,
per: per,
useOpenList: per.useOpenList,
hasRoaring: per.hasRoaring,
HolderPath: per.Dir,
//Path: per.Path(index, shard),
idx: idx,
per: per,
useOpenList: per.useOpenList,
hasRoaring: per.hasRoaring,
}
dbi.Shard[shard] = dbs
}
@ -411,7 +422,8 @@ func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs *
default:
panic(fmt.Sprintf("unknown txtyp: '%v'", ty))
}
w, err := registry.OpenDBWrapper(dbs.Path, DetectMemAccessPastTx)
w, err := registry.OpenDBWrapper(dbs.pathForType(ty), DetectMemAccessPastTx)
panicOn(err)
h := idx.Holder()
w.SetHolder(h)
@ -474,7 +486,8 @@ func TypedDBPerShardGetLocalShardsForIndex(ty txtype, idx *Index, roaringViewPat
Index: idx,
}
if roaringViewPath == "" {
for _, field := range idx.Fields() {
fields := idx.Fields()
for _, field := range fields {
for _, view := range field.views() {
sos, err := rx.SliceOfShards("", "", "", view.path)
if err != nil {

76
executor_test.go
View file

@ -535,7 +535,10 @@ func TestExecutor_Execute_Count(t *testing.T) {
}
func roaringOnlyTest(t *testing.T) {
if os.Getenv("PILOSA_TXSRC") != "roaring" {
src := os.Getenv("PILOSA_TXSRC")
if src == pilosa.RoaringTxn || (pilosa.DefaultTxsrc == pilosa.RoaringTxn && src == "") {
// okay to run, we are under roaring only
} else {
t.Skip("skip for everything but roaring")
}
}
@ -6230,7 +6233,18 @@ func TestExecutor_Execute_CountDistinct(t *testing.T) {
}
// AntitodePoint == row 1 b/c keys field.
writeQuery := `Set(100, type=AntidotePoint)Set(100, equip_id=100)Set(100, site_id=100)Set(100, id=100)`
// Note: type=TwoPoints should match 100/101, and no row in type
// matches 102, but 102 is present in equip_id at all.
writeQuery := `
Set(100, type=AntidotePoint)
Set(100, type=TwoPoints)
Set(101, type=TwoPoints)
Set(100, equip_id=100)
Set(101, equip_id=101)
Set(102, equip_id=102)
Set(100, site_id=100)
Set(100, id=100)
`
for k, i := range schema.Indexes {
_ = k
if _, err := api.Query(context.TODO(), &pilosa.QueryRequest{Index: i.Name, Query: writeQuery}); err != nil {
@ -6245,7 +6259,7 @@ func TestExecutor_Execute_CountDistinct(t *testing.T) {
Intersect(Row(type=AntidotePoint)),
index=equipment, field=equip_id),
Distinct(
Intersect(Row(type=AntidotePoint)),
Intersect(Row(type=TwoPoints)),
index=sites, field=equip_id)
), index=power_ts, field=site_id)`
@ -6305,11 +6319,61 @@ func TestExecutor_Execute_CountDistinct(t *testing.T) {
if !ok {
t.Fatalf("invalid response type, expected: []pilosa.GroupCount, got: %T", resp.Results[0])
}
if len(gc) != 1 {
t.Fatalf("invalid group count length, expected: 1, got: %v", len(gc))
if len(gc) != 2 {
t.Fatalf("invalid group count length, expected: 2, got: %v", len(gc))
}
if gc[0].Count != 1 {
t.Fatalf("invalid group count count, expected: 1, got: %v", gc[0].Count)
t.Fatalf("invalid group-by count for %d, expected: 1, got: %v", gc[0].Group[0].RowID, gc[0].Count)
}
if gc[1].Count != 1 {
t.Fatalf("invalid group-by count for %d, expected: 1, got: %v", gc[1].Group[0].RowID, gc[1].Count)
}
})
t.Run("Store(Distinct)", func(t *testing.T) {
_, err = api.Query(context.TODO(), &pilosa.QueryRequest{
Index: "sites",
Query: `Store(Distinct(field=equip_id), type="a")`,
})
if err != nil {
t.Fatal(err)
}
resp, err := api.Query(context.TODO(), &pilosa.QueryRequest{
Index: "sites",
Query: `Row(type="a")`,
})
if err != nil {
t.Fatal(err)
}
res, ok := resp.Results[0].(*pilosa.Row)
if !ok {
t.Fatalf("invalid response type, expected: *pilosa.Row, got: %T", resp.Results[0])
}
cols := res.Columns()
if !eq(cols, []uint64{100, 101, 102}) {
t.Fatalf("expected [100, 101, 102], got %d", cols)
}
_, err = api.Query(context.TODO(), &pilosa.QueryRequest{
Index: "sites",
Query: `Store(Distinct(Row(type="TwoPoints"), field=equip_id), type="b")`,
})
if err != nil {
t.Fatal(err)
}
resp, err = api.Query(context.TODO(), &pilosa.QueryRequest{
Index: "sites",
Query: `Row(type="b")`,
})
if err != nil {
t.Fatal(err)
}
res, ok = resp.Results[0].(*pilosa.Row)
if !ok {
t.Fatalf("invalid response type, expected: *pilosa.Row, got: %T", resp.Results[0])
}
cols = res.Columns()
if !eq(cols, []uint64{100, 101}) {
t.Fatalf("expected [100, 101], got %d", cols)
}
})
}

5
field.go
View file

@ -767,6 +767,11 @@ fileLoop:
if !fi.IsDir() {
continue
}
// Skip embedded db files too.
if f.holder.txf.IsTxDatabasePath(fi.Name()) {
continue
}
fieldQueue <- struct{}{}
eg.Go(func() error {
defer func() {

2
field_internal_test.go
View file

@ -707,7 +707,7 @@ func TestIntField_MinMaxForShard(t *testing.T) {
// Ensure we get errors when they are expected.
func TestDecimalField_MinMaxBoundaries(t *testing.T) {
th := newTestHolder(t)
defer th.Close()
for i, test := range []struct {
scale int64
min pql.Decimal

29
fragment_internal_test.go
View file

@ -1696,13 +1696,19 @@ func TestFragment_RankCache_Persistence(t *testing.T) {
}
func roaringOnlyTest(t *testing.T) {
if os.Getenv("PILOSA_TXSRC") != "roaring" {
src := os.Getenv("PILOSA_TXSRC")
if src == RoaringTxn || (DefaultTxsrc == RoaringTxn && src == "") {
// okay to run, we are under roaring only
} else {
t.Skip("skip for everything but roaring")
}
}
func roaringOnlyBenchmark(b *testing.B) {
if os.Getenv("PILOSA_TXSRC") != "roaring" {
src := os.Getenv("PILOSA_TXSRC")
if src == RoaringTxn || (DefaultTxsrc == RoaringTxn && src == "") {
// okay to run, we are under roaring only
} else {
b.Skip("skip for everything but roaring")
}
}
@ -3161,6 +3167,7 @@ func BenchmarkImportIntoLargeFragment(b *testing.B) {
}
panicOn(tx.Commit())
f.Clean(b)
h.Close()
}
}
@ -3434,6 +3441,9 @@ func newTestHolder(tb testing.TB) *Holder {
path, _ := testhook.TempDirInDir(tb, *TempDir, "holder-dir")
h := NewHolder(path, nil)
panicOn(h.Open())
testhook.Cleanup(tb, func() {
h.Close()
})
//h.SnapshotQueue = newSnapshotQueue(1, 1, nil)
return h
}
@ -3461,9 +3471,6 @@ func mustOpenFragmentFlags(tb testing.TB, index, field, view string, shard uint6
}
th := newTestHolder(tb)
testhook.Cleanup(tb, func() {
th.Close()
})
idx := fragTestMustOpenIndex(index, th, IndexOptions{})
if th.NeedsSnapshot() {
th.SnapshotQueue = newSnapshotQueue(1, 1, nil)
@ -4963,10 +4970,6 @@ func TestImportClearRestart(t *testing.T) {
// OVERWRITING the f.path with a new fragment
f2 := newFragment(h, f.path, "i", "f", viewStandard, 0, 0)
// f2, idx2 := mustOpenFragment(t, "i", "f", viewStandard, 0, "")
// _ = idx2
f2.MaxOpN = maxOpN
f2.CacheType = f.CacheType
@ -4975,7 +4978,7 @@ func TestImportClearRestart(t *testing.T) {
tx2 := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f2, Shard: f2.shard})
defer tx2.Rollback()
err = f.closeStorage()
err = f.Close()
if err != nil {
t.Fatalf("closing storage: %v", err)
}
@ -5010,6 +5013,10 @@ func TestImportClearRestart(t *testing.T) {
panicOn(tx2.Commit())
h3 := NewHolder(filepath.Dir(f2.path), nil)
testhook.Cleanup(t, func() {
h3.Close()
})
idx3, err := h3.CreateIndex("i", IndexOptions{})
_ = idx3
panicOn(err)
@ -5021,7 +5028,7 @@ func TestImportClearRestart(t *testing.T) {
tx3 := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f3, Shard: f3.shard})
defer tx3.Rollback()
err = f2.closeStorage()
err = f2.Close()
if err != nil {
t.Fatalf("f2 closing storage: %v", err)
}

10
holder.go
View file

@ -725,12 +725,15 @@ func (h *Holder) processForeignIndexFields() error {
// Close closes all open fragments.
func (h *Holder) Close() error {
if h == nil {
return nil
}
defer h.stopBkgr()
if globalUseStatTx {
fmt.Printf("%v\n", globalCallStats.report())
}
if h.txf.blueGreenReg != nil {
if h.txf != nil && h.txf.blueGreenReg != nil {
h.txf.blueGreenReg.Close()
}
@ -806,6 +809,11 @@ func (h *Holder) HasData() (bool, error) {
if !fi.IsDir() {
continue
}
// Skip embedded db files too.
if h.txf.IsTxDatabasePath(fi.Name()) {
continue
}
return true, nil
}
return false, nil

2
http/handler.go
View file

@ -1632,10 +1632,10 @@ func (h *Handler) handleGetMetricsJSON(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
metrics := make(map[string][]*prom2json.Family)
mfChan := make(chan *dto.MetricFamily, 60)
transport := http.DefaultTransport.(*http.Transport).Clone()
for _, node := range h.api.Hosts(r.Context()) {
metricsURI := node.URI.String() + "/metrics"
mfChan := make(chan *dto.MetricFamily, 60)
err := prom2json.FetchMetricFamilies(metricsURI, mfChan, transport)
if err != nil {
http.Error(w, "fetching metrics: "+err.Error(), http.StatusInternalServerError)

6
index.go
View file

@ -261,6 +261,11 @@ fileLoop:
if !fi.IsDir() {
continue
}
// Skip embedded db files too.
if i.holder.txf.IsTxDatabasePath(fi.Name()) {
continue
}
indexQueue <- struct{}{}
eg.Go(func() error {
defer func() {
@ -369,6 +374,7 @@ func (i *Index) saveMeta() error {
// Close closes the index and its fields.
func (i *Index) Close() error {
i.mu.Lock()
defer i.mu.Unlock()
defer func() {

31
lmdb.go
View file

@ -1520,20 +1520,19 @@ func (tx *LMDBTx) toContainer(typ byte, v []byte) (r *roaring.Container) {
return ToContainer(typ, w)
}
func ToContainer(typ byte, w []byte) (r *roaring.Container) {
func ToContainer(typ byte, w []byte) (c *roaring.Container) {
switch typ {
case roaring.ContainerArray:
c := roaring.NewContainerArray(toArray16(w))
return c
c = roaring.NewContainerArray(toArray16(w))
case roaring.ContainerBitmap:
c := roaring.NewContainerBitmap(-1, toArray64(w))
return c
c = roaring.NewContainerBitmap(-1, toArray64(w))
case roaring.ContainerRun:
c := roaring.NewContainerRun(toInterval16(w))
return c
c = roaring.NewContainerRun(toInterval16(w))
default:
panic(fmt.Sprintf("unknown container: %v", typ))
}
c.SetMapped(true)
return c
}
// StringifiedLMDBKeys returns a string with all the container
@ -1629,11 +1628,13 @@ func stringifiedLMDBKeysTx(tx *LMDBTx, short bool) (r string) {
}
func (w *LMDBWrapper) DeleteDBPath(dbs *DBShard) (err error) {
path := dbs.Path
path := dbs.pathForType(lmdbTxn)
err = os.RemoveAll(path)
if err != nil {
return errors.Wrap(err, "DeleteDBPath")
}
// if we go back to flat instead of inside its own directory,
// there will be a second -lock file needing deletion too.
lockfile := path + "-lock"
if FileExists(lockfile) {
err = os.RemoveAll(lockfile)
@ -1641,15 +1642,19 @@ func (w *LMDBWrapper) DeleteDBPath(dbs *DBShard) (err error) {
return
}
func (w *LMDBWrapper) DeleteField(index, field, fieldPath string) error {
func (w *LMDBWrapper) DeleteField(index, field, fieldPath string) (err error) {
// TODO(jea) cleanup: I think this fieldPath delete just goes away now.
// remove this commented stuff once we are sure.
//
// under blue-green roaring_lmdb, the directory will not be found, b/c roaring will have
// already done the os.RemoveAll(). BUT, RemoveAll returns nil error in this case. Docs:
// "If the path does not exist, RemoveAll returns nil (no error)"
err := w.DeleteDBPath(&DBShard{Path: fieldPath})
if err != nil {
return errors.Wrap(err, "removing directory")
}
//w.DeleteDBPath(&DBShard{Path: fieldPath})
//if err != nil {
//return errors.Wrap(err, "removing directory")
//}
prefix := txkey.FieldPrefix(index, field)
return w.DeletePrefix(prefix)
}

2
lmdb_test.go
View file

@ -101,7 +101,7 @@ func mustOpenEmptyLMDBWrapper(path string) (w *LMDBWrapper, cleaner func()) {
return w, func() {
w.Close()
panicOn(w.DeleteDBPath(&DBShard{Path: fn}))
panicOn(w.DeleteDBPath(&DBShard{HolderPath: fn}))
}
}

2
metrics.go
View file

@ -67,4 +67,6 @@ const (
MetricExclusiveTransactionActive = "transaction_exclusive_active"
MetricExclusiveTransactionEnd = "transaction_exclusive_end"
MetricExclusiveTransactionBlocked = "transaction_exclusive_blocked"
MetricPqlQueries = "pql_queries_total"
MetricSqlQueries = "sql_queries_total"
)

3
rbf.go
View file

@ -570,7 +570,8 @@ func (w *RbfDBWrapper) DeleteFragment(index, field, view string, shard uint64, f
}
func (w *RbfDBWrapper) DeleteDBPath(dbs *DBShard) error {
panic("TODO")
path := dbs.pathForType(rbfTxn)
return os.RemoveAll(path)
}
func (w *RbfDBWrapper) OpenListString() (r string) {

16
rbf/cursorx.go
View file

@ -150,22 +150,25 @@ func (c *Cursor) CurrentPageType() int {
return cell.Type
}
func toContainer(l leafCell, tx *Tx) *roaring.Container {
func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
orig := l.Data
var cpMaybe []byte
var mapped bool
if EnableRowCache || DoAllocZero {
// make a copy, otherwise the rowCache will see corrupted data
// or mmapped data that may disappear.
cpMaybe = make([]byte, len(orig))
copy(cpMaybe, orig)
mapped = false
} else {
// not a copy
cpMaybe = orig
mapped = true
}
switch l.Type {
case ContainerTypeArray:
return roaring.NewContainerArray(toArray16(cpMaybe))
c = roaring.NewContainerArray(toArray16(cpMaybe))
case ContainerTypeBitmapPtr:
_, bm, _ := tx.leafCellBitmap(toPgno(cpMaybe))
cloneMaybe := bm
@ -173,13 +176,14 @@ func toContainer(l leafCell, tx *Tx) *roaring.Container {
cloneMaybe = make([]uint64, len(bm))
copy(cloneMaybe, bm)
}
return roaring.NewContainerBitmap(l.N, cloneMaybe)
c = roaring.NewContainerBitmap(l.N, cloneMaybe)
case ContainerTypeBitmap:
return roaring.NewContainerBitmap(l.N, toArray64(cpMaybe))
c = roaring.NewContainerBitmap(l.N, toArray64(cpMaybe))
case ContainerTypeRLE:
return roaring.NewContainerRun(toInterval16(cpMaybe))
c = roaring.NewContainerRun(toInterval16(cpMaybe))
}
return nil
c.SetMapped(mapped)
return c
}
type Nodetype int

4
rbf/db.go
View file

@ -800,6 +800,8 @@ func (db *DB) removeTx(tx *Tx) error {
db.mu.Lock()
defer db.mu.Unlock()
delete(tx.db.txs, tx)
// Write pages from WAL to DB.
// TODO(bbj): Move this to an async goroutine.
if tx.writable {
@ -808,8 +810,6 @@ func (db *DB) removeTx(tx *Tx) error {
}
}
delete(tx.db.txs, tx)
// Disassociate from db.
tx.db = nil

19
rbf/db_test.go
View file

@ -92,14 +92,25 @@ func TestDB_Recovery(t *testing.T) {
}
// Add one additional bit in a second transaction.
if tx, err := db.Begin(true); err != nil {
tx0, err := db.Begin(true)
if err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", uint64(len(a))); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
} else if _, err := tx0.Add("x", uint64(len(a))); err != nil {
t.Fatal(err)
}
// Start a read-only transaction so the write tx does not checkpoint the WAL.
tx1, err := db.Begin(false)
if err != nil {
t.Fatal(err)
}
// Commit write transaction.
if err := tx0.Commit(); err != nil {
t.Fatal(err)
}
tx1.Rollback()
// Close database & truncate WAL to remove commit page & bitmap data page.
segment := db.ActiveWALSegment()
if err := db.Close(); err != nil {

7
roaring/container_stash.go
View file

@ -309,6 +309,13 @@ func (c *Container) setMapped(mapped bool) {
}
}
// SetMapped marks a container as "mapped"; do this if you're setting a
// container's storage to something that it shouldn't write to, like mmapped
// memory.
func (c *Container) SetMapped(mapped bool) {
c.setMapped(mapped)
}
// setDirty marks a container as "dirty" -- we don't trust container's n.
// this should never happen except for bitmaps.
func (c *Container) setDirty(dirty bool) {

29
roaring/roaring.go
View file

@ -3300,33 +3300,8 @@ func (c *Container) arrayRemove(v uint16) (*Container, bool) {
}
c = c.Thaw()
array = c.array()
const needCopyOnWriteDueToReadOnlyMmap = true
// TODO(jea) quick benchmarks don't show less performance if needCopyOnWriteDueToReadOnlyMmap
// is true, but we may need more rigorous measurement.
//
// Don't have a COW:
// all benchmarks in roaring/
// ok github.com/pilosa/pilosa/v2/roaring 217.138s
//
// ok, have a COW:
// all benchmarks in roaring/
// ok github.com/pilosa/pilosa/v2/roaring 214.295s
if needCopyOnWriteDueToReadOnlyMmap {
n := len(array)
array2 := make([]uint16, n-1)
copy(array2, array[:i])
copy(array2[i:], array[i+1:])
c.setArray(array2)
} else {
// seg fault here with read-only mmap; go 1.14.7 linux.
// example: cap = 7 i = 0 len = 7
// the append tries to write read-only memory?
array = append(array[:i], array[i+1:]...)
c.setArray(array)
}
array = append(array[:i], array[i+1:]...)
c.setArray(array)
return c, true
}

15
rrtx.go
View file

@ -21,6 +21,7 @@ import (
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"sync/atomic"
@ -81,12 +82,19 @@ func (tx *RoaringTx) SliceOfShards(index, field, view, optionalViewPath string)
}
for _, fi := range fis {
//vv("rrtx next fi = '%v'", fi.Name())
if fi.IsDir() {
continue
}
name := fi.Name()
if strings.HasSuffix(name, ".cache") {
continue
}
// Parse filename into integer.
shard, err := strconv.ParseUint(filepath.Base(fi.Name()), 10, 64)
shard, err := strconv.ParseUint(filepath.Base(name), 10, 64)
if err != nil {
//vv("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", index, field, view, fi.Name())
//panic(fmt.Sprintf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", index, field, view, fi.Name()))
//tx.Index.holder.Logger.Debugf("WARNING: couldn't use non-integer file as shard in index/field/view %s/%s/%s: %s", index, field, view, fi.Name())
continue
@ -552,10 +560,13 @@ func (w *RoaringWrapper) IsClosed() (closed bool) {
}
func (w *RoaringWrapper) DeleteDBPath(dbs *DBShard) (err error) {
return os.RemoveAll(dbs.Path)
//vv("RoaringWrapper.DeleteDBPath called on dbs = '%#v'", dbs)
path := dbs.pathForType(roaringTxn)
return os.RemoveAll(path)
}
func (w *RoaringWrapper) DeleteField(index, field, fieldPath string) error {
//vv("RoaringWrapper.DeleteField(index = '%v', field = '%v', fieldPath = '%v'", index, field, fieldPath)
// match txn sn count vs lmdb/etc.
atomic.AddInt64(&globalNextTxSnRoaring, 1)

3
server/grpc.go
View file

@ -167,6 +167,7 @@ func (h *GRPCHandler) DeleteVDS(ctx context.Context, req *pb.DeleteVDSRequest) (
}
func (h *GRPCHandler) execSQL(ctx context.Context, queryStr string) (pb.ToRowser, error) {
h.stats.Count(pilosa.MetricSqlQueries, 1, 1)
return execSQL(ctx, h.api, h.logger, queryStr)
}
@ -243,6 +244,7 @@ func (h *GRPCHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQ
durFormat := time.Since(t)
h.stats.Timing(pilosa.MetricGRPCStreamQueryDurationSeconds, durQuery, 0.1)
h.stats.Timing(pilosa.MetricGRPCStreamFormatDurationSeconds, durFormat, 0.1)
h.stats.Count(pilosa.MetricPqlQueries, 1, 1)
return errToStatusError(nil)
}
@ -284,6 +286,7 @@ func (h *GRPCHandler) QueryPQLUnary(ctx context.Context, req *pb.QueryPQLRequest
h.stats.Timing(pilosa.MetricGRPCUnaryQueryDurationSeconds, durQuery, 0.1)
h.stats.Timing(pilosa.MetricGRPCUnaryFormatDurationSeconds, durFormat, 0.1)
h.stats.Count(pilosa.MetricPqlQueries, 1, 1)
return table, errToStatusError(nil)
}

17
server/handler_test.go
View file

@ -381,6 +381,23 @@ func TestHandler_Endpoints(t *testing.T) {
}
})
t.Run("Metrics", func(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/metrics", nil))
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
})
t.Run("Metrics.json", func(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/metrics.json", nil))
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
mustJSONDecode(t, w.Body)
})
t.Run("Abort no resize job", func(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/cluster/resize/abort", nil))

2
testdata/schema.json vendored
View file

@ -287,7 +287,7 @@
{
"name": "sites",
"options": {
"keys": true,
"keys": false,
"trackExistence": true
},
"fields": [

12
tournament.sh Executable file
View file

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

2
tx_test.go
View file

@ -63,7 +63,7 @@ func skipForRoaring(t *testing.T) {
src := os.Getenv("PILOSA_TXSRC")
// once txfactory.go DefaultTxsrc != RoaringTxn, this
// will break, of course. Take out the src == "" below.
if src == "" || strings.Contains(src, "roaring") {
if (src == "" && pilosa.DefaultTxsrc == pilosa.RoaringTxn) || strings.Contains(src, "roaring") {
t.Skip("skip if roaring pseudo-txn involved -- won't show transactional rollback")
}
}

4
txfactory.go
View file

@ -435,6 +435,10 @@ func (ty txtype) FileSuffix() string {
}
func (txf *TxFactory) IsTxDatabasePath(path string) bool {
if strings.HasSuffix(filepath.Base(path), ".txstores@@@") {
// top level dir
return true
}
for _, ty := range allTypesWithSuffixes {
if strings.HasSuffix(path, ty.FileSuffix()) {
return true

2
txfactory_internal_test.go
View file

@ -112,6 +112,7 @@ func Test_TxFactory_Qcx_query_context(t *testing.T) {
// and b) we have an easy migration mechanism, to go from one storage format to another.
//
func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
t.Skip("TODO(jea) bring this back in. broken by the local vs remote shard determination for a cluster")
orig := os.Getenv("PILOSA_TXSRC")
defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests!
@ -231,6 +232,7 @@ func Test_TxFactory_UpdateBlueFromGreen_OnStartup(t *testing.T) {
// go to verify it but blue has more data than green.
// That will also cause query divergence.
func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
t.Skip("TODO(jea) bring this back in. broken by the local vs remote shard determination for a cluster")
orig := os.Getenv("PILOSA_TXSRC")
defer os.Setenv("PILOSA_TXSRC", orig) // must restore or will mess up other tests!

1
util.go
View file

@ -103,6 +103,7 @@ func mapDiff(mapA, mapB map[uint64]bool) (r []int) {
r = append(r, int(a))
}
}
sort.Ints(r)
return
}