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

Merge pull request #963 from molecula/rbf_config

Browse source 
rbf: add DBConfig
This commit is contained in:
jaten-molecula 2020-10-14 20:12:46 -05:00 • committed by GitHub
commit fa21c81a27
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 38 additions and 191 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
cmd/convert/main.go
View file

@ -1,43 +0,0 @@
// 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 main
import (
"log"
"os"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/rbf"
)
func main() {
if len(os.Args) != 3 {
log.Fatal("USAGE convert srcPath destPath")
}
holder := pilosa.NewHolder(os.Args[1], nil)
err := holder.Open()
if err != nil {
log.Fatal(err)
}
c := &pilosa.RBFConverter{
Dbs: make(map[string]*rbf.DB),
Base: os.Args[2],
}
holder.ConvertToRBF(c)
}

8
rbf.go
View file

@ -153,12 +153,8 @@ func (r *rbfDBRegistrar) OpenDBWrapper(path0 string, doAllocZero bool) (DBWrappe
// creates the effect of having only one DB open per pilosa node.
return w, nil
}
var db *rbf.DB
if doAllocZero {
db = rbf.NewDBWithAllocZero(path)
} else {
db = rbf.NewDB(path)
}
db := rbf.NewDB(path)
db.DoAllocZero = doAllocZero
w = &RbfDBWrapper{
reg: r,

9
rbf/cursorx.go
View file

@ -28,13 +28,6 @@ import (
// directly, but only a copy.
const EnableRowCache = true
// DoAllocZero means we copy mmap read data and
// wipe it afterwards to catch retention of data
// past Tx.Rollback which was a big problem.
// This should be set by NewDBWithAllocZero and never changed
// afterwards in order to avoid a data race.
var DoAllocZero bool
//probably should just implement the container interface
// but for now i'll do it
func (c *Cursor) Rows() ([]uint64, error) {
@ -158,7 +151,7 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
orig := l.Data
var cpMaybe []byte
var mapped bool
if EnableRowCache || DoAllocZero {
if EnableRowCache || tx.db.DoAllocZero {
// make a copy, otherwise the rowCache will see corrupted data
// or mmapped data that may disappear.
cpMaybe = make([]byte, len(orig))

31
rbf/db.go
View file

@ -38,6 +38,8 @@ const (
MaxWALSegmentFileSize = 10 * (1 << 20)
)
// DB options like MaxSize, FsyncEnabled, DoAllocZero
// can be set before calling DB.Open().
type DB struct {
data []byte // mmap data
file *os.File // file descriptor
@ -58,24 +60,23 @@ type DB struct {
// The maximum allowed database size. Required by mmap.
MaxSize int64
}
// NewDBWithAllocZero sets DoAllocZero true and
// returns a new instance of DB. We set it here
// to avoid a data race afterwards.
func NewDBWithAllocZero(path string) *DB {
DoAllocZero = true
return NewDB(path)
// Set before calling db.Open()
FsyncEnabled bool
// for mmap correctness testing.
DoAllocZero bool
}
// NewDB returns a new instance of DB.
func NewDB(path string) *DB {
db := &DB{
txs: make(map[*Tx]struct{}),
pageMap: immutable.NewMap(&uint32Hasher{}),
wcache: make([]byte, MaxWALSegmentFileSize+PageSize),
Path: path,
MaxSize: DefaultMaxSize,
txs: make(map[*Tx]struct{}),
pageMap: immutable.NewMap(&uint32Hasher{}),
wcache: make([]byte, MaxWALSegmentFileSize+PageSize),
Path: path,
MaxSize: DefaultMaxSize,
FsyncEnabled: true,
}
return db
}
@ -169,7 +170,7 @@ func (db *DB) openWALSegments() error {
continue
}
segment := NewWALSegment(filepath.Join(db.WALPath(), fi.Name()))
segment := db.NewWALSegment(filepath.Join(db.WALPath(), fi.Name()))
if err := segment.Open(); err != nil {
_ = db.closeWALSegments()
return err
@ -180,7 +181,7 @@ func (db *DB) openWALSegments() error {
// Truncate everything after the last successful meta page.
if walID, err := findLastWALMetaPage(db.segments); err != nil {
return err
} else if db.segments, err = truncateWALAfter(db.segments, walID); err != nil {
} else if db.segments, err = db.truncateWALAfter(db.segments, walID); err != nil {
return err
}
@ -306,7 +307,7 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
}
// Ensure WAL pages are fully copied & synced to DB file.
if err := fsync(db.file); err != nil {
if err := db.fsync(db.file); err != nil {
return fmt.Errorf("db file sync: %w", err)
}

13
rbf/rbf.go
View file

@ -91,11 +91,6 @@ var (
// Debug is just a temporary flag used for debugging.
var Debug bool
// Testing constants.
const (
SyncEnabled = true
)
// Magic32 returns the magic bytes as a big endian encoded uint32.
func Magic32() uint32 {
return binary.BigEndian.Uint32([]byte(Magic))
@ -655,7 +650,7 @@ func RowValues(b []uint64) []uint64 {
// }
// truncate truncates the file at path to sz bytes. File must exist.
func truncate(path string, sz int64) error {
func (db *DB) truncate(path string, sz int64) error {
f, err := os.OpenFile(path, os.O_WRONLY, 0666)
if err != nil {
return fmt.Errorf("open file: %w", err)
@ -664,14 +659,14 @@ func truncate(path string, sz int64) error {
if err := f.Truncate(sz); err != nil {
return fmt.Errorf("truncate: %w", err)
} else if err := fsync(f); err != nil {
} else if err := db.fsync(f); err != nil {
return fmt.Errorf("sync: %w", err)
}
return f.Close()
}
func fsync(f *os.File) error {
if !SyncEnabled {
func (db *DB) fsync(f *os.File) error {
if !db.FsyncEnabled {
return nil
}
return f.Sync()

6
rbf/tx.go
View file

@ -122,7 +122,7 @@ func (tx *Tx) Rollback() {
// TODO(bbj): Invalidate DB if rollback fails. Possibly attempt reopen?
if tx.dirty {
if _, err := truncateWALAfter(tx.segments, tx.walID); err != nil {
if _, err := tx.db.truncateWALAfter(tx.segments, tx.walID); err != nil {
panicOn(err)
}
tx.segments = nil
@ -1610,7 +1610,7 @@ func (tx *Tx) flushWALWriter() error {
// Flush cache to writer.
if _, err := w.WriteAt(tx.wcache, int64(segment.PageN)*PageSize); err != nil {
return fmt.Errorf("write wal segment: %w", err)
} else if err := fsync(w); err != nil {
} else if err := tx.db.fsync(w); err != nil {
return fmt.Errorf("sync wal segment: %w", err)
} else if err := w.Close(); err != nil {
return fmt.Errorf("close wal segment: %w", err)
@ -1686,7 +1686,7 @@ func (tx *Tx) ensureWritableWALSegment() error {
}
// Create new segment file.
s := NewWALSegment(filepath.Join(tx.db.WALPath(), FormatWALSegmentPath(base)))
s := tx.db.NewWALSegment(filepath.Join(tx.db.WALPath(), FormatWALSegmentPath(base)))
if err := s.Open(); err != nil {
return fmt.Errorf("add wal segment: %w", err)
}

10
rbf/wal.go
View file

@ -26,6 +26,7 @@ import (
// WALSegment represents a single file in the WAL.
type WALSegment struct {
db *DB
Path string // path to file
MinWALID int64 // base WALID; calculated from path
PageN int // number of written pages
@ -34,8 +35,9 @@ type WALSegment struct {
}
// NewWALSegment returns a new instance of WALSegment for a given path.
func NewWALSegment(path string) WALSegment {
func (db *DB) NewWALSegment(path string) WALSegment {
return WALSegment{
db: db,
Path: path,
}
}
@ -74,7 +76,7 @@ func (s *WALSegment) Open() (err error) {
s.PageN = int(sz / PageSize)
if sz%PageSize != 0 {
sz = int64(s.PageN * PageSize)
if err := truncate(s.Path, sz); err != nil {
if err := s.db.truncate(s.Path, sz); err != nil {
return fmt.Errorf("truncate wal file: %w", err)
}
}
@ -212,7 +214,7 @@ func findLastWALMetaPage(segments []WALSegment) (walID int64, err error) {
}
// truncateWALAfter removes all pages in the WAL after walID.
func truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error) {
func (db *DB) truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error) {
var newSegments []WALSegment
for i := range segments {
@ -229,7 +231,7 @@ func truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error)
newSegment := *segment
newSegment.PageN = int((walID - newSegment.MinWALID) + 1)
if err := truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
if err := db.truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
return segments, err
}
newSegments = append(newSegments, newSegment)

9
rbf/wal_test.go
View file

@ -28,7 +28,10 @@ import (
func TestWALSegment_Open(t *testing.T) {
t.Run("OK", func(t *testing.T) {
s := MustOpenWALSegment(t, 10)
db := MustOpenDB(t)
defer MustCloseDB(t, db)
s := MustOpenWALSegment(t, db, 10)
defer MustCloseWALSegment(t, s)
if got, want := s.MinWALID, int64(10); got != want {
t.Fatalf("Base()=%d, want %d", got, want)
@ -153,7 +156,7 @@ func benchmarkWALSegment_WriteWALPage(b *testing.B, flushSize int) {
*/
// MustOpenWALSegment opens a WAL segment in a temporary path. Fails on error.
func MustOpenWALSegment(tb testing.TB, walID int64) rbf.WALSegment {
func MustOpenWALSegment(tb testing.TB, db *rbf.DB, walID int64) rbf.WALSegment {
tb.Helper()
dir, err := ioutil.TempDir("", "")
@ -165,7 +168,7 @@ func MustOpenWALSegment(tb testing.TB, walID int64) rbf.WALSegment {
tb.Fatal(err)
}
s := rbf.NewWALSegment(path)
s := db.NewWALSegment(path)
if err := s.Open(); err != nil {
tb.Fatal(err)
}

100
xrbrsupport.go
View file

@ -1,100 +0,0 @@
// 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 pilosa
import (
"fmt"
"github.com/pilosa/pilosa/v2/rbf"
"github.com/pilosa/pilosa/v2/roaring"
)
type Converter interface {
Convert(index, field, view string, shard uint64, rb *roaring.Bitmap) error
Shutdown()
}
type RBFConverter struct {
Dbs map[string]*rbf.DB
Base string
}
func (rbc *RBFConverter) GetOrCreateDB(index string, shard uint64) (*rbf.DB, error) {
key := fmt.Sprintf("%s/%d", index, shard)
db, found := rbc.Dbs[key]
if found {
return db, nil
}
path := rbc.Base + "/" + key
db = rbf.NewDB(path)
err := db.Open()
if err != nil {
return nil, err
}
rbc.Dbs[key] = db
return db, nil
}
func (rbc *RBFConverter) Shutdown() {
for key, db := range rbc.Dbs {
fmt.Println("Shutdown", key)
db.Close()
}
}
func (rbc *RBFConverter) Convert(index, field, view string, shard uint64, rb *roaring.Bitmap) error {
fmt.Println("CONVERT", index, field, view, shard)
db, err := rbc.GetOrCreateDB(index, shard)
if err != nil {
return err
}
tx, err := db.Begin(writable)
if err != nil {
return err
}
defer tx.Rollback()
name := fmt.Sprintf("%s/%s", field, view)
err = tx.CreateBitmap(name)
if err != nil {
return err
}
_, err = tx.AddRoaring(name, rb)
if err != nil {
return err
}
return tx.Commit()
}
func (h *Holder) ConvertToRBF(c Converter) {
/*
for idxname, idx := range h.indexes {
for fieldName, field := range idx.fields {
for _, view := range field.views() {
for shard, fragment := range view.fragments {
panic("NEED bitmap from storage")
junk := roaring.NewBitmap()
err := c.Convert(idxname, fieldName, view.name, shard, junk)
if err != nil {
fmt.Println("ERR", err, fragment.shard) //just added shard for compile
}
}
}
}
}
c.Shutdown()
*/
}