scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Rename Index to Holder

Browse source 
Rename `index.go` to `holder.go` and `index_test.go` to `holder_test.go`
This commit is contained in:
Travis 2017-04-23 15:52:51 -05:00
parent eec02179c8
commit 2cacff8e09
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
9 changed files with 519 additions and 518 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

122
client_test.go
View file

@ -13,81 +13,81 @@ import (
"github.com/pilosa/pilosa/pql"
)
func createCluster(c *pilosa.Cluster) ([]*Server, []*Index) {
func createCluster(c *pilosa.Cluster) ([]*Server, []*Holder) {
numNodes := len(c.Nodes)
idx := make([]*Index, numNodes)
hldr := make([]*Holder, numNodes)
server := make([]*Server, numNodes)
for i := 0; i < numNodes; i++ {
idx[i] = MustOpenIndex()
hldr[i] = MustOpenHolder()
server[i] = NewServer()
server[i].Handler.Host = server[i].Host()
server[i].Handler.Cluster = c
server[i].Handler.Cluster.Nodes[i].Host = server[i].Host()
server[i].Handler.Index = idx[i].Index
server[i].Handler.Holder = hldr[i].Holder
}
return server, idx
return server, hldr
}
// Test distributed TopN Row count across 3 nodes.
func TestClient_MultiNode(t *testing.T) {
cluster := NewCluster(3)
s, idx := createCluster(cluster)
s, hldr := createCluster(cluster)
for i := 0; i < len(cluster.Nodes); i++ {
defer idx[i].Close()
defer hldr[i].Close()
defer s[i].Close()
}
s[0].Handler.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor()
e.Index = idx[0].Index
e.Holder = hldr[0].Holder
e.Host = cluster.Nodes[0].Host
e.Cluster = cluster
return e.Execute(ctx, db, query, slices, opt)
}
s[1].Handler.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor()
e.Index = idx[1].Index
e.Holder = hldr[1].Holder
e.Host = cluster.Nodes[1].Host
e.Cluster = cluster
return e.Execute(ctx, db, query, slices, opt)
}
s[2].Handler.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor()
e.Index = idx[2].Index
e.Holder = hldr[2].Holder
e.Host = cluster.Nodes[2].Host
e.Cluster = cluster
return e.Execute(ctx, db, query, slices, opt)
}
// Create a dispersed set of bitmaps across 3 nodes such that each individual node and slice width increment would reveal a different TopN.
idx[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(99, 1, 2, 3, 4)
idx[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(100, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
idx[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(98, 1, 2, 3, 4, 5, 6)
idx[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(1, 4)
idx[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(22, 1, 2, 3, 4, 5)
hldr[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(99, 1, 2, 3, 4)
hldr[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(100, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
hldr[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(98, 1, 2, 3, 4, 5, 6)
hldr[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(1, 4)
hldr[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).MustSetBits(22, 1, 2, 3, 4, 5)
idx[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(100, (SliceWidth*10)+10)
idx[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(4, (SliceWidth*10)+10, (SliceWidth*10)+11, (SliceWidth*10)+12)
idx[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(4, (SliceWidth*10)+10, (SliceWidth*10)+11, (SliceWidth*10)+12, (SliceWidth*10)+13, (SliceWidth*10)+14, (SliceWidth*10)+15)
idx[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(2, (SliceWidth*10)+1, (SliceWidth*10)+2, (SliceWidth*10)+3, (SliceWidth*10)+4)
idx[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(3, (SliceWidth*10)+1, (SliceWidth*10)+2, (SliceWidth*10)+3, (SliceWidth*10)+4, (SliceWidth*10)+5)
idx[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(22, (SliceWidth*10)+1, (SliceWidth*10)+2, (SliceWidth*10)+10)
hldr[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(100, (SliceWidth*10)+10)
hldr[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(4, (SliceWidth*10)+10, (SliceWidth*10)+11, (SliceWidth*10)+12)
hldr[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(4, (SliceWidth*10)+10, (SliceWidth*10)+11, (SliceWidth*10)+12, (SliceWidth*10)+13, (SliceWidth*10)+14, (SliceWidth*10)+15)
hldr[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(2, (SliceWidth*10)+1, (SliceWidth*10)+2, (SliceWidth*10)+3, (SliceWidth*10)+4)
hldr[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(3, (SliceWidth*10)+1, (SliceWidth*10)+2, (SliceWidth*10)+3, (SliceWidth*10)+4, (SliceWidth*10)+5)
hldr[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).MustSetBits(22, (SliceWidth*10)+1, (SliceWidth*10)+2, (SliceWidth*10)+10)
idx[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(24, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12, (SliceWidth*6)+13, (SliceWidth*6)+14)
idx[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(20, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12, (SliceWidth*6)+13)
idx[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(21, (SliceWidth*6)+10)
idx[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(100, (SliceWidth*6)+10)
idx[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(99, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12)
idx[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(98, (SliceWidth*6)+10, (SliceWidth*6)+11)
idx[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(22, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12)
hldr[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(24, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12, (SliceWidth*6)+13, (SliceWidth*6)+14)
hldr[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(20, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12, (SliceWidth*6)+13)
hldr[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(21, (SliceWidth*6)+10)
hldr[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(100, (SliceWidth*6)+10)
hldr[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(99, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12)
hldr[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(98, (SliceWidth*6)+10, (SliceWidth*6)+11)
hldr[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).MustSetBits(22, (SliceWidth*6)+10, (SliceWidth*6)+11, (SliceWidth*6)+12)
// Rebuild the RankCache.
// We have to do this to avoid the 10-second cache invalidation delay
// built into cache.Invalidate()
idx[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).RecalculateCache()
idx[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).RecalculateCache()
idx[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).RecalculateCache()
hldr[0].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 0).RecalculateCache()
hldr[1].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 10).RecalculateCache()
hldr[2].MustCreateFragmentIfNotExists("d", "f.n", pilosa.ViewStandard, 6).RecalculateCache()
// Connect to each node to compare results.
client := make([]*Client, 3)
@ -112,9 +112,9 @@ func TestClient_MultiNode(t *testing.T) {
}
// Set max slice to correct value.
idx[0].DB("d").SetRemoteMaxSlice(10)
idx[1].DB("d").SetRemoteMaxSlice(10)
idx[2].DB("d").SetRemoteMaxSlice(10)
hldr[0].DB("d").SetRemoteMaxSlice(10)
hldr[1].DB("d").SetRemoteMaxSlice(10)
hldr[2].DB("d").SetRemoteMaxSlice(10)
result, err = client[0].ExecuteQuery(context.Background(), "d", q, true)
if err != nil {
@ -157,11 +157,11 @@ func TestClient_MultiNode(t *testing.T) {
// Ensure client can bulk import data.
func TestClient_Import(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Load bitmap into cache to ensure cache gets updated.
f := idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
f := hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
f.Row(0)
s := NewServer()
@ -169,7 +169,7 @@ func TestClient_Import(t *testing.T) {
s.Handler.Host = s.Host()
s.Handler.Cluster = NewCluster(1)
s.Handler.Cluster.Nodes[0].Host = s.Host()
s.Handler.Index = idx.Index
s.Handler.Holder = hldr.Holder
// Send import request.
c := MustNewClient(s.Host())
@ -192,10 +192,10 @@ func TestClient_Import(t *testing.T) {
// Ensure client can bulk import data to an inverse frame.
func TestClient_ImportInverseEnabled(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
d := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
d := hldr.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
frameOpts := pilosa.FrameOptions{
InverseEnabled: true,
}
@ -220,7 +220,7 @@ func TestClient_ImportInverseEnabled(t *testing.T) {
s.Handler.Host = s.Host()
s.Handler.Cluster = NewCluster(1)
s.Handler.Cluster.Nodes[0].Host = s.Host()
s.Handler.Index = idx.Index
s.Handler.Holder = hldr.Holder
// Send import request.
c := MustNewClient(s.Host())
@ -247,20 +247,20 @@ func TestClient_ImportInverseEnabled(t *testing.T) {
// Ensure client backup and restore a frame.
func TestClient_BackupRestore(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).MustSetBits(100, 1, 2, 3, SliceWidth-1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(100, SliceWidth, SliceWidth+2)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 5).MustSetBits(100, (5*SliceWidth)+1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).MustSetBits(200, 20000)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).MustSetBits(100, 1, 2, 3, SliceWidth-1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(100, SliceWidth, SliceWidth+2)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 5).MustSetBits(100, (5*SliceWidth)+1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).MustSetBits(200, 20000)
s := NewServer()
defer s.Close()
s.Handler.Host = s.Host()
s.Handler.Cluster = NewCluster(1)
s.Handler.Cluster.Nodes[0].Host = s.Host()
s.Handler.Index = idx.Index
s.Handler.Holder = hldr.Holder
c := MustNewClient(s.Host())
@ -271,7 +271,7 @@ func TestClient_BackupRestore(t *testing.T) {
}
// Restore to a different frame.
if _, err := idx.MustCreateDBIfNotExists("x", pilosa.DBOptions{}).CreateFrameIfNotExists("y", pilosa.FrameOptions{}); err != nil {
if _, err := hldr.MustCreateDBIfNotExists("x", pilosa.DBOptions{}).CreateFrameIfNotExists("y", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
}
if err := c.RestoreFrom(context.Background(), &buf, "x", "y", pilosa.ViewStandard); err != nil {
@ -279,38 +279,38 @@ func TestClient_BackupRestore(t *testing.T) {
}
// Verify data.
if a := idx.Fragment("x", "y", pilosa.ViewStandard, 0).Row(100).Bits(); !reflect.DeepEqual(a, []uint64{1, 2, 3, SliceWidth - 1}) {
if a := hldr.Fragment("x", "y", pilosa.ViewStandard, 0).Row(100).Bits(); !reflect.DeepEqual(a, []uint64{1, 2, 3, SliceWidth - 1}) {
t.Fatalf("unexpected bits(0): %+v", a)
}
if a := idx.Fragment("x", "y", pilosa.ViewStandard, 1).Row(100).Bits(); !reflect.DeepEqual(a, []uint64{SliceWidth, SliceWidth + 2}) {
if a := hldr.Fragment("x", "y", pilosa.ViewStandard, 1).Row(100).Bits(); !reflect.DeepEqual(a, []uint64{SliceWidth, SliceWidth + 2}) {
t.Fatalf("unexpected bits(0): %+v", a)
}
if a := idx.Fragment("x", "y", pilosa.ViewStandard, 5).Row(100).Bits(); !reflect.DeepEqual(a, []uint64{(5 * SliceWidth) + 1}) {
if a := hldr.Fragment("x", "y", pilosa.ViewStandard, 5).Row(100).Bits(); !reflect.DeepEqual(a, []uint64{(5 * SliceWidth) + 1}) {
t.Fatalf("unexpected bits(0): %+v", a)
}
if a := idx.Fragment("x", "y", pilosa.ViewStandard, 0).Row(200).Bits(); !reflect.DeepEqual(a, []uint64{20000}) {
if a := hldr.Fragment("x", "y", pilosa.ViewStandard, 0).Row(200).Bits(); !reflect.DeepEqual(a, []uint64{20000}) {
t.Fatalf("unexpected bits: %+v", a)
}
}
// Ensure client can retrieve a list of all checksums for blocks in a fragment.
func TestClient_FragmentBlocks(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Set two bits on blocks 0 & 3.
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(pilosa.HashBlockSize*3, 100)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(pilosa.HashBlockSize*3, 100)
// Set a bit on a different slice.
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, 1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, 1)
s := NewServer()
defer s.Close()
s.Handler.Host = s.Host()
s.Handler.Cluster = NewCluster(1)
s.Handler.Cluster.Nodes[0].Host = s.Host()
s.Handler.Index = idx.Index
s.Handler.Holder = hldr.Holder
// Retrieve blocks.
c := MustNewClient(s.Host())
@ -326,7 +326,7 @@ func TestClient_FragmentBlocks(t *testing.T) {
}
// Verify data matches local blocks.
if a := idx.Fragment("d", "f", pilosa.ViewStandard, 0).Blocks(); !reflect.DeepEqual(a, blocks) {
if a := hldr.Fragment("d", "f", pilosa.ViewStandard, 0).Blocks(); !reflect.DeepEqual(a, blocks) {
t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks))
}
}

30
executor.go
View file

@ -27,7 +27,7 @@ const (
// Executor recursively executes calls in a PQL query across all slices.
type Executor struct {
Index *Index
Holder *Holder
// Local hostname & cluster configuration.
Host string
@ -60,7 +60,7 @@ func (e *Executor) Execute(ctx context.Context, db string, q *pql.Query, slices
if len(slices) == 0 {
if needsSlices(q.Calls) {
// Round up the number of slices.
maxSlice := e.Index.DB(db).MaxSlice()
maxSlice := e.Holder.DB(db).MaxSlice()
// Generate a slices of all slices.
slices = make([]uint64, maxSlice+1)
@ -160,7 +160,7 @@ func (e *Executor) executeBitmapCall(ctx context.Context, db string, c *pql.Call
bm, _ := other.(*Bitmap)
if c.Name == "Bitmap" {
d := e.Index.DB(db)
d := e.Holder.DB(db)
if d != nil {
columnLabel := d.ColumnLabel()
if columnID, ok, err := c.UintArg(columnLabel); ok && err == nil {
@ -315,7 +315,7 @@ func (e *Executor) executeTopNSlice(ctx context.Context, db string, c *pql.Call,
frame = DefaultFrame
}
f := e.Index.Fragment(db, frame, ViewStandard, slice)
f := e.Holder.Fragment(db, frame, ViewStandard, slice)
if f == nil {
return nil, nil
}
@ -362,7 +362,7 @@ func (e *Executor) executeDifferenceSlice(ctx context.Context, db string, c *pql
func (e *Executor) executeBitmapSlice(ctx context.Context, db string, c *pql.Call, slice uint64) (*Bitmap, error) {
// Fetch column label from database.
d := e.Index.DB(db)
d := e.Holder.DB(db)
if d == nil {
return nil, ErrDatabaseNotFound
}
@ -373,7 +373,7 @@ func (e *Executor) executeBitmapSlice(ctx context.Context, db string, c *pql.Cal
if frame == "" {
frame = DefaultFrame
}
f := e.Index.Frame(db, frame)
f := e.Holder.Frame(db, frame)
if f == nil {
return nil, ErrFrameNotFound
}
@ -400,7 +400,7 @@ func (e *Executor) executeBitmapSlice(ctx context.Context, db string, c *pql.Cal
}
}
frag := e.Index.Fragment(db, frame, view, slice)
frag := e.Holder.Fragment(db, frame, view, slice)
if frag == nil {
return NewBitmap(), nil
}
@ -438,7 +438,7 @@ func (e *Executor) executeRangeSlice(ctx context.Context, db string, c *pql.Call
}
// Retrieve base frame.
f := e.Index.Frame(db, frame)
f := e.Holder.Frame(db, frame)
if f == nil {
return nil, ErrFrameNotFound
}
@ -479,7 +479,7 @@ func (e *Executor) executeRangeSlice(ctx context.Context, db string, c *pql.Call
// Union bitmaps across all time-based subframes.
bm := &Bitmap{}
for _, view := range ViewsByTimeRange(ViewStandard, startTime, endTime, q) {
f := e.Index.Fragment(db, frame, view, slice)
f := e.Holder.Fragment(db, frame, view, slice)
if f == nil {
continue
}
@ -548,7 +548,7 @@ func (e *Executor) executeClearBit(ctx context.Context, db string, c *pql.Call,
}
// Retrieve frame.
d := e.Index.DB(db)
d := e.Holder.DB(db)
if d == nil {
return false, ErrDatabaseNotFound
}
@ -642,7 +642,7 @@ func (e *Executor) executeSetBit(ctx context.Context, db string, c *pql.Call, op
}
// Retrieve frame.
d := e.Index.DB(db)
d := e.Holder.DB(db)
if d == nil {
return false, ErrDatabaseNotFound
}
@ -747,7 +747,7 @@ func (e *Executor) executeSetRowAttrs(ctx context.Context, db string, c *pql.Cal
}
// Retrieve frame.
frame := e.Index.Frame(db, frameName)
frame := e.Holder.Frame(db, frameName)
if frame == nil {
return ErrFrameNotFound
}
@ -807,7 +807,7 @@ func (e *Executor) executeBulkSetRowAttrs(ctx context.Context, db string, calls
}
// Retrieve frame.
f := e.Index.Frame(db, frame)
f := e.Holder.Frame(db, frame)
if f == nil {
return nil, ErrFrameNotFound
}
@ -846,7 +846,7 @@ func (e *Executor) executeBulkSetRowAttrs(ctx context.Context, db string, calls
// Bulk insert attributes by frame.
for name, frameMap := range m {
// Retrieve frame.
frame := e.Index.Frame(db, name)
frame := e.Holder.Frame(db, name)
if frame == nil {
return nil, ErrFrameNotFound
}
@ -886,7 +886,7 @@ func (e *Executor) executeBulkSetRowAttrs(ctx context.Context, db string, calls
// executeSetColumnAttrs executes a SetColumnAttrs() call.
func (e *Executor) executeSetColumnAttrs(ctx context.Context, db string, c *pql.Call, opt *ExecOptions) error {
// Retrieve database.
d := e.Index.DB(db)
d := e.Holder.DB(db)
if d == nil {
return ErrDatabaseNotFound
}

312
executor_test.go
View file

@ -15,15 +15,15 @@ import (
// Ensure a bitmap query can be executed.
func TestExecutor_Execute_Bitmap(t *testing.T) {
t.Run("Row", func(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
db := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
hldr := MustOpenHolder()
defer hldr.Close()
db := hldr.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
f, err := db.CreateFrame("f", pilosa.FrameOptions{InverseEnabled: true})
if err != nil {
t.Fatal(err)
}
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
// Set bits.
if _, err := e.Execute(context.Background(), "d", MustParse(``+
@ -47,14 +47,14 @@ func TestExecutor_Execute_Bitmap(t *testing.T) {
})
t.Run("Column", func(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
db := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
hldr := MustOpenHolder()
defer hldr.Close()
db := hldr.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
if _, err := db.CreateFrame("f", pilosa.FrameOptions{InverseEnabled: true}); err != nil {
t.Fatal(err)
}
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
// Set bits.
if _, err := e.Execute(context.Background(), "d", MustParse(``+
@ -80,15 +80,15 @@ func TestExecutor_Execute_Bitmap(t *testing.T) {
// Ensure a difference query can be executed.
func TestExecutor_Execute_Difference(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 1)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 2)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 3)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 2)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 4)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 1)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 2)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 3)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 2)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 4)
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if res, err := e.Execute(context.Background(), "d", MustParse(`Difference(Bitmap(id=10), Bitmap(id=11))`), nil, nil); err != nil {
t.Fatal(err)
} else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{1, 3}) {
@ -98,11 +98,11 @@ func TestExecutor_Execute_Difference(t *testing.T) {
// Ensure an empty difference query behaves properly.
func TestExecutor_Execute_Empty_Difference(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 1)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 1)
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if res, err := e.Execute(context.Background(), "d", MustParse(`Difference()`), nil, nil); err == nil {
t.Fatalf("Empty Difference query should give error, but got %v", res)
}
@ -110,17 +110,17 @@ func TestExecutor_Execute_Empty_Difference(t *testing.T) {
// Ensure an intersect query can be executed.
func TestExecutor_Execute_Intersect(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 1)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+2)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 1)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+1)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+2)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 1)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 2)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(11, SliceWidth+2)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 1)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 2)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(11, SliceWidth+2)
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if res, err := e.Execute(context.Background(), "d", MustParse(`Intersect(Bitmap(id=10), Bitmap(id=11))`), nil, nil); err != nil {
t.Fatal(err)
} else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{1, SliceWidth + 2}) {
@ -130,10 +130,10 @@ func TestExecutor_Execute_Intersect(t *testing.T) {
// Ensure an empty intersect query behaves properly.
func TestExecutor_Execute_Empty_Intersect(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if res, err := e.Execute(context.Background(), "d", MustParse(`Intersect()`), nil, nil); err == nil {
t.Fatalf("Empty Intersect query should give error, but got %v", res)
}
@ -141,16 +141,16 @@ func TestExecutor_Execute_Empty_Intersect(t *testing.T) {
// Ensure a union query can be executed.
func TestExecutor_Execute_Union(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 0)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+2)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 0)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+1)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+2)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 2)
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(11, SliceWidth+2)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(11, 2)
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 1).MustSetBits(11, SliceWidth+2)
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if res, err := e.Execute(context.Background(), "d", MustParse(`Union(Bitmap(id=10), Bitmap(id=11))`), nil, nil); err != nil {
t.Fatal(err)
} else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{0, 2, SliceWidth + 1, SliceWidth + 2}) {
@ -160,11 +160,11 @@ func TestExecutor_Execute_Union(t *testing.T) {
// Ensure an empty union query behaves properly.
func TestExecutor_Execute_Empty_Union(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 0)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "general", pilosa.ViewStandard, 0).MustSetBits(10, 0)
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if res, err := e.Execute(context.Background(), "d", MustParse(`Union()`), nil, nil); err != nil {
t.Fatal(err)
} else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{}) {
@ -174,13 +174,13 @@ func TestExecutor_Execute_Empty_Union(t *testing.T) {
// Ensure a count query can be executed.
func TestExecutor_Execute_Count(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).MustSetBits(10, 3)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+2)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).MustSetBits(10, 3)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, SliceWidth+2)
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if res, err := e.Execute(context.Background(), "d", MustParse(`Count(Bitmap(id=10, frame=f))`), nil, nil); err != nil {
t.Fatal(err)
} else if res[0] != uint64(3) {
@ -190,11 +190,11 @@ func TestExecutor_Execute_Count(t *testing.T) {
// Ensure a set query can be executed.
func TestExecutor_Execute_SetBit(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
e := NewExecutor(idx.Index, NewCluster(1))
f := idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
e := NewExecutor(hldr.Holder, NewCluster(1))
f := hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
if n := f.Row(11).Count(); n != 0 {
t.Fatalf("unexpected bitmap count: %d", n)
}
@ -221,11 +221,11 @@ func TestExecutor_Execute_SetBit(t *testing.T) {
// Ensure a SetRowAttrs() query can be executed.
func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Create frames.
db := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
db := hldr.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
if _, err := db.CreateFrameIfNotExists("f", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
} else if _, err := db.CreateFrameIfNotExists("xxx", pilosa.FrameOptions{}); err != nil {
@ -234,7 +234,7 @@ func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
// Set two fields on f/10.
// Also set fields on other bitmaps and frames to test isolation.
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if _, err := e.Execute(context.Background(), "d", MustParse(`SetRowAttrs(id=10, frame=f, foo="bar")`), nil, nil); err != nil {
t.Fatal(err)
}
@ -248,7 +248,7 @@ func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
t.Fatal(err)
}
f := idx.Frame("d", "f")
f := hldr.Frame("d", "f")
if m, err := f.RowAttrStore().Attrs(10); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(m, map[string]interface{}{"foo": "bar", "baz": int64(123), "bat": true}) {
@ -258,22 +258,22 @@ func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
// Ensure a TopN() query can be executed.
func TestExecutor_Execute_TopN(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Set bits for rows 0, 10, & 20 across two slices.
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth+2)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 5).SetBit(0, (5*SliceWidth)+100)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 0)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(20, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "other", pilosa.ViewStandard, 0).SetBit(0, 0)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth+2)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 5).SetBit(0, (5*SliceWidth)+100)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 0)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(20, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "other", pilosa.ViewStandard, 0).SetBit(0, 0)
// Execute query.
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame=f, n=2)`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result[0], []pilosa.Pair{
@ -284,19 +284,19 @@ func TestExecutor_Execute_TopN(t *testing.T) {
}
}
func TestExecutor_Execute_TopN_fill(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Set bits for rows 0, 10, & 20 across two slices.
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 2)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(1, SliceWidth+2)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(1, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 2)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(1, SliceWidth+2)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(1, SliceWidth)
// Execute query.
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame=f, n=1)`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -308,29 +308,29 @@ func TestExecutor_Execute_TopN_fill(t *testing.T) {
// Ensure
func TestExecutor_Execute_TopN_fill_small(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 2).SetBit(0, 2*SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 3).SetBit(0, 3*SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 4).SetBit(0, 4*SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 2).SetBit(0, 2*SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 3).SetBit(0, 3*SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 4).SetBit(0, 4*SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(1, 0)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(1, 1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(1, 0)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(1, 1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(2, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(2, SliceWidth+1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(2, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(2, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 2).SetBit(3, 2*SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 2).SetBit(3, 2*SliceWidth+1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 2).SetBit(3, 2*SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 2).SetBit(3, 2*SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 3).SetBit(4, 3*SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 3).SetBit(4, 3*SliceWidth+1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 3).SetBit(4, 3*SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 3).SetBit(4, 3*SliceWidth+1)
// Execute query.
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame=f, n=1)`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -342,26 +342,26 @@ func TestExecutor_Execute_TopN_fill_small(t *testing.T) {
// Ensure a TopN() query with a source bitmap can be executed.
func TestExecutor_Execute_TopN_Src(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Set bits for rows 0, 10, & 20 across two slices.
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(20, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(20, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(20, SliceWidth+2)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth+1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(20, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(20, SliceWidth+1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(20, SliceWidth+2)
// Create an intersecting row.
idx.MustCreateFragmentIfNotExists("d", "other", pilosa.ViewStandard, 1).SetBit(100, SliceWidth)
idx.MustCreateFragmentIfNotExists("d", "other", pilosa.ViewStandard, 1).SetBit(100, SliceWidth+1)
idx.MustCreateFragmentIfNotExists("d", "other", pilosa.ViewStandard, 1).SetBit(100, SliceWidth+2)
hldr.MustCreateFragmentIfNotExists("d", "other", pilosa.ViewStandard, 1).SetBit(100, SliceWidth)
hldr.MustCreateFragmentIfNotExists("d", "other", pilosa.ViewStandard, 1).SetBit(100, SliceWidth+1)
hldr.MustCreateFragmentIfNotExists("d", "other", pilosa.ViewStandard, 1).SetBit(100, SliceWidth+2)
// Execute query.
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(Bitmap(id=100, frame=other), frame=f, n=3)`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -376,16 +376,16 @@ func TestExecutor_Execute_TopN_Src(t *testing.T) {
//Ensure TopN handles Attribute filters
func TestExecutor_Execute_TopN_Attr(t *testing.T) {
//
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth)
if err := idx.Frame("d", "f").RowAttrStore().SetAttrs(10, map[string]interface{}{"category": int64(123)}); err != nil {
if err := hldr.Frame("d", "f").RowAttrStore().SetAttrs(10, map[string]interface{}{"category": int64(123)}); err != nil {
t.Fatal(err)
}
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame="f", n=1, field="category", filters=[123])`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -399,16 +399,16 @@ func TestExecutor_Execute_TopN_Attr(t *testing.T) {
//Ensure TopN handles Attribute filters with source bitmap
func TestExecutor_Execute_TopN_Attr_Src(t *testing.T) {
//
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(10, SliceWidth)
if err := idx.Frame("d", "f").RowAttrStore().SetAttrs(10, map[string]interface{}{"category": uint64(123)}); err != nil {
if err := hldr.Frame("d", "f").RowAttrStore().SetAttrs(10, map[string]interface{}{"category": uint64(123)}); err != nil {
t.Fatal(err)
}
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(Bitmap(id=10,frame=f),frame="f", n=1, field="category", filters=[123])`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -421,11 +421,11 @@ func TestExecutor_Execute_TopN_Attr_Src(t *testing.T) {
// Ensure a range query can be executed.
func TestExecutor_Execute_Range(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Create database.
db := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
db := hldr.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
// Create frame.
f, err := db.CreateFrameIfNotExists("f", pilosa.FrameOptions{})
@ -447,7 +447,7 @@ func TestExecutor_Execute_Range(t *testing.T) {
f.MustSetBit(pilosa.ViewStandard, 1, 2, MustParseTimePtr("2002-02-01 00:00")) // too late
f.MustSetBit(pilosa.ViewStandard, 10, 2, MustParseTimePtr("2001-01-01 00:00")) // different row
e := NewExecutor(idx.Index, NewCluster(1))
e := NewExecutor(hldr.Holder, NewCluster(1))
if res, err := e.Execute(context.Background(), "d", MustParse(`Range(id=1, frame=f, start="1999-12-31T00:00", end="2002-01-01T03:00")`), nil, nil); err != nil {
t.Fatal(err)
} else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{2, 3, 4, 5, 6, 7}) {
@ -487,11 +487,11 @@ func TestExecutor_Execute_Remote_Bitmap(t *testing.T) {
// Create local executor data.
// The local node owns slice 1.
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, (1*SliceWidth)+1)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, (1*SliceWidth)+1)
e := NewExecutor(idx.Index, c)
e := NewExecutor(hldr.Holder, c)
if res, err := e.Execute(context.Background(), "d", MustParse(`Bitmap(id=10, frame=f)`), nil, nil); err != nil {
t.Fatal(err)
} else if bits := res[0].(*pilosa.Bitmap).Bits(); !reflect.DeepEqual(bits, []uint64{1, 2, (1 * SliceWidth) + 1, 2*SliceWidth + 4}) {
@ -514,12 +514,12 @@ func TestExecutor_Execute_Remote_Count(t *testing.T) {
}
// Create local executor data. The local node owns slice 1.
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, (1*SliceWidth)+1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, (1*SliceWidth)+2)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, (1*SliceWidth)+1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(10, (1*SliceWidth)+2)
e := NewExecutor(idx.Index, c)
e := NewExecutor(hldr.Holder, c)
if res, err := e.Execute(context.Background(), "d", MustParse(`Count(Bitmap(id=10, frame=f))`), nil, nil); err != nil {
t.Fatal(err)
} else if res[0] != uint64(12) {
@ -550,21 +550,21 @@ func TestExecutor_Execute_Remote_SetBit(t *testing.T) {
}
// Create local executor data.
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Create frame.
if _, err := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{}).CreateFrame("f", pilosa.FrameOptions{}); err != nil {
if _, err := hldr.MustCreateDBIfNotExists("d", pilosa.DBOptions{}).CreateFrame("f", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
}
e := NewExecutor(idx.Index, c)
e := NewExecutor(hldr.Holder, c)
if _, err := e.Execute(context.Background(), "d", MustParse(`SetBit(id=10, frame=f, columnID=2)`), nil, nil); err != nil {
t.Fatal(err)
}
// Verify that one bit is set on both node's index.
if n := idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).Row(10).Count(); n != 1 {
// Verify that one bit is set on both node's holder.
if n := hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).Row(10).Count(); n != 1 {
t.Fatalf("unexpected local count: %d", n)
}
if !remoteCalled {
@ -595,23 +595,23 @@ func TestExecutor_Execute_Remote_SetBit_With_Timestamp(t *testing.T) {
}
// Create local executor data.
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Create frame.
if f, err := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{}).CreateFrame("f", pilosa.FrameOptions{}); err != nil {
if f, err := hldr.MustCreateDBIfNotExists("d", pilosa.DBOptions{}).CreateFrame("f", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
} else if err := f.SetTimeQuantum("Y"); err != nil {
t.Fatal(err)
}
e := NewExecutor(idx.Index, c)
e := NewExecutor(hldr.Holder, c)
if _, err := e.Execute(context.Background(), "d", MustParse(`SetBit(id=10, frame=f, columnID=2, timestamp="2016-12-11T10:09")`), nil, nil); err != nil {
t.Fatal(err)
}
// Verify that one bit is set on both node's index.
if n := idx.MustCreateFragmentIfNotExists("d", "f", "standard_2016", 0).Row(10).Count(); n != 1 {
// Verify that one bit is set on both node's holder.
if n := hldr.MustCreateFragmentIfNotExists("d", "f", "standard_2016", 0).Row(10).Count(); n != 1 {
t.Fatalf("unexpected local count: %d", n)
}
if !remoteCalled {
@ -662,12 +662,12 @@ func TestExecutor_Execute_Remote_TopN(t *testing.T) {
}
// Create local executor data on slice 1 & 3.
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(30, (1*SliceWidth)+1)
idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 3).MustSetBits(30, (3*SliceWidth)+2)
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).MustSetBits(30, (1*SliceWidth)+1)
hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 3).MustSetBits(30, (3*SliceWidth)+2)
e := NewExecutor(idx.Index, c)
e := NewExecutor(hldr.Holder, c)
if res, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame=f, n=3)`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(res, []interface{}{[]pilosa.Pair{
@ -686,9 +686,9 @@ type Executor struct {
// NewExecutor returns a new instance of Executor.
// The executor always matches the hostname of the first cluster node.
func NewExecutor(index *pilosa.Index, cluster *pilosa.Cluster) *Executor {
func NewExecutor(holder *pilosa.Holder, cluster *pilosa.Cluster) *Executor {
e := &Executor{Executor: pilosa.NewExecutor()}
e.Index = index
e.Holder = holder
e.Cluster = cluster
e.Host = cluster.Nodes[0].Host
return e

59
handler.go
View file

@ -17,16 +17,17 @@ import (
"strings"
"time"
"reflect"
"github.com/gogo/protobuf/proto"
"github.com/gorilla/mux"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"reflect"
)
// Handler represents an HTTP handler.
type Handler struct {
Index *Index
Holder *Holder
Broadcaster Broadcaster
// Local hostname & cluster configuration.
@ -107,7 +108,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// handleGetSchema handles GET /schema requests.
func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
if err := json.NewEncoder(w).Encode(getSchemaResponse{
DBs: h.Index.Schema(),
DBs: h.Holder.Schema(),
}); err != nil {
h.logger().Printf("write schema response error: %s", err)
}
@ -172,7 +173,7 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
}
// Retrieve column attributes across all calls.
columnAttrSets, err := h.readColumnAttrSets(h.Index.DB(dbName), columnIDs)
columnAttrSets, err := h.readColumnAttrSets(h.Holder.DB(dbName), columnIDs)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
h.writeQueryResponse(w, r, &QueryResponse{Err: err})
@ -195,9 +196,9 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) {
func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) {
var ms map[string]uint64
if inverse, _ := strconv.ParseBool(r.URL.Query().Get("inverse")); inverse {
ms = h.Index.MaxInverseSlices()
ms = h.Holder.MaxInverseSlices()
} else {
ms = h.Index.MaxSlices()
ms = h.Holder.MaxSlices()
}
if strings.Contains(r.Header.Get("Accept"), "application/x-protobuf") {
pb := &internal.MaxSlicesResponse{
@ -226,7 +227,7 @@ func (h *Handler) handleGetDBs(w http.ResponseWriter, r *http.Request) {
// handleGetDB handles GET /db/<dbname> requests.
func (h *Handler) handleGetDB(w http.ResponseWriter, r *http.Request) {
dbName := mux.Vars(r)["db"]
db := h.Index.DB(dbName)
db := h.Holder.DB(dbName)
if db == nil {
http.Error(w, ErrDatabaseNotFound.Error(), http.StatusNotFound)
return
@ -311,8 +312,8 @@ type postDBResponse struct{}
func (h *Handler) handleDeleteDB(w http.ResponseWriter, r *http.Request) {
dbName := mux.Vars(r)["db"]
// Delete database from the index.
if err := h.Index.DeleteDB(dbName); err != nil {
// Delete database from the holder.
if err := h.Holder.DeleteDB(dbName); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
@ -350,7 +351,7 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) {
}
// Create database.
_, err = h.Index.CreateDB(dbName, req.Options)
_, err = h.Holder.CreateDB(dbName, req.Options)
if err == ErrDatabaseExists {
http.Error(w, err.Error(), http.StatusConflict)
return
@ -394,7 +395,7 @@ func (h *Handler) handlePatchDBTimeQuantum(w http.ResponseWriter, r *http.Reques
}
// Retrieve database by name.
database := h.Index.DB(dbName)
database := h.Holder.DB(dbName)
if database == nil {
http.Error(w, ErrDatabaseNotFound.Error(), http.StatusNotFound)
return
@ -429,8 +430,8 @@ func (h *Handler) handlePostDBAttrDiff(w http.ResponseWriter, r *http.Request) {
return
}
// Retrieve database from index.
db := h.Index.DB(dbName)
// Retrieve database from holder.
db := h.Holder.DB(dbName)
if db == nil {
http.Error(w, ErrDatabaseNotFound.Error(), http.StatusNotFound)
return
@ -492,7 +493,7 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) {
}
// Find database.
db := h.Index.DB(dbName)
db := h.Holder.DB(dbName)
if db == nil {
http.Error(w, ErrDatabaseNotFound.Error(), http.StatusNotFound)
return
@ -576,7 +577,7 @@ func (h *Handler) handleDeleteFrame(w http.ResponseWriter, r *http.Request) {
frameName := mux.Vars(r)["frame"]
// Find database.
db := h.Index.DB(dbName)
db := h.Holder.DB(dbName)
if db == nil {
if err := json.NewEncoder(w).Encode(deleteDBResponse{}); err != nil {
h.logger().Printf("response encoding error: %s", err)
@ -628,7 +629,7 @@ func (h *Handler) handlePatchFrameTimeQuantum(w http.ResponseWriter, r *http.Req
}
// Retrieve database by name.
f := h.Index.Frame(dbName, frameName)
f := h.Holder.Frame(dbName, frameName)
if f == nil {
http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound)
return
@ -658,7 +659,7 @@ func (h *Handler) handleGetFrameViews(w http.ResponseWriter, r *http.Request) {
frameName := mux.Vars(r)["frame"]
// Retrieve views.
f := h.Index.Frame(dbName, frameName)
f := h.Holder.Frame(dbName, frameName)
if f == nil {
http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound)
return
@ -693,8 +694,8 @@ func (h *Handler) handlePostFrameAttrDiff(w http.ResponseWriter, r *http.Request
return
}
// Retrieve database from index.
f := h.Index.Frame(dbName, frameName)
// Retrieve database from holder.
f := h.Holder.Frame(dbName, frameName)
if f == nil {
http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound)
return
@ -891,7 +892,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
// Find the DB.
h.logger().Println("importing:", req.DB, req.Frame, req.Slice)
db := h.Index.DB(req.DB)
db := h.Holder.DB(req.DB)
if db == nil {
h.logger().Printf("fragment error: db=%s, frame=%s, slice=%d, err=%s", req.DB, req.Frame, req.Slice, ErrDatabaseNotFound.Error())
http.Error(w, ErrDatabaseNotFound.Error(), http.StatusNotFound)
@ -956,7 +957,7 @@ func (h *Handler) handleGetExportCSV(w http.ResponseWriter, r *http.Request) {
}
// Find the fragment.
f := h.Index.Fragment(db, frame, view, slice)
f := h.Holder.Fragment(db, frame, view, slice)
if f == nil {
return
}
@ -1010,8 +1011,8 @@ func (h *Handler) handleGetFragmentData(w http.ResponseWriter, r *http.Request)
return
}
// Retrieve fragment from index.
f := h.Index.Fragment(q.Get("db"), q.Get("frame"), q.Get("view"), slice)
// Retrieve fragment from holder.
f := h.Holder.Fragment(q.Get("db"), q.Get("frame"), q.Get("view"), slice)
if f == nil {
http.Error(w, "fragment not found", http.StatusNotFound)
return
@ -1034,7 +1035,7 @@ func (h *Handler) handlePostFragmentData(w http.ResponseWriter, r *http.Request)
}
// Retrieve frame.
f := h.Index.Frame(q.Get("db"), q.Get("frame"))
f := h.Holder.Frame(q.Get("db"), q.Get("frame"))
if f == nil {
http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound)
return
@ -1073,8 +1074,8 @@ func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Requ
return
}
// Retrieve fragment from index.
f := h.Index.Fragment(req.DB, req.Frame, req.View, req.Slice)
// Retrieve fragment from holder.
f := h.Holder.Fragment(req.DB, req.Frame, req.View, req.Slice)
if f == nil {
http.Error(w, ErrFragmentNotFound.Error(), http.StatusNotFound)
return
@ -1109,8 +1110,8 @@ func (h *Handler) handleGetFragmentBlocks(w http.ResponseWriter, r *http.Request
return
}
// Retrieve fragment from index.
f := h.Index.Fragment(q.Get("db"), q.Get("frame"), q.Get("view"), slice)
// Retrieve fragment from holder.
f := h.Holder.Fragment(q.Get("db"), q.Get("frame"), q.Get("view"), slice)
if f == nil {
http.Error(w, "fragment not found", http.StatusNotFound)
return
@ -1160,7 +1161,7 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request)
}
// Retrieve frame.
f := h.Index.Frame(dbName, frameName)
f := h.Holder.Frame(dbName, frameName)
if f == nil {
http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound)
return

122
handler_test.go
View file

@ -31,11 +31,11 @@ func TestHandler_NotFound(t *testing.T) {
// Ensure the handler can return the schema.
func TestHandler_Schema(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
d0 := idx.MustCreateDBIfNotExists("d0", pilosa.DBOptions{})
d1 := idx.MustCreateDBIfNotExists("d1", pilosa.DBOptions{})
d0 := hldr.MustCreateDBIfNotExists("d0", pilosa.DBOptions{})
d1 := hldr.MustCreateDBIfNotExists("d1", pilosa.DBOptions{})
if f, err := d0.CreateFrameIfNotExists("f1", pilosa.FrameOptions{InverseEnabled: true}); err != nil {
t.Fatal(err)
@ -54,7 +54,7 @@ func TestHandler_Schema(t *testing.T) {
}
h := NewHandler()
h.Index = idx.Index
h.Holder = hldr.Holder
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("GET", "/schema", nil))
if w.Code != http.StatusOK {
@ -66,19 +66,19 @@ func TestHandler_Schema(t *testing.T) {
// Ensure the handler can return the maxslice map.
func TestHandler_MaxSlices(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
idx.MustCreateFragmentIfNotExists("d0", "f0", pilosa.ViewStandard, 1).MustSetBits(30, (1*SliceWidth)+1)
idx.MustCreateFragmentIfNotExists("d0", "f0", pilosa.ViewStandard, 1).MustSetBits(30, (1*SliceWidth)+2)
idx.MustCreateFragmentIfNotExists("d0", "f0", pilosa.ViewStandard, 3).MustSetBits(30, (3*SliceWidth)+4)
hldr.MustCreateFragmentIfNotExists("d0", "f0", pilosa.ViewStandard, 1).MustSetBits(30, (1*SliceWidth)+1)
hldr.MustCreateFragmentIfNotExists("d0", "f0", pilosa.ViewStandard, 1).MustSetBits(30, (1*SliceWidth)+2)
hldr.MustCreateFragmentIfNotExists("d0", "f0", pilosa.ViewStandard, 3).MustSetBits(30, (3*SliceWidth)+4)
idx.MustCreateFragmentIfNotExists("d1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*SliceWidth)+1)
idx.MustCreateFragmentIfNotExists("d1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*SliceWidth)+2)
idx.MustCreateFragmentIfNotExists("d1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*SliceWidth)+8)
hldr.MustCreateFragmentIfNotExists("d1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*SliceWidth)+1)
hldr.MustCreateFragmentIfNotExists("d1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*SliceWidth)+2)
hldr.MustCreateFragmentIfNotExists("d1", "f1", pilosa.ViewStandard, 0).MustSetBits(40, (0*SliceWidth)+8)
h := NewHandler()
h.Index = idx.Index
h.Holder = hldr.Holder
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("GET", "/slices/max", nil))
if w.Code != http.StatusOK {
@ -90,10 +90,10 @@ func TestHandler_MaxSlices(t *testing.T) {
// Ensure the handler can return the maxslice map for the inverse views.
func TestHandler_MaxSlices_Inverse(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
f0, err := idx.MustCreateDBIfNotExists("d0", pilosa.DBOptions{}).CreateFrame("f0", pilosa.FrameOptions{InverseEnabled: true})
f0, err := hldr.MustCreateDBIfNotExists("d0", pilosa.DBOptions{}).CreateFrame("f0", pilosa.FrameOptions{InverseEnabled: true})
if err != nil {
t.Fatal(err)
}
@ -105,7 +105,7 @@ func TestHandler_MaxSlices_Inverse(t *testing.T) {
t.Fatal(err)
}
f1, err := idx.MustCreateDBIfNotExists("d1", pilosa.DBOptions{}).CreateFrame("f1", pilosa.FrameOptions{InverseEnabled: true})
f1, err := hldr.MustCreateDBIfNotExists("d1", pilosa.DBOptions{}).CreateFrame("f1", pilosa.FrameOptions{InverseEnabled: true})
if err != nil {
t.Fatal(err)
}
@ -118,7 +118,7 @@ func TestHandler_MaxSlices_Inverse(t *testing.T) {
}
h := NewHandler()
h.Index = idx.Index
h.Holder = hldr.Holder
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("GET", "/slices/max?inverse=true", nil))
if w.Code != http.StatusOK {
@ -255,11 +255,11 @@ func TestHandler_Query_Bitmap_JSON(t *testing.T) {
// Ensure the handler can execute a query that returns a bitmap with column attributes as JSON.
func TestHandler_Query_Bitmap_ColumnAttrs_JSON(t *testing.T) {
idx := NewIndex()
defer idx.Close()
hldr := NewHolder()
defer hldr.Close()
// Create database and set column attributes.
db, err := idx.CreateDBIfNotExists("d", pilosa.DBOptions{})
db, err := hldr.CreateDBIfNotExists("d", pilosa.DBOptions{})
if err != nil {
t.Fatal(err)
} else if err := db.ColumnAttrStore().SetAttrs(3, map[string]interface{}{"x": "y"}); err != nil {
@ -269,7 +269,7 @@ func TestHandler_Query_Bitmap_ColumnAttrs_JSON(t *testing.T) {
}
h := NewHandler()
h.Index = idx.Index
h.Holder = hldr.Holder
h.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
bm := pilosa.NewBitmap(1, 3, 66, pilosa.SliceWidth+1)
bm.Attrs = map[string]interface{}{"a": "b", "c": 1, "d": true}
@ -320,11 +320,11 @@ func TestHandler_Query_Bitmap_Protobuf(t *testing.T) {
// Ensure the handler can execute a query that returns a bitmap with column attributes as protobuf.
func TestHandler_Query_Bitmap_ColumnAttrs_Protobuf(t *testing.T) {
idx := NewIndex()
defer idx.Close()
hldr := NewHolder()
defer hldr.Close()
// Create database and set column attributes.
db, err := idx.CreateDBIfNotExists("d", pilosa.DBOptions{})
db, err := hldr.CreateDBIfNotExists("d", pilosa.DBOptions{})
if err != nil {
t.Fatal(err)
} else if err := db.ColumnAttrStore().SetAttrs(1, map[string]interface{}{"x": "y"}); err != nil {
@ -332,7 +332,7 @@ func TestHandler_Query_Bitmap_ColumnAttrs_Protobuf(t *testing.T) {
}
h := NewHandler()
h.Index = idx.Index
h.Holder = hldr.Holder
h.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
bm := pilosa.NewBitmap(1, pilosa.SliceWidth+1)
bm.Attrs = map[string]interface{}{"a": "b", "c": int64(1), "d": true}
@ -491,15 +491,15 @@ func TestHandler_Query_ErrParse(t *testing.T) {
// Ensure the handler can delete a database.
func TestHandler_DB_Delete(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
s := NewServer()
s.Handler.Index = idx.Index
s.Handler.Holder = hldr.Holder
defer s.Close()
// Create database.
if _, err := idx.CreateDBIfNotExists("d", pilosa.DBOptions{}); err != nil {
if _, err := hldr.CreateDBIfNotExists("d", pilosa.DBOptions{}); err != nil {
t.Fatal(err)
}
@ -520,86 +520,86 @@ func TestHandler_DB_Delete(t *testing.T) {
}
// Verify database is gone.
if idx.DB("d") != nil {
if hldr.DB("d") != nil {
t.Fatal("expected nil database")
}
}
// Ensure handler can delete a frame.
func TestHandler_DeleteFrame(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
d0 := idx.MustCreateDBIfNotExists("d0", pilosa.DBOptions{})
hldr := MustOpenHolder()
defer hldr.Close()
d0 := hldr.MustCreateDBIfNotExists("d0", pilosa.DBOptions{})
if _, err := d0.CreateFrameIfNotExists("f1", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
}
h := NewHandler()
h.Index = idx.Index
h.Holder = hldr.Holder
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("DELETE", "/db/d0/frame/f1", strings.NewReader("")))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{}`+"\n" {
t.Fatalf("unexpected body: %s", body)
} else if f := idx.DB("d0").Frame("f1"); f != nil {
} else if f := hldr.DB("d0").Frame("f1"); f != nil {
t.Fatal("expected nil frame")
}
}
// Ensure handler can set the DB time quantum.
func TestHandler_SetDBTimeQuantum(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
idx.MustCreateDBIfNotExists("d0", pilosa.DBOptions{})
hldr := MustOpenHolder()
defer hldr.Close()
hldr.MustCreateDBIfNotExists("d0", pilosa.DBOptions{})
h := NewHandler()
h.Index = idx.Index
h.Holder = hldr.Holder
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/time-quantum", strings.NewReader(`{"timeQuantum":"ymdh"}`)))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{}`+"\n" {
t.Fatalf("unexpected body: %s", body)
} else if q := idx.DB("d0").TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
} else if q := hldr.DB("d0").TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
t.Fatalf("unexpected time quantum: %s", q)
}
}
// Ensure handler can set the frame time quantum.
func TestHandler_SetFrameTimeQuantum(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
// Create frame.
if _, err := idx.MustCreateDBIfNotExists("d0", pilosa.DBOptions{}).CreateFrame("f1", pilosa.FrameOptions{}); err != nil {
if _, err := hldr.MustCreateDBIfNotExists("d0", pilosa.DBOptions{}).CreateFrame("f1", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
}
h := NewHandler()
h.Index = idx.Index
h.Holder = hldr.Holder
w := httptest.NewRecorder()
h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/frame/f1/time-quantum", strings.NewReader(`{"timeQuantum":"ymdh"}`)))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{}`+"\n" {
t.Fatalf("unexpected body: %s", body)
} else if q := idx.DB("d0").Frame("f1").TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
} else if q := hldr.DB("d0").Frame("f1").TimeQuantum(); q != pilosa.TimeQuantum("YMDH") {
t.Fatalf("unexpected time quantum: %s", q)
}
}
// Ensure the handler can return data in differing blocks for a database.
func TestHandler_DB_AttrStore_Diff(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
s := NewServer()
s.Handler.Index = idx.Index
s.Handler.Holder = hldr.Holder
defer s.Close()
// Set attributes on the database.
db, err := idx.CreateDBIfNotExists("d", pilosa.DBOptions{})
db, err := hldr.CreateDBIfNotExists("d", pilosa.DBOptions{})
if err != nil {
t.Fatal(err)
}
@ -640,15 +640,15 @@ func TestHandler_DB_AttrStore_Diff(t *testing.T) {
// Ensure the handler can return data in differing blocks for a frame.
func TestHandler_Frame_AttrStore_Diff(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
s := NewServer()
s.Handler.Index = idx.Index
s.Handler.Holder = hldr.Holder
defer s.Close()
// Set attributes on the database.
d := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
d := hldr.MustCreateDBIfNotExists("d", pilosa.DBOptions{})
f, err := d.CreateFrameIfNotExists("meta", pilosa.FrameOptions{})
if err != nil {
t.Fatal(err)
@ -690,15 +690,15 @@ func TestHandler_Frame_AttrStore_Diff(t *testing.T) {
// Ensure the handler can backup a fragment and then restore it.
func TestHandler_Fragment_BackupRestore(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
hldr := MustOpenHolder()
defer hldr.Close()
s := NewServer()
s.Handler.Index = idx.Index
s.Handler.Holder = hldr.Holder
defer s.Close()
// Set bits in the index.
f0 := idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
f0 := hldr.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
f0.MustSetBits(100, 1, 2, 3)
// Begin backing up from slice d/f/0.
@ -714,7 +714,7 @@ func TestHandler_Fragment_BackupRestore(t *testing.T) {
}
// Create frame.
if _, err := idx.MustCreateDBIfNotExists("x", pilosa.DBOptions{}).CreateFrame("y", pilosa.FrameOptions{}); err != nil {
if _, err := hldr.MustCreateDBIfNotExists("x", pilosa.DBOptions{}).CreateFrame("y", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
}
@ -729,7 +729,7 @@ func TestHandler_Fragment_BackupRestore(t *testing.T) {
}
// Verify data is correctly restored.
f1 := idx.Fragment("x", "y", pilosa.ViewStandard, 0)
f1 := hldr.Fragment("x", "y", pilosa.ViewStandard, 0)
if f1 == nil {
t.Fatal("fragment x/y/standard/0 not created")
} else if bits := f1.Row(100).Bits(); !reflect.DeepEqual(bits, []uint64{1, 2, 3}) {

192
index.go → holder.go
View file

@ -16,8 +16,8 @@ import (
// DefaultCacheFlushInterval is the default value for Fragment.CacheFlushInterval.
const DefaultCacheFlushInterval = 1 * time.Minute
// Index represents a container for fragments.
type Index struct {
// Holder represents a container for indexes.
type Holder struct {
mu sync.Mutex
// Databases by name.
@ -40,9 +40,9 @@ type Index struct {
LogOutput io.Writer
}
// NewIndex returns a new instance of Index.
func NewIndex() *Index {
return &Index{
// NewHolder returns a new instance of Holder.
func NewHolder() *Holder {
return &Holder{
dbs: make(map[string]*DB),
closing: make(chan struct{}, 0),
@ -54,14 +54,14 @@ func NewIndex() *Index {
}
}
// Open initializes the root data directory for the index.
func (i *Index) Open() error {
if err := os.MkdirAll(i.Path, 0777); err != nil {
// Open initializes the root data directory for the holder.
func (h *Holder) Open() error {
if err := os.MkdirAll(h.Path, 0777); err != nil {
return err
}
// Open path to read all database directories.
f, err := os.Open(i.Path)
f, err := os.Open(h.Path)
if err != nil {
return err
}
@ -77,68 +77,68 @@ func (i *Index) Open() error {
continue
}
i.logger().Printf("opening database: %s", filepath.Base(fi.Name()))
h.logger().Printf("opening database: %s", filepath.Base(fi.Name()))
db, err := i.newDB(i.DBPath(filepath.Base(fi.Name())), filepath.Base(fi.Name()))
db, err := h.newDB(h.DBPath(filepath.Base(fi.Name())), filepath.Base(fi.Name()))
if err == ErrName {
i.logger().Printf("ERROR opening database: %s, err=%s", fi.Name(), err)
h.logger().Printf("ERROR opening database: %s, err=%s", fi.Name(), err)
continue
} else if err != nil {
return err
}
if err := db.Open(); err != nil {
if err == ErrName {
i.logger().Printf("ERROR opening database: %s, err=%s", db.Name(), err)
h.logger().Printf("ERROR opening database: %s, err=%s", db.Name(), err)
continue
}
return fmt.Errorf("open db: name=%s, err=%s", db.Name(), err)
}
i.dbs[db.Name()] = db
h.dbs[db.Name()] = db
i.Stats.Count("dbN", 1)
h.Stats.Count("dbN", 1)
}
// Periodically flush cache.
i.wg.Add(1)
go func() { defer i.wg.Done(); i.monitorCacheFlush() }()
h.wg.Add(1)
go func() { defer h.wg.Done(); h.monitorCacheFlush() }()
return nil
}
// Close closes all open fragments.
func (i *Index) Close() error {
func (h *Holder) Close() error {
// Notify goroutines of closing and wait for completion.
close(i.closing)
i.wg.Wait()
close(h.closing)
h.wg.Wait()
for _, db := range i.dbs {
for _, db := range h.dbs {
db.Close()
}
return nil
}
// MaxSlices returns MaxSlice map for all databases.
func (i *Index) MaxSlices() map[string]uint64 {
func (h *Holder) MaxSlices() map[string]uint64 {
a := make(map[string]uint64)
for _, db := range i.DBs() {
for _, db := range h.DBs() {
a[db.Name()] = db.MaxSlice()
}
return a
}
// MaxInverseSlices returns MaxInverseSlice map for all databases.
func (i *Index) MaxInverseSlices() map[string]uint64 {
func (h *Holder) MaxInverseSlices() map[string]uint64 {
a := make(map[string]uint64)
for _, db := range i.DBs() {
for _, db := range h.DBs() {
a[db.Name()] = db.MaxInverseSlice()
}
return a
}
// Schema returns schema data for all databases and frames.
func (i *Index) Schema() []*DBInfo {
func (h *Holder) Schema() []*DBInfo {
var a []*DBInfo
for _, db := range i.DBs() {
for _, db := range h.DBs() {
di := &DBInfo{Name: db.Name()}
for _, frame := range db.Frames() {
fi := &FrameInfo{Name: frame.Name()}
@ -156,24 +156,24 @@ func (i *Index) Schema() []*DBInfo {
}
// DBPath returns the path where a given database is stored.
func (i *Index) DBPath(name string) string { return filepath.Join(i.Path, name) }
func (h *Holder) DBPath(name string) string { return filepath.Join(h.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 (h *Holder) DB(name string) *DB {
h.mu.Lock()
defer h.mu.Unlock()
return h.db(name)
}
func (i *Index) db(name string) *DB { return i.dbs[name] }
func (h *Holder) db(name string) *DB { return h.dbs[name] }
// DBs returns a list of all databases in the index.
func (i *Index) DBs() []*DB {
i.mu.Lock()
defer i.mu.Unlock()
// DBs returns a list of all databases in the holder.
func (h *Holder) DBs() []*DB {
h.mu.Lock()
defer h.mu.Unlock()
a := make([]*DB, 0, len(i.dbs))
for _, db := range i.dbs {
a := make([]*DB, 0, len(h.dbs))
for _, db := range h.dbs {
a = append(a, db)
}
sort.Sort(dbSlice(a))
@ -183,43 +183,43 @@ func (i *Index) DBs() []*DB {
// CreateDB creates a database.
// An error is returned if the database already exists.
func (i *Index) CreateDB(name string, opt DBOptions) (*DB, error) {
i.mu.Lock()
defer i.mu.Unlock()
func (h *Holder) CreateDB(name string, opt DBOptions) (*DB, error) {
h.mu.Lock()
defer h.mu.Unlock()
// Ensure db doesn't already exist.
if i.dbs[name] != nil {
if h.dbs[name] != nil {
return nil, ErrDatabaseExists
}
return i.createDB(name, opt)
return h.createDB(name, opt)
}
// CreateDBIfNotExists returns a database by name.
// The database is created if it does not already exist.
func (i *Index) CreateDBIfNotExists(name string, opt DBOptions) (*DB, error) {
i.mu.Lock()
defer i.mu.Unlock()
func (h *Holder) CreateDBIfNotExists(name string, opt DBOptions) (*DB, error) {
h.mu.Lock()
defer h.mu.Unlock()
// Find database in cache first.
if db := i.dbs[name]; db != nil {
if db := h.dbs[name]; db != nil {
return db, nil
}
return i.createDB(name, opt)
return h.createDB(name, opt)
}
func (i *Index) createDB(name string, opt DBOptions) (*DB, error) {
func (h *Holder) createDB(name string, opt DBOptions) (*DB, error) {
if name == "" {
return nil, errors.New("database name required")
}
// Return database if it exists.
if db := i.db(name); db != nil {
if db := h.db(name); db != nil {
return db, nil
}
// Otherwise create a new database.
db, err := i.newDB(i.DBPath(name), name)
db, err := h.newDB(h.DBPath(name), name)
if err != nil {
return nil, err
}
@ -232,31 +232,31 @@ func (i *Index) createDB(name string, opt DBOptions) (*DB, error) {
db.SetColumnLabel(opt.ColumnLabel)
db.SetTimeQuantum(opt.TimeQuantum)
i.dbs[db.Name()] = db
h.dbs[db.Name()] = db
i.Stats.Count("dbN", 1)
h.Stats.Count("dbN", 1)
return db, nil
}
func (i *Index) newDB(path, name string) (*DB, error) {
func (h *Holder) newDB(path, name string) (*DB, error) {
db, err := NewDB(path, name)
if err != nil {
return nil, err
}
db.LogOutput = i.LogOutput
db.stats = i.Stats.WithTags(fmt.Sprintf("db:%s", db.Name()))
db.broadcaster = i.Broadcaster
db.LogOutput = h.LogOutput
db.stats = h.Stats.WithTags(fmt.Sprintf("db:%s", db.Name()))
db.broadcaster = h.Broadcaster
return db, nil
}
// DeleteDB removes a database from the index.
func (i *Index) DeleteDB(name string) error {
i.mu.Lock()
defer i.mu.Unlock()
// DeleteDB removes a database from the holder.
func (h *Holder) DeleteDB(name string) error {
h.mu.Lock()
defer h.mu.Unlock()
// Ignore if database doesn't exist.
db := i.db(name)
db := h.db(name)
if db == nil {
return nil
}
@ -267,21 +267,21 @@ func (i *Index) DeleteDB(name string) error {
}
// Delete database directory.
if err := os.RemoveAll(i.DBPath(name)); err != nil {
if err := os.RemoveAll(h.DBPath(name)); err != nil {
return err
}
// Remove reference.
delete(i.dbs, name)
delete(h.dbs, name)
i.Stats.Count("dbN", -1)
h.Stats.Count("dbN", -1)
return nil
}
// Frame returns the frame for a database and name.
func (i *Index) Frame(db, name string) *Frame {
d := i.DB(db)
func (h *Holder) Frame(db, name string) *Frame {
d := h.DB(db)
if d == nil {
return nil
}
@ -289,8 +289,8 @@ func (i *Index) Frame(db, name string) *Frame {
}
// View returns the view for a database, frame, and name.
func (i *Index) View(db, frame, name string) *View {
f := i.Frame(db, frame)
func (h *Holder) View(db, frame, name string) *View {
f := h.Frame(db, frame)
if f == nil {
return nil
}
@ -298,8 +298,8 @@ func (i *Index) View(db, frame, name string) *View {
}
// Fragment returns the fragment for a database, frame & slice.
func (i *Index) Fragment(db, frame, view string, slice uint64) *Fragment {
v := i.View(db, frame, view)
func (h *Holder) Fragment(db, frame, view string, slice uint64) *Fragment {
v := h.View(db, frame, view)
if v == nil {
return nil
}
@ -308,33 +308,33 @@ func (i *Index) Fragment(db, frame, view string, slice uint64) *Fragment {
// monitorCacheFlush periodically flushes all fragment caches sequentially.
// This is run in a goroutine.
func (i *Index) monitorCacheFlush() {
ticker := time.NewTicker(i.CacheFlushInterval)
func (h *Holder) monitorCacheFlush() {
ticker := time.NewTicker(h.CacheFlushInterval)
defer ticker.Stop()
for {
select {
case <-i.closing:
case <-h.closing:
return
case <-ticker.C:
i.flushCaches()
h.flushCaches()
}
}
}
func (i *Index) flushCaches() {
for _, db := range i.DBs() {
func (h *Holder) flushCaches() {
for _, db := range h.DBs() {
for _, frame := range db.Frames() {
for _, view := range frame.Views() {
for _, fragment := range view.Fragments() {
select {
case <-i.closing:
case <-h.closing:
return
default:
}
if err := fragment.FlushCache(); err != nil {
i.logger().Printf("error flushing cache: err=%s, path=%s", err, fragment.CachePath())
h.logger().Printf("error flushing cache: err=%s, path=%s", err, fragment.CachePath())
}
}
}
@ -342,12 +342,12 @@ func (i *Index) flushCaches() {
}
}
func (i *Index) logger() *log.Logger { return log.New(i.LogOutput, "", log.LstdFlags) }
func (h *Holder) logger() *log.Logger { return log.New(h.LogOutput, "", log.LstdFlags) }
// IndexSyncer is an active anti-entropy tool that compares the local index
// with a remote index based on block checksums and resolves differences.
type IndexSyncer struct {
Index *Index
// HolderSyncer is an active anti-entropy tool that compares the local holder
// with a remote holder based on block checksums and resolves differences.
type HolderSyncer struct {
Holder *Holder
Host string
Cluster *Cluster
@ -357,7 +357,7 @@ type IndexSyncer struct {
}
// Returns true if the syncer has been marked to close.
func (s *IndexSyncer) IsClosing() bool {
func (s *HolderSyncer) IsClosing() bool {
select {
case <-s.Closing:
return true
@ -366,10 +366,10 @@ func (s *IndexSyncer) IsClosing() bool {
}
}
// SyncIndex compares the index on host with the local index and resolves differences.
func (s *IndexSyncer) SyncIndex() error {
// SyncHolder compares the holder on host with the local holder and resolves differences.
func (s *HolderSyncer) SyncHolder() error {
// Iterate over schema in sorted order.
for _, di := range s.Index.Schema() {
for _, di := range s.Holder.Schema() {
// Verify syncer has not closed.
if s.IsClosing() {
return nil
@ -397,7 +397,7 @@ func (s *IndexSyncer) SyncIndex() error {
return nil
}
for slice := uint64(0); slice <= s.Index.DB(di.Name).MaxSlice(); slice++ {
for slice := uint64(0); slice <= s.Holder.DB(di.Name).MaxSlice(); slice++ {
// Ignore slices that this host doesn't own.
if !s.Cluster.OwnsFragment(s.Host, di.Name, slice) {
continue
@ -421,9 +421,9 @@ func (s *IndexSyncer) SyncIndex() error {
}
// syncDatabase synchronizes database attributes with the rest of the cluster.
func (s *IndexSyncer) syncDatabase(db string) error {
func (s *HolderSyncer) syncDatabase(db string) error {
// Retrieve database reference.
d := s.Index.DB(db)
d := s.Holder.DB(db)
if d == nil {
return nil
}
@ -466,9 +466,9 @@ func (s *IndexSyncer) syncDatabase(db string) error {
}
// syncFrame synchronizes frame attributes with the rest of the cluster.
func (s *IndexSyncer) syncFrame(db, name string) error {
func (s *HolderSyncer) syncFrame(db, name string) error {
// Retrieve database reference.
f := s.Index.Frame(db, name)
f := s.Holder.Frame(db, name)
if f == nil {
return nil
}
@ -513,9 +513,9 @@ func (s *IndexSyncer) syncFrame(db, name string) error {
}
// syncFragment synchronizes a fragment with the rest of the cluster.
func (s *IndexSyncer) syncFragment(db, frame, view string, slice uint64) error {
func (s *HolderSyncer) syncFragment(db, frame, view string, slice uint64) error {
// Retrieve local frame.
f := s.Index.Frame(db, frame)
f := s.Holder.Frame(db, frame)
if f == nil {
return ErrFrameNotFound
}

128
index_test.go → holder_test.go
View file

@ -12,56 +12,56 @@ import (
"github.com/pilosa/pilosa/pql"
)
// Ensure index can delete a database and its underlying files.
func TestIndex_DeleteDB(t *testing.T) {
idx := MustOpenIndex()
defer idx.Close()
// Ensure holder can delete a database and its underlying files.
func TestHolder_DeleteDB(t *testing.T) {
hldr := MustOpenHolder()
defer hldr.Close()
// Write bits to separate databases.
f0 := idx.MustCreateFragmentIfNotExists("d0", "f", pilosa.ViewStandard, 0)
f0 := hldr.MustCreateFragmentIfNotExists("d0", "f", pilosa.ViewStandard, 0)
if _, err := f0.SetBit(100, 200); err != nil {
t.Fatal(err)
}
f1 := idx.MustCreateFragmentIfNotExists("d1", "f", pilosa.ViewStandard, 0)
f1 := hldr.MustCreateFragmentIfNotExists("d1", "f", pilosa.ViewStandard, 0)
if _, err := f1.SetBit(100, 200); err != nil {
t.Fatal(err)
}
// Ensure d0 exists.
if _, err := os.Stat(idx.DBPath("d0")); err != nil {
if _, err := os.Stat(hldr.DBPath("d0")); err != nil {
t.Fatal(err)
}
// Delete d0.
if err := idx.DeleteDB("d0"); err != nil {
if err := hldr.DeleteDB("d0"); err != nil {
t.Fatal(err)
}
// Ensure d0 files are removed & d1 still exists.
if _, err := os.Stat(idx.DBPath("d0")); !os.IsNotExist(err) {
if _, err := os.Stat(hldr.DBPath("d0")); !os.IsNotExist(err) {
t.Fatal("expected d0 file deletion")
} else if _, err := os.Stat(idx.DBPath("d1")); err != nil {
} else if _, err := os.Stat(hldr.DBPath("d1")); err != nil {
t.Fatal("expected d1 files to still exist", err)
}
}
// Ensure index can sync with a remote index.
func TestIndexSyncer_SyncIndex(t *testing.T) {
// Ensure holder can sync with a remote holder.
func TestHolderSyncer_SyncHolder(t *testing.T) {
cluster := NewCluster(2)
// Create a local index.
idx0 := MustOpenIndex()
defer idx0.Close()
// Create a local holder.
hldr0 := MustOpenHolder()
defer hldr0.Close()
// Create a remote index wrapped by an HTTP
idx1 := MustOpenIndex()
defer idx1.Close()
// Create a remote holder wrapped by an HTTP
hldr1 := MustOpenHolder()
defer hldr1.Close()
s := NewServer()
defer s.Close()
s.Handler.Index = idx1.Index
s.Handler.Holder = hldr1.Holder
s.Handler.Executor.ExecuteFn = func(ctx context.Context, db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor()
e.Index = idx1.Index
e.Holder = hldr1.Holder
e.Host = cluster.Nodes[1].Host
e.Cluster = cluster
return e.Execute(ctx, db, query, slices, opt)
@ -73,14 +73,14 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
cluster.Nodes[1].Host = MustParseURLHost(s.URL)
// Create frames on nodes.
for _, idx := range []*Index{idx0, idx1} {
idx.MustCreateFrameIfNotExists("d", "f")
idx.MustCreateFrameIfNotExists("d", "f0")
idx.MustCreateFrameIfNotExists("y", "z")
for _, hldr := range []*Holder{hldr0, hldr1} {
hldr.MustCreateFrameIfNotExists("d", "f")
hldr.MustCreateFrameIfNotExists("d", "f0")
hldr.MustCreateFrameIfNotExists("y", "z")
}
// Set data on the local index.
f := idx0.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
// Set data on the local holder.
f := hldr0.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
if _, err := f.SetBit(0, 10); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(2, 20); err != nil {
@ -91,15 +91,15 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
t.Fatal(err)
}
f = idx0.MustCreateFragmentIfNotExists("d", "f0", pilosa.ViewStandard, 1)
f = hldr0.MustCreateFragmentIfNotExists("d", "f0", pilosa.ViewStandard, 1)
if _, err := f.SetBit(9, SliceWidth+5); err != nil {
t.Fatal(err)
}
idx0.MustCreateFragmentIfNotExists("y", "z", pilosa.ViewStandard, 0)
hldr0.MustCreateFragmentIfNotExists("y", "z", pilosa.ViewStandard, 0)
// Set data on the remote index.
f = idx1.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
// Set data on the remote holder.
f = hldr1.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0)
if _, err := f.SetBit(0, 4000); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(3, 10); err != nil {
@ -108,7 +108,7 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
t.Fatal(err)
}
f = idx1.MustCreateFragmentIfNotExists("y", "z", pilosa.ViewStandard, 3)
f = hldr1.MustCreateFragmentIfNotExists("y", "z", pilosa.ViewStandard, 3)
if _, err := f.SetBit(10, (3*SliceWidth)+4); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(10, (3*SliceWidth)+5); err != nil {
@ -118,23 +118,23 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
}
// Set highest slice.
idx0.DB("d").SetRemoteMaxSlice(1)
idx0.DB("y").SetRemoteMaxSlice(3)
hldr0.DB("d").SetRemoteMaxSlice(1)
hldr0.DB("y").SetRemoteMaxSlice(3)
// Set up syncer.
syncer := pilosa.IndexSyncer{
Index: idx0.Index,
syncer := pilosa.HolderSyncer{
Holder: hldr0.Holder,
Host: cluster.Nodes[0].Host,
Cluster: cluster,
}
if err := syncer.SyncIndex(); err != nil {
if err := syncer.SyncHolder(); err != nil {
t.Fatal(err)
}
// Verify data is the same on both nodes.
for i, idx := range []*Index{idx0, idx1} {
f := idx.Fragment("d", "f", pilosa.ViewStandard, 0)
for i, hldr := range []*Holder{hldr0, hldr1} {
f := hldr.Fragment("d", "f", pilosa.ViewStandard, 0)
if a := f.Row(0).Bits(); !reflect.DeepEqual(a, []uint64{10, 4000}) {
t.Fatalf("unexpected bits(%d/0): %+v", i, a)
} else if a := f.Row(2).Bits(); !reflect.DeepEqual(a, []uint64{20}) {
@ -147,7 +147,7 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
t.Fatalf("unexpected bits(%d/200): %+v", i, a)
}
f = idx.Fragment("d", "f0", pilosa.ViewStandard, 1)
f = hldr.Fragment("d", "f0", pilosa.ViewStandard, 1)
a := f.Row(9).Bits()
if !reflect.DeepEqual(a, []uint64{SliceWidth + 5}) {
t.Fatalf("unexpected bits(%d/d/f0): %+v", i, a)
@ -155,51 +155,51 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
if a := f.Row(9).Bits(); !reflect.DeepEqual(a, []uint64{SliceWidth + 5}) {
t.Fatalf("unexpected bits(%d/d/f0): %+v", i, a)
}
f = idx.Fragment("y", "z", pilosa.ViewStandard, 3)
f = hldr.Fragment("y", "z", pilosa.ViewStandard, 3)
if a := f.Row(10).Bits(); !reflect.DeepEqual(a, []uint64{(3 * SliceWidth) + 4, (3 * SliceWidth) + 5, (3 * SliceWidth) + 7}) {
t.Fatalf("unexpected bits(%d/y/z): %+v", i, a)
}
}
}
// Index is a test wrapper for pilosa.Index.
type Index struct {
*pilosa.Index
// Holder is a test wrapper for pilosa.Holder.
type Holder struct {
*pilosa.Holder
LogOutput bytes.Buffer
}
// NewIndex returns a new instance of Index with a temporary path.
func NewIndex() *Index {
// NewHolder returns a new instance of Holder with a temporary path.
func NewHolder() *Holder {
path, err := ioutil.TempDir("", "pilosa-")
if err != nil {
panic(err)
}
i := &Index{Index: pilosa.NewIndex()}
i.Path = path
i.Index.LogOutput = &i.LogOutput
h := &Holder{Holder: pilosa.NewHolder()}
h.Path = path
h.Holder.LogOutput = &h.LogOutput
return i
return h
}
// MustOpenIndex creates and opens an index at a temporary path. Panic on error.
func MustOpenIndex() *Index {
i := NewIndex()
if err := i.Open(); err != nil {
// MustOpenHolder creates and opens a holder at a temporary path. Panic on error.
func MustOpenHolder() *Holder {
h := NewHolder()
if err := h.Open(); err != nil {
panic(err)
}
return i
return h
}
// Close closes the index and removes all underlying data.
func (i *Index) Close() error {
defer os.RemoveAll(i.Path)
return i.Index.Close()
// Close closes the holder and removes all underlying data.
func (h *Holder) Close() error {
defer os.RemoveAll(h.Path)
return h.Holder.Close()
}
// MustCreateDBIfNotExists returns a given db. Panic on error.
func (i *Index) MustCreateDBIfNotExists(db string, opt pilosa.DBOptions) *DB {
d, err := i.Index.CreateDBIfNotExists(db, opt)
func (h *Holder) MustCreateDBIfNotExists(db string, opt pilosa.DBOptions) *DB {
d, err := h.Holder.CreateDBIfNotExists(db, opt)
if err != nil {
panic(err)
}
@ -207,8 +207,8 @@ func (i *Index) MustCreateDBIfNotExists(db string, opt pilosa.DBOptions) *DB {
}
// MustCreateFrameIfNotExists returns a given frame. Panic on error.
func (i *Index) MustCreateFrameIfNotExists(db, frame string) *Frame {
f, err := i.MustCreateDBIfNotExists(db, pilosa.DBOptions{}).CreateFrameIfNotExists(frame, pilosa.FrameOptions{})
func (h *Holder) MustCreateFrameIfNotExists(db, frame string) *Frame {
f, err := h.MustCreateDBIfNotExists(db, pilosa.DBOptions{}).CreateFrameIfNotExists(frame, pilosa.FrameOptions{})
if err != nil {
panic(err)
}
@ -216,8 +216,8 @@ func (i *Index) MustCreateFrameIfNotExists(db, frame string) *Frame {
}
// MustCreateFragmentIfNotExists returns a given fragment. Panic on error.
func (i *Index) MustCreateFragmentIfNotExists(db, frame, view string, slice uint64) *Fragment {
d := i.MustCreateDBIfNotExists(db, pilosa.DBOptions{})
func (h *Holder) MustCreateFragmentIfNotExists(db, frame, view string, slice uint64) *Fragment {
d := h.MustCreateDBIfNotExists(db, pilosa.DBOptions{})
f, err := d.CreateFrameIfNotExists(frame, pilosa.FrameOptions{})
if err != nil {
panic(err)

66
server.go
View file

@ -24,7 +24,7 @@ const (
DefaultPollingInterval = 60 * time.Second
)
// Server represents an index wrapped by a running HTTP server.
// Server represents a holder wrapped by a running HTTP server.
type Server struct {
ln net.Listener
@ -33,7 +33,7 @@ type Server struct {
closing chan struct{}
// Data storage and HTTP interface.
Index *Index
Holder *Holder
Handler *Handler
Broadcaster Broadcaster
BroadcastReceiver BroadcastReceiver
@ -55,7 +55,7 @@ func NewServer() *Server {
s := &Server{
closing: make(chan struct{}),
Index: NewIndex(),
Holder: NewHolder(),
Handler: NewHandler(),
Broadcaster: NopBroadcaster,
BroadcastReceiver: NopBroadcastReceiver,
@ -66,7 +66,7 @@ func NewServer() *Server {
LogOutput: os.Stderr,
}
s.Handler.Index = s.Index
s.Handler.Holder = s.Holder
return s
}
@ -96,8 +96,8 @@ func (s *Server) Open() error {
s.Cluster.Nodes = []*Node{{Host: s.Host}}
}
// Open index.
if err := s.Index.Open(); err != nil {
// Open holder.
if err := s.Holder.Open(); err != nil {
return err
}
@ -112,7 +112,7 @@ func (s *Server) Open() error {
// Create executor for executing queries.
e := NewExecutor()
e.Index = s.Index
e.Holder = s.Holder
e.Host = s.Host
e.Cluster = s.Cluster
@ -123,9 +123,9 @@ func (s *Server) Open() error {
s.Handler.Executor = e
s.Handler.LogOutput = s.LogOutput
// Initialize Index.
s.Index.Broadcaster = s.Broadcaster
s.Index.LogOutput = s.LogOutput
// Initialize Holder.
s.Holder.Broadcaster = s.Broadcaster
s.Holder.LogOutput = s.LogOutput
// Serve HTTP.
go func() { http.Serve(ln, s.Handler) }()
@ -147,8 +147,8 @@ func (s *Server) Close() error {
if s.ln != nil {
s.ln.Close()
}
if s.Index != nil {
s.Index.Close()
if s.Holder != nil {
s.Holder.Close()
}
return nil
@ -168,7 +168,7 @@ func (s *Server) monitorAntiEntropy() {
ticker := time.NewTicker(s.AntiEntropyInterval)
defer ticker.Stop()
s.logger().Printf("index sync monitor initializing (%s interval)", s.AntiEntropyInterval)
s.logger().Printf("holder sync monitor initializing (%s interval)", s.AntiEntropyInterval)
for {
// Wait for tick or a close.
@ -178,23 +178,23 @@ func (s *Server) monitorAntiEntropy() {
case <-ticker.C:
}
s.logger().Printf("index sync beginning")
s.logger().Printf("holder sync beginning")
// Initialize syncer with local index and remote client.
var syncer IndexSyncer
syncer.Index = s.Index
// Initialize syncer with local holder and remote client.
var syncer HolderSyncer
syncer.Holder = s.Holder
syncer.Host = s.Host
syncer.Cluster = s.Cluster
syncer.Closing = s.closing
// Sync indexes.
if err := syncer.SyncIndex(); err != nil {
s.logger().Printf("index sync error: err=%s", err)
// Sync holders.
if err := syncer.SyncHolder(); err != nil {
s.logger().Printf("holder sync error: err=%s", err)
continue
}
// Record successful sync in log.
s.logger().Printf("index sync complete")
s.logger().Printf("holder sync complete")
}
}
@ -215,14 +215,14 @@ func (s *Server) monitorMaxSlices() {
case <-ticker.C:
}
oldmaxslices := s.Index.MaxSlices()
oldmaxslices := s.Holder.MaxSlices()
for _, node := range s.Cluster.Nodes {
if s.Host != node.Host {
maxSlices, _ := checkMaxSlices(node.Host)
for db, newmax := range maxSlices {
// if we don't know about a db locally, log an error because
// db's should be created and synced prior to slice creation
if localdb := s.Index.DB(db); localdb != nil {
if localdb := s.Holder.DB(db); localdb != nil {
if newmax > oldmaxslices[db] {
oldmaxslices[db] = newmax
localdb.SetRemoteMaxSlice(newmax)
@ -240,30 +240,30 @@ func (s *Server) monitorMaxSlices() {
func (s *Server) ReceiveMessage(pb proto.Message) error {
switch obj := pb.(type) {
case *internal.CreateSliceMessage:
d := s.Index.DB(obj.DB)
d := s.Holder.DB(obj.DB)
if d == nil {
return fmt.Errorf("Local DB not found: %s", obj.DB)
}
d.SetRemoteMaxSlice(obj.Slice)
case *internal.CreateDBMessage:
opt := DBOptions{ColumnLabel: obj.Meta.ColumnLabel}
_, err := s.Index.CreateDB(obj.DB, opt)
_, err := s.Holder.CreateDB(obj.DB, opt)
if err != nil {
return err
}
case *internal.DeleteDBMessage:
if err := s.Index.DeleteDB(obj.DB); err != nil {
if err := s.Holder.DeleteDB(obj.DB); err != nil {
return err
}
case *internal.CreateFrameMessage:
db := s.Index.DB(obj.DB)
db := s.Holder.DB(obj.DB)
opt := FrameOptions{RowLabel: obj.Meta.RowLabel}
_, err := db.CreateFrame(obj.Frame, opt)
if err != nil {
return err
}
case *internal.DeleteFrameMessage:
db := s.Index.DB(obj.DB)
db := s.Holder.DB(obj.DB)
if err := db.DeleteFrame(obj.Frame); err != nil {
return err
}
@ -273,16 +273,16 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
// Server implements gossip.StateHandler.
// LocalState returns the state of the local node as well as the
// index (dbs/frames) according to the local node.
// holder (dbs/frames) according to the local node.
// In a gossip implementation, memberlist.Delegate.LocalState() uses this.
func (s *Server) LocalState() (proto.Message, error) {
if s.Index == nil {
return nil, errors.New("Server.Index is nil.")
if s.Holder == nil {
return nil, errors.New("Server.Holder is nil.")
}
return &internal.NodeState{
Host: s.Host,
State: "OK", // TODO: make this work, pull from s.Cluster.Node
DBs: encodeDBs(s.Index.DBs()),
DBs: encodeDBs(s.Holder.DBs()),
}, nil
}
@ -300,7 +300,7 @@ func (s *Server) mergeRemoteState(ns *internal.NodeState) error {
ColumnLabel: db.Meta.ColumnLabel,
TimeQuantum: TimeQuantum(db.Meta.TimeQuantum),
}
d, err := s.Index.CreateDBIfNotExists(db.Name, opt)
d, err := s.Holder.CreateDBIfNotExists(db.Name, opt)
if err != nil {
return err
}

6
server/server.go
View file

@ -115,10 +115,10 @@ func (m *Command) SetupServer() error {
m.Server.LogOutput = logFile
}
// Configure index.
// Configure holder.
fmt.Fprintf(m.Stderr, "Using data from: %s\n", m.Config.DataDir)
m.Server.Index.Path = m.Config.DataDir
m.Server.Index.Stats = pilosa.NewExpvarStatsClient()
m.Server.Holder.Path = m.Config.DataDir
m.Server.Holder.Stats = pilosa.NewExpvarStatsClient()
var err error
m.Server.Host, err = normalizeHost(m.Config.Host)