From 411a7dbcd461c4bc6a79b1eab8630fc8bba46da5 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Thu, 14 Jan 2016 10:44:50 -0800 Subject: [PATCH] bitmap attributes This commit adds storage for bitmap attributes and adds a Frame type to contain them. --- frame.go | 227 ++++++++++++++++++++++++++++++++++++++++++++++++++ frame_test.go | 127 ++++++++++++++++++++++++++++ handler.go | 2 +- index.go | 51 +++++------- 4 files changed, 376 insertions(+), 31 deletions(-) create mode 100644 frame.go create mode 100644 frame_test.go diff --git a/frame.go b/frame.go new file mode 100644 index 000000000..09f77e9ef --- /dev/null +++ b/frame.go @@ -0,0 +1,227 @@ +package pilosa + +import ( + "encoding/binary" + "encoding/json" + "os" + "path/filepath" + "strconv" + "sync" + "time" + + "github.com/boltdb/bolt" +) + +// Frame represents a container for fragments. +type Frame struct { + mu sync.Mutex + path string + db string + name string + + // Fragments + fragments map[uint64]*Fragment + + // Attribute storage and cache + store *bolt.DB + battrs map[uint64]map[string]interface{} +} + +// NewFrame returns a new instance of frame. +func NewFrame(path, db, name string) *Frame { + return &Frame{ + path: path, + db: db, + name: name, + + fragments: make(map[uint64]*Fragment), + + battrs: make(map[uint64]map[string]interface{}), + } +} + +// Open opens and initializes the frame. +func (f *Frame) Open() error { + // Ensure the frame's path exists. + if err := os.MkdirAll(f.path, 0777); err != nil { + return err + } + + // Open attribute store. + store, err := bolt.Open(filepath.Join(f.path, "data"), 0666, &bolt.Options{Timeout: 1 * time.Second}) + if err != nil { + return err + } + f.store = store + + // Initialize database. + if err := f.store.Update(func(tx *bolt.Tx) error { + if _, err := tx.CreateBucketIfNotExists([]byte("battrs")); err != nil { + return err + } + return nil + }); err != nil { + _ = f.Close() + return err + } + + return nil +} + +// Close closes the frame and its fragments. +func (f *Frame) Close() error { + f.mu.Lock() + defer f.mu.Unlock() + + // Close the attribute store. + if f.store != nil { + _ = f.store.Close() + } + + // Close all fragments. + for _, frag := range f.fragments { + _ = frag.Close() + } + f.fragments = make(map[uint64]*Fragment) + + return nil +} + +// Path returns the path the frame was initialized with. +func (f *Frame) Path() string { return f.path } + +// FragmentPath returns the path to a fragment in the frame. +func (f *Frame) FragmentPath(slice uint64) string { + return filepath.Join(f.path, strconv.FormatUint(slice, 10)) +} + +// Fragment returns a fragment in the frame by slice. +func (f *Frame) Fragment(slice uint64) *Fragment { + f.mu.Lock() + defer f.mu.Unlock() + return f.fragment(slice) +} + +func (f *Frame) fragment(slice uint64) *Fragment { return f.fragments[slice] } + +// CreateFragmentIfNotExists returns a fragment in the frame by slice. +func (f *Frame) CreateFragmentIfNotExists(slice uint64) (*Fragment, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.createFragmentIfNotExists(slice) +} + +func (f *Frame) createFragmentIfNotExists(slice uint64) (*Fragment, error) { + // Find fragment in cache first. + if frag := f.fragments[slice]; frag != nil { + return frag, nil + } + + // Initialize and open fragment. + frag := NewFragment(f.FragmentPath(slice), f.db, f.name, slice) + if err := frag.Open(); err != nil { + return nil, err + } + f.fragments[slice] = frag + + return frag, nil +} + +// BitmapAttrs returns the value of the attribute for a bitmap. +func (f *Frame) BitmapAttrs(id uint64) (m map[string]interface{}, err error) { + f.mu.Lock() + defer f.mu.Unlock() + + // Check cache for map. + if m = f.battrs[id]; m != nil { + return m, nil + } + + // Find attributes from storage. + if err = f.store.View(func(tx *bolt.Tx) error { + m, err = f.bitmapAttrs(tx, id) + if err != nil { + return err + } + return nil + }); err != nil { + return nil, err + } + + // Add to cache. + f.battrs[id] = m + + return +} + +// SetBitmapAttr sets attribute values for a bitmap. +func (f *Frame) SetBitmapAttrs(id uint64, m map[string]interface{}) error { + f.mu.Lock() + defer f.mu.Unlock() + + var battr map[string]interface{} + if err := f.store.Update(func(tx *bolt.Tx) error { + attr, err := f.bitmapAttrs(tx, id) + if err != nil { + return err + } + battr = attr + + // Create a new map if it is empty so we don't update emptyMap. + if len(battr) == 0 { + battr = make(map[string]interface{}, len(m)) + } + + // Merge attributes with original values. + // Nil values should delete keys. + for k, v := range m { + if v == nil { + delete(battr, k) + } else { + battr[k] = v + } + } + + // Marshal and save new values. + buf, err := json.Marshal(battr) + if err != nil { + return err + } + if err := tx.Bucket([]byte("battrs")).Put(u64tob(id), buf); err != nil { + return err + } + return nil + }); err != nil { + return err + } + + // Swap attributes map in cache. + f.battrs[id] = battr + + return nil +} + +// bitmapAttrs returns a map of attributes for a bitmap. +func (f *Frame) bitmapAttrs(tx *bolt.Tx, id uint64) (map[string]interface{}, error) { + if v := tx.Bucket([]byte("battrs")).Get(u64tob(id)); v != nil { + m := make(map[string]interface{}) + if err := json.Unmarshal(v, &m); err != nil { + return nil, err + } + return m, nil + } + return emptyMap, nil +} + +// u64tob encodes v to big endian encoding. +func u64tob(v uint64) []byte { + b := make([]byte, 8) + binary.BigEndian.PutUint64(b, v) + return b +} + +// btou64 decodes b from big endian encoding. +func btou64(b []byte) uint64 { return binary.BigEndian.Uint64(b) } + +// emptyMap is a reusable map that contains no keys. +var emptyMap = make(map[string]interface{}) diff --git a/frame_test.go b/frame_test.go new file mode 100644 index 000000000..391004982 --- /dev/null +++ b/frame_test.go @@ -0,0 +1,127 @@ +package pilosa_test + +import ( + "io/ioutil" + "os" + "reflect" + "testing" + + "github.com/umbel/pilosa" +) + +// Ensure frame can open and retrieve a fragment. +func TestFrame_CreateFragmentIfNotExists(t *testing.T) { + f := MustOpenFrame() + defer f.Close() + + // Create fragment. + frag, err := f.CreateFragmentIfNotExists(100) + if err != nil { + t.Fatal(err) + } else if frag == nil { + t.Fatal("expected fragment") + } + + // Retrieve existing fragment. + frag2, err := f.CreateFragmentIfNotExists(100) + if err != nil { + t.Fatal(err) + } else if frag != frag2 { + t.Fatal("fragment mismatch") + } + + if frag != f.Fragment(100) { + t.Fatal("fragment mismatch") + } +} + +// Ensure frame can set and retrieve bitmap attributes. +func TestFrame_BitmapAttrs(t *testing.T) { + f := MustOpenFrame() + defer f.Close() + + // Set attributes. + if err := f.SetBitmapAttrs(1, map[string]interface{}{"A": float64(100)}); err != nil { + t.Fatal(err) + } else if err := f.SetBitmapAttrs(2, map[string]interface{}{"A": float64(200)}); err != nil { + t.Fatal(err) + } else if err := f.SetBitmapAttrs(1, map[string]interface{}{"B": "VALUE"}); err != nil { + t.Fatal(err) + } + + // Retrieve attributes for bitmap #1. + if m, err := f.BitmapAttrs(1); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(m, map[string]interface{}{"A": float64(100), "B": "VALUE"}) { + t.Fatalf("unexpected attrs(1): %#v", m) + } + + // Retrieve attributes for bitmap #2. + if m, err := f.BitmapAttrs(2); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(m, map[string]interface{}{"A": float64(200)}) { + t.Fatalf("unexpected attrs(2): %#v", m) + } +} + +// Ensure frame returns a non-nil empty map if unset. +func TestFrame_BitmapAttrs_Empty(t *testing.T) { + f := MustOpenFrame() + defer f.Close() + + if m, err := f.BitmapAttrs(100); err != nil { + t.Fatal(err) + } else if m == nil || len(m) > 0 { + t.Fatalf("unexpected attrs: %#v", m) + } +} + +// Ensure frame can unset attributes if explicitly set to nil. +func TestFrame_BitmapAttrs_Unset(t *testing.T) { + f := MustOpenFrame() + defer f.Close() + + // Set attributes. + if err := f.SetBitmapAttrs(1, map[string]interface{}{"A": "X", "B": "Y"}); err != nil { + t.Fatal(err) + } else if err := f.SetBitmapAttrs(1, map[string]interface{}{"B": nil}); err != nil { + t.Fatal(err) + } + + // Verify attributes. + if m, err := f.BitmapAttrs(1); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(m, map[string]interface{}{"A": "X"}) { + t.Fatalf("unexpected attrs: %#v", m) + } +} + +// Frame represents a test wrapper for pilosa.Frame. +type Frame struct { + *pilosa.Frame +} + +// NewFrame returns a new instance of Frame d/0. +func NewFrame() *Frame { + path, err := ioutil.TempDir("", "pilosa-frame-") + if err != nil { + panic(err) + } + + return &Frame{Frame: pilosa.NewFrame(path, "d", "f")} +} + +// MustOpenFrame returns a new, opened frame at a temporary path. Panic on error. +func MustOpenFrame() *Frame { + f := NewFrame() + if err := f.Open(); err != nil { + panic(err) + } + return f +} + +// Close closes the frame and removes the underlying data. +func (f *Frame) Close() error { + defer os.RemoveAll(f.Path()) + return f.Frame.Close() +} diff --git a/handler.go b/handler.go index 1937a4519..cacbeef51 100644 --- a/handler.go +++ b/handler.go @@ -267,7 +267,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { // Find the correct fragment. f, err := h.Index.CreateFragmentIfNotExists(db, frame, slice) if err != nil { - h.logger().Printf("fragment error: db=%s, frame=%s, slice=%s, err=%s", db, frame, slice, err) + h.logger().Printf("fragment error: db=%s, frame=%s, slice=%d, err=%s", db, frame, slice, err) http.Error(w, "fragment error", http.StatusInternalServerError) return } diff --git a/index.go b/index.go index 5bf25de59..ad1aa6270 100644 --- a/index.go +++ b/index.go @@ -11,17 +11,18 @@ import ( // Index represents a container for fragments. type Index struct { - mu sync.Mutex - path string - sliceN uint64 - fragments map[fragmentKey]*Fragment + mu sync.Mutex + path string + sliceN uint64 + + frames map[frameKey]*Frame } // NewIndex returns a new instance of Index. func NewIndex(path string) *Index { return &Index{ - path: path, - fragments: make(map[fragmentKey]*Fragment), + path: path, + frames: make(map[frameKey]*Frame), } } @@ -112,9 +113,9 @@ func (i *Index) openFrame(db, frame string) error { // Close closes all open fragments. func (i *Index) Close() error { - for key, f := range i.fragments { + for key, f := range i.frames { if err := f.Close(); err != nil { - log.Println("error closing fragment(%v): %s", key, err) + log.Println("error closing frame(%s/%s): %s", key.db, key.frame, err) } } return nil @@ -130,16 +131,14 @@ func (i *Index) SliceN() uint64 { return i.sliceN } -// FragmentPath returns the path where a given fragment is stored. -func (i *Index) FragmentPath(db, frame string, slice uint64) string { - return filepath.Join(i.path, db, frame, strconv.FormatUint(slice, 10)) -} +// FramePath returns the path where a given frame is stored. +func (i *Index) FramePath(db, frame string) string { return filepath.Join(i.path, db, frame) } // Fragment returns the fragment for a database, frame & slice. func (i *Index) Fragment(db, frame string, slice uint64) *Fragment { i.mu.Lock() defer i.mu.Unlock() - return i.fragments[fragmentKey{db, frame, slice}] + return i.frames[frameKey{db, frame}].fragment(slice) } // CreateFragmentIfNotExists returns the fragment for a database, frame & slice. @@ -153,30 +152,22 @@ func (i *Index) CreateFragmentIfNotExists(db, frame string, slice uint64) (*Frag i.sliceN = slice } - // Create fragment, if not exists. - key := fragmentKey{db, frame, slice} - if i.fragments[key] == nil { - path := i.FragmentPath(db, frame, slice) - - // Create parent directory, if necessary. - if err := os.MkdirAll(filepath.Dir(path), 0777); err != nil { - return nil, fmt.Errorf("parent fragment dir: %s", err) - } - - // Initialize and open fragment. - f := NewFragment(path, db, frame, slice) + // Create frame, if not exists. + key := frameKey{db, frame} + if i.frames[key] == nil { + f := NewFrame(i.FramePath(db, frame), db, frame) if err := f.Open(); err != nil { return nil, err } - i.fragments[key] = f + i.frames[key] = f } - return i.fragments[key], nil + // Create fragment, if not exists. + return i.frames[key].createFragmentIfNotExists(slice) } -// fragmentKey is the map key for fragment look ups. -type fragmentKey struct { +// frameKey is the map key for frame look ups. +type frameKey struct { db string frame string - slice uint64 }