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

Merge pull request #44 from benbjohnson/pql

Browse source 
Add bitmap and profile attribute support
This commit is contained in:
tgruben 2016-02-01 15:16:31 -06:00
commit 6a3296457a
18 changed files with 1873 additions and 498 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
bitmap.go
View file

@ -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))
}

111
cmd/pilosa/main_test.go
View file

@ -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)
}

264
db.go Normal file
View file

@ -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
}

127
db_test.go Normal file
View file

@ -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()
}

79
executor.go
View file

@ -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))

65
executor_test.go
View file

@ -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)
}

24
fragment.go
View file

@ -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()

121
frame.go
View file

@ -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

219
handler.go
View file

@ -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

145
handler_test.go
View file

@ -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 {

187
index.go
View file

@ -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)
}

100
internal/internal.pb.go
View file

@ -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"`

29
internal/internal.proto
View file

@ -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 {

103
pilosa.go
View file

@ -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
}

209
pql/ast.go
View file

@ -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())
}

71
pql/ast_test.go
View file

@ -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)
}
}

295
pql/parser.go
View file

@ -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)

191
pql/parser_test.go
View file

@ -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},
},
},
}) {