diff --git a/bitmap.go b/bitmap.go index 4e0db6d99..c9b93bfb7 100644 --- a/bitmap.go +++ b/bitmap.go @@ -25,6 +25,9 @@ var CounterKey = int64(-1) type Bitmap struct { tree *rbtree.Tree bcount uint64 + + // Attributes associated with the bitmap. + Attrs map[string]interface{} } // NewBitmap returns a new instance of Bitmap. @@ -309,7 +312,20 @@ func (b *Bitmap) ReadFrom(r io.Reader) (n int64, err error) { } // MarshalJSON returns a JSON-encoded byte slice of b. -func (b *Bitmap) MarshalJSON() ([]byte, error) { return json.Marshal(b.Bits()) } +func (b *Bitmap) MarshalJSON() ([]byte, error) { + var o struct { + Attrs map[string]interface{} `json:"attrs"` + Bits []uint64 `json:"bits"` + } + o.Bits = b.Bits() + + o.Attrs = b.Attrs + if o.Attrs == nil { + o.Attrs = make(map[string]interface{}) + } + + return json.Marshal(&o) +} // MarshalBinary returns a gob-encoded byte slice of b. func (b *Bitmap) MarshalBinary() ([]byte, error) { @@ -402,7 +418,13 @@ func (b *Bitmap) BitCount() uint64 { // encodeBitmap converts b into its internal representation. func encodeBitmap(b *Bitmap) *internal.Bitmap { - pb := &internal.Bitmap{} + if b == nil { + return nil + } + + pb := &internal.Bitmap{ + Attrs: encodeAttrs(b.Attrs), + } for i := b.tree.Min(); !i.Limit(); i = i.Next() { pb.Chunks = append(pb.Chunks, encodeChunk(i.Item().(*Chunk))) } @@ -411,7 +433,12 @@ func encodeBitmap(b *Bitmap) *internal.Bitmap { // decodeBitmap converts b from its internal representation. func decodeBitmap(pb *internal.Bitmap) *Bitmap { + if pb == nil { + return nil + } + b := NewBitmap() + b.Attrs = decodeAttrs(pb.GetAttrs()) for _, chunk := range pb.GetChunks() { b.AddChunk(decodeChunk(chunk)) } diff --git a/cmd/pilosa/main_test.go b/cmd/pilosa/main_test.go index fe38a231f..9b99e3b72 100644 --- a/cmd/pilosa/main_test.go +++ b/cmd/pilosa/main_test.go @@ -29,9 +29,9 @@ func TestMain_Set_Quick(t *testing.T) { m := MustRunMain() defer m.Close() - // Execute set() commands. + // Execute SetBit() commands. for _, cmd := range cmds { - if res, err := m.Query("d", fmt.Sprintf(`set(id=%d, frame=%q, profile_id=%d)`, cmd.ID, cmd.Frame, cmd.ProfileID)); err != nil { + if res, err := m.Query("db=d", fmt.Sprintf(`SetBit(id=%d, frame=%q, profileID=%d)`, cmd.ID, cmd.Frame, cmd.ProfileID)); err != nil { t.Fatal(err) } else if res != `{}`+"\n" { t.Fatalf("unexpected result: %s", res) @@ -41,8 +41,13 @@ func TestMain_Set_Quick(t *testing.T) { // Validate data. for frame, frameSet := range SetCommands(cmds).Frames() { for id, profileIDs := range frameSet { - exp := MustMarshalJSON(map[string]interface{}{"result": profileIDs}) + "\n" - if res, err := m.Query("d", fmt.Sprintf(`get(id=%d, frame=%q)`, id, frame)); err != nil { + exp := MustMarshalJSON(map[string]interface{}{ + "result": map[string]interface{}{ + "bits": profileIDs, + "attrs": map[string]interface{}{}, + }, + }) + "\n" + if res, err := m.Query("db=d", fmt.Sprintf(`Bitmap(id=%d, frame=%q)`, id, frame)); err != nil { t.Fatal(err) } else if res != exp { t.Fatalf("unexpected result:\n\ngot=%s\n\nexp=%s\n\n", res, exp) @@ -57,8 +62,13 @@ func TestMain_Set_Quick(t *testing.T) { // Validate data after reopening. for frame, frameSet := range SetCommands(cmds).Frames() { for id, profileIDs := range frameSet { - exp := MustMarshalJSON(map[string]interface{}{"result": profileIDs}) + "\n" - if res, err := m.Query("d", fmt.Sprintf(`get(id=%d, frame=%q)`, id, frame)); err != nil { + exp := MustMarshalJSON(map[string]interface{}{ + "result": map[string]interface{}{ + "bits": profileIDs, + "attrs": map[string]interface{}{}, + }, + }) + "\n" + if res, err := m.Query("db=d", fmt.Sprintf(`Bitmap(id=%d, frame=%q)`, id, frame)); err != nil { t.Fatal(err) } else if res != exp { t.Fatalf("unexpected result (reopen):\n\ngot=%s\n\nexp=%s\n\n", res, exp) @@ -76,6 +86,91 @@ func TestMain_Set_Quick(t *testing.T) { } } +// Ensure program can set bitmap attributes and retrieve them. +func TestMain_SetBitmapAttrs(t *testing.T) { + m := MustRunMain() + defer m.Close() + + // Set bits on different bitmaps in different frames. + if _, err := m.Query("db=d", `SetBit(id=1, frame="x.n", profileID=100)`); err != nil { + t.Fatal(err) + } else if _, err := m.Query("db=d", `SetBit(id=2, frame="x.n", profileID=100)`); err != nil { + t.Fatal(err) + } else if _, err := m.Query("db=d", `SetBit(id=2, frame="z", profileID=100)`); err != nil { + t.Fatal(err) + } + + // Set bitmap attributes. + if _, err := m.Query("db=d", `SetBitmapAttrs(id=1, frame="x.n", x=100)`); err != nil { + t.Fatal(err) + } else if _, err := m.Query("db=d", `SetBitmapAttrs(id=2, frame="x.n", x=200)`); err != nil { + t.Fatal(err) + } else if _, err := m.Query("db=d", `SetBitmapAttrs(id=2, frame="z", x=300)`); err != nil { + t.Fatal(err) + } + + // Query bitmap x.n/1. + if res, err := m.Query("db=d", `Bitmap(id=1, frame="x.n")`); err != nil { + t.Fatal(err) + } else if res != `{"result":{"attrs":{"x":100},"bits":[100]}}`+"\n" { + t.Fatalf("unexpected result: %s", res) + } + + // Query bitmap x.n/2. + if res, err := m.Query("db=d", `Bitmap(id=2, frame="x.n")`); err != nil { + t.Fatal(err) + } else if res != `{"result":{"attrs":{"x":200},"bits":[100]}}`+"\n" { + t.Fatalf("unexpected result: %s", res) + } + + if err := m.Reopen(); err != nil { + t.Fatal(err) + } + + // Query bitmap after reopening. + if res, err := m.Query("db=d&profiles=true", `Bitmap(id=1, frame="x.n")`); err != nil { + t.Fatal(err) + } else if res != `{"result":{"attrs":{"x":100},"bits":[100]}}`+"\n" { + t.Fatalf("unexpected result(reopen): %s", res) + } +} + +// Ensure program can set profile attributes and retrieve them. +func TestMain_SetProfileAttrs(t *testing.T) { + m := MustRunMain() + defer m.Close() + + // Set bits on bitmap. + if _, err := m.Query("db=d", `SetBit(id=1, frame="x.n", profileID=100)`); err != nil { + t.Fatal(err) + } else if _, err := m.Query("db=d", `SetBit(id=1, frame="x.n", profileID=101)`); err != nil { + t.Fatal(err) + } + + // Set profile attributes. + if _, err := m.Query("db=d", `SetProfileAttrs(id=100, foo="bar")`); err != nil { + t.Fatal(err) + } + + // Query bitmap. + if res, err := m.Query("db=d&profiles=true", `Bitmap(id=1, frame="x.n")`); err != nil { + t.Fatal(err) + } else if res != `{"result":{"attrs":{},"bits":[100,101]},"profiles":[{"id":100,"attrs":{"foo":"bar"}}]}`+"\n" { + t.Fatalf("unexpected result: %s", res) + } + + if err := m.Reopen(); err != nil { + t.Fatal(err) + } + + // Query bitmap after reopening. + if res, err := m.Query("db=d&profiles=true", `Bitmap(id=1, frame="x.n")`); err != nil { + t.Fatal(err) + } else if res != `{"result":{"attrs":{},"bits":[100,101]},"profiles":[{"id":100,"attrs":{"foo":"bar"}}]}`+"\n" { + t.Fatalf("unexpected result(reopen): %s", res) + } +} + // Ensure the host can be parsed. func TestConfig_Parse_Host(t *testing.T) { if c, err := ParseConfig(`host = "local"`); err != nil { @@ -174,8 +269,8 @@ func (m *Main) Reopen() error { func (m *Main) URL() string { return "http://" + m.Addr().String() } // Query executes a query against the program through the HTTP API. -func (m *Main) Query(db, query string) (string, error) { - resp := MustDo("POST", m.URL()+"/query?db="+db, query) +func (m *Main) Query(rawQuery, query string) (string, error) { + resp := MustDo("POST", m.URL()+"/query?"+rawQuery, query) if resp.StatusCode != http.StatusOK { return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } diff --git a/db.go b/db.go new file mode 100644 index 000000000..246d5f289 --- /dev/null +++ b/db.go @@ -0,0 +1,264 @@ +package pilosa + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sync" + "time" + + "github.com/boltdb/bolt" +) + +// DB represents a container for frames. +type DB struct { + mu sync.Mutex + path string + name string + + // Frames by name. + frames map[string]*Frame + + // Profile attribute storage and cache + store *bolt.DB + attrs map[uint64]map[string]interface{} +} + +// NewDB returns a new instance of DB. +func NewDB(path, name string) *DB { + return &DB{ + path: path, + name: name, + frames: make(map[string]*Frame), + attrs: make(map[uint64]map[string]interface{}), + } +} + +// Name returns name of the database. +func (db *DB) Name() string { return db.name } + +// Path returns the path the database was initialized with. +func (db *DB) Path() string { return db.path } + +// Open opens and initializes the database. +func (db *DB) Open() error { + // Ensure the path exists. + if err := os.MkdirAll(db.path, 0777); err != nil { + return err + } + + if err := db.openFrames(); err != nil { + return err + } + + if err := db.openStore(); err != nil { + return err + } + + return nil +} + +// openFrames opens and initializes the frames inside the database. +func (db *DB) openFrames() error { + f, err := os.Open(db.path) + if err != nil { + return err + } + defer f.Close() + + fis, err := f.Readdir(0) + if err != nil { + return err + } + + for _, fi := range fis { + if !fi.IsDir() { + continue + } + + fr := NewFrame(db.FramePath(filepath.Base(fi.Name())), db.name, filepath.Base(fi.Name())) + if err := fr.Open(); err != nil { + return fmt.Errorf("open frame: name=%s, err=%s", fr.Name(), err) + } + db.frames[fr.Name()] = fr + } + return nil +} + +// openStore opens and initializes the attribute store. +func (db *DB) openStore() error { + // Open attribute store. + store, err := bolt.Open(filepath.Join(db.path, "data"), 0666, &bolt.Options{Timeout: 1 * time.Second}) + if err != nil { + return err + } + db.store = store + + // Initialize database. + if err := db.store.Update(func(tx *bolt.Tx) error { + if _, err := tx.CreateBucketIfNotExists([]byte("attrs")); err != nil { + return err + } + return nil + }); err != nil { + _ = db.Close() + return err + } + + return nil +} + +// Close closes the database and its frames. +func (db *DB) Close() error { + db.mu.Lock() + defer db.mu.Unlock() + + // Close the attribute store. + if db.store != nil { + db.store.Close() + } + + // Close all frames. + for _, f := range db.frames { + f.Close() + } + db.frames = make(map[string]*Frame) + + return nil +} + +// SliceN returns the max slice in the database. +func (db *DB) SliceN() uint64 { + db.mu.Lock() + defer db.mu.Unlock() + + var max uint64 + for _, f := range db.frames { + if slice := f.SliceN(); slice > max { + max = slice + } + } + return max +} + +// FramePath returns the path to a frame in the database. +func (db *DB) FramePath(name string) string { return filepath.Join(db.path, name) } + +// Frame returns a frame in the database by name. +func (db *DB) Frame(name string) *Frame { + db.mu.Lock() + defer db.mu.Unlock() + return db.frame(name) +} + +func (db *DB) frame(name string) *Frame { return db.frames[name] } + +// CreateFrameIfNotExists returns a frame in the database by name. +func (db *DB) CreateFrameIfNotExists(name string) (*Frame, error) { + db.mu.Lock() + defer db.mu.Unlock() + return db.createFrameIfNotExists(name) +} + +func (db *DB) createFrameIfNotExists(name string) (*Frame, error) { + // Find frame in cache first. + if f := db.frames[name]; f != nil { + return f, nil + } + + // Initialize and open frame. + f := NewFrame(db.FramePath(name), db.name, name) + if err := f.Open(); err != nil { + return nil, err + } + db.frames[name] = f + + return f, nil +} + +// ProfileAttrs returns the value of the attribute for a profile. +func (db *DB) ProfileAttrs(id uint64) (m map[string]interface{}, err error) { + db.mu.Lock() + defer db.mu.Unlock() + + // Check cache for map. + if m = db.attrs[id]; m != nil { + return m, nil + } + + // Find attributes from storage. + if err = db.store.View(func(tx *bolt.Tx) error { + m, err = txProfileAttrs(tx, id) + if err != nil { + return err + } + return nil + }); err != nil { + return nil, err + } + + // Add to cache. + db.attrs[id] = m + + return +} + +// SetProfileAttrs sets attribute values for a profile. +func (db *DB) SetProfileAttrs(id uint64, m map[string]interface{}) error { + db.mu.Lock() + defer db.mu.Unlock() + + var attr map[string]interface{} + if err := db.store.Update(func(tx *bolt.Tx) error { + tmp, err := txProfileAttrs(tx, id) + if err != nil { + return err + } + attr = tmp + + // Create a new map if it is empty so we don't update emptyMap. + if len(attr) == 0 { + attr = 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(attr, k) + } else { + attr[k] = v + } + } + + // Marshal and save new values. + buf, err := json.Marshal(attr) + if err != nil { + return err + } + if err := tx.Bucket([]byte("attrs")).Put(u64tob(id), buf); err != nil { + return err + } + return nil + }); err != nil { + return err + } + + // Swap attributes map in cache. + db.attrs[id] = attr + + return nil +} + +// txProfileAttrs returns a map of attributes for a profile. +func txProfileAttrs(tx *bolt.Tx, id uint64) (map[string]interface{}, error) { + if v := tx.Bucket([]byte("attrs")).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 +} diff --git a/db_test.go b/db_test.go new file mode 100644 index 000000000..d18fb2cb9 --- /dev/null +++ b/db_test.go @@ -0,0 +1,127 @@ +package pilosa_test + +import ( + "io/ioutil" + "os" + "reflect" + "testing" + + "github.com/umbel/pilosa" +) + +// Ensure database can open and retrieve a frame. +func TestDB_CreateFrameIfNotExists(t *testing.T) { + db := MustOpenDB() + defer db.Close() + + // Create frame. + f, err := db.CreateFrameIfNotExists("f") + if err != nil { + t.Fatal(err) + } else if f == nil { + t.Fatal("expected frame") + } + + // Retrieve existing frame. + other, err := db.CreateFrameIfNotExists("f") + if err != nil { + t.Fatal(err) + } else if f != other { + t.Fatal("frame mismatch") + } + + if f != db.Frame("f") { + t.Fatal("frame mismatch") + } +} + +// Ensure database can set and retrieve profile attributes. +func TestDB_ProfileAttrs(t *testing.T) { + db := MustOpenDB() + defer db.Close() + + // Set attributes. + if err := db.SetProfileAttrs(1, map[string]interface{}{"A": float64(100)}); err != nil { + t.Fatal(err) + } else if err := db.SetProfileAttrs(2, map[string]interface{}{"A": float64(200)}); err != nil { + t.Fatal(err) + } else if err := db.SetProfileAttrs(1, map[string]interface{}{"B": "VALUE"}); err != nil { + t.Fatal(err) + } + + // Retrieve attributes for profile #1. + if m, err := db.ProfileAttrs(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 profile #2. + if m, err := db.ProfileAttrs(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 database returns a non-nil empty map if unset. +func TestDB_ProfileAttrs_Empty(t *testing.T) { + db := MustOpenDB() + defer db.Close() + + if m, err := db.ProfileAttrs(100); err != nil { + t.Fatal(err) + } else if m == nil || len(m) > 0 { + t.Fatalf("unexpected attrs: %#v", m) + } +} + +// Ensure database can unset attributes if explicitly set to nil. +func TestDB_ProfileAttrs_Unset(t *testing.T) { + db := MustOpenDB() + defer db.Close() + + // Set attributes. + if err := db.SetProfileAttrs(1, map[string]interface{}{"A": "X", "B": "Y"}); err != nil { + t.Fatal(err) + } else if err := db.SetProfileAttrs(1, map[string]interface{}{"B": nil}); err != nil { + t.Fatal(err) + } + + // Verify attributes. + if m, err := db.ProfileAttrs(1); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(m, map[string]interface{}{"A": "X"}) { + t.Fatalf("unexpected attrs: %#v", m) + } +} + +// DB represents a test wrapper for pilosa.DB. +type DB struct { + *pilosa.DB +} + +// NewDB returns a new instance of DB d. +func NewDB() *DB { + path, err := ioutil.TempDir("", "pilosa-db-") + if err != nil { + panic(err) + } + + return &DB{DB: pilosa.NewDB(path, "d")} +} + +// MustOpenDB returns a new, opened database at a temporary path. Panic on error. +func MustOpenDB() *DB { + db := NewDB() + if err := db.Open(); err != nil { + panic(err) + } + return db +} + +// Close closes the database and removes the underlying data. +func (db *DB) Close() error { + defer os.RemoveAll(db.Path()) + return db.DB.Close() +} diff --git a/executor.go b/executor.go index d1a84f783..72c9b8e7b 100644 --- a/executor.go +++ b/executor.go @@ -48,8 +48,12 @@ func (e *Executor) Execute(db string, q *pql.Query, slices []uint64) (interface{ // Ignore slices for set calls. switch root := q.Root.(type) { - case *pql.Set: - return nil, e.executeSet(db, root) + case *pql.SetBit: + return nil, e.executeSetBit(db, root) + case *pql.SetBitmapAttrs: + return nil, e.executeSetBitmapAttrs(db, root) + case *pql.SetProfileAttrs: + return nil, e.executeSetProfileAttrs(db, root) } // If slices aren't specified, then include all of them. @@ -75,6 +79,8 @@ func (e *Executor) executeCall(db string, c pql.Call, slices []uint64) (interfac return e.executeBitmapCall(db, c, slices) case *pql.Count: return e.executeCount(db, c, slices) + case *pql.Profile: + return e.executeProfile(db, c) case *pql.TopN: return e.executeTopN(db, c, slices) default: @@ -105,16 +111,29 @@ func (e *Executor) executeBitmapCall(db string, c pql.BitmapCall, slices []uint6 } other.Merge(res.(*Bitmap)) } + + // Attach bitmap attributes for Bitmap() calls. + if c, ok := c.(*pql.Bitmap); ok { + fr := e.Index().Frame(db, c.Frame) + if fr != nil { + attrs, err := fr.BitmapAttrs(c.ID) + if err != nil { + return nil, err + } + other.Attrs = attrs + } + } + return other, nil } // executeBitmapCallSlice executes a bitmap call for a single slice. func (e *Executor) executeBitmapCallSlice(db string, c pql.BitmapCall, slice uint64) (*Bitmap, error) { switch c := c.(type) { + case *pql.Bitmap: + return e.executeBitmapSlice(db, c, slice) case *pql.Difference: return e.executeDifferenceSlice(db, c, slice) - case *pql.Get: - return e.executeGetSlice(db, c, slice) case *pql.Intersect: return e.executeIntersectSlice(db, c, slice) case *pql.Range: @@ -126,7 +145,7 @@ func (e *Executor) executeBitmapCallSlice(db string, c pql.BitmapCall, slice uin } } -// executeTopN executes a top-n() call. +// executeTopN executes a TopN() call. func (e *Executor) executeTopN(db string, c *pql.TopN, slices []uint64) ([]Pair, error) { panic("FIXME: calculate top n from each slice") } @@ -150,7 +169,7 @@ func (e *Executor) executeDifferenceSlice(db string, c *pql.Difference, slice ui return other, nil } -func (e *Executor) executeGetSlice(db string, c *pql.Get, slice uint64) (*Bitmap, error) { +func (e *Executor) executeBitmapSlice(db string, c *pql.Bitmap, slice uint64) (*Bitmap, error) { frame := c.Frame if frame == "" { frame = DefaultFrame @@ -232,8 +251,14 @@ func (e *Executor) executeCount(db string, c *pql.Count, slices []uint64) (uint6 return n, nil } -// executeSet executes a set call. -func (e *Executor) executeSet(db string, c *pql.Set) error { +// executeProfile executes a Profile() call. +// This call only executes locally since the profile attibutes are stored locally. +func (e *Executor) executeProfile(db string, c *pql.Profile) (*Profile, error) { + panic("FIXME: impl: e.Index().ProfileAttr(c.ID)") +} + +// executeSetBit executes a SetBit() call. +func (e *Executor) executeSetBit(db string, c *pql.SetBit) error { slice := c.ProfileID / SliceWidth for _, node := range e.Cluster.SliceNodes(slice) { @@ -256,6 +281,42 @@ func (e *Executor) executeSet(db string, c *pql.Set) error { return nil } +// executeSetBitmapAttrs executes a SetBitmapAttrs() call. +func (e *Executor) executeSetBitmapAttrs(db string, c *pql.SetBitmapAttrs) error { + // Retrieve frame. + frame, err := e.Index().CreateFrameIfNotExists(db, c.Frame) + if err != nil { + return err + } + + // Set attributes. + if err := frame.SetBitmapAttrs(c.ID, c.Attrs); err != nil { + return err + } + + // TODO: Propagate attributes to other servers in cluster. + + return nil +} + +// executeSetProfileAttrs executes a SetProfileAttrs() call. +func (e *Executor) executeSetProfileAttrs(db string, c *pql.SetProfileAttrs) error { + // Retrieve database. + d, err := e.Index().CreateDBIfNotExists(db) + if err != nil { + return err + } + + // Set attributes. + if err := d.SetProfileAttrs(c.ID, c.Attrs); err != nil { + return err + } + + // TODO: Propagate attributes to other servers in cluster. + + return nil +} + // exec executes a PQL query remotely for a set of slices on a node. func (e *Executor) exec(node *Node, db string, q *pql.Query, slices []uint64) (result interface{}, err error) { // Encode request object. @@ -319,7 +380,7 @@ func (e *Executor) exec(node *Node, db string, q *pql.Query, slices []uint64) (r return decodePairs(pb.GetPairs()), nil case *pql.Count: return pb.GetN(), nil - case *pql.Set: + case *pql.SetBit: return nil, nil default: panic(fmt.Sprintf("invalid node for remote exec: %T", q.Root)) diff --git a/executor_test.go b/executor_test.go index 530f4c162..004957a31 100644 --- a/executor_test.go +++ b/executor_test.go @@ -10,15 +10,19 @@ import ( "github.com/umbel/pilosa/pql" ) -// Ensure a get query can be executed. -func TestExecutor_Execute_Get(t *testing.T) { +// Ensure a bitmap query can be executed. +func TestExecutor_Execute_Bitmap(t *testing.T) { idx := MustOpenIndex() defer idx.Close() idx.MustCreateFragmentIfNotExists("d", "f", 0).MustSetBit(10, 3) idx.MustCreateFragmentIfNotExists("d", "f", 1).MustSetBit(10, SliceWidth+1) + if err := idx.Frame("d", "f").SetBitmapAttrs(10, map[string]interface{}{"foo": "bar", "baz": 123}); err != nil { + t.Fatal(err) + } + e := NewExecutor(idx.Index, NewCluster(1)) - if res, err := e.Execute("d", MustParse(`get(id=10, frame=f)`), nil); err != nil { + if res, err := e.Execute("d", MustParse(`Bitmap(id=10, frame=f)`), nil); err != nil { t.Fatal(err) } else if chunks := res.(*pilosa.Bitmap).Chunks(); len(chunks) != 2 { t.Fatalf("unexpected chunk length: %s", spew.Sdump(chunks)) @@ -26,6 +30,8 @@ func TestExecutor_Execute_Get(t *testing.T) { t.Fatalf("unexpected chunk(0): %s", spew.Sdump(chunks[0])) } else if chunks[1].Value[0] != 2 { t.Fatalf("unexpected chunk(1): %s", spew.Sdump(chunks[1])) + } else if attrs := res.(*pilosa.Bitmap).Attrs; !reflect.DeepEqual(attrs, map[string]interface{}{"foo": "bar", "baz": 123}) { + t.Fatalf("unexpected attrs: %s", spew.Sdump(attrs)) } } @@ -39,7 +45,7 @@ func TestExecutor_Execute_Difference(t *testing.T) { idx.MustCreateFragmentIfNotExists("d", "general", 0).MustSetBit(11, 2) e := NewExecutor(idx.Index, NewCluster(1)) - if res, err := e.Execute("d", MustParse(`difference(get(id=10), get(id=11))`), nil); err != nil { + if res, err := e.Execute("d", MustParse(`Difference(Bitmap(id=10), Bitmap(id=11))`), nil); err != nil { t.Fatal(err) } else if chunks := res.(*pilosa.Bitmap).Chunks(); len(chunks) != 1 { t.Fatalf("unexpected chunk length: %s", spew.Sdump(chunks)) @@ -61,7 +67,7 @@ func TestExecutor_Execute_Intersect(t *testing.T) { idx.MustCreateFragmentIfNotExists("d", "general", 1).MustSetBit(11, SliceWidth+2) e := NewExecutor(idx.Index, NewCluster(1)) - if res, err := e.Execute("d", MustParse(`intersect(get(id=10), get(id=11))`), nil); err != nil { + if res, err := e.Execute("d", MustParse(`Intersect(Bitmap(id=10), Bitmap(id=11))`), nil); err != nil { t.Fatal(err) } else if chunks := res.(*pilosa.Bitmap).Chunks(); len(chunks) != 2 { t.Fatalf("unexpected chunk length: %s", spew.Sdump(chunks)) @@ -84,7 +90,7 @@ func TestExecutor_Execute_Union(t *testing.T) { idx.MustCreateFragmentIfNotExists("d", "general", 1).MustSetBit(11, SliceWidth+2) e := NewExecutor(idx.Index, NewCluster(1)) - if res, err := e.Execute("d", MustParse(`union(get(id=10), get(id=11))`), nil); err != nil { + if res, err := e.Execute("d", MustParse(`Union(Bitmap(id=10), Bitmap(id=11))`), nil); err != nil { t.Fatal(err) } else if chunks := res.(*pilosa.Bitmap).Chunks(); len(chunks) != 2 { t.Fatalf("unexpected chunk length: %s", spew.Sdump(chunks)) @@ -104,7 +110,7 @@ func TestExecutor_Execute_Count(t *testing.T) { idx.MustCreateFragmentIfNotExists("d", "f", 1).MustSetBit(10, SliceWidth+2) e := NewExecutor(idx.Index, NewCluster(1)) - if n, err := e.Execute("d", MustParse(`count(get(id=10, frame=f))`), nil); err != nil { + if n, err := e.Execute("d", MustParse(`Count(Bitmap(id=10, frame=f))`), nil); err != nil { t.Fatal(err) } else if n != uint64(3) { t.Fatalf("unexpected n: %d", n) @@ -112,12 +118,12 @@ func TestExecutor_Execute_Count(t *testing.T) { } // Ensure a set query can be executed. -func TestExecutor_Execute_Set(t *testing.T) { +func TestExecutor_Execute_SetBit(t *testing.T) { idx := MustOpenIndex() defer idx.Close() e := NewExecutor(idx.Index, NewCluster(1)) - if _, err := e.Execute("d", MustParse(`set(id=10, frame=f, profile_id=1)`), nil); err != nil { + if _, err := e.Execute("d", MustParse(`SetBit(id=10, frame=f, profileID=1)`), nil); err != nil { t.Fatal(err) } @@ -127,6 +133,35 @@ func TestExecutor_Execute_Set(t *testing.T) { } } +// Ensure a SetBitmapAttrs() query can be executed. +func TestExecutor_Execute_SetBitmapAttrs(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + // Set two fields on f/10. + // Also set fields on other bitmaps and frames to test isolation. + e := NewExecutor(idx.Index, NewCluster(1)) + if _, err := e.Execute("d", MustParse(`SetBitmapAttrs(id=10, frame=f, foo="bar")`), nil); err != nil { + t.Fatal(err) + } + if _, err := e.Execute("d", MustParse(`SetBitmapAttrs(id=200, frame=f, YYY=1)`), nil); err != nil { + t.Fatal(err) + } + if _, err := e.Execute("d", MustParse(`SetBitmapAttrs(id=10, frame=XXX, YYY=1)`), nil); err != nil { + t.Fatal(err) + } + if _, err := e.Execute("d", MustParse(`SetBitmapAttrs(id=10, frame=f, baz=123, bat=true)`), nil); err != nil { + t.Fatal(err) + } + + f := idx.Frame("d", "f") + if m, err := f.BitmapAttrs(10); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(m, map[string]interface{}{"foo": "bar", "baz": int64(123), "bat": true}) { + t.Fatalf("unexpected bitmap attr: %#v", m) + } +} + // Ensure a remote query can return a bitmap. func TestExecutor_Execute_Remote_Bitmap(t *testing.T) { c := NewCluster(2) @@ -140,7 +175,7 @@ func TestExecutor_Execute_Remote_Bitmap(t *testing.T) { s.Handler.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64) (interface{}, error) { if db != `d` { t.Fatalf("unexpected db: %s", db) - } else if query.String() != `get(id=10, frame=f)` { + } else if query.String() != `Bitmap(id=10, frame=f)` { t.Fatalf("unexpected query: %s", query.String()) } else if !reflect.DeepEqual(slices, []uint64{0, 2, 4}) { t.Fatalf("unexpected slices: %+v", slices) @@ -162,7 +197,7 @@ func TestExecutor_Execute_Remote_Bitmap(t *testing.T) { idx.MustCreateFragmentIfNotExists("d", "f", 1).MustSetBit(10, (1*SliceWidth)+1) e := NewExecutor(idx.Index, c) - if res, err := e.Execute("d", MustParse(`get(id=10, frame=f)`), nil); err != nil { + if res, err := e.Execute("d", MustParse(`Bitmap(id=10, frame=f)`), nil); err != nil { t.Fatal(err) } else if chunks := res.(*pilosa.Bitmap).Chunks(); len(chunks) != 3 { t.Fatalf("unexpected chunk length: %s", spew.Sdump(chunks)) @@ -194,7 +229,7 @@ func TestExecutor_Execute_Remote_Count(t *testing.T) { idx.MustCreateFragmentIfNotExists("d", "f", 1).MustSetBit(10, (1*SliceWidth)+2) e := NewExecutor(idx.Index, c) - if n, err := e.Execute("d", MustParse(`count(get(id=10, frame=f))`), nil); err != nil { + if n, err := e.Execute("d", MustParse(`Count(Bitmap(id=10, frame=f))`), nil); err != nil { t.Fatal(err) } else if n != uint64(12) { t.Fatalf("unexpected n: %d", n) @@ -202,7 +237,7 @@ func TestExecutor_Execute_Remote_Count(t *testing.T) { } // Ensure a remote query can set bits on multiple nodes. -func TestExecutor_Execute_Remote_Set(t *testing.T) { +func TestExecutor_Execute_Remote_SetBit(t *testing.T) { c := NewCluster(2) c.ReplicaN = 2 @@ -216,7 +251,7 @@ func TestExecutor_Execute_Remote_Set(t *testing.T) { s.Handler.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64) (interface{}, error) { if db != `d` { t.Fatalf("unexpected db: %s", db) - } else if query.String() != `set(id=10, frame=f, profile_id=2)` { + } else if query.String() != `SetBit(id=10, frame=f, profileID=2)` { t.Fatalf("unexpected query: %s", query.String()) } remoteCalled = true @@ -228,7 +263,7 @@ func TestExecutor_Execute_Remote_Set(t *testing.T) { defer idx.Close() e := NewExecutor(idx.Index, c) - if _, err := e.Execute("d", MustParse(`set(id=10, frame=f, profile_id=2)`), nil); err != nil { + if _, err := e.Execute("d", MustParse(`SetBit(id=10, frame=f, profileID=2)`), nil); err != nil { t.Fatal(err) } diff --git a/fragment.go b/fragment.go index 680e01737..944eaad62 100644 --- a/fragment.go +++ b/fragment.go @@ -61,6 +61,18 @@ func NewFragment(path, db, frame string, slice uint64) *Fragment { return f } +// Path returns the path the fragment was initialized with. +func (f *Fragment) Path() string { return f.path } + +// DB returns the database the fragment was initialized with. +func (f *Fragment) DB() string { return f.db } + +// Frame returns the frame the fragment was initialized with. +func (f *Fragment) Frame() string { return f.frame } + +// Slice returns the slice the fragment was initialized with. +func (f *Fragment) Slice() uint64 { return f.slice } + // Open opens the underlying storage. func (f *Fragment) Open() error { f.mu.Lock() @@ -174,18 +186,6 @@ func (f *Fragment) closeStorage() error { return nil } -// Path returns the path the fragment was initialized with. -func (f *Fragment) Path() string { return f.path } - -// DB returns the database the fragment was initialized with. -func (f *Fragment) DB() string { return f.db } - -// Frame returns the frame the fragment was initialized with. -func (f *Fragment) Frame() string { return f.frame } - -// Slice returns the slice the fragment was initialized with. -func (f *Fragment) Slice() uint64 { return f.slice } - // Bitmap returns a bitmap by ID. func (f *Fragment) Bitmap(bitmapID uint64) *Bitmap { f.mu.Lock() diff --git a/frame.go b/frame.go index 09f77e9ef..3ada6a707 100644 --- a/frame.go +++ b/frame.go @@ -3,6 +3,7 @@ package pilosa import ( "encoding/binary" "encoding/json" + "fmt" "os" "path/filepath" "strconv" @@ -19,12 +20,12 @@ type Frame struct { db string name string - // Fragments + // Fragments by slice. fragments map[uint64]*Fragment - // Attribute storage and cache - store *bolt.DB - battrs map[uint64]map[string]interface{} + // Bitmap attribute storage and cache + store *bolt.DB + attrs map[uint64]map[string]interface{} } // NewFrame returns a new instance of frame. @@ -35,11 +36,33 @@ func NewFrame(path, db, name string) *Frame { name: name, fragments: make(map[uint64]*Fragment), - - battrs: make(map[uint64]map[string]interface{}), + attrs: make(map[uint64]map[string]interface{}), } } +// Name returns the name the frame was initialized with. +func (f *Frame) Name() string { return f.name } + +// DB returns the database name the frame was initialized with. +func (f *Frame) DB() string { return f.db } + +// Path returns the path the frame was initialized with. +func (f *Frame) Path() string { return f.path } + +// SliceN returns the max slice in the frame. +func (f *Frame) SliceN() uint64 { + f.mu.Lock() + defer f.mu.Unlock() + + var max uint64 + for slice := range f.fragments { + if slice > max { + max = slice + } + } + return max +} + // Open opens and initializes the frame. func (f *Frame) Open() error { // Ensure the frame's path exists. @@ -47,6 +70,53 @@ func (f *Frame) Open() error { return err } + if err := f.openFragments(); err != nil { + return err + } + + if err := f.openStore(); err != nil { + return err + } + + return nil +} + +// openFragments opens and initializes the fragments inside the frame. +func (f *Frame) openFragments() error { + file, err := os.Open(f.path) + if err != nil { + return err + } + defer file.Close() + + fis, err := file.Readdir(0) + if err != nil { + return err + } + + for _, fi := range fis { + if fi.IsDir() { + continue + } + + // Parse filename into integer. + slice, err := strconv.ParseUint(filepath.Base(fi.Name()), 10, 64) + if err != nil { + continue + } + + frag := NewFragment(f.FragmentPath(slice), f.db, f.name, slice) + if err := frag.Open(); err != nil { + return fmt.Errorf("open fragment: slice=%s, err=%s", frag.Slice(), err) + } + f.fragments[frag.Slice()] = frag + } + + return nil +} + +// openStore opens and initializes the attribute store. +func (f *Frame) openStore() error { // Open attribute store. store, err := bolt.Open(filepath.Join(f.path, "data"), 0666, &bolt.Options{Timeout: 1 * time.Second}) if err != nil { @@ -56,7 +126,7 @@ func (f *Frame) Open() error { // Initialize database. if err := f.store.Update(func(tx *bolt.Tx) error { - if _, err := tx.CreateBucketIfNotExists([]byte("battrs")); err != nil { + if _, err := tx.CreateBucketIfNotExists([]byte("attrs")); err != nil { return err } return nil @@ -87,9 +157,6 @@ func (f *Frame) Close() error { 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)) @@ -133,13 +200,13 @@ func (f *Frame) BitmapAttrs(id uint64) (m map[string]interface{}, err error) { defer f.mu.Unlock() // Check cache for map. - if m = f.battrs[id]; m != nil { + if m = f.attrs[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) + m, err = txBitmapAttrs(tx, id) if err != nil { return err } @@ -149,45 +216,45 @@ func (f *Frame) BitmapAttrs(id uint64) (m map[string]interface{}, err error) { } // Add to cache. - f.battrs[id] = m + f.attrs[id] = m return } -// SetBitmapAttr sets attribute values for a bitmap. +// SetBitmapAttrs 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{} + var attr map[string]interface{} if err := f.store.Update(func(tx *bolt.Tx) error { - attr, err := f.bitmapAttrs(tx, id) + tmp, err := txBitmapAttrs(tx, id) if err != nil { return err } - battr = attr + attr = tmp // 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)) + if len(attr) == 0 { + attr = 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) + delete(attr, k) } else { - battr[k] = v + attr[k] = v } } // Marshal and save new values. - buf, err := json.Marshal(battr) + buf, err := json.Marshal(attr) if err != nil { return err } - if err := tx.Bucket([]byte("battrs")).Put(u64tob(id), buf); err != nil { + if err := tx.Bucket([]byte("attrs")).Put(u64tob(id), buf); err != nil { return err } return nil @@ -196,14 +263,14 @@ func (f *Frame) SetBitmapAttrs(id uint64, m map[string]interface{}) error { } // Swap attributes map in cache. - f.battrs[id] = battr + f.attrs[id] = attr 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 { +// txBitmapAttrs returns a map of attributes for a bitmap. +func txBitmapAttrs(tx *bolt.Tx, id uint64) (map[string]interface{}, error) { + if v := tx.Bucket([]byte("attrs")).Get(u64tob(id)); v != nil { m := make(map[string]interface{}) if err := json.Unmarshal(v, &m); err != nil { return nil, err diff --git a/handler.go b/handler.go index cacbeef51..a57b95f8e 100644 --- a/handler.go +++ b/handler.go @@ -88,40 +88,72 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // handlePostQuery handles /query requests. func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) { // Parse incoming request. - db, query, slices, err := h.readQueryRequest(r) - - // h.logger().Printf("%s %s db=%s q=%s slices=%v", r.Method, r.URL.Path, db, query, slices) - + req, err := h.readQueryRequest(r) if err != nil { w.WriteHeader(http.StatusBadRequest) - h.writeQueryResponse(w, r, nil, err) + h.writeQueryResponse(w, r, &QueryResponse{Err: err}) return } // Parse query string. - q, err := pql.NewParser(strings.NewReader(query)).Parse() + q, err := pql.NewParser(strings.NewReader(req.Query)).Parse() if err != nil { w.WriteHeader(http.StatusBadRequest) - h.writeQueryResponse(w, r, nil, err) + h.writeQueryResponse(w, r, &QueryResponse{Err: err}) return } // Execute the query. - res, e := h.Executor.Execute(db, q, slices) + result, err := h.Executor.Execute(req.DB, q, req.Slices) + resp := &QueryResponse{Result: result, Err: err} + + // Fill profile attributes if requested. + if bm, ok := result.(*Bitmap); ok && req.Profiles { + profiles, err := h.readProfiles(h.Index.DB(req.DB), bm.Bits()) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + h.writeQueryResponse(w, r, &QueryResponse{Err: err}) + return + } + resp.Profiles = profiles + } // Set appropriate status code, if there is an error. - if e != nil { + if resp.Err != nil { w.WriteHeader(http.StatusInternalServerError) } // Write response back to client. - if err := h.writeQueryResponse(w, r, res, e); err != nil { + if err := h.writeQueryResponse(w, r, resp); err != nil { h.logger().Printf("write query response error: %s", err) } } +// readProfiles returns a list of profile objects by id. +func (h *Handler) readProfiles(db *DB, ids []uint64) ([]*Profile, error) { + if db == nil { + return nil, nil + } + + a := make([]*Profile, 0, len(ids)) + for _, id := range ids { + // Read attributes for profile. Skip profile if empty. + attrs, err := db.ProfileAttrs(id) + if err != nil { + return nil, err + } else if len(attrs) == 0 { + continue + } + + // Append profile with attributes. + a = append(a, &Profile{ID: id, Attrs: attrs}) + } + + return a, nil +} + // readQueryRequest parses an query parameters from r. -func (h *Handler) readQueryRequest(r *http.Request) (db, query string, slices []uint64, err error) { +func (h *Handler) readQueryRequest(r *http.Request) (*QueryRequest, error) { switch r.Header.Get("Content-Type") { case "application/x-protobuf": return h.readProtobufQueryRequest(r) @@ -131,105 +163,68 @@ func (h *Handler) readQueryRequest(r *http.Request) (db, query string, slices [] } // readProtobufQueryRequest parses query parameters in protobuf from r. -func (h *Handler) readProtobufQueryRequest(r *http.Request) (db, query string, slices []uint64, err error) { +func (h *Handler) readProtobufQueryRequest(r *http.Request) (*QueryRequest, error) { // Slurp the body. body, err := ioutil.ReadAll(r.Body) if err != nil { - return + return nil, err } // Unmarshal into object. var req internal.QueryRequest - if err = proto.Unmarshal(body, &req); err != nil { - return + if err := proto.Unmarshal(body, &req); err != nil { + return nil, err } - return req.GetDB(), req.GetQuery(), req.GetSlices(), nil + return decodeQueryRequest(&req), nil } // readURLQueryRequest parses query parameters from URL parameters from r. -func (h *Handler) readURLQueryRequest(r *http.Request) (db, query string, slices []uint64, err error) { +func (h *Handler) readURLQueryRequest(r *http.Request) (*QueryRequest, error) { q := r.URL.Query() - // Read DB argument. - db = q.Get("db") - // Parse query string. buf, err := ioutil.ReadAll(r.Body) if err != nil { - return + return nil, err } - query = string(buf) + query := string(buf) // Parse list of slices. - slices, err = parseUint64Slice(q.Get("slices")) + slices, err := parseUint64Slice(q.Get("slices")) if err != nil { - err = errors.New("invalid slice argument") - return + return nil, errors.New("invalid slice argument") } - return + return &QueryRequest{ + DB: q.Get("db"), + Query: query, + Slices: slices, + Profiles: q.Get("profiles") == "true", + }, nil } // writeQueryResponse writes the response from the executor to w. -func (h *Handler) writeQueryResponse(w http.ResponseWriter, r *http.Request, res interface{}, err error) error { +func (h *Handler) writeQueryResponse(w http.ResponseWriter, r *http.Request, resp *QueryResponse) error { if strings.Contains(r.Header.Get("Accept"), "application/x-protobuf") { - return h.writeProtobufQueryResponse(w, res, err) + return h.writeProtobufQueryResponse(w, resp) } - return h.writeJSONQueryResponse(w, res, err) + return h.writeJSONQueryResponse(w, resp) } // writeProtobufQueryResponse writes the response from the executor to w as protobuf. -func (h *Handler) writeProtobufQueryResponse(w http.ResponseWriter, res interface{}, e error) error { - var resp internal.QueryResponse - - // Set the result on the appropriate field. - if res != nil { - switch res := res.(type) { - case *Bitmap: - resp.Bitmap = encodeBitmap(res) - case Pairs: - resp.Pairs = encodePairs(res) - case uint64: - resp.N = proto.Uint64(res) - default: - panic(fmt.Sprintf("invalid query response type: %T", res)) - } - } - - // Set the error if there is one. - if e != nil { - resp.Err = proto.String(e.Error()) - } - - // Encode response. - buf, err := proto.Marshal(&resp) - if err != nil { +func (h *Handler) writeProtobufQueryResponse(w http.ResponseWriter, resp *QueryResponse) error { + if buf, err := proto.Marshal(encodeQueryResponse(resp)); err != nil { + return err + } else if _, err := w.Write(buf); err != nil { return err } - - // Write response back to client. - if _, err := w.Write(buf); err != nil { - return err - } - return nil } // writeJSONQueryResponse writes the response from the executor to w as JSON. -func (h *Handler) writeJSONQueryResponse(w http.ResponseWriter, res interface{}, e error) error { - var o struct { - Result interface{} `json:"result,omitempty"` - Error string `json:"error,omitempty"` - } - o.Result = res - - if e != nil { - o.Error = e.Error() - } - - // Otherwise marshal the result as JSON. - return json.NewEncoder(w).Encode(o) +func (h *Handler) writeJSONQueryResponse(w http.ResponseWriter, resp *QueryResponse) error { + return json.NewEncoder(w).Encode(resp) } // handlePostImport handles /import requests. @@ -343,6 +338,84 @@ func (h *Handler) logger() *log.Logger { return log.New(h.LogOutput, "", log.LstdFlags) } +// QueryRequest represent a request to process a query. +type QueryRequest struct { + // Database to execute query against. + DB string + + // The query string to parse and execute. + Query string + + // The slices to include in the query execution. + // If empty, all slices are included. + Slices []uint64 + + // Return profile attributes, if true. + Profiles bool +} + +func decodeQueryRequest(pb *internal.QueryRequest) *QueryRequest { + return &QueryRequest{ + DB: pb.GetDB(), + Query: pb.GetQuery(), + Slices: pb.GetSlices(), + Profiles: pb.GetProfiles(), + } +} + +// QueryResponse represent a response from a processed query. +type QueryResponse struct { + // Query execution results. + // Can be a Bitmap, Pairs, or uint64. + Result interface{} + + // Set of profiles matching IDs returned in Result. + Profiles []*Profile + + // Error during parsing or execution. + Err error +} + +func (resp *QueryResponse) MarshalJSON() ([]byte, error) { + var output struct { + Result interface{} `json:"result,omitempty"` + Profiles []*Profile `json:"profiles,omitempty"` + Err string `json:"error,omitempty"` + } + output.Result = resp.Result + output.Profiles = resp.Profiles + + if resp.Err != nil { + output.Err = resp.Err.Error() + } + return json.Marshal(output) +} + +func encodeQueryResponse(resp *QueryResponse) *internal.QueryResponse { + pb := &internal.QueryResponse{ + Profiles: encodeProfiles(resp.Profiles), + } + + if resp.Result != nil { + switch result := resp.Result.(type) { + case *Bitmap: + pb.Bitmap = encodeBitmap(result) + case Pairs: + pb.Pairs = encodePairs(result) + case uint64: + pb.N = proto.Uint64(result) + default: + panic(fmt.Sprintf("invalid query result type: %T", resp.Result)) + } + } + + if resp.Err != nil { + pb.Err = proto.String(resp.Err.Error()) + } + + return pb +} + // parseUint64Slice returns a slice of uint64s from a comma-delimited string. func parseUint64Slice(s string) ([]uint64, error) { var a []uint64 diff --git a/handler_test.go b/handler_test.go index 1f944f446..e54dcd679 100644 --- a/handler_test.go +++ b/handler_test.go @@ -32,7 +32,7 @@ func TestHandler_Query_Args_URL(t *testing.T) { h.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64) (interface{}, error) { if db != "db0" { t.Fatalf("unexpected db: %s", db) - } else if query.String() != `count(get(id=100))` { + } else if query.String() != `Count(Bitmap(id=100))` { t.Fatalf("unexpected query: %s", query.String()) } else if !reflect.DeepEqual(slices, []uint64{0, 1}) { t.Fatalf("unexpected slices: %+v", slices) @@ -41,7 +41,7 @@ func TestHandler_Query_Args_URL(t *testing.T) { } w := httptest.NewRecorder() - h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=db0&slices=0,1", strings.NewReader("count( get( 100))"))) + h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=db0&slices=0,1", strings.NewReader("Count( Bitmap( 100))"))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{"result":100}`+"\n" { @@ -55,7 +55,7 @@ func TestHandler_Query_Args_Protobuf(t *testing.T) { h.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64) (interface{}, error) { if db != "db0" { t.Fatalf("unexpected db: %s", db) - } else if query.String() != `count(get(id=100))` { + } else if query.String() != `Count(Bitmap(id=100))` { t.Fatalf("unexpected query: %s", query.String()) } else if !reflect.DeepEqual(slices, []uint64{0, 1}) { t.Fatalf("unexpected slices: %+v", slices) @@ -66,7 +66,7 @@ func TestHandler_Query_Args_Protobuf(t *testing.T) { // Generate request body. reqBody, err := proto.Marshal(&internal.QueryRequest{ DB: proto.String("db0"), - Query: proto.String("count(get(100))"), + Query: proto.String("Count(Bitmap(100))"), Slices: []uint64{0, 1}, }) if err != nil { @@ -87,7 +87,7 @@ func TestHandler_Query_Args_Protobuf(t *testing.T) { // Ensure the handler returns an error when parsing bad arguments. func TestHandler_Query_Args_Err(t *testing.T) { w := httptest.NewRecorder() - NewHandler().ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=db0&slices=a,b", strings.NewReader("get(100)"))) + NewHandler().ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=db0&slices=a,b", strings.NewReader("Bitmap(100)"))) if w.Code != http.StatusBadRequest { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{"error":"invalid slice argument"}`+"\n" { @@ -103,7 +103,7 @@ func TestHandler_Query_Uint64_JSON(t *testing.T) { } w := httptest.NewRecorder() - h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=db0&slices=0,1", strings.NewReader("count( get( 100))"))) + h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=db0&slices=0,1", strings.NewReader("Count( Bitmap( 100))"))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{"result":100}`+"\n" { @@ -119,7 +119,7 @@ func TestHandler_Query_Uint64_Protobuf(t *testing.T) { } w := httptest.NewRecorder() - r := MustNewHTTPRequest("POST", "/query", strings.NewReader("count(get(100))")) + r := MustNewHTTPRequest("POST", "/query", strings.NewReader("Count(Bitmap(100))")) r.Header.Set("Accept", "application/x-protobuf") h.ServeHTTP(w, r) if w.Code != http.StatusOK { @@ -138,15 +138,49 @@ func TestHandler_Query_Uint64_Protobuf(t *testing.T) { func TestHandler_Query_Bitmap_JSON(t *testing.T) { h := NewHandler() h.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64) (interface{}, error) { - return pilosa.NewBitmap(1, 3, 66, pilosa.SliceWidth+1), nil + bm := pilosa.NewBitmap(1, 3, 66, pilosa.SliceWidth+1) + bm.Attrs = map[string]interface{}{"a": "b", "c": 1, "d": true} + return bm, nil } w := httptest.NewRecorder() - h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query", strings.NewReader("get(100)"))) + h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=d", strings.NewReader("Bitmap(100)"))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) - } else if body := w.Body.String(); body != `{"result":[1,3,66,65537]}`+"\n" { - t.Fatalf("unexpected body: %q", body) + } else if body := w.Body.String(); body != `{"result":{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,65537]}}`+"\n" { + t.Fatalf("unexpected body: %s", body) + } +} + +// Ensure the handler can execute a query that returns a bitmap with profiles as JSON. +func TestHandler_Query_Bitmap_Profiles_JSON(t *testing.T) { + idx := NewIndex() + defer idx.Close() + + // Create database and set profile attributes. + db, err := idx.CreateDBIfNotExists("d") + if err != nil { + t.Fatal(err) + } else if err := db.SetProfileAttrs(3, map[string]interface{}{"x": "y"}); err != nil { + t.Fatal(err) + } else if err := db.SetProfileAttrs(66, map[string]interface{}{"y": 123, "z": false}); err != nil { + t.Fatal(err) + } + + h := NewHandler() + h.Index = idx.Index + h.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64) (interface{}, error) { + bm := pilosa.NewBitmap(1, 3, 66, pilosa.SliceWidth+1) + bm.Attrs = map[string]interface{}{"a": "b", "c": 1, "d": true} + return bm, nil + } + + w := httptest.NewRecorder() + h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=d&profiles=true", strings.NewReader("Bitmap(100)"))) + if w.Code != http.StatusOK { + t.Fatalf("unexpected status code: %d", w.Code) + } else if body := w.Body.String(); body != `{"result":{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,65537]},"profiles":[{"id":3,"attrs":{"x":"y"}},{"id":66,"attrs":{"y":123,"z":false}}]}`+"\n" { + t.Fatalf("unexpected body: %s", body) } } @@ -154,11 +188,13 @@ func TestHandler_Query_Bitmap_JSON(t *testing.T) { func TestHandler_Query_Bitmap_Protobuf(t *testing.T) { h := NewHandler() h.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64) (interface{}, error) { - return pilosa.NewBitmap(1, pilosa.SliceWidth+1), nil + bm := pilosa.NewBitmap(1, pilosa.SliceWidth+1) + bm.Attrs = map[string]interface{}{"a": "b", "c": int64(1), "d": true} + return bm, nil } w := httptest.NewRecorder() - r := MustNewHTTPRequest("POST", "/query", strings.NewReader("get(100)")) + r := MustNewHTTPRequest("POST", "/query", strings.NewReader("Bitmap(100)")) r.Header.Set("Accept", "application/x-protobuf") h.ServeHTTP(w, r) if w.Code != http.StatusOK { @@ -170,6 +206,81 @@ func TestHandler_Query_Bitmap_Protobuf(t *testing.T) { t.Fatal(err) } else if a := resp.GetBitmap().GetChunks(); len(a) != 2 { t.Fatalf("unexpected bitmap chunk length: %d", len(a)) + } else if attrs := resp.GetBitmap().GetAttrs(); len(attrs) != 3 { + t.Fatalf("unexpected attr length: %d", len(attrs)) + } else if k, v := attrs[0].GetKey(), attrs[0].GetStringValue(); k != "a" || v != "b" { + t.Fatalf("unexpected attr[0]: %s=%v", k, v) + } else if k, v := attrs[1].GetKey(), attrs[1].GetIntValue(); k != "c" || v != int64(1) { + t.Fatalf("unexpected attr[1]: %s=%v", k, v) + } else if k, v := attrs[2].GetKey(), attrs[2].GetBoolValue(); k != "d" || v != true { + t.Fatalf("unexpected attr[2]: %s=%v", k, v) + } +} + +// Ensure the handler can execute a query that returns a bitmap with profiles as protobuf. +func TestHandler_Query_Bitmap_Profiles_Protobuf(t *testing.T) { + idx := NewIndex() + defer idx.Close() + + // Create database and set profile attributes. + db, err := idx.CreateDBIfNotExists("d") + if err != nil { + t.Fatal(err) + } else if err := db.SetProfileAttrs(1, map[string]interface{}{"x": "y"}); err != nil { + t.Fatal(err) + } + + h := NewHandler() + h.Index = idx.Index + h.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64) (interface{}, error) { + bm := pilosa.NewBitmap(1, pilosa.SliceWidth+1) + bm.Attrs = map[string]interface{}{"a": "b", "c": int64(1), "d": true} + return bm, nil + } + + // Encode request body. + buf, err := proto.Marshal(&internal.QueryRequest{ + DB: proto.String("d"), + Query: proto.String("Bitmap(100)"), + Profiles: proto.Bool(true), + }) + if err != nil { + t.Fatal(err) + } + + w := httptest.NewRecorder() + r := MustNewHTTPRequest("POST", "/query", bytes.NewReader(buf)) + r.Header.Set("Content-Type", "application/x-protobuf") + r.Header.Set("Accept", "application/x-protobuf") + h.ServeHTTP(w, r) + if w.Code != http.StatusOK { + t.Fatalf("unexpected status code: %d", w.Code) + } + + var resp internal.QueryResponse + if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if a := resp.GetBitmap().GetChunks(); len(a) != 2 { + t.Fatalf("unexpected bitmap chunk length: %d", len(a)) + } else if attrs := resp.GetBitmap().GetAttrs(); len(attrs) != 3 { + t.Fatalf("unexpected attr length: %d", len(attrs)) + } else if k, v := attrs[0].GetKey(), attrs[0].GetStringValue(); k != "a" || v != "b" { + t.Fatalf("unexpected attr[0]: %s=%v", k, v) + } else if k, v := attrs[1].GetKey(), attrs[1].GetIntValue(); k != "c" || v != int64(1) { + t.Fatalf("unexpected attr[1]: %s=%v", k, v) + } else if k, v := attrs[2].GetKey(), attrs[2].GetBoolValue(); k != "d" || v != true { + t.Fatalf("unexpected attr[2]: %s=%v", k, v) + } + + if a := resp.GetProfiles(); len(a) != 1 { + t.Fatalf("unexpected profiles length: %d", len(a)) + } else if a[0].GetID() != 1 { + t.Fatalf("unexpected id: %d", a[0].GetID()) + } else if len(a[0].GetAttrs()) != 1 { + t.Fatalf("unexpected profile attr length: %d", len(a)) + } else if k, v := a[0].GetAttrs()[0].GetKey(), a[0].GetAttrs()[0].GetStringValue(); k != "x" || v != "y" { + t.Fatalf("unexpected attr[0]: %s=%v", k, v) } } @@ -184,7 +295,7 @@ func TestHandler_Query_Pairs_JSON(t *testing.T) { } w := httptest.NewRecorder() - h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query", strings.NewReader(`top-n(frame=x, n=2)`))) + h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query", strings.NewReader(`TopN(frame=x, n=2)`))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{"result":[{"key":1,"count":2},{"key":3,"count":4}]}`+"\n" { @@ -203,7 +314,7 @@ func TestHandler_Query_Pairs_Protobuf(t *testing.T) { } w := httptest.NewRecorder() - r := MustNewHTTPRequest("POST", "/query", strings.NewReader(`top-n(frame=x, n=2)`)) + r := MustNewHTTPRequest("POST", "/query", strings.NewReader(`TopN(frame=x, n=2)`)) r.Header.Set("Accept", "application/x-protobuf") h.ServeHTTP(w, r) if w.Code != http.StatusOK { @@ -226,7 +337,7 @@ func TestHandler_Query_Err_JSON(t *testing.T) { } w := httptest.NewRecorder() - h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query", strings.NewReader(`get(100)`))) + h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query", strings.NewReader(`Bitmap(100)`))) if w.Code != http.StatusInternalServerError { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{"error":"marker"}`+"\n" { @@ -242,7 +353,7 @@ func TestHandler_Query_Err_Protobuf(t *testing.T) { } w := httptest.NewRecorder() - r := MustNewHTTPRequest("POST", "/query", strings.NewReader(`top-n(frame=x, n=2)`)) + r := MustNewHTTPRequest("POST", "/query", strings.NewReader(`TopN(frame=x, n=2)`)) r.Header.Set("Accept", "application/x-protobuf") h.ServeHTTP(w, r) if w.Code != http.StatusInternalServerError { diff --git a/index.go b/index.go index ad1aa6270..8192d0010 100644 --- a/index.go +++ b/index.go @@ -2,27 +2,25 @@ package pilosa import ( "fmt" - "log" "os" "path/filepath" - "strconv" "sync" ) // Index represents a container for fragments. type Index struct { - mu sync.Mutex - path string - sliceN uint64 + mu sync.Mutex + path string - frames map[frameKey]*Frame + // Databases by name. + dbs map[string]*DB } // NewIndex returns a new instance of Index. func NewIndex(path string) *Index { return &Index{ - path: path, - frames: make(map[frameKey]*Frame), + path: path, + dbs: make(map[string]*DB), } } @@ -32,16 +30,7 @@ func (i *Index) Open() error { return err } - // Open all databases. - if err := i.openDatabases(); err != nil { - return err - } - - return nil -} - -// openDatabases recursively opens all directories within the data directory. -func (i *Index) openDatabases() error { + // Open path to read all database directories. f, err := os.Open(i.path) if err != nil { return err @@ -56,67 +45,21 @@ func (i *Index) openDatabases() error { for _, fi := range fis { if !fi.IsDir() { continue - } else if err := i.openDatabase(filepath.Base(fi.Name())); err != nil { - return err } - } - return nil -} -// openDatabase recursively opens all frames within a database directory. -func (i *Index) openDatabase(db string) error { - f, err := os.Open(filepath.Join(i.path, db)) - if err != nil { - return err - } - defer f.Close() - - fis, err := f.Readdir(0) - if err != nil { - return err - } - - for _, fi := range fis { - if !fi.IsDir() { - continue - } else if err := i.openFrame(db, filepath.Base(fi.Name())); err != nil { - return err - } - } - return nil -} - -// openFrame recursively opens all fragments within a frame directory. -func (i *Index) openFrame(db, frame string) error { - f, err := os.Open(filepath.Join(i.path, db, frame)) - if err != nil { - return err - } - defer f.Close() - - fis, err := f.Readdir(0) - if err != nil { - return err - } - - for _, fi := range fis { - slice, err := strconv.ParseUint(filepath.Base(fi.Name()), 10, 64) - if err != nil || fi.IsDir() { - continue - } - if _, err := i.CreateFragmentIfNotExists(db, frame, slice); err != nil { - return fmt.Errorf("open fragment: db=%s, frame=%d, slice=%d, err=%s", db, frame, slice, err) + db := NewDB(i.DBPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) + if err := db.Open(); err != nil { + return fmt.Errorf("open db: name=%s, err=%s", db.Name(), err) } + i.dbs[db.Name()] = db } return nil } // Close closes all open fragments. func (i *Index) Close() error { - for key, f := range i.frames { - if err := f.Close(); err != nil { - log.Println("error closing frame(%s/%s): %s", key.db, key.frame, err) - } + for _, db := range i.dbs { + db.Close() } return nil } @@ -124,50 +67,90 @@ func (i *Index) Close() error { // Path returns the path the index was initialized with. func (i *Index) Path() string { return i.path } -// SliceN returs the total number of slices managed by the index. +// SliceN returns the highest slice across all frames. func (i *Index) SliceN() uint64 { i.mu.Lock() defer i.mu.Unlock() - return i.sliceN + + var sliceN uint64 + for _, db := range i.dbs { + if n := db.SliceN(); n > sliceN { + sliceN = n + } + } + return sliceN } -// 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) } +// DBPath returns the path where a given database is stored. +func (i *Index) DBPath(name string) string { return filepath.Join(i.path, name) } + +// DB returns the database by name. +func (i *Index) DB(name string) *DB { + i.mu.Lock() + defer i.mu.Unlock() + return i.db(name) +} + +func (i *Index) db(name string) *DB { return i.dbs[name] } + +// CreateDBIfNotExists returns a database by name. +// The database is created if it does not already exist. +func (i *Index) CreateDBIfNotExists(name string) (*DB, error) { + i.mu.Lock() + defer i.mu.Unlock() + return i.createDBIfNotExists(name) +} + +func (i *Index) createDBIfNotExists(name string) (*DB, error) { + // Return database if it exists. + if db := i.db(name); db != nil { + return db, nil + } + + // Otherwise create a new database. + db := NewDB(i.DBPath(name), name) + if err := db.Open(); err != nil { + return nil, err + } + i.dbs[db.Name()] = db + + return db, nil +} + +// Frame returns the frame for a database and name. +func (i *Index) Frame(db, name string) *Frame { + d := i.DB(db) + if d == nil { + return nil + } + return d.Frame(name) +} + +// CreateFrameIfNotExists returns the frame for a database & name. +// The frame is created if it doesn't already exist. +func (i *Index) CreateFrameIfNotExists(db, name string) (*Frame, error) { + d, err := i.CreateDBIfNotExists(db) + if err != nil { + return nil, err + } + return d.CreateFrameIfNotExists(name) +} // 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.frames[frameKey{db, frame}].fragment(slice) + f := i.Frame(db, frame) + if f == nil { + return nil + } + return f.fragment(slice) } // CreateFragmentIfNotExists returns the fragment for a database, frame & slice. // The fragment is created if it doesn't already exist. func (i *Index) CreateFragmentIfNotExists(db, frame string, slice uint64) (*Fragment, error) { - i.mu.Lock() - defer i.mu.Unlock() - - // Track the highest slice. - if slice > i.sliceN { - i.sliceN = slice + f, err := i.CreateFrameIfNotExists(db, frame) + if err != nil { + return nil, err } - - // 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.frames[key] = f - } - - // Create fragment, if not exists. - return i.frames[key].createFragmentIfNotExists(slice) -} - -// frameKey is the map key for frame look ups. -type frameKey struct { - db string - frame string + return f.CreateFragmentIfNotExists(slice) } diff --git a/internal/internal.pb.go b/internal/internal.pb.go index 0a557144b..19f0cfa1a 100644 --- a/internal/internal.pb.go +++ b/internal/internal.pb.go @@ -13,6 +13,8 @@ It has these top-level messages: Chunk Pair Bit + Profile + Attr QueryRequest QueryResponse ImportRequest @@ -29,6 +31,7 @@ var _ = math.Inf type Bitmap struct { Chunks []*Chunk `protobuf:"bytes,1,rep" json:"Chunks,omitempty"` + Attrs []*Attr `protobuf:"bytes,2,rep" json:"Attrs,omitempty"` XXX_unrecognized []byte `json:"-"` } @@ -43,6 +46,13 @@ func (m *Bitmap) GetChunks() []*Chunk { return nil } +func (m *Bitmap) GetAttrs() []*Attr { + if m != nil { + return m.Attrs + } + return nil +} + type Chunk struct { Key *uint64 `protobuf:"varint,1,req" json:"Key,omitempty"` Value []uint64 `protobuf:"varint,2,rep" json:"Value,omitempty"` @@ -115,10 +125,75 @@ func (m *Bit) GetProfileID() uint64 { return 0 } +type Profile struct { + ID *uint64 `protobuf:"varint,1,req" json:"ID,omitempty"` + Attrs []*Attr `protobuf:"bytes,2,rep" json:"Attrs,omitempty"` + XXX_unrecognized []byte `json:"-"` +} + +func (m *Profile) Reset() { *m = Profile{} } +func (m *Profile) String() string { return proto.CompactTextString(m) } +func (*Profile) ProtoMessage() {} + +func (m *Profile) GetID() uint64 { + if m != nil && m.ID != nil { + return *m.ID + } + return 0 +} + +func (m *Profile) GetAttrs() []*Attr { + if m != nil { + return m.Attrs + } + return nil +} + +type Attr struct { + Key *string `protobuf:"bytes,1,req" json:"Key,omitempty"` + StringValue *string `protobuf:"bytes,2,opt" json:"StringValue,omitempty"` + IntValue *int64 `protobuf:"varint,3,opt" json:"IntValue,omitempty"` + BoolValue *bool `protobuf:"varint,4,opt" json:"BoolValue,omitempty"` + XXX_unrecognized []byte `json:"-"` +} + +func (m *Attr) Reset() { *m = Attr{} } +func (m *Attr) String() string { return proto.CompactTextString(m) } +func (*Attr) ProtoMessage() {} + +func (m *Attr) GetKey() string { + if m != nil && m.Key != nil { + return *m.Key + } + return "" +} + +func (m *Attr) GetStringValue() string { + if m != nil && m.StringValue != nil { + return *m.StringValue + } + return "" +} + +func (m *Attr) GetIntValue() int64 { + if m != nil && m.IntValue != nil { + return *m.IntValue + } + return 0 +} + +func (m *Attr) GetBoolValue() bool { + if m != nil && m.BoolValue != nil { + return *m.BoolValue + } + return false +} + type QueryRequest struct { DB *string `protobuf:"bytes,1,req" json:"DB,omitempty"` Query *string `protobuf:"bytes,2,req" json:"Query,omitempty"` Slices []uint64 `protobuf:"varint,3,rep" json:"Slices,omitempty"` + Profiles *bool `protobuf:"varint,4,opt" json:"Profiles,omitempty"` XXX_unrecognized []byte `json:"-"` } @@ -147,12 +222,20 @@ func (m *QueryRequest) GetSlices() []uint64 { return nil } +func (m *QueryRequest) GetProfiles() bool { + if m != nil && m.Profiles != nil { + return *m.Profiles + } + return false +} + type QueryResponse struct { - Err *string `protobuf:"bytes,1,opt" json:"Err,omitempty"` - Bitmap *Bitmap `protobuf:"bytes,2,opt" json:"Bitmap,omitempty"` - N *uint64 `protobuf:"varint,3,opt" json:"N,omitempty"` - Pairs []*Pair `protobuf:"bytes,4,rep" json:"Pairs,omitempty"` - XXX_unrecognized []byte `json:"-"` + Err *string `protobuf:"bytes,1,opt" json:"Err,omitempty"` + Bitmap *Bitmap `protobuf:"bytes,2,opt" json:"Bitmap,omitempty"` + N *uint64 `protobuf:"varint,3,opt" json:"N,omitempty"` + Pairs []*Pair `protobuf:"bytes,4,rep" json:"Pairs,omitempty"` + Profiles []*Profile `protobuf:"bytes,5,rep" json:"Profiles,omitempty"` + XXX_unrecognized []byte `json:"-"` } func (m *QueryResponse) Reset() { *m = QueryResponse{} } @@ -187,6 +270,13 @@ func (m *QueryResponse) GetPairs() []*Pair { return nil } +func (m *QueryResponse) GetProfiles() []*Profile { + if m != nil { + return m.Profiles + } + return nil +} + type ImportRequest struct { DB *string `protobuf:"bytes,1,req" json:"DB,omitempty"` Frame *string `protobuf:"bytes,2,req" json:"Frame,omitempty"` diff --git a/internal/internal.proto b/internal/internal.proto index ac1b97bca..7e89869a1 100644 --- a/internal/internal.proto +++ b/internal/internal.proto @@ -2,6 +2,7 @@ package internal; message Bitmap { repeated Chunk Chunks = 1; + repeated Attr Attrs = 2; } message Chunk { @@ -19,17 +20,31 @@ message Bit { required uint64 ProfileID = 2; } +message Profile { + required uint64 ID = 1; + repeated Attr Attrs = 2; +} + +message Attr { + required string Key = 1; + optional string StringValue = 2; + optional int64 IntValue = 3; + optional bool BoolValue = 4; +} + message QueryRequest { - required string DB = 1; - required string Query = 2; - repeated uint64 Slices = 3; + required string DB = 1; + required string Query = 2; + repeated uint64 Slices = 3; + optional bool Profiles = 4; } message QueryResponse { - optional string Err = 1; - optional Bitmap Bitmap = 2; - optional uint64 N = 3; - repeated Pair Pairs = 4; + optional string Err = 1; + optional Bitmap Bitmap = 2; + optional uint64 N = 3; + repeated Pair Pairs = 4; + repeated Profile Profiles = 5; } message ImportRequest { diff --git a/pilosa.go b/pilosa.go index 89e672f81..a6767dc11 100644 --- a/pilosa.go +++ b/pilosa.go @@ -2,6 +2,10 @@ package pilosa import ( "errors" + "sort" + + "github.com/gogo/protobuf/proto" + "github.com/umbel/pilosa/internal" ) //go:generate protoc --gogo_out=. internal/internal.proto @@ -19,3 +23,102 @@ var ( // Version represents the current running version of Pilosa. var Version string + +// Profile represents vertical column in a database. +// A profile can have a set of attributes attached to it. +type Profile struct { + ID uint64 `json:"id"` + Attrs map[string]interface{} `json:"attrs,omitempty"` +} + +// encodeProfiles converts a into its internal representation. +func encodeProfiles(a []*Profile) []*internal.Profile { + other := make([]*internal.Profile, len(a)) + for i := range a { + other[i] = encodeProfile(a[i]) + } + return other +} + +// decodeProfiles converts a from its internal representation. +func decodeProfiles(a []*internal.Profile) []*Profile { + other := make([]*Profile, len(a)) + for i := range a { + other[i] = decodeProfile(a[i]) + } + return other +} + +// encodeProfile converts p into its internal representation. +func encodeProfile(p *Profile) *internal.Profile { + return &internal.Profile{ + ID: proto.Uint64(p.ID), + Attrs: encodeAttrs(p.Attrs), + } +} + +// decodeProfile converts b from its internal representation. +func decodeProfile(pb *internal.Profile) *Profile { + p := &Profile{ + ID: pb.GetID(), + } + + if len(pb.GetAttrs()) > 0 { + p.Attrs = make(map[string]interface{}, len(pb.GetAttrs())) + for _, attr := range pb.GetAttrs() { + k, v := decodeAttr(attr) + p.Attrs[k] = v + } + } + + return p +} + +func encodeAttrs(m map[string]interface{}) []*internal.Attr { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + + a := make([]*internal.Attr, len(keys)) + for i := range keys { + a[i] = encodeAttr(keys[i], m[keys[i]]) + } + return a +} + +func decodeAttrs(pb []*internal.Attr) map[string]interface{} { + m := make(map[string]interface{}, len(pb)) + for i := range pb { + key, value := decodeAttr(pb[i]) + m[key] = value + } + return m +} + +// encodeAttr converts a key/value pair into an Attr internal representation. +func encodeAttr(key string, value interface{}) *internal.Attr { + pb := &internal.Attr{Key: proto.String(key)} + switch value := value.(type) { + case string: + pb.StringValue = proto.String(value) + case int64: + pb.IntValue = proto.Int64(value) + case bool: + pb.BoolValue = proto.Bool(value) + } + return pb +} + +// decodeAttr converts from an Attr internal representation to a key/value pair. +func decodeAttr(attr *internal.Attr) (key string, value interface{}) { + if attr.StringValue != nil { + return attr.GetKey(), attr.GetStringValue() + } else if attr.IntValue != nil { + return attr.GetKey(), attr.GetIntValue() + } else if attr.BoolValue != nil { + return attr.GetKey(), attr.GetBoolValue() + } + return attr.GetKey(), nil +} diff --git a/pql/ast.go b/pql/ast.go index 11a53aae8..0aeaa0abc 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -2,6 +2,7 @@ package pql import ( "fmt" + "sort" "strings" "time" ) @@ -20,15 +21,18 @@ type Node interface { String() string } -func (*Clear) node() {} -func (*Count) node() {} -func (*Difference) node() {} -func (*Get) node() {} -func (*Intersect) node() {} -func (*Range) node() {} -func (*Set) node() {} -func (*TopN) node() {} -func (*Union) node() {} +func (*Bitmap) node() {} +func (*ClearBit) node() {} +func (*Count) node() {} +func (*Difference) node() {} +func (*Intersect) node() {} +func (*Profile) node() {} +func (*Range) node() {} +func (*SetBit) node() {} +func (*SetBitmapAttrs) node() {} +func (*SetProfileAttrs) node() {} +func (*TopN) node() {} +func (*Union) node() {} // Call represents a function call in the AST. type Call interface { @@ -36,15 +40,18 @@ type Call interface { call() } -func (*Clear) call() {} -func (*Count) call() {} -func (*Difference) call() {} -func (*Get) call() {} -func (*Intersect) call() {} -func (*Range) call() {} -func (*Set) call() {} -func (*TopN) call() {} -func (*Union) call() {} +func (*Bitmap) call() {} +func (*ClearBit) call() {} +func (*Count) call() {} +func (*Difference) call() {} +func (*Intersect) call() {} +func (*Profile) call() {} +func (*Range) call() {} +func (*SetBit) call() {} +func (*SetBitmapAttrs) call() {} +func (*SetProfileAttrs) call() {} +func (*TopN) call() {} +func (*Union) call() {} // Calls represents a list of calls. type Calls []Call @@ -77,13 +84,31 @@ func (a BitmapCalls) String() string { } func (*Difference) bitmapCall() {} -func (*Get) bitmapCall() {} +func (*Bitmap) bitmapCall() {} func (*Intersect) bitmapCall() {} func (*Range) bitmapCall() {} func (*Union) bitmapCall() {} -// Clear represents a clear() function call. -type Clear struct { +// Bitmap represents a Bitmap() function call. +type Bitmap struct { + ID uint64 + Frame string +} + +// String returns the string representation of the call. +func (c *Bitmap) String() string { + args := make([]string, 0, 2) + if c.ID != 0 { + args = append(args, fmt.Sprintf("id=%d", c.ID)) + } + if c.Frame != "" { + args = append(args, fmt.Sprintf("frame=%s", c.Frame)) + } + return fmt.Sprintf("Bitmap(%s)", strings.Join(args, ", ")) +} + +// ClearBit represents a ClearBit() function call. +type ClearBit struct { ID uint64 Frame string Filter uint64 @@ -91,11 +116,9 @@ type Clear struct { } // String returns the string representation of the call. -func (c *Clear) String() string { +func (c *ClearBit) String() string { args := make([]string, 0, 4) - if c.ID != 0 { - args = append(args, fmt.Sprintf("id=%d", c.ID)) - } + args = append(args, fmt.Sprintf("id=%d", c.ID)) if c.Frame != "" { args = append(args, fmt.Sprintf("frame=%s", c.Frame)) } @@ -103,9 +126,9 @@ func (c *Clear) String() string { args = append(args, fmt.Sprintf("filter=%d", c.Filter)) } if c.ProfileID != 0 { - args = append(args, fmt.Sprintf("profile_id=%d", c.ProfileID)) + args = append(args, fmt.Sprintf("profileID=%d", c.ProfileID)) } - return fmt.Sprintf("clear(%s)", strings.Join(args, ", ")) + return fmt.Sprintf("ClearBit(%s)", strings.Join(args, ", ")) } // Count represents a count() function call. @@ -115,7 +138,7 @@ type Count struct { // String returns the string representation of the call. func (c *Count) String() string { - return fmt.Sprintf("count(%s)", c.Input.String()) + return fmt.Sprintf("Count(%s)", c.Input.String()) } // Difference represents an difference() function call. @@ -125,25 +148,7 @@ type Difference struct { // String returns the string representation of the call. func (c *Difference) String() string { - return fmt.Sprintf("difference(%s)", c.Inputs.String()) -} - -// Get represents a get() function call. -type Get struct { - ID uint64 - Frame string -} - -// String returns the string representation of the call. -func (c *Get) String() string { - args := make([]string, 0, 2) - if c.ID != 0 { - args = append(args, fmt.Sprintf("id=%d", c.ID)) - } - if c.Frame != "" { - args = append(args, fmt.Sprintf("frame=%s", c.Frame)) - } - return fmt.Sprintf("get(%s)", strings.Join(args, ", ")) + return fmt.Sprintf("Difference(%s)", c.Inputs.String()) } // Intersect represents an intersect() function call. @@ -153,7 +158,19 @@ type Intersect struct { // String returns the string representation of the call. func (c *Intersect) String() string { - return fmt.Sprintf("intersect(%s)", c.Inputs.String()) + return fmt.Sprintf("Intersect(%s)", c.Inputs.String()) +} + +// Profile represents a Profile() function call. +type Profile struct { + ID uint64 +} + +// String returns the string representation of the call. +func (c *Profile) String() string { + args := make([]string, 0, 1) + args = append(args, fmt.Sprintf("id=%d", c.ID)) + return fmt.Sprintf("Profile(%s)", strings.Join(args, ", ")) } // Range represents a range() function call. @@ -179,11 +196,11 @@ func (c *Range) String() string { if !c.EndTime.IsZero() { args = append(args, fmt.Sprintf("end=%s", c.EndTime.Format(TimeFormat))) } - return fmt.Sprintf("range(%s)", strings.Join(args, ", ")) + return fmt.Sprintf("Range(%s)", strings.Join(args, ", ")) } -// Set represents a set() function call. -type Set struct { +// SetBit represents a SetBit() function call. +type SetBit struct { ID uint64 Frame string Filter uint64 @@ -191,11 +208,9 @@ type Set struct { } // String returns the string representation of the call. -func (c *Set) String() string { +func (c *SetBit) String() string { args := make([]string, 0, 2) - if c.ID != 0 { - args = append(args, fmt.Sprintf("id=%d", c.ID)) - } + args = append(args, fmt.Sprintf("id=%d", c.ID)) if c.Frame != "" { args = append(args, fmt.Sprintf("frame=%s", c.Frame)) } @@ -203,12 +218,88 @@ func (c *Set) String() string { args = append(args, fmt.Sprintf("filter=%d", c.Filter)) } if c.ProfileID != 0 { - args = append(args, fmt.Sprintf("profile_id=%d", c.ProfileID)) + args = append(args, fmt.Sprintf("profileID=%d", c.ProfileID)) } - return fmt.Sprintf("set(%s)", strings.Join(args, ", ")) + return fmt.Sprintf("SetBit(%s)", strings.Join(args, ", ")) } -// TopN represents a top-n() function call. +// SetBitmapAttrs represents a SetBitmapAttrs() function call. +type SetBitmapAttrs struct { + ID uint64 + Frame string + Attrs map[string]interface{} +} + +// String returns the string representation of the call. +func (c *SetBitmapAttrs) String() string { + args := make([]string, 0, 2) + args = append(args, fmt.Sprintf("id=%d", c.ID)) + if c.Frame != "" { + args = append(args, fmt.Sprintf("frame=%s", c.Frame)) + } + + // Sort keys. + keys := make([]string, 0, len(c.Attrs)) + for k := range c.Attrs { + keys = append(keys, k) + } + sort.Strings(keys) + + // Write key/value pairs. + for _, k := range keys { + if c.Attrs[k] == nil { + args = append(args, fmt.Sprintf("%s=null", k)) + continue + } + + switch v := c.Attrs[k].(type) { + case string: + args = append(args, fmt.Sprintf("%s=\"%s\"", k, v)) + default: + args = append(args, fmt.Sprintf("%s=%v", k, v)) + } + } + + return fmt.Sprintf("SetBitmapAttrs(%s)", strings.Join(args, ", ")) +} + +// SetProfileAttrs represents a SetProfileAttrs() function call. +type SetProfileAttrs struct { + ID uint64 + Attrs map[string]interface{} +} + +// String returns the string representation of the call. +func (c *SetProfileAttrs) String() string { + args := make([]string, 0, 2) + args = append(args, fmt.Sprintf("id=%d", c.ID)) + + // Sort keys. + keys := make([]string, 0, len(c.Attrs)) + for k := range c.Attrs { + keys = append(keys, k) + } + sort.Strings(keys) + + // Write key/value pairs. + for _, k := range keys { + if c.Attrs[k] == nil { + args = append(args, fmt.Sprintf("%s=null", k)) + continue + } + + switch v := c.Attrs[k].(type) { + case string: + args = append(args, fmt.Sprintf("%s=\"%s\"", k, v)) + default: + args = append(args, fmt.Sprintf("%s=%v", k, v)) + } + } + + return fmt.Sprintf("SetProfileAttrs(%s)", strings.Join(args, ", ")) +} + +// TopN represents a TopN() function call. type TopN struct { Frame string N int @@ -224,5 +315,5 @@ type Union struct { // String returns the string representation of the call. func (c *Union) String() string { - return fmt.Sprintf("union(%s)", c.Inputs.String()) + return fmt.Sprintf("Union(%s)", c.Inputs.String()) } diff --git a/pql/ast_test.go b/pql/ast_test.go index a390f5263..e3974013a 100644 --- a/pql/ast_test.go +++ b/pql/ast_test.go @@ -7,18 +7,26 @@ import ( "github.com/umbel/pilosa/pql" ) -// Ensure the Clear call can be converted into a string. -func TestClear_String(t *testing.T) { - s := (&pql.Clear{ID: 1, Frame: "x.n", Filter: 2, ProfileID: 3}).String() - if s != `clear(id=1, frame=x.n, filter=2, profile_id=3)` { +// Ensure the Bitmap call can be converted into a string. +func TestBitmap_String(t *testing.T) { + s := (&pql.Bitmap{ID: 1, Frame: "x.n"}).String() + if s != `Bitmap(id=1, frame=x.n)` { + t.Fatalf("unexpected string: %s", s) + } +} + +// Ensure the ClearBit call can be converted into a string. +func TestClearBit_String(t *testing.T) { + s := (&pql.ClearBit{ID: 1, Frame: "x.n", Filter: 2, ProfileID: 3}).String() + if s != `ClearBit(id=1, frame=x.n, filter=2, profileID=3)` { t.Fatalf("unexpected string: %s", s) } } // Ensure the Count call can be converted into a string. func TestCount_String(t *testing.T) { - s := (&pql.Count{Input: &pql.Get{ID: 1, Frame: "x.n"}}).String() - if s != `count(get(id=1, frame=x.n))` { + s := (&pql.Count{Input: &pql.Bitmap{ID: 1, Frame: "x.n"}}).String() + if s != `Count(Bitmap(id=1, frame=x.n))` { t.Fatalf("unexpected string: %s", s) } } @@ -26,19 +34,11 @@ func TestCount_String(t *testing.T) { // Ensure the Difference call can be converted into a string. func TestDifference_String(t *testing.T) { s := (&pql.Difference{Inputs: pql.BitmapCalls{ - &pql.Get{ID: 1, Frame: "x.n"}, - &pql.Get{ID: 2}, + &pql.Bitmap{ID: 1, Frame: "x.n"}, + &pql.Bitmap{ID: 2}, }, }).String() - if s != `difference(get(id=1, frame=x.n), get(id=2))` { - t.Fatalf("unexpected string: %s", s) - } -} - -// Ensure the Get call can be converted into a string. -func TestGet_String(t *testing.T) { - s := (&pql.Get{ID: 1, Frame: "x.n"}).String() - if s != `get(id=1, frame=x.n)` { + if s != `Difference(Bitmap(id=1, frame=x.n), Bitmap(id=2))` { t.Fatalf("unexpected string: %s", s) } } @@ -46,11 +46,18 @@ func TestGet_String(t *testing.T) { // Ensure the Intersect call can be converted into a string. func TestIntersect_String(t *testing.T) { s := (&pql.Intersect{Inputs: pql.BitmapCalls{ - &pql.Get{ID: 1, Frame: "x.n"}, - &pql.Get{ID: 2}, + &pql.Bitmap{ID: 1, Frame: "x.n"}, + &pql.Bitmap{ID: 2}, }, }).String() - if s != `intersect(get(id=1, frame=x.n), get(id=2))` { + if s != `Intersect(Bitmap(id=1, frame=x.n), Bitmap(id=2))` { + t.Fatalf("unexpected string: %s", s) + } +} + +// Ensure the Profile call can be converted into a string. +func TestProfile_String(t *testing.T) { + if s := (&pql.Profile{ID: 1}).String(); s != `Profile(id=1)` { t.Fatalf("unexpected string: %s", s) } } @@ -63,15 +70,23 @@ func TestRange_String(t *testing.T) { StartTime: time.Unix(0, 0).UTC(), EndTime: time.Date(2000, 1, 2, 3, 4, 0, 0, time.UTC), }).String() - if s != `range(id=1, frame=x.n, start=1970-01-01T00:00, end=2000-01-02T03:04)` { + if s != `Range(id=1, frame=x.n, start=1970-01-01T00:00, end=2000-01-02T03:04)` { t.Fatalf("unexpected string: %s", s) } } -// Ensure the Set call can be converted into a string. -func TestSet_String(t *testing.T) { - s := (&pql.Set{ID: 1, Frame: "x.n", Filter: 2, ProfileID: 3}).String() - if s != `set(id=1, frame=x.n, filter=2, profile_id=3)` { +// Ensure the SetBit call can be converted into a string. +func TestSetBit_String(t *testing.T) { + s := (&pql.SetBit{ID: 1, Frame: "x.n", Filter: 2, ProfileID: 3}).String() + if s != `SetBit(id=1, frame=x.n, filter=2, profileID=3)` { + t.Fatalf("unexpected string: %s", s) + } +} + +// Ensure the SetBitmapAttrs call can be converted into a string. +func TestSetBitmapAttrs_String(t *testing.T) { + s := (&pql.SetBitmapAttrs{ID: 1, Frame: "x.n", Attrs: map[string]interface{}{"foo": "bar", "baz": 123, "bat": true, "x": nil}}).String() + if s != `SetBitmapAttrs(id=1, frame=x.n, bat=true, baz=123, foo="bar", x=null)` { t.Fatalf("unexpected string: %s", s) } } @@ -79,11 +94,11 @@ func TestSet_String(t *testing.T) { // Ensure the Union call can be converted into a string. func TestUnion_String(t *testing.T) { s := (&pql.Union{Inputs: pql.BitmapCalls{ - &pql.Get{ID: 1, Frame: "x.n"}, - &pql.Get{ID: 2}, + &pql.Bitmap{ID: 1, Frame: "x.n"}, + &pql.Bitmap{ID: 2}, }, }).String() - if s != `union(get(id=1, frame=x.n), get(id=2))` { + if s != `Union(Bitmap(id=1, frame=x.n), Bitmap(id=2))` { t.Fatalf("unexpected string: %s", s) } } diff --git a/pql/parser.go b/pql/parser.go index 1fb41a5c3..5f30d3280 100644 --- a/pql/parser.go +++ b/pql/parser.go @@ -45,32 +45,73 @@ func (p *Parser) parseCall() (Call, error) { } switch lit { - case "count": + case "Bitmap": + return p.parseBitmapCall() + case "Count": return p.parseCountCall() - case "clear": - return p.parseClearCall() - case "difference": + case "ClearBit": + return p.parseClearBitCall() + case "Difference": return p.parseDifferenceCall() - case "get": - return p.parseGetCall() - case "intersect": + case "Intersect": return p.parseIntersectCall() - case "range": + case "Profile": + return p.parseProfileCall() + case "Range": return p.parseRangeCall() - case "set": - return p.parseSetCall() - case "top-n": + case "SetBit": + return p.parseSetBitCall() + case "SetBitmapAttrs": + return p.parseSetBitmapAttrsCall() + case "SetProfileAttrs": + return p.parseSetProfileAttrsCall() + case "TopN": return p.parseTopNCall() - case "union": + case "Union": return p.parseUnionCall() default: return nil, &ParseError{Message: fmt.Sprintf("function not found: %s", lit), Pos: pos} } } -// parseClearCall parses a clear() function call. -func (p *Parser) parseClearCall() (*Clear, error) { - c := &Clear{} +// parseBitmapCall parses a Bitmap() function call. +func (p *Parser) parseBitmapCall() (*Bitmap, error) { + c := &Bitmap{} + pos := p.pos() + + // Scan opening parenthesis. + if err := p.expect(LPAREN); err != nil { + return nil, err + } + + // Parse arguments. + args, err := p.parseArgs() + if err != nil { + return nil, err + } + + // Copy arguments to AST. + for _, arg := range args { + switch arg.key { + case 0, "id": + if err := decodeUint64(arg.value, &c.ID); err != nil { + return nil, parseErrorf(pos, "id: %s", err) + } + case 1, "frame": + if err := decodeString(arg.value, &c.Frame); err != nil { + return nil, parseErrorf(pos, "frame: %s", err) + } + default: + return nil, parseErrorf(pos, "invalid Bitmap() arg: %v", arg.key) + } + } + + return c, nil +} + +// parseClearBitCall parses a ClearBit() function call. +func (p *Parser) parseClearBitCall() (*ClearBit, error) { + c := &ClearBit{} pos := p.pos() // Scan opening parenthesis. @@ -99,19 +140,19 @@ func (p *Parser) parseClearCall() (*Clear, error) { if err := decodeUint64(arg.value, &c.Filter); err != nil { return nil, parseErrorf(pos, "filter: %s", err) } - case 3, "profile_id": + case 3, "profileID": if err := decodeUint64(arg.value, &c.ProfileID); err != nil { - return nil, parseErrorf(pos, "profile_id: %s", err) + return nil, parseErrorf(pos, "profileID: %s", err) } default: - return nil, parseErrorf(pos, "invalid arg: %v", arg.key) + return nil, parseErrorf(pos, "invalid ClearBit() arg: %v", arg.key) } } return c, nil } -// parseCount parses a count() function call. +// parseCount parses a Count() function call. func (p *Parser) parseCountCall() (*Count, error) { c := &Count{} pos := p.pos() @@ -139,7 +180,7 @@ func (p *Parser) parseCountCall() (*Count, error) { return c, nil } -// parseDifference parses a difference() function call. +// parseDifference parses a Difference() function call. func (p *Parser) parseDifferenceCall() (*Difference, error) { c := &Difference{} pos := p.pos() @@ -160,49 +201,14 @@ func (p *Parser) parseDifferenceCall() (*Difference, error) { if v, ok := arg.value.(BitmapCall); ok { c.Inputs = append(c.Inputs, v) } else { - return nil, parseErrorf(pos, "invalid arg: %v", arg.value) + return nil, parseErrorf(pos, "invalid Difference() arg: %v", arg.value) } } return c, nil } -// parseGetCall parses a get() function call. -func (p *Parser) parseGetCall() (*Get, error) { - c := &Get{} - pos := p.pos() - - // Scan opening parenthesis. - if err := p.expect(LPAREN); err != nil { - return nil, err - } - - // Parse arguments. - args, err := p.parseArgs() - if err != nil { - return nil, err - } - - // Copy arguments to AST. - for _, arg := range args { - switch arg.key { - case 0, "id": - if err := decodeUint64(arg.value, &c.ID); err != nil { - return nil, parseErrorf(pos, "id: %s", err) - } - case 1, "frame": - if err := decodeString(arg.value, &c.Frame); err != nil { - return nil, parseErrorf(pos, "frame: %s", err) - } - default: - return nil, parseErrorf(pos, "invalid arg: %v", arg.key) - } - } - - return c, nil -} - -// parseIntersect parses a intersect() function call. +// parseIntersect parses a Intersect() function call. func (p *Parser) parseIntersectCall() (*Intersect, error) { c := &Intersect{} pos := p.pos() @@ -223,14 +229,45 @@ func (p *Parser) parseIntersectCall() (*Intersect, error) { if v, ok := arg.value.(BitmapCall); ok { c.Inputs = append(c.Inputs, v) } else { - return nil, parseErrorf(pos, "invalid arg: %v", arg.value) + return nil, parseErrorf(pos, "invalid Intersect() arg: %v", arg.value) } } return c, nil } -// parseRangeCall parses a range() function call. +// parseProfileCall parses a Profile() function call. +func (p *Parser) parseProfileCall() (*Profile, error) { + c := &Profile{} + pos := p.pos() + + // Scan opening parenthesis. + if err := p.expect(LPAREN); err != nil { + return nil, err + } + + // Parse arguments. + args, err := p.parseArgs() + if err != nil { + return nil, err + } + + // Copy arguments to AST. + for _, arg := range args { + switch arg.key { + case 0, "id": + if err := decodeUint64(arg.value, &c.ID); err != nil { + return nil, parseErrorf(pos, "id: %s", err) + } + default: + return nil, parseErrorf(pos, "invalid Profile() arg: %v", arg.key) + } + } + + return c, nil +} + +// parseRangeCall parses a Range() function call. func (p *Parser) parseRangeCall() (*Range, error) { c := &Range{} pos := p.pos() @@ -266,16 +303,16 @@ func (p *Parser) parseRangeCall() (*Range, error) { return nil, parseErrorf(pos, "end: %s", err) } default: - return nil, parseErrorf(pos, "invalid arg: %v", arg.key) + return nil, parseErrorf(pos, "invalid Range() arg: %v", arg.key) } } return c, nil } -// parseSetCall parses a set() function call. -func (p *Parser) parseSetCall() (*Set, error) { - c := &Set{} +// parseSetBitCall parses a SetBit() function call. +func (p *Parser) parseSetBitCall() (*SetBit, error) { + c := &SetBit{} pos := p.pos() // Scan opening parenthesis. @@ -304,19 +341,125 @@ func (p *Parser) parseSetCall() (*Set, error) { if err := decodeUint64(arg.value, &c.Filter); err != nil { return nil, parseErrorf(pos, "filter: %s", err) } - case 3, "profile_id": + case 3, "profileID": if err := decodeUint64(arg.value, &c.ProfileID); err != nil { - return nil, parseErrorf(pos, "profile_id: %s", err) + return nil, parseErrorf(pos, "profileID: %s", err) } default: - return nil, parseErrorf(pos, "invalid arg: %v", arg.key) + return nil, parseErrorf(pos, "invalid SetBit() arg: %v", arg.key) } } return c, nil } -// parseTopNCall parses a top-n() function call. +// parseSetBitmapAttrsCall parses a SetBitmapAttrs() function call. +func (p *Parser) parseSetBitmapAttrsCall() (*SetBitmapAttrs, error) { + c := &SetBitmapAttrs{ + Attrs: make(map[string]interface{}), + } + pos := p.pos() + + // Scan opening parenthesis. + if err := p.expect(LPAREN); err != nil { + return nil, err + } + + // Parse arguments. + args, err := p.parseArgs() + if err != nil { + return nil, err + } + + // Copy arguments to AST. + for _, arg := range args { + switch arg.key { + case 0, "id": + if err := decodeUint64(arg.value, &c.ID); err != nil { + return nil, parseErrorf(pos, "id: %s", err) + } + case 1, "frame": + if err := decodeString(arg.value, &c.Frame); err != nil { + return nil, parseErrorf(pos, "frame: %s", err) + } + default: + key, ok := arg.key.(string) + if !ok { + return nil, parseErrorf(pos, "invalid attr arg: %v", arg.key) + } + + // Special handling for nil values. + if arg.value == nil { + c.Attrs[key] = nil + continue + } + + switch v := arg.value.(type) { + case string, bool: + c.Attrs[key] = v + case uint64: + c.Attrs[key] = int64(v) + default: + return nil, parseErrorf(pos, "invalid SetBitmapAttrs() arg: %v", arg.key) + } + } + } + + return c, nil +} + +// parseSetProfileAttrsCall parses a SetProfileAttrs() function call. +func (p *Parser) parseSetProfileAttrsCall() (*SetProfileAttrs, error) { + c := &SetProfileAttrs{ + Attrs: make(map[string]interface{}), + } + pos := p.pos() + + // Scan opening parenthesis. + if err := p.expect(LPAREN); err != nil { + return nil, err + } + + // Parse arguments. + args, err := p.parseArgs() + if err != nil { + return nil, err + } + + // Copy arguments to AST. + for _, arg := range args { + switch arg.key { + case 0, "id": + if err := decodeUint64(arg.value, &c.ID); err != nil { + return nil, parseErrorf(pos, "id: %s", err) + } + default: + key, ok := arg.key.(string) + if !ok { + return nil, parseErrorf(pos, "invalid attr arg: %v", arg.key) + } + + // Special handling for nil values. + if arg.value == nil { + c.Attrs[key] = nil + continue + } + + switch v := arg.value.(type) { + case string, bool: + c.Attrs[key] = v + case uint64: + c.Attrs[key] = int64(v) + default: + return nil, parseErrorf(pos, "invalid SetProfileAttrs() arg: %v", arg.key) + } + } + } + + return c, nil +} + +// parseTopNCall parses a TopN() function call. func (p *Parser) parseTopNCall() (*TopN, error) { c := &TopN{} pos := p.pos() @@ -344,14 +487,14 @@ func (p *Parser) parseTopNCall() (*TopN, error) { return nil, parseErrorf(pos, "n: %s", err) } default: - return nil, parseErrorf(pos, "invalid arg: %v", arg.key) + return nil, parseErrorf(pos, "invalid TopN() arg: %v", arg.key) } } return c, nil } -// parseUnion parses a union() function call. +// parseUnion parses a Union() function call. func (p *Parser) parseUnionCall() (*Union, error) { c := &Union{} pos := p.pos() @@ -372,7 +515,7 @@ func (p *Parser) parseUnionCall() (*Union, error) { if v, ok := arg.value.(BitmapCall); ok { c.Inputs = append(c.Inputs, v) } else { - return nil, parseErrorf(pos, "invalid arg: %v", arg.value) + return nil, parseErrorf(pos, "invalid Union() arg: %v", arg.value) } } @@ -446,7 +589,17 @@ func (p *Parser) parseArg() (arg, error) { // Read value token. tok, pos, lit = p.scanIgnoreWhitespace() switch tok { - case IDENT, STRING: + case IDENT: + if lit == "true" { + value = true + } else if lit == "false" { + value = false + } else if lit == "null" { + value = nil + } else { + value = lit + } + case STRING: value = lit case NUMBER: v, err := strconv.ParseUint(lit, 10, 64) diff --git a/pql/parser_test.go b/pql/parser_test.go index 9e7333dfb..f0ff1e333 100644 --- a/pql/parser_test.go +++ b/pql/parser_test.go @@ -9,13 +9,43 @@ import ( "github.com/umbel/pilosa/pql" ) -// Ensure the parser can parse a "clear()" function with keyed args. -func TestParser_Parse_Clear_Key(t *testing.T) { - q, err := pql.ParseString(`clear(id=1, frame="b.n", filter=2, profile_id = 3)`) +// Ensure the parser can parse a "Bitmap()" function with keyed args. +func TestParser_Parse_Bitmap_Key(t *testing.T) { + q, err := pql.ParseString(`Bitmap(id=1, frame="b.n")`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ - Root: &pql.Clear{ + Root: &pql.Bitmap{ + ID: 1, + Frame: "b.n", + }, + }) { + t.Fatalf("unexpected query: %s", spew.Sdump(q)) + } +} + +// Ensure the parser can parse a "Bitmap()" function with array args. +func TestParser_Parse_Bitmap_Array(t *testing.T) { + q, err := pql.ParseString(`Bitmap(1, "b.n")`) + if err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(q, &pql.Query{ + Root: &pql.Bitmap{ + ID: 1, + Frame: "b.n", + }, + }) { + t.Fatalf("unexpected query: %s", spew.Sdump(q)) + } +} + +// Ensure the parser can parse a "ClearBit()" function with keyed args. +func TestParser_Parse_ClearBit_Key(t *testing.T) { + q, err := pql.ParseString(`ClearBit(id=1, frame="b.n", filter=2, profileID = 3)`) + if err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(q, &pql.Query{ + Root: &pql.ClearBit{ ID: 1, Frame: "b.n", Filter: 2, @@ -26,13 +56,13 @@ func TestParser_Parse_Clear_Key(t *testing.T) { } } -// Ensure the parser can parse a "clear()" function with array args. -func TestParser_Parse_Clear_Array(t *testing.T) { - q, err := pql.ParseString(`clear(1, "b.n", 2, 3)`) +// Ensure the parser can parse a "ClearBit()" function with array args. +func TestParser_Parse_ClearBit_Array(t *testing.T) { + q, err := pql.ParseString(`ClearBit(1, "b.n", 2, 3)`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ - Root: &pql.Clear{ + Root: &pql.ClearBit{ ID: 1, Frame: "b.n", Filter: 2, @@ -45,12 +75,12 @@ func TestParser_Parse_Clear_Array(t *testing.T) { // Ensure the parser can parse a "count()" function. func TestParser_Parse_Count(t *testing.T) { - q, err := pql.ParseString(`count(get(1))`) + q, err := pql.ParseString(`Count(Bitmap(1))`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ Root: &pql.Count{ - Input: &pql.Get{ + Input: &pql.Bitmap{ ID: 1, }, }, @@ -61,14 +91,14 @@ func TestParser_Parse_Count(t *testing.T) { // Ensure the parser can parse a "difference()" function. func TestParser_Parse_Difference(t *testing.T) { - q, err := pql.ParseString(`difference(get(1), get(2))`) + q, err := pql.ParseString(`Difference(Bitmap(1), Bitmap(2))`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ Root: &pql.Difference{ Inputs: pql.BitmapCalls{ - &pql.Get{ID: 1}, - &pql.Get{ID: 2}, + &pql.Bitmap{ID: 1}, + &pql.Bitmap{ID: 2}, }, }, }) { @@ -76,46 +106,16 @@ func TestParser_Parse_Difference(t *testing.T) { } } -// Ensure the parser can parse a "get()" function with keyed args. -func TestParser_Parse_Get_Key(t *testing.T) { - q, err := pql.ParseString(`get(id=1, frame="b.n")`) - if err != nil { - t.Fatal(err) - } else if !reflect.DeepEqual(q, &pql.Query{ - Root: &pql.Get{ - ID: 1, - Frame: "b.n", - }, - }) { - t.Fatalf("unexpected query: %s", spew.Sdump(q)) - } -} - -// Ensure the parser can parse a "get()" function with array args. -func TestParser_Parse_Get_Array(t *testing.T) { - q, err := pql.ParseString(`get(1, "b.n")`) - if err != nil { - t.Fatal(err) - } else if !reflect.DeepEqual(q, &pql.Query{ - Root: &pql.Get{ - ID: 1, - Frame: "b.n", - }, - }) { - t.Fatalf("unexpected query: %s", spew.Sdump(q)) - } -} - // Ensure the parser can parse a "intersect()" function. func TestParser_Parse_Intersect(t *testing.T) { - q, err := pql.ParseString(`intersect(get(1), get(2))`) + q, err := pql.ParseString(`Intersect(Bitmap(1), Bitmap(2))`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ Root: &pql.Intersect{ Inputs: pql.BitmapCalls{ - &pql.Get{ID: 1}, - &pql.Get{ID: 2}, + &pql.Bitmap{ID: 1}, + &pql.Bitmap{ID: 2}, }, }, }) { @@ -123,9 +123,33 @@ func TestParser_Parse_Intersect(t *testing.T) { } } +// Ensure the parser can parse a "Profile()" function with keyed args. +func TestParser_Parse_Profile_Key(t *testing.T) { + q, err := pql.ParseString(`Profile(id=1)`) + if err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(q, &pql.Query{ + Root: &pql.Profile{ID: 1}, + }) { + t.Fatalf("unexpected query: %s", spew.Sdump(q)) + } +} + +// Ensure the parser can parse a "Profile()" function with array args. +func TestParser_Parse_Profile_Array(t *testing.T) { + q, err := pql.ParseString(`Profile(1)`) + if err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(q, &pql.Query{ + Root: &pql.Profile{ID: 1}, + }) { + t.Fatalf("unexpected query: %s", spew.Sdump(q)) + } +} + // Ensure the parser can parse a "range()" function with keyed args. func TestParser_Parse_Range_Key(t *testing.T) { - q, err := pql.ParseString(`range(start="2000-01-02T03:04", id=20, frame="b.n", end="2001-01-02T03:04")`) + q, err := pql.ParseString(`Range(start="2000-01-02T03:04", id=20, frame="b.n", end="2001-01-02T03:04")`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ @@ -142,7 +166,7 @@ func TestParser_Parse_Range_Key(t *testing.T) { // Ensure the parser can parse a "range()" function with array args. func TestParser_Parse_Range_Array(t *testing.T) { - q, err := pql.ParseString(`range(20, "b.n", "2000-01-02T03:04", "2001-01-02T03:04")`) + q, err := pql.ParseString(`Range(20, "b.n", "2000-01-02T03:04", "2001-01-02T03:04")`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ @@ -157,13 +181,13 @@ func TestParser_Parse_Range_Array(t *testing.T) { } } -// Ensure the parser can parse a "set()" function with keyed args. -func TestParser_Parse_Set_Key(t *testing.T) { - q, err := pql.ParseString(`set(id=1, frame="b.n", filter=2, profile_id = 3)`) +// Ensure the parser can parse a "SetBit()" function with keyed args. +func TestParser_Parse_SetBit_Key(t *testing.T) { + q, err := pql.ParseString(`SetBit(id=1, frame="b.n", filter=2, profileID = 3)`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ - Root: &pql.Set{ + Root: &pql.SetBit{ ID: 1, Frame: "b.n", Filter: 2, @@ -174,13 +198,13 @@ func TestParser_Parse_Set_Key(t *testing.T) { } } -// Ensure the parser can parse a "set()" function with array args. -func TestParser_Parse_Set_Array(t *testing.T) { - q, err := pql.ParseString(`set(1, "b.n", 2, 3)`) +// Ensure the parser can parse a "SetBit()" function with array args. +func TestParser_Parse_SetBit_Array(t *testing.T) { + q, err := pql.ParseString(`SetBit(1, "b.n", 2, 3)`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ - Root: &pql.Set{ + Root: &pql.SetBit{ ID: 1, Frame: "b.n", Filter: 2, @@ -191,9 +215,50 @@ func TestParser_Parse_Set_Array(t *testing.T) { } } -// Ensure the parser can parse a "top-n()" function with keyed args. +// Ensure the parser can parse a "SetBitmapAttrs()" function with keyed args. +func TestParser_Parse_SetBitmapAttrs_Key(t *testing.T) { + q, err := pql.ParseString(`SetBitmapAttrs(id=1, frame="b.n", foo="bar", bar=123, baz=true, bat=false, x=null)`) + if err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(q, &pql.Query{ + Root: &pql.SetBitmapAttrs{ + ID: 1, + Frame: "b.n", + Attrs: map[string]interface{}{ + "foo": "bar", + "bar": int64(123), + "baz": true, + "bat": false, + "x": nil, + }, + }, + }) { + t.Fatalf("unexpected query: %s", spew.Sdump(q)) + } +} + +// Ensure the parser can parse a "SetBitmapAttrs()" function with array args. +func TestParser_Parse_SetBitmapAttrs_Array(t *testing.T) { + q, err := pql.ParseString(`SetBitmapAttrs(1, "b.n", foo=bar, bar=123)`) + if err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(q, &pql.Query{ + Root: &pql.SetBitmapAttrs{ + ID: 1, + Frame: "b.n", + Attrs: map[string]interface{}{ + "foo": "bar", + "bar": int64(123), + }, + }, + }) { + t.Fatalf("unexpected query: %s", spew.Sdump(q)) + } +} + +// Ensure the parser can parse a "TopN()" function with keyed args. func TestParser_Parse_TopN_Key(t *testing.T) { - q, err := pql.ParseString(`top-n(frame="b.n", n=2)`) + q, err := pql.ParseString(`TopN(frame="b.n", n=2)`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ @@ -206,9 +271,9 @@ func TestParser_Parse_TopN_Key(t *testing.T) { } } -// Ensure the parser can parse a "top-n()" function with array args. +// Ensure the parser can parse a "TopN()" function with array args. func TestParser_Parse_TopN_Array(t *testing.T) { - q, err := pql.ParseString(`top-n("b.n", 2)`) + q, err := pql.ParseString(`TopN("b.n", 2)`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ @@ -223,14 +288,14 @@ func TestParser_Parse_TopN_Array(t *testing.T) { // Ensure the parser can parse a "union()" function. func TestParser_Parse_Union(t *testing.T) { - q, err := pql.ParseString(`union(get(1), get(2))`) + q, err := pql.ParseString(`Union(Bitmap(1), Bitmap(2))`) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(q, &pql.Query{ Root: &pql.Union{ Inputs: pql.BitmapCalls{ - &pql.Get{ID: 1}, - &pql.Get{ID: 2}, + &pql.Bitmap{ID: 1}, + &pql.Bitmap{ID: 2}, }, }, }) {