scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-08 03:47:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'develop' into newpql

Browse source
This commit is contained in:
Matt Jaffee 2018-06-20 09:31:42 -05:00
commit 1a2fa51b63
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
44 changed files with 3045 additions and 513 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
Gopkg.lock generated
View file

@ -70,6 +70,18 @@
packages = ["proto"]
revision = "1643683e1b54a9e88ad26d98f81400c8c9d9f4f9"
[[projects]]
name = "github.com/google/go-cmp"
packages = [
"cmp",
"cmp/cmpopts",
"cmp/internal/diff",
"cmp/internal/function",
"cmp/internal/value"
]
revision = "3af367b6b30c263d47e8895973edcca9a49cf029"
version = "v0.2.0"
[[projects]]
name = "github.com/gorilla/context"
packages = ["."]
@ -304,6 +316,6 @@
[solve-meta]
analyzer-name = "dep"
analyzer-version = 1
inputs-digest = "325d0fb217ec7f1509186ff947e184f6c8e65941f06000eb110180e65816b1a4"
inputs-digest = "40bd9c0a1a403580ad77f9ae84e81a97da1d1622b3f620bd000271c52b50b8b5"
solver-name = "gps-cdcl"
solver-version = 1

13
api.go
View file

@ -44,6 +44,7 @@ type API struct {
BroadcastHandler BroadcastHandler
StatusHandler StatusHandler
Cluster *Cluster
TranslateStore TranslateStore
Logger Logger
}
@ -124,6 +125,18 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
if err != nil {
return resp, errors.Wrap(err, "reading column attrs")
}
// Translate column attributes, if necessary.
if api.TranslateStore != nil {
for _, col := range resp.ColumnAttrSets {
v, err := api.TranslateStore.TranslateColumnToString(req.Index, col.ID)
if err != nil {
return resp, err
}
col.Key, col.ID = v, 0
}
}
resp.ColumnAttrSets = columnAttrSets
}
return resp, nil

79
attr_test.go
View file

@ -15,15 +15,20 @@
package pilosa_test
import (
"io/ioutil"
"os"
"reflect"
"runtime"
"sync"
"testing"
"github.com/pilosa/pilosa/test"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/boltdb"
)
// Ensure database can set and retrieve column attributes.
func TestAttrStore_Attrs(t *testing.T) {
s := test.MustOpenAttrStore()
s := MustOpenAttrStore()
defer s.Close()
// Set attributes.
@ -52,7 +57,7 @@ func TestAttrStore_Attrs(t *testing.T) {
// Ensure database returns a non-nil empty map if unset.
func TestAttrStore_Attrs_Empty(t *testing.T) {
s := test.MustOpenAttrStore()
s := MustOpenAttrStore()
defer s.Close()
if m, err := s.Attrs(100); err != nil {
@ -64,7 +69,7 @@ func TestAttrStore_Attrs_Empty(t *testing.T) {
// Ensure database can unset attributes if explicitly set to nil.
func TestAttrStore_Attrs_Unset(t *testing.T) {
s := test.MustOpenAttrStore()
s := MustOpenAttrStore()
defer s.Close()
// Set attributes.
@ -84,7 +89,7 @@ func TestAttrStore_Attrs_Unset(t *testing.T) {
// Ensure attribute block checksums can be returned.
func TestAttrStore_Blocks(t *testing.T) {
s := test.MustOpenAttrStore()
s := MustOpenAttrStore()
defer s.Close()
// Set attributes.
@ -123,3 +128,67 @@ func TestAttrStore_Blocks(t *testing.T) {
t.Fatalf("block 2 mismatch: %#v != %#v", blks0[2], blks1[2])
}
}
// AttrStore represents a test wrapper for pilosa.AttrStore.
type AttrStore struct {
pilosa.AttrStore
}
// NewAttrStore returns a new instance of AttrStore.
func NewAttrStore(string) pilosa.AttrStore {
f, err := ioutil.TempFile("", "pilosa-attr-")
if err != nil {
panic(err)
}
f.Close()
os.Remove(f.Name())
return &AttrStore{boltdb.NewAttrStore(f.Name())}
}
func BenchmarkAttrStore_Duplicate(b *testing.B) {
s := MustOpenAttrStore()
defer s.Close()
// Set attributes.
const n = 5
for i := 0; i < n; i++ {
if err := s.SetAttrs(uint64(i), map[string]interface{}{"A": 100, "B": "foo", "C": true, "D": 100.2}); err != nil {
b.Fatal(err)
}
}
b.ReportAllocs()
b.ResetTimer()
// Update attributes with an existing subset.
cpuN := runtime.GOMAXPROCS(0)
var wg sync.WaitGroup
for i := 0; i < cpuN; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := 0; j < b.N/cpuN; j++ {
if err := s.SetAttrs(uint64(j%n), map[string]interface{}{"A": int64(100), "B": "foo", "D": 100.2}); err != nil {
b.Fatal(err)
}
}
}()
}
wg.Wait()
}
// MustOpenAttrStore returns a new, opened attribute store at a temporary path. Panic on error.
func MustOpenAttrStore() pilosa.AttrStore {
s := NewAttrStore("")
if err := s.Open(); err != nil {
panic(err)
}
return s
}
// Close closes the database and removes the underlying data.
func (s *AttrStore) Close() error {
defer os.RemoveAll(s.Path())
return s.AttrStore.Close()
}

16
cluster.go
View file

@ -910,7 +910,7 @@ func (c *Cluster) open() error {
Event: uint32(NodeJoin),
Node: EncodeNode(c.Node),
}
if err := c.Broadcaster.SendAsync(msg); err != nil {
if err := c.Broadcaster.SendSync(msg); err != nil {
return fmt.Errorf("sending restart NodeJoin: %v", err)
}
@ -1801,3 +1801,17 @@ func (c *Cluster) mergeClusterStatus(cs *internal.ClusterStatus) error {
return nil
}
func (c *Cluster) setStatic(hosts []string) error {
c.Static = true
c.Coordinator = c.Node.ID
for _, address := range hosts {
uri, err := NewURIFromAddress(address)
if err != nil {
return errors.Wrap(err, "getting URI")
}
c.Nodes = append(c.Nodes, &Node{URI: *uri})
}
c.MemberSet = NewStaticMemberSet(c.Nodes)
return nil
}

32
cluster_internal_test.go
View file

@ -153,19 +153,19 @@ func TestFragSources(t *testing.T) {
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 101, nil)
_, err = field.SetBit(1, 101, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 1300000, nil)
_, err = field.SetBit(1, 1300000, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 2600000, nil)
_, err = field.SetBit(1, 2600000, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 3900000, nil)
_, err = field.SetBit(1, 3900000, nil)
if err != nil {
t.Fatal(err)
}
@ -552,7 +552,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Single node, in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
tc.addNode()
node := tc.Clusters[0]
@ -580,7 +580,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Single node, not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
tc.addNode()
node := tc.Clusters[0]
@ -605,14 +605,14 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Multiple nodes, no data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
tc.addNode()
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
tc.AddNode(false)
tc.addNode()
node0 := tc.Clusters[0]
node1 := tc.Clusters[1]
@ -643,7 +643,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Multiple nodes, in/not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
tc.addNode()
node0 := tc.Clusters[0]
// write topology to data file
@ -664,12 +664,12 @@ func TestCluster_ResizeStates(t *testing.T) {
// Expect an error by adding a node not in the topology.
expectedError := "host is not in topology: node1"
err := tc.AddNode(false)
err := tc.addNode()
if err == nil || err.Error() != expectedError {
t.Errorf("did not receive expected error: %s", expectedError)
}
tc.AddNode(false)
tc.addNode()
node2 := tc.Clusters[2]
// Ensure that node comes up in state NORMAL.
@ -687,7 +687,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Multiple nodes, with data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
tc.addNode()
node0 := tc.Clusters[0]
// Open TestCluster.
@ -699,8 +699,8 @@ func TestCluster_ResizeStates(t *testing.T) {
if err := tc.CreateField("i", "f", FieldOptions{}); err != nil {
t.Fatal(err)
}
tc.SetBit("i", "f", "standard", 1, 101, nil)
tc.SetBit("i", "f", "standard", 1, 1300000, nil)
tc.SetBit("i", "f", 1, 101, nil)
tc.SetBit("i", "f", 1, 1300000, nil)
// Before starting the resize, get the CheckSum to use for
// comparison later.
@ -709,8 +709,8 @@ func TestCluster_ResizeStates(t *testing.T) {
node0Fragment := node0View.Fragment(1)
node0Checksum := node0Fragment.Checksum()
// AddNode needs to block until the resize process has completed.
tc.AddNode(false)
// addNode needs to block until the resize process has completed.
tc.addNode()
node1 := tc.Clusters[1]
// Ensure that nodes come up in state NORMAL.

3
ctl/server.go
View file

@ -43,6 +43,9 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.StringSliceVarP(&srv.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster. Only used for testing.")
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Duration that will trigger log and stat messages for slow queries.")
// Translation
flags.StringVarP(&srv.Config.Translation.PrimaryURL, "translation.primary-url", "", srv.Config.Translation.PrimaryURL, "URL for primary translation node for replication.")
// Gossip
flags.StringVarP(&srv.Config.Gossip.Port, "gossip.port", "", srv.Config.Gossip.Port, "Port to which pilosa should bind for internal state sharing.")
flags.StringSliceVarP(&srv.Config.Gossip.Seeds, "gossip.seeds", "", srv.Config.Gossip.Seeds, "Host with which to seed the gossip membership.")

10
enterprise/b/containers_btree.go
View file

@ -121,6 +121,16 @@ func (btc *BTreeContainers) GetOrCreate(key uint64) *roaring.Container {
return btc.lastContainer
}
func (btc *BTreeContainers) Count() (n uint64) {
e, _ := btc.tree.Seek(0)
_, c, err := e.Next()
for err != io.EOF {
n += uint64(c.N())
_, c, err = e.Next()
}
return
}
func (btc *BTreeContainers) Clone() roaring.Containers {
nbtc := NewBTreeContainers()

128
executor.go
View file

@ -50,6 +50,9 @@ type Executor struct {
// Maximum number of SetBit() or ClearBit() commands per request.
MaxWritesPerRequest int
// Stores key/id translation data.
TranslateStore TranslateStore
}
// ExecutorOption is a functional option type for pilosa.Executor
@ -83,6 +86,11 @@ func (e *Executor) Execute(ctx context.Context, index string, q *pql.Query, slic
return nil, ErrIndexRequired
}
idx := e.Holder.Index(index)
if idx == nil {
return nil, ErrIndexNotFound
}
// Verify that the number of writes do not exceed the maximum.
if e.MaxWritesPerRequest > 0 && q.WriteCallN() > e.MaxWritesPerRequest {
return nil, ErrTooManyWrites
@ -93,6 +101,29 @@ func (e *Executor) Execute(ctx context.Context, index string, q *pql.Query, slic
opt = &ExecOptions{}
}
// Translate query keys to ids, if necessary.
for i := range q.Calls {
if err := e.translateCall(index, idx, q.Calls[i]); err != nil {
return nil, err
}
}
results, err := e.execute(ctx, index, q, slices, opt)
if err != nil {
return nil, err
}
// Translate response objects from ids to keys, if necessary.
for i := range results {
results[i], err = e.translateResult(index, idx, q.Calls[i], results[i])
if err != nil {
return nil, err
}
}
return results, nil
}
func (e *Executor) execute(ctx context.Context, index string, q *pql.Query, slices []uint64, opt *ExecOptions) ([]interface{}, error) {
// Don't bother calculating slices for query types that don't require it.
needsSlices := needsSlices(q.Calls)
@ -995,17 +1026,17 @@ func (e *Executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
return false, fmt.Errorf("ClearBit col field '%v' required", columnLabel)
}
return e.executeClearBitView(ctx, index, c, f, ViewStandard, colID, rowID, opt)
return e.executeClearBitField(ctx, index, c, f, colID, rowID, opt)
}
// executeClearBitView executes a ClearBit() call for a single view.
func (e *Executor) executeClearBitView(ctx context.Context, index string, c *pql.Call, f *Field, view string, colID, rowID uint64, opt *ExecOptions) (bool, error) {
// executeClearBitField executes a ClearBit() call for a single view.
func (e *Executor) executeClearBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, opt *ExecOptions) (bool, error) {
slice := colID / SliceWidth
ret := false
for _, node := range e.Cluster.sliceNodes(index, slice) {
// Update locally if host matches.
if node.ID == e.Node.ID {
val, err := f.ClearBit(view, rowID, colID, nil)
val, err := f.ClearBit(rowID, colID, nil)
if err != nil {
return false, err
} else if val {
@ -1070,18 +1101,18 @@ func (e *Executor) executeSetBit(ctx context.Context, index string, c *pql.Call,
timestamp = &t
}
return e.executeSetBitView(ctx, index, c, f, ViewStandard, colID, rowID, timestamp, opt)
return e.executeSetBitField(ctx, index, c, f, colID, rowID, timestamp, opt)
}
// executeSetBitView executes a SetBit() call for a specific view.
func (e *Executor) executeSetBitView(ctx context.Context, index string, c *pql.Call, f *Field, view string, colID, rowID uint64, timestamp *time.Time, opt *ExecOptions) (bool, error) {
// executeSetBitField executes a SetBit() call for a specific view.
func (e *Executor) executeSetBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, timestamp *time.Time, opt *ExecOptions) (bool, error) {
slice := colID / SliceWidth
ret := false
for _, node := range e.Cluster.sliceNodes(index, slice) {
// Update locally if host matches.
if node.ID == e.Node.ID {
val, err := f.SetBit(view, rowID, colID, timestamp)
val, err := f.SetBit(rowID, colID, timestamp)
if err != nil {
return false, err
} else if val {
@ -1559,6 +1590,78 @@ func (e *Executor) mapperLocal(ctx context.Context, slices []uint64, mapFn mapFu
}
}
func (e *Executor) translateCall(index string, idx *Index, c *pql.Call) error {
// Translate column key.
if idx.Keys() {
if value := callArgString(c, "col"); value != "" {
ids, err := e.TranslateStore.TranslateColumnsToUint64(index, []string{value})
if err != nil {
return err
}
c.Args["col"] = ids[0]
}
}
// Translate row key, if field is specified & key exists.
if fieldName := callArgString(c, "field"); fieldName != "" {
field := idx.Field(fieldName)
if field.Keys() {
if value := callArgString(c, "row"); value != "" {
ids, err := e.TranslateStore.TranslateRowsToUint64(index, fieldName, []string{value})
if err != nil {
return err
}
c.Args["row"] = ids[0]
}
}
}
// Translate child calls.
for _, child := range c.Children {
if err := e.translateCall(index, idx, child); err != nil {
return err
}
}
return nil
}
func (e *Executor) translateResult(index string, idx *Index, call *pql.Call, result interface{}) (interface{}, error) {
switch result := result.(type) {
case *Row:
if idx.Keys() {
other := &Row{Attrs: result.Attrs}
for _, segment := range result.Segments() {
for _, col := range segment.Columns() {
key, err := e.TranslateStore.TranslateColumnToString(index, col)
if err != nil {
return nil, err
}
other.Keys = append(other.Keys, key)
}
}
return other, nil
}
case []Pair:
if fieldName := callArgString(call, "field"); fieldName != "" {
field := idx.Field(fieldName)
if field.Keys() {
other := make([]Pair, len(result))
for i := range result {
key, err := e.TranslateStore.TranslateRowToString(index, fieldName, result[i].ID)
if err != nil {
return nil, err
}
other[i] = Pair{Key: key, Count: result[i].Count}
}
return other, nil
}
}
}
return result, nil
}
// errSliceUnavailable is a marker error if no nodes are available.
var errSliceUnavailable = errors.New("slice unavailable")
@ -1670,3 +1773,12 @@ func (vc *ValCount) Larger(other ValCount) ValCount {
Count: vc.Count,
}
}
func callArgString(call *pql.Call, key string) string {
value, ok := call.Args[key]
if !ok {
return ""
}
s, _ := value.(string)
return s
}

126
executor_test.go
View file

@ -22,6 +22,8 @@ import (
"testing"
"github.com/davecgh/go-spew/spew"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/test"
@ -101,6 +103,35 @@ func TestExecutor_Execute_Bitmap(t *testing.T) {
t.Fatal(err)
}
})
t.Run("Keys", func(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{Keys: true})
if _, err := index.CreateField("f", pilosa.FieldOptions{Keys: true}); err != nil {
t.Fatal(err)
}
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
// Set bits.
if _, err := e.Execute(context.Background(), "i", test.MustParse(``+
`SetBit(field=f, row="bar", col="foo")`+"\n"+
`SetBit(field=f, row="baz", col="foo")`+"\n"+
`SetBit(field=f, row="bar", col="bat")`+"\n"+
`SetBit(field=f, row="bbb", col="aaa")`+"\n",
), nil, nil); err != nil {
t.Fatal(err)
}
if results, err := e.Execute(context.Background(), "i", test.MustParse(`Bitmap(row="bar", field=f)`), nil, nil); err != nil {
t.Fatal(err)
} else if diff := cmp.Diff(results, []interface{}{
&pilosa.Row{Keys: []string{"foo", "bat"}, Attrs: map[string]interface{}{}},
}, cmpopts.IgnoreUnexported(pilosa.Row{})); diff != "" {
t.Fatal(diff)
}
})
}
// Ensure a difference query can be executed.
@ -383,36 +414,36 @@ func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
// Ensure a TopN() query can be executed.
func TestExecutor_Execute_TopN(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
t.Run("ID", func(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
// Set columns for rows 0, 10, & 20 across two slices.
if idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("f", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("other", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := e.Execute(context.Background(), "i", test.MustParse(`
SetBit(field=f, row=0, col=0)
SetBit(field=f, row=0, col=1)
SetBit(field=f, row=0, col=`+strconv.Itoa(SliceWidth)+`)
SetBit(field=f, row=0, col=`+strconv.Itoa(SliceWidth+2)+`)
SetBit(field=f, row=0, col=`+strconv.Itoa((5*SliceWidth)+100)+`)
SetBit(field=f, row=10, col=0)
SetBit(field=f, row=10, col=`+strconv.Itoa(SliceWidth)+`)
SetBit(field=f, row=20, col=`+strconv.Itoa(SliceWidth)+`)
SetBit(field=other, row=0, col=0)
`), nil, nil); err != nil {
t.Fatal(err)
}
// Set columns for rows 0, 10, & 20 across two slices.
if idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("f", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("other", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := e.Execute(context.Background(), "i", test.MustParse(`
SetBit(field=f, row=0, col=0)
SetBit(field=f, row=0, col=1)
SetBit(field=f, row=0, col=`+strconv.Itoa(SliceWidth)+`)
SetBit(field=f, row=0, col=`+strconv.Itoa(SliceWidth+2)+`)
SetBit(field=f, row=0, col=`+strconv.Itoa((5*SliceWidth)+100)+`)
SetBit(field=f, row=10, col=0)
SetBit(field=f, row=10, col=`+strconv.Itoa(SliceWidth)+`)
SetBit(field=f, row=20, col=`+strconv.Itoa(SliceWidth)+`)
SetBit(field=other, row=0, col=0)
`), nil, nil); err != nil {
t.Fatal(err)
}
hldr.MustCreateRankedFragmentIfNotExists("i", "f", pilosa.ViewStandard, 0).RecalculateCache()
hldr.MustCreateRankedFragmentIfNotExists("i", "f", pilosa.ViewStandard, 1).RecalculateCache()
hldr.MustCreateRankedFragmentIfNotExists("i", "f", pilosa.ViewStandard, 5).RecalculateCache()
hldr.MustCreateRankedFragmentIfNotExists("i", "f", pilosa.ViewStandard, 0).RecalculateCache()
hldr.MustCreateRankedFragmentIfNotExists("i", "f", pilosa.ViewStandard, 1).RecalculateCache()
hldr.MustCreateRankedFragmentIfNotExists("i", "f", pilosa.ViewStandard, 5).RecalculateCache()
t.Run("Standard", func(t *testing.T) {
if result, err := e.Execute(context.Background(), "i", test.MustParse(`TopN(field=f, n=2)`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result[0], []pilosa.Pair{
@ -422,6 +453,46 @@ func TestExecutor_Execute_TopN(t *testing.T) {
t.Fatalf("unexpected result: %s", spew.Sdump(result))
}
})
t.Run("Keys", func(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
// Set columns for rows 0, 10, & 20 across two slices.
if idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{Keys: true}); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("f", pilosa.FieldOptions{Keys: true}); err != nil {
t.Fatal(err)
} else if _, err := idx.CreateField("other", pilosa.FieldOptions{Keys: true}); err != nil {
t.Fatal(err)
} else if _, err := e.Execute(context.Background(), "i", test.MustParse(`
SetBit(field=f, row="foo", col="a")
SetBit(field=f, row="foo", col="b")
SetBit(field=f, row="foo", col="c")
SetBit(field=f, row="foo", col="d")
SetBit(field=f, row="foo", col="e")
SetBit(field=f, row="bar", col="a")
SetBit(field=f, row="bar", col="b")
SetBit(field=f, row="baz", col="b")
SetBit(field=other, row="foo", col="a")
`), nil, nil); err != nil {
t.Fatal(err)
}
hldr.MustCreateRankedFragmentIfNotExists("i", "f", pilosa.ViewStandard, 0).RecalculateCache()
if result, err := e.Execute(context.Background(), "i", test.MustParse(`TopN(field=f, n=2)`), nil, nil); err != nil {
t.Fatal(err)
} else if diff := cmp.Diff(result, []interface{}{
[]pilosa.Pair{
{Key: "foo", Count: 5},
{Key: "bar", Count: 2},
},
}); diff != "" {
t.Fatal(diff)
}
})
}
func TestExecutor_Execute_TopN_fill(t *testing.T) {
@ -1213,6 +1284,7 @@ func TestExecutor_Execute_Remote_TopN(t *testing.T) {
func TestExecutor_Execute_ErrMaxWritesPerRequest(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{})
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
e.MaxWritesPerRequest = 3
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetBit() ClearBit() SetBit() SetBit()`), nil, nil); err != pilosa.ErrTooManyWrites {

36
field.go
View file

@ -282,6 +282,7 @@ func (f *Field) loadMeta() error {
f.options.Min = pb.Min
f.options.Max = pb.Max
f.options.TimeQuantum = TimeQuantum(pb.TimeQuantum)
f.options.Keys = pb.Keys
return nil
}
@ -317,6 +318,7 @@ func (f *Field) applyOptions(opt FieldOptions) error {
f.options.Min = 0
f.options.Max = 0
f.options.TimeQuantum = ""
f.options.Keys = opt.Keys
case FieldTypeInt:
f.options.Type = opt.Type
f.options.CacheType = CacheTypeNone
@ -324,6 +326,7 @@ func (f *Field) applyOptions(opt FieldOptions) error {
f.options.Min = opt.Min
f.options.Max = opt.Max
f.options.TimeQuantum = ""
f.options.Keys = opt.Keys
// Create new bsiGroup.
bsig := &bsiGroup{
@ -345,6 +348,7 @@ func (f *Field) applyOptions(opt FieldOptions) error {
f.options.CacheSize = 0
f.options.Min = 0
f.options.Max = 0
f.options.Keys = opt.Keys
// Set the time quantum.
if err := f.SetTimeQuantum(opt.TimeQuantum); err != nil {
f.Close()
@ -378,6 +382,13 @@ func (f *Field) Close() error {
return nil
}
// Keys returns true if the field uses string keys.
func (f *Field) Keys() bool {
f.mu.RLock()
defer f.mu.RUnlock()
return f.options.Keys
}
// bsiGroup returns a bsiGroup by name.
func (f *Field) bsiGroup(name string) *bsiGroup {
f.mu.RLock()
@ -643,14 +654,11 @@ func (f *Field) ViewRow(viewName string, rowID uint64) (*Row, error) {
}
// SetBit sets a bit on a view within the field.
func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed bool, err error) {
// Validate view name.
if !isValidView(name) {
return false, ErrInvalidView
}
func (f *Field) SetBit(rowID, colID uint64, t *time.Time) (changed bool, err error) {
viewName := ViewStandard
// Retrieve view. Exit if it doesn't exist.
view, err := f.CreateViewIfNotExists(name)
view, err := f.CreateViewIfNotExists(viewName)
if err != nil {
return changed, errors.Wrap(err, "creating view")
}
@ -668,7 +676,7 @@ func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
}
// If a timestamp is specified then set bits across all views for the quantum.
for _, subname := range viewsByTime(name, *t, f.TimeQuantum()) {
for _, subname := range viewsByTime(viewName, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, errors.Wrapf(err, "creating view %s", subname)
@ -685,14 +693,11 @@ func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
}
// ClearBit clears a bit within the field.
func (f *Field) ClearBit(name string, rowID, colID uint64, t *time.Time) (changed bool, err error) {
// Validate view name.
if !isValidView(name) {
return false, ErrInvalidView
}
func (f *Field) ClearBit(rowID, colID uint64, t *time.Time) (changed bool, err error) {
viewName := ViewStandard
// Retrieve view. Exit if it doesn't exist.
view, err := f.CreateViewIfNotExists(name)
view, err := f.CreateViewIfNotExists(viewName)
if err != nil {
return changed, errors.Wrap(err, "creating view")
}
@ -710,7 +715,7 @@ func (f *Field) ClearBit(name string, rowID, colID uint64, t *time.Time) (change
}
// If a timestamp is specified then clear bits across all views for the quantum.
for _, subname := range viewsByTime(name, *t, f.TimeQuantum()) {
for _, subname := range viewsByTime(viewName, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, errors.Wrapf(err, "creating view %s", subname)
@ -1038,6 +1043,7 @@ type FieldOptions struct {
Min int64 `json:"min,omitempty"`
Max int64 `json:"max,omitempty"`
TimeQuantum TimeQuantum `json:"timeQuantum,omitempty"`
Keys bool `json:"keys,omitempty"`
}
// Validate ensures that FieldOption values are valid.
@ -1075,6 +1081,7 @@ func encodeFieldOptions(o *FieldOptions) *internal.FieldOptions {
Min: o.Min,
Max: o.Max,
TimeQuantum: string(o.TimeQuantum),
Keys: o.Keys,
}
}
@ -1089,6 +1096,7 @@ func decodeFieldOptions(options *internal.FieldOptions) *FieldOptions {
Min: options.Min,
Max: options.Max,
TimeQuantum: TimeQuantum(options.TimeQuantum),
Keys: options.Keys,
}
}

46
gossip/gossip.go
View file

@ -24,8 +24,6 @@ import (
"sync"
"time"
"golang.org/x/sync/errgroup"
"github.com/gogo/protobuf/proto"
"github.com/hashicorp/memberlist"
"github.com/pilosa/pilosa"
@ -36,7 +34,6 @@ import (
// Ensure GossipMemberSet implements interfaces.
var _ pilosa.BroadcastReceiver = &GossipMemberSet{}
var _ pilosa.Gossiper = &GossipMemberSet{}
var _ memberlist.Delegate = &GossipMemberSet{}
// GossipMemberSet represents a gossip implementation of MemberSet using memberlist.
@ -240,49 +237,6 @@ func NewGossipMemberSet(name string, host string, cfg Config, ger *GossipEventRe
return g, nil
}
// SendSync implementation of the Broadcaster interface.
func (g *GossipMemberSet) SendSync(pb proto.Message) error {
msg, err := pilosa.MarshalMessage(pb)
if err != nil {
return fmt.Errorf("marshal message: %s", err)
}
mlist := g.memberlist
// Direct sends the message directly to every node.
// An error from any node raises an error on the entire operation.
//
// Gossip uses the gossip protocol to eventually deliver the message
// to every node.
var eg errgroup.Group
for _, n := range mlist.Members() {
// Don't send the message to the local node.
if n == mlist.LocalNode() {
continue
}
node := n
eg.Go(func() error {
return mlist.SendToTCP(node, msg)
})
}
return eg.Wait()
}
// SendAsync implementation of the Gossiper interface.
func (g *GossipMemberSet) SendAsync(pb proto.Message) error {
msg, err := pilosa.MarshalMessage(pb)
if err != nil {
return fmt.Errorf("marshal message: %s", err)
}
b := &broadcast{
msg: msg,
notify: nil,
}
g.broadcasts.QueueBroadcast(b)
return nil
}
// NodeMeta implementation of the memberlist.Delegate interface.
func (g *GossipMemberSet) NodeMeta(limit int) []byte {
buf, err := proto.Marshal(pilosa.EncodeNode(g.node))

8
holder.go
View file

@ -111,7 +111,8 @@ func (h *Holder) Open() error {
}
for _, fi := range fis {
if !fi.IsDir() {
// Skip files or hidden directories.
if !fi.IsDir() || strings.HasPrefix(fi.Name(), ".") {
continue
}
@ -338,12 +339,15 @@ func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) {
return nil, errors.Wrap(err, "creating")
}
index.keys = opt.Keys
if err := index.Open(); err != nil {
return nil, errors.Wrap(err, "opening")
} else if err := index.saveMeta(); err != nil {
return nil, errors.Wrap(err, "meta")
}
// Update options.
h.indexes[index.Name()] = index
return index, nil

6
holder_test.go
View file

@ -210,7 +210,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
} else if field, err := idx.CreateField("bar", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := field.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
@ -231,7 +231,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
} else if field, err := idx.CreateField("bar", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := field.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
@ -257,7 +257,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
} else if view, err := field.CreateViewIfNotExists(pilosa.ViewStandard); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := field.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
} else if err := view.Fragment(0).FlushCache(); err != nil {
t.Fatal(err)

28
http/client_test.go
View file

@ -148,10 +148,10 @@ func TestClient_MultiNode(t *testing.T) {
hldr[2].MustCreateRankedFragmentIfNotExists("i", "f", pilosa.ViewStandard, sliceNums[2]).RecalculateCache()
// Connect to each node to compare results.
client := make([]*test.Client, 3)
client[0] = test.MustNewClient(s[0].Host(), defaultClient)
client[1] = test.MustNewClient(s[1].Host(), defaultClient)
client[2] = test.MustNewClient(s[2].Host(), defaultClient)
client := make([]*Client, 3)
client[0] = MustNewClient(s[0].Host(), defaultClient)
client[1] = MustNewClient(s[1].Host(), defaultClient)
client[2] = MustNewClient(s[2].Host(), defaultClient)
topN := 4
queryRequest := &internal.QueryRequest{
@ -231,7 +231,7 @@ func TestClient_Import(t *testing.T) {
s.Handler.API.Holder = hldr.Holder
// Send import request.
c := test.MustNewClient(s.Host(), defaultClient)
c := MustNewClient(s.Host(), defaultClient)
if err := c.Import(context.Background(), "i", "f", 0, []pilosa.Bit{
{RowID: 0, ColumnID: 1},
{RowID: 0, ColumnID: 5},
@ -276,7 +276,7 @@ func TestClient_ImportValue(t *testing.T) {
s.Handler.API.Holder = hldr.Holder
// Send import request.
c := test.MustNewClient(s.Host(), defaultClient)
c := MustNewClient(s.Host(), defaultClient)
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnID: 1, Value: -10},
{ColumnID: 2, Value: 20},
@ -345,7 +345,7 @@ func TestClient_FragmentBlocks(t *testing.T) {
s.Handler.API.Holder = hldr.Holder
// Retrieve blocks.
c := test.MustNewClient(s.Host(), defaultClient)
c := MustNewClient(s.Host(), defaultClient)
blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", 0)
if err != nil {
t.Fatal(err)
@ -362,3 +362,17 @@ func TestClient_FragmentBlocks(t *testing.T) {
t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks))
}
}
// Client represents a test wrapper for pilosa.Client.
type Client struct {
*http.InternalClient
}
// MustNewClient returns a new instance of Client. Panic on error.
func MustNewClient(host string, h *gohttp.Client) *Client {
c, err := http.NewInternalClient(host, h)
if err != nil {
panic(err)
}
return &Client{InternalClient: c}
}

74
http/handler.go
View file

@ -203,6 +203,8 @@ func NewRouter(handler *Handler) *mux.Router {
// For now we just do it for the most commonly used handler, /query
router.HandleFunc("/index/{index}/query", handler.methodNotAllowedHandler).Methods("GET")
router.HandleFunc("/translate/data", handler.handleGetTranslateData).Methods("GET")
router.Use(handler.queryArgValidator)
return router
}
@ -350,21 +352,15 @@ func (h *Handler) handleGetIndexes(w http.ResponseWriter, r *http.Request) {
// handleGetIndex handles GET /index/<indexname> requests.
func (h *Handler) handleGetIndex(w http.ResponseWriter, r *http.Request) {
indexName := mux.Vars(r)["index"]
index, err := h.API.Index(r.Context(), indexName)
if err != nil {
http.Error(w, err.Error(), http.StatusNotFound)
return
for _, idx := range h.API.Schema(r.Context()) {
if idx.Name == indexName {
if err := json.NewEncoder(w).Encode(idx); err != nil {
h.Logger.Printf("write response error: %s", err)
}
return
}
}
if err := json.NewEncoder(w).Encode(getIndexResponse{
map[string]string{"name": index.Name()},
}); err != nil {
h.Logger.Printf("write response error: %s", err)
}
}
type getIndexResponse struct {
Index map[string]string `json:"index"`
http.Error(w, fmt.Sprintf("Index %s Not Found", indexName), http.StatusNotFound)
}
type postIndexRequest struct {
@ -1184,6 +1180,56 @@ func (h *Handler) GetAPI() *pilosa.API {
type defaultClusterMessageResponse struct{}
// TranslateStoreBufferSize is the buffer size used for streaming data.
const TranslateStoreBufferSize = 65536
func (h *Handler) handleGetTranslateData(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
offset, _ := strconv.ParseInt(q.Get("offset"), 10, 64)
rc, err := h.API.TranslateStore.Reader(r.Context(), offset)
if err == pilosa.ErrNotImplemented {
http.Error(w, err.Error(), http.StatusNotImplemented)
return
} else if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
defer rc.Close()
// Ensure reader is closed when the client disconnects.
go func() { <-r.Context().Done(); rc.Close() }()
// Flush header so client can continue.
w.WriteHeader(http.StatusOK)
if w, ok := w.(http.Flusher); ok {
w.Flush()
}
// Copy from reader to client until store or client disconnect.
buf := make([]byte, TranslateStoreBufferSize)
for {
// Read from store.
n, err := rc.Read(buf)
if err == io.EOF {
return
} else if err != nil {
h.Logger.Printf("http: translate store read error: %s", err)
return
} else if n == 0 {
continue
}
// Write to response & flush.
if _, err := w.Write(buf[:n]); err != nil {
h.Logger.Printf("http: translate store response write error: %s", err)
return
} else if w, ok := w.(http.Flusher); ok {
w.Flush()
}
}
}
type queryValidationSpec struct {
required []string
args map[string]struct{}

8
http/handler_test.go
View file

@ -85,12 +85,12 @@ func TestHandler_Schema(t *testing.T) {
if f, err := i0.CreateFieldIfNotExists("f1", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if f, err := i1.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if _, err := i0.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
@ -122,12 +122,12 @@ func TestHandler_Status(t *testing.T) {
if f, err := i0.CreateFieldIfNotExists("f1", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if f, err := i1.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if _, err := i0.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {

87
http/translator.go Normal file
View file

@ -0,0 +1,87 @@
package http
import (
"bytes"
"context"
"fmt"
"io"
"io/ioutil"
"net/http"
"net/url"
"strconv"
"github.com/pilosa/pilosa"
)
// Ensure implementation implements inteface.
var _ pilosa.TranslateStore = (*TranslateStore)(nil)
// TranslateStore represents an implementation of TranslateStore that
// communicates over HTTP. This is used with the TranslateHandler.
type TranslateStore struct {
URL string
}
// NewTranslateStore returns a new instance of TranslateStore.
func NewTranslateStore(rawurl string) *TranslateStore {
return &TranslateStore{URL: rawurl}
}
// TranslateColumnsToUint64 is not currently implemented.
func (s *TranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
return nil, pilosa.ErrNotImplemented
}
// TranslateColumnToString is not currently implemented.
func (s *TranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
return "", pilosa.ErrNotImplemented
}
// TranslateRowsToUint64 is not currently implemented.
func (s *TranslateStore) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
return nil, pilosa.ErrNotImplemented
}
// TranslateRowToString is not currently implemented.
func (s *TranslateStore) TranslateRowToString(index, frame string, values uint64) (string, error) {
return "", pilosa.ErrNotImplemented
}
// Reader returns a reader that can stream data from a remote store.
func (s *TranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
// Generate remote URL.
u, err := url.Parse(s.URL)
if err != nil {
return nil, err
}
u.Path = "/translate/data"
u.RawQuery = (url.Values{
"offset": {strconv.FormatInt(off, 10)},
}).Encode()
// Connect a stream to the remote server.
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return nil, err
}
req = req.WithContext(ctx)
// Connect a stream to the remote server.
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, fmt.Errorf("http: cannot connect to translate store endpoint: %s", err)
}
// Handle error codes or return body as stream.
switch resp.StatusCode {
case http.StatusOK:
return resp.Body, nil
case http.StatusNotImplemented:
resp.Body.Close()
return nil, pilosa.ErrNotImplemented
default:
body, _ := ioutil.ReadAll(resp.Body)
resp.Body.Close()
return nil, fmt.Errorf("http: invalid translate store endpoint status: code=%d url=%s body=%q", resp.StatusCode, u.String(), bytes.TrimSpace(body))
}
}

134
http/translator_test.go Normal file
View file

@ -0,0 +1,134 @@
package http_test
import (
"context"
"io"
"io/ioutil"
"net/http/httptest"
"testing"
"time"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/mock"
"github.com/pilosa/pilosa/test"
)
func TestTranslateStore_Reader(t *testing.T) {
// Ensure client can connect and stream the translate store data.
t.Run("OK", func(t *testing.T) {
t.Run("ServerDisconnect", func(t *testing.T) {
var mrc mock.ReadCloser
var readN int
mrc.ReadFunc = func(p []byte) (int, error) {
readN++
switch readN {
case 1:
copy(p, []byte("foo"))
return 3, nil
case 2:
copy(p, []byte("barbaz"))
return 6, nil
case 3:
return 0, io.EOF
default:
t.Fatal("unexpected read")
return 0, nil
}
}
var closeInvoked bool
mrc.CloseFunc = func() error {
closeInvoked = true
return nil
}
// Setup handler on test server.
var translateStore mock.TranslateStore
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
if off != 100 {
t.Fatalf("unexpected off: %d", off)
}
return &mrc, nil
}
h := test.MustNewHandler()
h.API.TranslateStore = &translateStore
s := httptest.NewServer(h)
defer s.Close()
// Connect to server and stream all available data.
store := http.NewTranslateStore(s.URL)
rc, err := store.Reader(context.Background(), 100)
if err != nil {
t.Fatal(err)
} else if data, err := ioutil.ReadAll(rc); err != nil {
t.Fatal(err)
} else if string(data) != `foobarbaz` {
t.Fatalf("unexpected data: %q", data)
} else if err := rc.Close(); err != nil {
t.Fatal(err)
}
if !closeInvoked {
t.Fatal("expected server close")
}
})
// Ensure server closes store reader if client disconnects.
t.Run("ClientDisconnect", func(t *testing.T) {
// Setup mock so that Read() hangs.
done := make(chan struct{})
var mrc mock.ReadCloser
mrc.ReadFunc = func(p []byte) (int, error) {
<-done
return 0, io.EOF
}
var closeInvoked bool
mrc.CloseFunc = func() error {
closeInvoked = true
return nil
}
var translateStore mock.TranslateStore
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
return &mrc, nil
}
h := test.MustNewHandler()
h.API.TranslateStore = &translateStore
s := httptest.NewServer(h)
defer s.Close()
defer close(done)
// Connect to server and begin streaming.
ctx, cancel := context.WithCancel(context.Background())
store := http.NewTranslateStore(s.URL)
if _, err := store.Reader(ctx, 0); err != nil {
t.Fatal(err)
}
// Cancel the context and check if server is closed.
cancel()
time.Sleep(100 * time.Millisecond)
if !closeInvoked {
t.Fatal("expected server-side close")
}
})
})
// Ensure client is notified if the server doesn't support streaming replication.
t.Run("ErrNotImplemented", func(t *testing.T) {
var translateStore mock.TranslateStore
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
return nil, pilosa.ErrNotImplemented
}
h := test.MustNewHandler()
h.API.TranslateStore = &translateStore
s := httptest.NewServer(h)
defer s.Close()
_, err := http.NewTranslateStore(s.URL).Reader(context.Background(), 0)
if err != pilosa.ErrNotImplemented {
t.Fatalf("unexpected error: %s", err)
}
})
}

22
index.go
View file

@ -33,6 +33,7 @@ type Index struct {
mu sync.RWMutex
path string
name string
keys bool // use string keys
// Fields by name.
fields map[string]*Field
@ -80,6 +81,9 @@ func (i *Index) Name() string { return i.name }
// Path returns the path the index was initialized with.
func (i *Index) Path() string { return i.path }
// Keys returns true if the index uses string keys.
func (i *Index) Keys() bool { return i.keys }
// ColumnAttrStore returns the storage for column attributes.
func (i *Index) ColumnAttrStore() AttrStore { return i.columnAttrStore }
@ -164,18 +168,17 @@ func (i *Index) loadMeta() error {
}
// Copy metadata fields.
i.keys = pb.Keys
return nil
}
// NOTE: Until we introduce new attributes to store in the index .meta file,
// we don't need to actually write the file. The code related to index.options
// and the index meta file are left in place for future use.
/*
// saveMeta writes meta data for the index.
func (i *Index) saveMeta() error {
// Marshal metadata.
buf, err := proto.Marshal(&internal.IndexMeta{})
buf, err := proto.Marshal(&internal.IndexMeta{
Keys: i.keys,
})
if err != nil {
return errors.Wrap(err, "marshalling")
}
@ -187,7 +190,6 @@ func (i *Index) saveMeta() error {
return nil
}
*/
// Close closes the index and its fields.
func (i *Index) Close() error {
@ -407,11 +409,15 @@ func encodeIndex(d *Index) *internal.Index {
}
// IndexOptions represents options to set when initializing an index.
type IndexOptions struct{}
type IndexOptions struct {
Keys bool `json:"keys"`
}
// Encode converts i into its internal representation.
func (i *IndexOptions) Encode() *internal.IndexMeta {
return &internal.IndexMeta{}
return &internal.IndexMeta{
Keys: i.Keys,
}
}
// hasTime returns true if a contains a non-nil time.

215
inmem/translator.go Normal file
View file

@ -0,0 +1,215 @@
package inmem
import (
"context"
"io"
"sync"
"github.com/pilosa/pilosa"
)
// Ensure type implements interface.
var _ pilosa.TranslateStore = &TranslateStore{}
// TranslateStore is an in-memory storage engine for translating string-to-uint64 values.
type TranslateStore struct {
mu sync.RWMutex
cols map[string]*translateIndex
rows map[frameKey]*translateIndex
}
// NewTranslateStore returns a new instance of TranslateStore.
func NewTranslateStore() *TranslateStore {
return &TranslateStore{
cols: make(map[string]*translateIndex),
rows: make(map[frameKey]*translateIndex),
}
}
// Reader returns an error because it is not supported by the inmem store.
func (s *TranslateStore) Reader(ctx context.Context, offset int64) (io.ReadCloser, error) {
return nil, pilosa.ErrReplicationNotSupported
}
// TranslateColumnsToUint64 converts value to a uint64 id.
// If value does not have an associated id then one is created.
func (s *TranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
ret := make([]uint64, len(values))
// Read value under read lock.
s.mu.RLock()
if idx := s.cols[index]; idx != nil {
var writeRequired bool
for i := range values {
v, ok := idx.lookup[values[i]]
if !ok {
writeRequired = true
}
ret[i] = v
}
if !writeRequired {
s.mu.RUnlock()
return ret, nil
}
}
s.mu.RUnlock()
// If any values not found then recheck and then add under a write lock.
s.mu.Lock()
defer s.mu.Unlock()
// Recheck if value was created between the read lock and write lock.
idx := s.cols[index]
if idx != nil {
var writeRequired bool
for i := range values {
if ret[i] != 0 {
continue
}
v, ok := idx.lookup[values[i]]
if !ok {
writeRequired = true
continue
}
ret[i] = v
}
if !writeRequired {
return ret, nil
}
}
// Create index map if it doesn't exists.
if idx == nil {
idx = newTranslateIndex()
s.cols[index] = idx
}
// Add new identifiers.
for i := range values {
if ret[i] != 0 {
continue
}
idx.seq++
v := idx.seq
ret[i] = v
idx.lookup[values[i]] = v
idx.reverse[v] = values[i]
}
return ret, nil
}
// TranslateColumnToString converts a uint64 id to its associated string value.
// If the id is not associated with a string value then a blank string is returned.
func (s *TranslateStore) TranslateColumnToString(index string, value uint64) (string, error) {
s.mu.RLock()
if idx := s.cols[index]; idx != nil {
if ret, ok := idx.reverse[value]; ok {
s.mu.RUnlock()
return ret, nil
}
}
s.mu.RUnlock()
return "", nil
}
func (s *TranslateStore) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
key := frameKey{index, frame}
ret := make([]uint64, len(values))
// Read value under read lock.
s.mu.RLock()
if idx := s.rows[key]; idx != nil {
var writeRequired bool
for i := range values {
v, ok := idx.lookup[values[i]]
if !ok {
writeRequired = true
}
ret[i] = v
}
if !writeRequired {
s.mu.RUnlock()
return ret, nil
}
}
s.mu.RUnlock()
// If any values not found then recheck and then add under a write lock.
s.mu.Lock()
defer s.mu.Unlock()
// Recheck if value was created between the read lock and write lock.
idx := s.rows[key]
if idx != nil {
var writeRequired bool
for i := range values {
if ret[i] != 0 {
continue
}
v, ok := idx.lookup[values[i]]
if !ok {
writeRequired = true
continue
}
ret[i] = v
}
if !writeRequired {
return ret, nil
}
}
// Create map if it doesn't exists.
if idx == nil {
idx = newTranslateIndex()
s.rows[key] = idx
}
// Add new identifiers.
for i := range values {
if ret[i] != 0 {
continue
}
idx.seq++
v := idx.seq
ret[i] = v
idx.lookup[values[i]] = v
idx.reverse[v] = values[i]
}
return ret, nil
}
func (s *TranslateStore) TranslateRowToString(index, frame string, value uint64) (string, error) {
s.mu.RLock()
if idx := s.rows[frameKey{index, frame}]; idx != nil {
if ret, ok := idx.reverse[value]; ok {
s.mu.RUnlock()
return ret, nil
}
}
s.mu.RUnlock()
return "", nil
}
type frameKey struct {
index string
frame string
}
type translateIndex struct {
seq uint64
lookup map[string]uint64
reverse map[uint64]string
}
func newTranslateIndex() *translateIndex {
return &translateIndex{
lookup: make(map[string]uint64),
reverse: make(map[uint64]string),
}
}

132
inmem/translator_test.go Normal file
View file

@ -0,0 +1,132 @@
package inmem_test
import (
"fmt"
"math/rand"
"reflect"
"testing"
"github.com/pilosa/pilosa/inmem"
)
func TestTranslateStore_TranslateColumn(t *testing.T) {
s := inmem.NewTranslateStore()
// First translation should start id at zero.
if ids, err := s.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Next translation on the same index should move to one.
if ids, err := s.TranslateColumnsToUint64("IDX0", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{2}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Translation on a different index restarts at 0.
if ids, err := s.TranslateColumnsToUint64("IDX1", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Ensure that string values can be looked up by ID.
if value, err := s.TranslateColumnToString("IDX0", 2); err != nil {
t.Fatal(err)
} else if value != "bar" {
t.Fatalf("unexpected value: %s", value)
}
}
func TestTranslateStore_TranslateRow(t *testing.T) {
s := inmem.NewTranslateStore()
// First translation should start id at zero.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Next translation on the same index should move to one.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME0", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{2}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Translation on a different index restarts at 0.
if ids, err := s.TranslateRowsToUint64("IDX1", "FRAME0", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Translation on a different frame restarts at 0.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME1", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Ensure that string values can be looked up by ID.
if value, err := s.TranslateRowToString("IDX0", "FRAME0", 2); err != nil {
t.Fatal(err)
} else if value != "bar" {
t.Fatalf("unexpected value: %s", value)
}
}
func BenchmarkTranslateStore_TranslateColumnsToUint64(b *testing.B) {
const batchSize = 1000
s := inmem.NewTranslateStore()
// Generate keys before benchmark begins
keySets := make([][]string, b.N/1000)
for i := range keySets {
keySets[i] = make([]string, batchSize)
for j, jv := range rand.New(rand.NewSource(0)).Perm(batchSize) {
keySets[i][j] = fmt.Sprintf("%08d%08d", jv, i)
}
}
b.ResetTimer()
for _, keySet := range keySets {
if _, err := s.TranslateColumnsToUint64("IDX0", keySet); err != nil {
b.Fatal(err)
}
}
}
func BenchmarkTranslateStore_TranslateColumnToString(b *testing.B) {
const batchSize = 1000
s := inmem.NewTranslateStore()
// Generate keys before benchmark begins
for i := 0; i < b.N; i += batchSize {
keySet := make([]string, batchSize)
for j, jv := range rand.New(rand.NewSource(0)).Perm(batchSize) {
keySet[j] = fmt.Sprintf("%08d%08d", jv, i)
}
if _, err := s.TranslateColumnsToUint64("IDX0", keySet); err != nil {
b.Fatal(err)
}
}
// Generate random key access.
perm := rand.New(rand.NewSource(0)).Perm(b.N)
b.ResetTimer()
for i := 0; i < b.N; i++ {
if _, err := s.TranslateColumnToString("IDX0", uint64(perm[i])); err != nil {
b.Fatal(err)
}
}
}

349
internal/private.pb.go
View file

@ -1,6 +1,5 @@
// Code generated by protoc-gen-gogo.
// Code generated by protoc-gen-gogo. DO NOT EDIT.
// source: private.proto
// DO NOT EDIT!
/*
Package internal is a generated protocol buffer package.
@ -61,6 +60,7 @@ var _ = math.Inf
const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
type IndexMeta struct {
Keys bool `protobuf:"varint,3,opt,name=Keys,proto3" json:"Keys,omitempty"`
}
func (m *IndexMeta) Reset() { *m = IndexMeta{} }
@ -68,6 +68,13 @@ func (m *IndexMeta) String() string { return proto.CompactTextString(
func (*IndexMeta) ProtoMessage() {}
func (*IndexMeta) Descriptor() ([]byte, []int) { return fileDescriptorPrivate, []int{0} }
func (m *IndexMeta) GetKeys() bool {
if m != nil {
return m.Keys
}
return false
}
type FieldOptions struct {
Type string `protobuf:"bytes,8,opt,name=Type,proto3" json:"Type,omitempty"`
CacheType string `protobuf:"bytes,3,opt,name=CacheType,proto3" json:"CacheType,omitempty"`
@ -75,6 +82,7 @@ type FieldOptions struct {
Min int64 `protobuf:"varint,9,opt,name=Min,proto3" json:"Min,omitempty"`
Max int64 `protobuf:"varint,10,opt,name=Max,proto3" json:"Max,omitempty"`
TimeQuantum string `protobuf:"bytes,5,opt,name=TimeQuantum,proto3" json:"TimeQuantum,omitempty"`
Keys bool `protobuf:"varint,11,opt,name=Keys,proto3" json:"Keys,omitempty"`
}
func (m *FieldOptions) Reset() { *m = FieldOptions{} }
@ -124,6 +132,13 @@ func (m *FieldOptions) GetTimeQuantum() string {
return ""
}
func (m *FieldOptions) GetKeys() bool {
if m != nil {
return m.Keys
}
return false
}
type ImportResponse struct {
Err string `protobuf:"bytes,1,opt,name=Err,proto3" json:"Err,omitempty"`
}
@ -966,6 +981,16 @@ func (m *IndexMeta) MarshalTo(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.Keys {
dAtA[i] = 0x18
i++
if m.Keys {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i++
}
return i, nil
}
@ -1017,6 +1042,16 @@ func (m *FieldOptions) MarshalTo(dAtA []byte) (int, error) {
i++
i = encodeVarintPrivate(dAtA, i, uint64(m.Max))
}
if m.Keys {
dAtA[i] = 0x58
i++
if m.Keys {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i++
}
return i, nil
}
@ -2105,24 +2140,6 @@ func (m *RecalculateCaches) MarshalTo(dAtA []byte) (int, error) {
return i, nil
}
func encodeFixed64Private(dAtA []byte, offset int, v uint64) int {
dAtA[offset] = uint8(v)
dAtA[offset+1] = uint8(v >> 8)
dAtA[offset+2] = uint8(v >> 16)
dAtA[offset+3] = uint8(v >> 24)
dAtA[offset+4] = uint8(v >> 32)
dAtA[offset+5] = uint8(v >> 40)
dAtA[offset+6] = uint8(v >> 48)
dAtA[offset+7] = uint8(v >> 56)
return offset + 8
}
func encodeFixed32Private(dAtA []byte, offset int, v uint32) int {
dAtA[offset] = uint8(v)
dAtA[offset+1] = uint8(v >> 8)
dAtA[offset+2] = uint8(v >> 16)
dAtA[offset+3] = uint8(v >> 24)
return offset + 4
}
func encodeVarintPrivate(dAtA []byte, offset int, v uint64) int {
for v >= 1<<7 {
dAtA[offset] = uint8(v&0x7f | 0x80)
@ -2135,6 +2152,9 @@ func encodeVarintPrivate(dAtA []byte, offset int, v uint64) int {
func (m *IndexMeta) Size() (n int) {
var l int
_ = l
if m.Keys {
n += 2
}
return n
}
@ -2162,6 +2182,9 @@ func (m *FieldOptions) Size() (n int) {
if m.Max != 0 {
n += 1 + sovPrivate(uint64(m.Max))
}
if m.Keys {
n += 2
}
return n
}
@ -2675,6 +2698,26 @@ func (m *IndexMeta) Unmarshal(dAtA []byte) error {
return fmt.Errorf("proto: IndexMeta: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 3:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Keys", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
m.Keys = bool(v != 0)
default:
iNdEx = preIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
@ -2869,6 +2912,26 @@ func (m *FieldOptions) Unmarshal(dAtA []byte) error {
break
}
}
case 11:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Keys", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
m.Keys = bool(v != 0)
default:
iNdEx = preIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
@ -3485,51 +3548,14 @@ func (m *MaxSlices) Unmarshal(dAtA []byte) error {
if postIndex > l {
return io.ErrUnexpectedEOF
}
var keykey uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
keykey |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
var stringLenmapkey uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLenmapkey |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
intStringLenmapkey := int(stringLenmapkey)
if intStringLenmapkey < 0 {
return ErrInvalidLengthPrivate
}
postStringIndexmapkey := iNdEx + intStringLenmapkey
if postStringIndexmapkey > l {
return io.ErrUnexpectedEOF
}
mapkey := string(dAtA[iNdEx:postStringIndexmapkey])
iNdEx = postStringIndexmapkey
if m.Standard == nil {
m.Standard = make(map[string]uint64)
}
if iNdEx < postIndex {
var valuekey uint64
var mapkey string
var mapvalue uint64
for iNdEx < postIndex {
entryPreIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
@ -3539,31 +3565,69 @@ func (m *MaxSlices) Unmarshal(dAtA []byte) error {
}
b := dAtA[iNdEx]
iNdEx++
valuekey |= (uint64(b) & 0x7F) << shift
wire |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
var mapvalue uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
fieldNum := int32(wire >> 3)
if fieldNum == 1 {
var stringLenmapkey uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLenmapkey |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
if iNdEx >= l {
intStringLenmapkey := int(stringLenmapkey)
if intStringLenmapkey < 0 {
return ErrInvalidLengthPrivate
}
postStringIndexmapkey := iNdEx + intStringLenmapkey
if postStringIndexmapkey > l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
mapvalue |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
mapkey = string(dAtA[iNdEx:postStringIndexmapkey])
iNdEx = postStringIndexmapkey
} else if fieldNum == 2 {
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
mapvalue |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
} else {
iNdEx = entryPreIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPrivate
}
if (iNdEx + skippy) > postIndex {
return io.ErrUnexpectedEOF
}
iNdEx += skippy
}
m.Standard[mapkey] = mapvalue
} else {
var mapvalue uint64
m.Standard[mapkey] = mapvalue
}
m.Standard[mapkey] = mapvalue
iNdEx = postIndex
default:
iNdEx = preIndex
@ -6617,69 +6681,70 @@ var (
func init() { proto.RegisterFile("private.proto", fileDescriptorPrivate) }
var fileDescriptorPrivate = []byte{
// 1011 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x56, 0xcb, 0x6f, 0x1c, 0x35,
0x18, 0x67, 0x1e, 0xbb, 0xd9, 0xfd, 0xd2, 0x0d, 0x89, 0x0b, 0x61, 0x8a, 0x50, 0x58, 0xac, 0x4a,
0x0d, 0x3d, 0x44, 0xa5, 0xbd, 0xf0, 0xaa, 0x14, 0x25, 0x1b, 0x60, 0x10, 0x09, 0xe0, 0x49, 0x7a,
0xeb, 0xc1, 0xdd, 0xb5, 0xda, 0x51, 0x66, 0xc7, 0xc3, 0x8c, 0x27, 0xc9, 0xf6, 0xc0, 0x15, 0x2e,
0xdc, 0x11, 0x67, 0xfe, 0x18, 0x8e, 0xfc, 0x09, 0x28, 0xfc, 0x23, 0xc8, 0x9f, 0x3d, 0x8f, 0x64,
0x37, 0x4d, 0x15, 0x7a, 0xf3, 0xf7, 0x7e, 0xfd, 0x3e, 0xdb, 0x30, 0xc8, 0xf2, 0xf8, 0x84, 0x2b,
0xb1, 0x95, 0xe5, 0x52, 0x49, 0xd2, 0x8b, 0x53, 0x25, 0xf2, 0x94, 0x27, 0x74, 0x19, 0xfa, 0x61,
0x3a, 0x11, 0x67, 0xfb, 0x42, 0x71, 0xfa, 0xa7, 0x03, 0xb7, 0xbe, 0x8a, 0x45, 0x32, 0xf9, 0x3e,
0x53, 0xb1, 0x4c, 0x0b, 0xf2, 0x01, 0xf4, 0x77, 0xf9, 0xf8, 0x85, 0x38, 0x9c, 0x65, 0x22, 0xf0,
0x86, 0xce, 0x66, 0x9f, 0x35, 0x8c, 0x5a, 0x1a, 0xc5, 0x2f, 0x45, 0xe0, 0x0f, 0x9d, 0xcd, 0x01,
0x6b, 0x18, 0x64, 0x08, 0xcb, 0x87, 0xf1, 0x54, 0xfc, 0x58, 0xf2, 0x54, 0x95, 0xd3, 0xa0, 0x83,
0xd6, 0x6d, 0x16, 0x21, 0xe0, 0xa3, 0xe3, 0x1e, 0x8a, 0xf0, 0x4c, 0x56, 0xc1, 0xdb, 0x8f, 0xd3,
0xa0, 0x3f, 0x74, 0x36, 0x3d, 0xa6, 0x8f, 0xc8, 0xe1, 0x67, 0x01, 0x58, 0x0e, 0x3f, 0xa3, 0x14,
0x56, 0xc2, 0x69, 0x26, 0x73, 0xc5, 0x44, 0x91, 0xc9, 0xb4, 0x40, 0xab, 0xbd, 0x3c, 0x0f, 0x1c,
0x74, 0xa4, 0x8f, 0xf4, 0x67, 0x58, 0xdd, 0x49, 0xe4, 0xf8, 0x78, 0xc4, 0x15, 0x67, 0xe2, 0xa7,
0x52, 0x14, 0x8a, 0xbc, 0x03, 0x1d, 0xac, 0xd5, 0xea, 0x19, 0x42, 0x73, 0xb1, 0xe6, 0xc0, 0x35,
0x5c, 0x24, 0x34, 0x17, 0xed, 0xb1, 0x6a, 0x9f, 0x19, 0x42, 0x73, 0xa3, 0x24, 0x1e, 0x9b, 0x6a,
0x7d, 0x66, 0x08, 0x5d, 0xc7, 0x93, 0x58, 0x9c, 0xda, 0x12, 0xf1, 0x4c, 0x43, 0x58, 0x6b, 0xc5,
0xb7, 0x69, 0xae, 0x43, 0x97, 0xc9, 0xd3, 0x70, 0x54, 0x04, 0xce, 0xd0, 0xdb, 0xf4, 0x99, 0xa5,
0xb0, 0x91, 0x32, 0x29, 0xa7, 0xa9, 0x16, 0xb9, 0x28, 0x6a, 0x18, 0xf4, 0x0e, 0x74, 0xb0, 0xab,
0xba, 0xca, 0xc6, 0x56, 0x1f, 0xe9, 0x2f, 0x0e, 0xf4, 0xf7, 0xf9, 0x19, 0xa6, 0x51, 0x90, 0xc7,
0xd0, 0x8b, 0x14, 0x4f, 0x27, 0x3c, 0x9f, 0xa0, 0xd2, 0xf2, 0xc3, 0x8f, 0xb6, 0xaa, 0x41, 0x6f,
0xd5, 0x6a, 0x5b, 0x95, 0xce, 0x5e, 0xaa, 0xf2, 0x19, 0xab, 0x4d, 0xde, 0xff, 0x02, 0x06, 0x17,
0x44, 0x3a, 0xde, 0xb1, 0x98, 0x55, 0x5d, 0x3d, 0x16, 0x33, 0x5d, 0xff, 0x09, 0x4f, 0x4a, 0x81,
0xbd, 0xf2, 0x99, 0x21, 0x3e, 0x77, 0x3f, 0x75, 0xe8, 0x36, 0x90, 0xdd, 0x5c, 0x70, 0x25, 0x30,
0xc8, 0xbe, 0x28, 0x0a, 0xfe, 0x5c, 0x5c, 0xdd, 0x71, 0xd3, 0x45, 0xb7, 0xd5, 0x45, 0x7a, 0x1f,
0xc8, 0x48, 0x24, 0x42, 0x09, 0x8b, 0xc7, 0x57, 0x78, 0xa0, 0x51, 0x15, 0xed, 0x7a, 0x5d, 0x72,
0x0f, 0x7c, 0x0d, 0x6e, 0x0c, 0xb6, 0xfc, 0xf0, 0x76, 0xd3, 0x91, 0x1a, 0xf7, 0x0c, 0x15, 0x68,
0x52, 0x39, 0x45, 0x04, 0x5c, 0x5b, 0xc2, 0x02, 0xd0, 0xdc, 0xb7, 0xa1, 0x3c, 0x0c, 0xb5, 0xde,
0x84, 0x6a, 0x2f, 0x95, 0x8d, 0xb6, 0x5d, 0x95, 0x7b, 0xd3, 0x68, 0xf4, 0xa9, 0xe5, 0x6a, 0xfc,
0x1d, 0xf0, 0xa9, 0xb0, 0x36, 0x78, 0xae, 0x53, 0x71, 0xaf, 0x4f, 0x45, 0xbb, 0xd7, 0x98, 0x2d,
0x02, 0x6f, 0xe8, 0x69, 0xf7, 0x48, 0xd0, 0x47, 0xd0, 0x8d, 0xc6, 0x2f, 0xc4, 0x94, 0x93, 0x8f,
0x61, 0x09, 0xf3, 0x10, 0x85, 0x85, 0xd5, 0xdb, 0x97, 0x9a, 0xc8, 0x2a, 0x39, 0x1d, 0xd9, 0xfc,
0x17, 0xe6, 0x74, 0x0f, 0xba, 0x18, 0xbd, 0x08, 0xfc, 0xcb, 0x6e, 0x90, 0xcf, 0xac, 0x98, 0xee,
0x81, 0x77, 0xc4, 0x42, 0xbd, 0x2e, 0x98, 0x41, 0xe5, 0xc5, 0x52, 0xda, 0xf7, 0x37, 0xb2, 0x50,
0xb6, 0x1b, 0x78, 0xd6, 0xbc, 0x1f, 0x64, 0xae, 0xb0, 0xf5, 0x03, 0x86, 0x67, 0xfa, 0x14, 0xfc,
0x03, 0x39, 0x11, 0x64, 0x05, 0xdc, 0x70, 0x64, 0x7d, 0xb8, 0xe1, 0x88, 0x7c, 0x88, 0xee, 0x6d,
0x6b, 0x06, 0x4d, 0x12, 0x47, 0x2c, 0x64, 0x18, 0xf8, 0x2e, 0x0c, 0xc2, 0x62, 0x57, 0xca, 0x7c,
0x12, 0xa7, 0x5c, 0xc9, 0x1c, 0xbd, 0xf6, 0xd8, 0x45, 0x26, 0xdd, 0x86, 0x55, 0xed, 0x3e, 0x52,
0x5c, 0xd5, 0x80, 0x5f, 0x87, 0xae, 0xe6, 0xd5, 0xe1, 0x2c, 0x85, 0x90, 0xd7, 0x7a, 0xd5, 0x04,
0x91, 0xa0, 0xdf, 0x19, 0x0f, 0x7b, 0x27, 0x22, 0x55, 0x2d, 0x04, 0x20, 0x8d, 0x0e, 0x06, 0xcc,
0x10, 0x84, 0x9a, 0x52, 0x6c, 0xce, 0x2b, 0x4d, 0xce, 0x9a, 0xcb, 0x50, 0x46, 0x7f, 0x73, 0x00,
0xaa, 0x84, 0xca, 0xa2, 0x36, 0x71, 0xae, 0x36, 0x21, 0x9f, 0xb4, 0xae, 0x8f, 0xf9, 0x05, 0xa9,
0x45, 0xac, 0x75, 0xc9, 0x6c, 0x56, 0xb0, 0xb0, 0x28, 0x5f, 0x6d, 0xf4, 0x0d, 0xdf, 0x8e, 0x89,
0xd3, 0x18, 0x06, 0xbb, 0x49, 0x59, 0x28, 0x91, 0xdb, 0x8c, 0xf4, 0x35, 0x67, 0x18, 0x75, 0x7f,
0x1a, 0xc6, 0xe2, 0x16, 0x91, 0xbb, 0xd0, 0xd1, 0x99, 0x1a, 0x6c, 0xce, 0x97, 0x61, 0x84, 0xf4,
0x09, 0xf4, 0x76, 0xa2, 0xf0, 0xeb, 0x5c, 0x96, 0xd9, 0x42, 0xe4, 0x55, 0x2f, 0x8d, 0x3b, 0xff,
0xd2, 0x78, 0x73, 0x2f, 0x8d, 0xdf, 0xbc, 0x34, 0x11, 0xac, 0x99, 0x2b, 0x41, 0xaf, 0xc4, 0x4d,
0x6e, 0x84, 0xea, 0x69, 0xf0, 0x5a, 0x4f, 0x43, 0x04, 0x6b, 0x66, 0xf3, 0xdf, 0xa4, 0xd3, 0x3f,
0x5c, 0x58, 0x63, 0xa2, 0x88, 0x5f, 0x8a, 0x30, 0x2d, 0x54, 0x5e, 0x8e, 0xf5, 0x82, 0x6b, 0xfb,
0x6f, 0xe5, 0x33, 0xdb, 0x6d, 0x8f, 0x19, 0xe2, 0x75, 0xc0, 0x44, 0x1e, 0xc0, 0xf2, 0xe5, 0x05,
0x98, 0x57, 0x6d, 0xab, 0x90, 0x07, 0xb0, 0x14, 0xc9, 0x32, 0xd7, 0x48, 0x32, 0xeb, 0xdd, 0xba,
0x74, 0x4c, 0x66, 0x46, 0xcc, 0x2a, 0xb5, 0x16, 0x94, 0x3a, 0xaf, 0x86, 0x12, 0x79, 0x7c, 0x09,
0x4a, 0x41, 0x17, 0x0d, 0xde, 0x6b, 0x0c, 0x2e, 0x88, 0xd9, 0x45, 0x6d, 0xfa, 0xab, 0x03, 0xb7,
0xda, 0x29, 0xbc, 0xd6, 0x6e, 0xd4, 0x13, 0x71, 0x17, 0x4e, 0xc4, 0x5b, 0x34, 0x11, 0xbf, 0x99,
0x48, 0xf3, 0xca, 0x75, 0xda, 0xaf, 0xdc, 0x31, 0xdc, 0x99, 0x1b, 0xd3, 0xae, 0x9c, 0x66, 0x1a,
0x0f, 0xff, 0x63, 0x5c, 0xfa, 0xd6, 0xc8, 0x73, 0x3b, 0xa8, 0x3e, 0x33, 0x04, 0xfd, 0x0c, 0xde,
0x8d, 0x84, 0x6a, 0x0d, 0xa9, 0x42, 0xdb, 0x10, 0xbc, 0x03, 0x71, 0x7a, 0x45, 0xf9, 0x5a, 0x44,
0xbf, 0x84, 0xe0, 0x28, 0x9b, 0x70, 0x25, 0x6e, 0x64, 0xbd, 0x03, 0xbd, 0x43, 0x99, 0xc9, 0x44,
0x3e, 0x9f, 0x5d, 0xb3, 0xf5, 0x01, 0x2c, 0x99, 0x2b, 0xd2, 0x7c, 0x7c, 0xfa, 0xac, 0x22, 0xe9,
0x6d, 0x0d, 0xe8, 0x31, 0x4f, 0xc6, 0x65, 0xa2, 0xd3, 0xd0, 0x3f, 0xa0, 0x62, 0x67, 0xf5, 0xaf,
0xf3, 0x0d, 0xe7, 0xef, 0xf3, 0x0d, 0xe7, 0x9f, 0xf3, 0x0d, 0xe7, 0xf7, 0x7f, 0x37, 0xde, 0x7a,
0xd6, 0xc5, 0x1f, 0xed, 0xa3, 0xff, 0x02, 0x00, 0x00, 0xff, 0xff, 0x97, 0xf0, 0x12, 0xfd, 0xe2,
0x0a, 0x00, 0x00,
// 1028 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x56, 0xcb, 0x72, 0x1c, 0x35,
0x17, 0xfe, 0xfb, 0x32, 0xe3, 0x99, 0xe3, 0x8c, 0x7f, 0x5b, 0x01, 0xd3, 0xa1, 0x28, 0x67, 0x50,
0xa5, 0x2a, 0x26, 0x0b, 0x57, 0x48, 0x36, 0xdc, 0x52, 0xe5, 0xb2, 0xc7, 0x40, 0x03, 0x36, 0xa0,
0xb6, 0xb3, 0xcb, 0x42, 0x99, 0x51, 0x25, 0x5d, 0xee, 0x69, 0x35, 0xdd, 0x6a, 0xdb, 0x93, 0x05,
0x5b, 0xd8, 0xb0, 0xa7, 0x78, 0x12, 0x1e, 0x81, 0x25, 0x8f, 0x40, 0x99, 0x17, 0xa1, 0x74, 0xa4,
0xbe, 0xd8, 0x33, 0x8e, 0x53, 0x86, 0x9d, 0xce, 0xfd, 0xd3, 0xd1, 0x77, 0x24, 0xc1, 0x20, 0xcb,
0xe3, 0x13, 0xae, 0xc4, 0x56, 0x96, 0x4b, 0x25, 0x49, 0x2f, 0x4e, 0x95, 0xc8, 0x53, 0x9e, 0xd0,
0xbb, 0xd0, 0x0f, 0xd3, 0x89, 0x38, 0xdb, 0x17, 0x8a, 0x13, 0x02, 0xfe, 0xd7, 0x62, 0x56, 0x04,
0xde, 0xd0, 0xd9, 0xec, 0x31, 0x5c, 0xd3, 0xdf, 0x1d, 0xb8, 0xf5, 0x79, 0x2c, 0x92, 0xc9, 0xb7,
0x99, 0x8a, 0x65, 0x5a, 0x90, 0xf7, 0xa0, 0xbf, 0xcb, 0xc7, 0x2f, 0xc5, 0xe1, 0x2c, 0x13, 0xe8,
0xd9, 0x67, 0x8d, 0xa2, 0xb6, 0x46, 0xf1, 0x2b, 0x11, 0xf8, 0x43, 0x67, 0x73, 0xc0, 0x1a, 0x05,
0x19, 0xc2, 0xf2, 0x61, 0x3c, 0x15, 0xdf, 0x97, 0x3c, 0x55, 0xe5, 0x34, 0xe8, 0x60, 0x74, 0x5b,
0xa5, 0x21, 0x60, 0xe2, 0x1e, 0x9a, 0x70, 0x4d, 0x56, 0xc1, 0xdb, 0x8f, 0xd3, 0xa0, 0x3f, 0x74,
0x36, 0x3d, 0xa6, 0x97, 0xa8, 0xe1, 0x67, 0x01, 0x58, 0x0d, 0x3f, 0xab, 0xa1, 0x2f, 0xb7, 0xa0,
0x53, 0x58, 0x09, 0xa7, 0x99, 0xcc, 0x15, 0x13, 0x45, 0x26, 0xd3, 0x02, 0x33, 0xed, 0xe5, 0x79,
0xe0, 0x60, 0x72, 0xbd, 0xa4, 0x3f, 0xc2, 0xea, 0x4e, 0x22, 0xc7, 0xc7, 0x23, 0xae, 0x38, 0x13,
0x3f, 0x94, 0xa2, 0x50, 0xe4, 0x2d, 0xe8, 0x60, 0x4f, 0xac, 0x9f, 0x11, 0xb4, 0x16, 0xfb, 0x10,
0xb8, 0x46, 0x8b, 0x82, 0xd6, 0x62, 0x3c, 0x76, 0xc2, 0x67, 0x46, 0xd0, 0xda, 0x28, 0x89, 0xc7,
0xa6, 0x03, 0x3e, 0x33, 0x82, 0xc6, 0xf8, 0x34, 0x16, 0xa7, 0x76, 0xdb, 0xb8, 0xa6, 0x21, 0xac,
0xb5, 0xea, 0x5b, 0x98, 0xeb, 0xd0, 0x65, 0xf2, 0x34, 0x1c, 0x15, 0x81, 0x33, 0xf4, 0x36, 0x7d,
0x66, 0x25, 0x6c, 0xae, 0x4c, 0xca, 0x69, 0xaa, 0x4d, 0x2e, 0x9a, 0x1a, 0x05, 0xbd, 0x03, 0x1d,
0xec, 0xb4, 0xde, 0x65, 0x13, 0xab, 0x97, 0xf4, 0x27, 0x07, 0xfa, 0xfb, 0xfc, 0x0c, 0x61, 0x14,
0xe4, 0x09, 0xf4, 0x22, 0xc5, 0xd3, 0x09, 0xcf, 0x27, 0xe8, 0xb4, 0xfc, 0xe8, 0xfd, 0xad, 0x8a,
0x10, 0x5b, 0xb5, 0xdb, 0x56, 0xe5, 0xb3, 0x97, 0xaa, 0x7c, 0xc6, 0xea, 0x90, 0x77, 0x3f, 0x85,
0xc1, 0x05, 0x93, 0xae, 0x77, 0x2c, 0x66, 0x55, 0x57, 0x8f, 0xc5, 0x4c, 0xef, 0xff, 0x84, 0x27,
0xa5, 0xc0, 0x5e, 0xf9, 0xcc, 0x08, 0x9f, 0xb8, 0x1f, 0x39, 0x74, 0x1b, 0xc8, 0x6e, 0x2e, 0xb8,
0x12, 0x58, 0x64, 0x5f, 0x14, 0x05, 0x7f, 0x21, 0xae, 0xee, 0xb8, 0xe9, 0xa2, 0xdb, 0xea, 0x22,
0x7d, 0x00, 0x64, 0x24, 0x12, 0xa1, 0x84, 0xe5, 0xed, 0x6b, 0x32, 0xd0, 0xa8, 0xaa, 0x76, 0xbd,
0x2f, 0xb9, 0x0f, 0xbe, 0x1e, 0x02, 0x2c, 0xb6, 0xfc, 0xe8, 0x76, 0xd3, 0x91, 0x7a, 0x3e, 0x18,
0x3a, 0xd0, 0xa4, 0x4a, 0x8a, 0x0c, 0xb8, 0x76, 0x0b, 0x0b, 0x48, 0xf3, 0xc0, 0x96, 0xf2, 0xb0,
0xd4, 0x7a, 0x53, 0xaa, 0x3d, 0x68, 0xb6, 0xda, 0x76, 0xb5, 0xdd, 0x9b, 0x56, 0xa3, 0xcf, 0xac,
0x56, 0xf3, 0xef, 0x80, 0x4f, 0x85, 0x8d, 0xc1, 0x75, 0x0d, 0xc5, 0xbd, 0x1e, 0x8a, 0x4e, 0xaf,
0x39, 0xab, 0xef, 0x07, 0x4f, 0xa7, 0x47, 0x81, 0x3e, 0x86, 0x6e, 0x34, 0x7e, 0x29, 0xa6, 0x9c,
0x7c, 0x00, 0x4b, 0x88, 0x43, 0x14, 0x96, 0x56, 0xff, 0xbf, 0xd4, 0x44, 0x56, 0xd9, 0xe9, 0xc8,
0xe2, 0x5f, 0x88, 0xe9, 0x3e, 0x74, 0xb1, 0x7a, 0x11, 0xf8, 0x97, 0xd3, 0xa0, 0x9e, 0x59, 0x33,
0xdd, 0x03, 0xef, 0x88, 0x85, 0x7a, 0x5c, 0x10, 0x41, 0x95, 0xc5, 0x4a, 0x3a, 0xf7, 0x97, 0xb2,
0x50, 0xb6, 0x1b, 0xb8, 0xd6, 0xba, 0xef, 0x64, 0xae, 0xb0, 0xf5, 0x03, 0x86, 0x6b, 0xfa, 0x0c,
0xfc, 0x03, 0x39, 0x11, 0x64, 0x05, 0xdc, 0x70, 0x64, 0x73, 0xb8, 0xe1, 0x88, 0xdc, 0xc5, 0xf4,
0xb6, 0x35, 0x83, 0x06, 0xc4, 0x11, 0x0b, 0x19, 0x16, 0xbe, 0x07, 0x83, 0xb0, 0xd8, 0x95, 0x32,
0x9f, 0xc4, 0x29, 0x57, 0x32, 0xb7, 0x17, 0xe7, 0x45, 0x25, 0xdd, 0x86, 0x55, 0x9d, 0x3e, 0x52,
0x5c, 0xd5, 0x84, 0x5f, 0x87, 0xae, 0xd6, 0xd5, 0xe5, 0xac, 0x84, 0x94, 0xd7, 0x7e, 0xd5, 0x09,
0xa2, 0x40, 0xbf, 0x31, 0x19, 0xf6, 0x4e, 0x44, 0xaa, 0x5a, 0x0c, 0x40, 0x19, 0x13, 0x0c, 0x98,
0x11, 0x08, 0x35, 0x5b, 0xb1, 0x98, 0x57, 0x1a, 0xcc, 0x5a, 0xcb, 0xd0, 0x46, 0x7f, 0x71, 0x00,
0x2a, 0x40, 0x65, 0x51, 0x87, 0x38, 0x57, 0x87, 0x90, 0x0f, 0x5b, 0xd7, 0xc7, 0xfc, 0x80, 0xd4,
0x26, 0xd6, 0xba, 0x64, 0x36, 0x2b, 0x5a, 0x58, 0x96, 0xaf, 0x36, 0xfe, 0x46, 0x6f, 0x8f, 0x89,
0xd3, 0x18, 0x06, 0xbb, 0x49, 0x59, 0x28, 0x91, 0x5b, 0x44, 0xfa, 0x9a, 0x33, 0x8a, 0xba, 0x3f,
0x8d, 0x62, 0x71, 0x8b, 0xc8, 0x3d, 0xe8, 0x68, 0xa4, 0x86, 0x9b, 0xf3, 0xdb, 0x30, 0x46, 0xfa,
0x14, 0x7a, 0x3b, 0x51, 0xf8, 0x45, 0x2e, 0xcb, 0x6c, 0x21, 0xf3, 0xaa, 0xd7, 0xc7, 0x9d, 0x7f,
0x7d, 0xbc, 0xb9, 0xd7, 0xc7, 0xaf, 0x5f, 0x1f, 0x1a, 0xc1, 0x9a, 0xb9, 0x12, 0xf4, 0x48, 0xdc,
0xe4, 0x46, 0xa8, 0x9e, 0x06, 0xaf, 0xf5, 0x34, 0x44, 0xb0, 0x66, 0x26, 0xff, 0xbf, 0x4c, 0xfa,
0x9b, 0x0b, 0x6b, 0x4c, 0x14, 0xf1, 0x2b, 0x11, 0xa6, 0x85, 0xca, 0xcb, 0xb1, 0x1e, 0x70, 0x1d,
0xff, 0x95, 0x7c, 0x6e, 0xbb, 0xed, 0x31, 0x23, 0xbc, 0x09, 0x99, 0xc8, 0x43, 0x58, 0xbe, 0x3c,
0x00, 0xf3, 0xae, 0x6d, 0x17, 0xf2, 0x10, 0x96, 0x22, 0x59, 0xe6, 0x9a, 0x49, 0x66, 0xbc, 0x5b,
0x97, 0x8e, 0x41, 0x66, 0xcc, 0xac, 0x72, 0x6b, 0x51, 0xa9, 0xf3, 0x7a, 0x2a, 0x91, 0x27, 0x97,
0xa8, 0x14, 0x74, 0x31, 0xe0, 0x9d, 0x26, 0xe0, 0x82, 0x99, 0x5d, 0xf4, 0xa6, 0x3f, 0x3b, 0x70,
0xab, 0x0d, 0xe1, 0x8d, 0x66, 0xa3, 0x3e, 0x11, 0x77, 0xe1, 0x89, 0x78, 0x8b, 0x4e, 0xc4, 0x6f,
0x4e, 0xa4, 0x79, 0xe5, 0x3a, 0xed, 0x57, 0xee, 0x18, 0xee, 0xcc, 0x1d, 0xd3, 0xae, 0x9c, 0x66,
0x9a, 0x0f, 0xff, 0xe2, 0xb8, 0xf4, 0xad, 0x91, 0xe7, 0xf6, 0xa0, 0xfa, 0xcc, 0x08, 0xf4, 0x63,
0x78, 0x3b, 0x12, 0xaa, 0x75, 0x48, 0x15, 0xdb, 0x86, 0xe0, 0x1d, 0x88, 0xd3, 0x2b, 0xb6, 0xaf,
0x4d, 0xf4, 0x33, 0x08, 0x8e, 0xb2, 0x09, 0x57, 0xe2, 0x46, 0xd1, 0x3b, 0xd0, 0x3b, 0x94, 0x99,
0x4c, 0xe4, 0x8b, 0xd9, 0x35, 0x53, 0x1f, 0xc0, 0x92, 0xb9, 0x22, 0xcd, 0xc7, 0xa7, 0xcf, 0x2a,
0x91, 0xde, 0xd6, 0x84, 0x1e, 0xf3, 0x64, 0x5c, 0x26, 0x1a, 0x86, 0xfe, 0x01, 0x15, 0x3b, 0xab,
0x7f, 0x9c, 0x6f, 0x38, 0x7f, 0x9e, 0x6f, 0x38, 0x7f, 0x9d, 0x6f, 0x38, 0xbf, 0xfe, 0xbd, 0xf1,
0xbf, 0xe7, 0x5d, 0xfc, 0xf9, 0x3e, 0xfe, 0x27, 0x00, 0x00, 0xff, 0xff, 0xa3, 0x25, 0x40, 0x21,
0x0a, 0x0b, 0x00, 0x00,
}

2
internal/private.proto
View file

@ -3,6 +3,7 @@ syntax = "proto3";
package internal;
message IndexMeta {
bool Keys = 3;
}
message FieldOptions {
@ -12,6 +13,7 @@ message FieldOptions {
int64 Min = 9;
int64 Max = 10;
string TimeQuantum = 5;
bool Keys = 11;
}
message ImportResponse {

35
internal/public.pb.go
View file

@ -1,6 +1,5 @@
// Code generated by protoc-gen-gogo.
// Code generated by protoc-gen-gogo. DO NOT EDIT.
// source: public.proto
// DO NOT EDIT!
/*
Package internal is a generated protocol buffer package.
@ -28,6 +27,8 @@ import proto "github.com/golang/protobuf/proto"
import fmt "fmt"
import math "math"
import encoding_binary "encoding/binary"
import io "io"
// Reference imports to suppress errors if they are not otherwise used.
@ -799,7 +800,8 @@ func (m *Attr) MarshalTo(dAtA []byte) (int, error) {
if m.FloatValue != 0 {
dAtA[i] = 0x31
i++
i = encodeFixed64Public(dAtA, i, uint64(math.Float64bits(float64(m.FloatValue))))
encoding_binary.LittleEndian.PutUint64(dAtA[i:], uint64(math.Float64bits(float64(m.FloatValue))))
i += 8
}
return i, nil
}
@ -1235,24 +1237,6 @@ func (m *ImportValueRequest) MarshalTo(dAtA []byte) (int, error) {
return i, nil
}
func encodeFixed64Public(dAtA []byte, offset int, v uint64) int {
dAtA[offset] = uint8(v)
dAtA[offset+1] = uint8(v >> 8)
dAtA[offset+2] = uint8(v >> 16)
dAtA[offset+3] = uint8(v >> 24)
dAtA[offset+4] = uint8(v >> 32)
dAtA[offset+5] = uint8(v >> 40)
dAtA[offset+6] = uint8(v >> 48)
dAtA[offset+7] = uint8(v >> 56)
return offset + 8
}
func encodeFixed32Public(dAtA []byte, offset int, v uint32) int {
dAtA[offset] = uint8(v)
dAtA[offset+1] = uint8(v >> 8)
dAtA[offset+2] = uint8(v >> 16)
dAtA[offset+3] = uint8(v >> 24)
return offset + 4
}
func encodeVarintPublic(dAtA []byte, offset int, v uint64) int {
for v >= 1<<7 {
dAtA[offset] = uint8(v&0x7f | 0x80)
@ -2333,15 +2317,8 @@ func (m *Attr) Unmarshal(dAtA []byte) error {
if (iNdEx + 8) > l {
return io.ErrUnexpectedEOF
}
v = uint64(encoding_binary.LittleEndian.Uint64(dAtA[iNdEx:]))
iNdEx += 8
v = uint64(dAtA[iNdEx-8])
v |= uint64(dAtA[iNdEx-7]) << 8
v |= uint64(dAtA[iNdEx-6]) << 16
v |= uint64(dAtA[iNdEx-5]) << 24
v |= uint64(dAtA[iNdEx-4]) << 32
v |= uint64(dAtA[iNdEx-3]) << 40
v |= uint64(dAtA[iNdEx-2]) << 48
v |= uint64(dAtA[iNdEx-1]) << 56
m.FloatValue = float64(math.Float64frombits(v))
default:
iNdEx = preIndex

14
mock/mock.go Normal file
View file

@ -0,0 +1,14 @@
package mock
type ReadCloser struct {
ReadFunc func(p []byte) (int, error)
CloseFunc func() error
}
func (rc *ReadCloser) Read(p []byte) (int, error) {
return rc.ReadFunc(p)
}
func (rc *ReadCloser) Close() error {
return rc.CloseFunc()
}

38
mock/translator.go Normal file
View file

@ -0,0 +1,38 @@
package mock
import (
"context"
"io"
"github.com/pilosa/pilosa"
)
var _ pilosa.TranslateStore = (*TranslateStore)(nil)
type TranslateStore struct {
TranslateColumnsToUint64Func func(index string, values []string) ([]uint64, error)
TranslateColumnToStringFunc func(index string, values uint64) (string, error)
TranslateRowsToUint64Func func(index, frame string, values []string) ([]uint64, error)
TranslateRowToStringFunc func(index, frame string, values uint64) (string, error)
ReaderFunc func(ctx context.Context, off int64) (io.ReadCloser, error)
}
func (s *TranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
return s.TranslateColumnsToUint64Func(index, values)
}
func (s *TranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
return s.TranslateColumnToStringFunc(index, values)
}
func (s *TranslateStore) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
return s.TranslateRowsToUint64Func(index, frame, values)
}
func (s *TranslateStore) TranslateRowToString(index, frame string, value uint64) (string, error) {
return s.TranslateRowToStringFunc(index, frame, value)
}
func (s *TranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
return s.ReaderFunc(ctx, off)
}

3
pilosa.go
View file

@ -61,6 +61,8 @@ var (
ErrNodeIDNotExists = errors.New("node with provided ID does not exist")
ErrNodeNotCoordinator = errors.New("node is not the coordinator")
ErrResizeNotRunning = errors.New("no resize job currently running")
ErrNotImplemented = errors.New("not implemented")
)
// ApiMethodNotAllowedError wraps an error value indicating that a particular
@ -83,6 +85,7 @@ var nameRegexp = regexp.MustCompile(`^[a-z][a-z0-9_-]{0,63}$`)
// Can have a set of attributes attached to it.
type ColumnAttrSet struct {
ID uint64 `json:"id"`
Key string `json:"key,omitempty"`
Attrs map[string]interface{} `json:"attrs,omitempty"`
}

33
pql/ast.go
View file

@ -220,6 +220,23 @@ func (q *Query) WriteCallN() int {
return n
}
// HasKeys returns true if any call in the query uses keys and requires translation to ids.
func (q *Query) HasKeys() bool {
for _, call := range q.Calls {
if call.Args["col"] != nil {
if _, ok := call.Args["col"].(string); ok {
return true
}
}
if call.Args["row"] != nil {
if _, ok := call.Args["row"].(string); ok {
return true
}
}
}
return false
}
// String returns a string representation of the query.
func (q *Query) String() string {
a := make([]string, len(q.Calls))
@ -280,6 +297,22 @@ func (c *Call) UintSliceArg(key string) ([]uint64, bool, error) {
}
}
// StringArg is for reading the value at key from call.Args as a string. If the
// key is not in Call.Args, the value of the returned bool will be false, and
// the error will be nil. An error is returned if the value is not a string.
func (c *Call) StringArg(key string) (string, bool, error) {
val, ok := c.Args[key]
if !ok {
return "", false, nil
}
switch tval := val.(type) {
case string:
return tval, true, nil
default:
return "", true, fmt.Errorf("could not convert %v of type %T to string in Call.StringArg", tval, tval)
}
}
// Keys returns a list of argument keys in sorted order.
func (c *Call) Keys() []string {
a := make([]string, 0, len(c.Args))

9
roaring/containers.go
View file

@ -124,6 +124,14 @@ func (sc *SliceContainers) Size() int {
}
func (sc *SliceContainers) Count() uint64 {
n := uint64(0)
for i := range sc.containers {
n += uint64(sc.containers[i].n)
}
return n
}
func (sc *SliceContainers) seek(key uint64) (int, bool) {
i := search64(sc.keys, key)
found := true
@ -153,6 +161,7 @@ func (si *SliceIterator) Next() bool {
si.key = si.e.keys[si.i]
si.value = si.e.containers[si.i]
si.i++
return true
}

13
roaring/roaring.go
View file

@ -93,6 +93,7 @@ type Containers interface {
// return the first container at or after key. found will be true if a
// container is found at key.
Iterator(key uint64) (citer ContainerIterator, found bool)
Count() uint64
}
type ContainerIterator interface {
@ -218,12 +219,7 @@ func (b *Bitmap) Max() uint64 {
// Count returns the number of bits set in the bitmap.
func (b *Bitmap) Count() (n uint64) {
citer, _ := b.Containers.Iterator(0)
for citer.Next() {
_, c := citer.Value()
n += uint64(c.n)
}
return n
return b.Containers.Count()
}
// CountRange returns the number of bits set between [start, end).
@ -1030,6 +1026,11 @@ func (c *Container) Mapped() bool {
return c.mapped
}
// N returns the cached bit count of the container
func (c *Container) N() int {
return c.n
}
// Update updates the container
func (c *Container) Update(containerType byte, n int, mapped bool) {
c.containerType = containerType

10
row.go
View file

@ -27,6 +27,9 @@ import (
type Row struct {
segments []RowSegment
// String keys translated to/from segment columns.
Keys []string
// Attributes associated with the row.
Attrs map[string]interface{}
}
@ -166,6 +169,11 @@ func (r *Row) ClearBit(i uint64) (changed bool) {
return s.ClearBit(i)
}
// Segments returns a list of all segments in the row.
func (r *Row) Segments() []RowSegment {
return r.segments
}
// segment returns a segment for a given slice.
// Returns nil if segment does not exist.
func (r *Row) segment(slice uint64) *RowSegment {
@ -241,8 +249,10 @@ func (r *Row) MarshalJSON() ([]byte, error) {
var o struct {
Attrs map[string]interface{} `json:"attrs"`
Columns []uint64 `json:"columns"`
Keys []string `json:"keys,omitempty"`
}
o.Columns = r.Columns()
o.Keys = r.Keys
o.Attrs = r.Attrs
if o.Attrs == nil {

66
server.go
View file

@ -52,16 +52,18 @@ type Server struct {
closing chan struct{}
// Internal
Holder *Holder
Cluster *Cluster
diagnostics *DiagnosticsCollector
executor *Executor
Holder *Holder
Cluster *Cluster
TranslateFile *TranslateFile
diagnostics *DiagnosticsCollector
executor *Executor
hosts []string
clusterDisabled bool
// External
handler Handler
Broadcaster Broadcaster
BroadcastReceiver BroadcastReceiver
Gossiper Gossiper
systemInfo SystemInfo
gcNotifier GCNotifier
NewAttrStore func(string) AttrStore
@ -75,6 +77,8 @@ type Server struct {
diagnosticInterval time.Duration
maxWritesPerRequest int
primaryTranslateStore TranslateStore
defaultClient InternalClient
dataDir string
}
@ -169,6 +173,13 @@ func OptServerInternalClient(c InternalClient) ServerOption {
}
}
func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption {
return func(s *Server) error {
s.primaryTranslateStore = store
return nil
}
}
func OptServerStatsClient(sc StatsClient) ServerOption {
return func(s *Server) error {
s.Holder.Stats = sc
@ -198,6 +209,16 @@ func OptServerURI(uri *URI) ServerOption {
}
}
// OptClusterDisabled tells the server whether to use a static cluster with the
// defined hosts. Mostly used for testing.
func OptServerClusterDisabled(disabled bool, hosts []string) ServerOption {
return func(s *Server) error {
s.hosts = hosts
s.clusterDisabled = disabled
return nil
}
}
// NewServer returns a new instance of Server.
func NewServer(opts ...ServerOption) (*Server, error) {
s := &Server{
@ -241,11 +262,20 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.Cluster.Logger = s.logger
s.Cluster.Holder = s.Holder
// Initialize translation database.
s.TranslateFile = NewTranslateFile()
s.TranslateFile.Path = filepath.Join(path, "keys")
s.TranslateFile.PrimaryTranslateStore = s.primaryTranslateStore
if err := s.TranslateFile.Open(); err != nil {
return nil, err
}
// update URI port with actual listener port. TODO this should probably be done outside of here.
if s.URI.Port() == 0 {
s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port))
}
// Get or create NodeID.
s.NodeID = s.LoadNodeID()
// Set Cluster Node.
node := &Node{
@ -254,13 +284,23 @@ func NewServer(opts ...ServerOption) (*Server, error) {
IsCoordinator: s.Cluster.Coordinator == s.NodeID,
}
s.Cluster.Node = node
if s.clusterDisabled {
err := s.Cluster.setStatic(s.hosts)
if err != nil {
return nil, errors.Wrap(err, "setting cluster static")
}
}
// Append the NodeID tag to stats.
s.Holder.Stats = s.Holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID))
s.executor.Holder = s.Holder
s.executor.Node = node
s.executor.Cluster = s.Cluster
s.executor.TranslateStore = s.TranslateFile
s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
s.handler.GetAPI().Executor = s.executor
s.handler.GetAPI().TranslateStore = s.TranslateFile
return s, nil
}
@ -268,9 +308,8 @@ func NewServer(opts ...ServerOption) (*Server, error) {
// Open opens and initializes the server.
func (s *Server) Open() error {
s.logger.Printf("open server")
// s.ln can be configured prior to Open() via s.OpenListener().
if s.ln == nil {
return errors.New("Must pass a listener option to NewServer")
return errors.New("must pass a listener option to NewServer")
}
// Log startup
@ -279,14 +318,6 @@ func (s *Server) Open() error {
log.Println(errors.Wrap(err, "logging startup"))
}
// Get or create NodeID.
// Append the NodeID tag to stats.
// Create default HTTP client
// Create executor for executing queries.
// Cluster settings.
s.Cluster.Broadcaster = s.Broadcaster
s.Cluster.MaxWritesPerRequest = s.maxWritesPerRequest
@ -354,6 +385,9 @@ func (s *Server) Close() error {
if s.Holder != nil {
s.Holder.Close()
}
if s.TranslateFile != nil {
s.TranslateFile.Close()
}
return nil
}
@ -524,7 +558,7 @@ func (s *Server) SendSync(pb proto.Message) error {
// SendAsync represents an implementation of Broadcaster.
func (s *Server) SendAsync(pb proto.Message) error {
return s.Gossiper.SendAsync(pb)
return ErrNotImplemented
}
// SendTo represents an implementation of Broadcaster.

5
server/config.go
View file

@ -78,6 +78,11 @@ type Config struct {
// Gossip config is based around memberlist.Config.
Gossip gossip.Config `toml:"gossip"`
// Translation config supports translation store replication.
Translation struct {
PrimaryURL string `toml:"primary-url"`
}
AntiEntropy struct {
Interval toml.Duration `toml:"interval"`
} `toml:"anti-entropy"`

28
server/server.go
View file

@ -217,6 +217,12 @@ func (m *Command) SetupServer() error {
c := http.GetHTTPClient(TLSConfig)
// Setup connection to primary store if this is a replica.
var primaryTranslateStore pilosa.TranslateStore
if m.Config.Translation.PrimaryURL != "" {
primaryTranslateStore = http.NewTranslateStore(m.Config.Translation.PrimaryURL)
}
m.Server, err = pilosa.NewServer(
pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)),
pilosa.OptServerLongQueryTime(time.Duration(m.Config.Cluster.LongQueryTime)),
@ -235,6 +241,8 @@ func (m *Command) SetupServer() error {
pilosa.OptServerListener(ln),
pilosa.OptServerURI(uri),
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore),
pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts),
)
return errors.Wrap(err, "new server")
@ -242,26 +250,7 @@ func (m *Command) SetupServer() error {
// SetupNetworking sets up internode communication based on the configuration.
func (m *Command) SetupNetworking() error {
m.Server.NodeID = m.Server.LoadNodeID()
if m.Config.Cluster.Disabled {
m.Server.Cluster.Static = true
m.Server.Cluster.Coordinator = m.Server.NodeID
for _, address := range m.Config.Cluster.Hosts {
uri, err := pilosa.NewURIFromAddress(address)
if err != nil {
return errors.Wrap(err, "getting URI")
}
m.Server.Cluster.Nodes = append(m.Server.Cluster.Nodes, &pilosa.Node{
URI: *uri,
})
}
m.Server.Broadcaster = pilosa.NopBroadcaster
m.Server.Cluster.MemberSet = pilosa.NewStaticMemberSet(m.Server.Cluster.Nodes)
m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver
m.Server.Gossiper = pilosa.NopGossiper
return nil
}
@ -306,7 +295,6 @@ func (m *Command) SetupNetworking() error {
m.Server.Cluster.MemberSet = gossipMemberSet
m.Server.Broadcaster = m.Server
m.Server.BroadcastReceiver = gossipMemberSet
m.Server.Gossiper = gossipMemberSet
return nil
}

10
statik/statik.go Normal file
View file

File diff suppressed because one or more lines are too long

90
test/attr.go
View file

@ -1,90 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package test
import (
"io/ioutil"
"os"
"runtime"
"sync"
"testing"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/boltdb"
)
// AttrStore represents a test wrapper for pilosa.AttrStore.
type AttrStore struct {
pilosa.AttrStore
}
// NewAttrStore returns a new instance of AttrStore.
func NewAttrStore(string) pilosa.AttrStore {
f, err := ioutil.TempFile("", "pilosa-attr-")
if err != nil {
panic(err)
}
f.Close()
os.Remove(f.Name())
return &AttrStore{boltdb.NewAttrStore(f.Name())}
}
func BenchmarkAttrStore_Duplicate(b *testing.B) {
s := MustOpenAttrStore()
defer s.Close()
// Set attributes.
const n = 5
for i := 0; i < n; i++ {
if err := s.SetAttrs(uint64(i), map[string]interface{}{"A": 100, "B": "foo", "C": true, "D": 100.2}); err != nil {
b.Fatal(err)
}
}
b.ReportAllocs()
b.ResetTimer()
// Update attributes with an existing subset.
cpuN := runtime.GOMAXPROCS(0)
var wg sync.WaitGroup
for i := 0; i < cpuN; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for j := 0; j < b.N/cpuN; j++ {
if err := s.SetAttrs(uint64(j%n), map[string]interface{}{"A": int64(100), "B": "foo", "D": 100.2}); err != nil {
b.Fatal(err)
}
}
}()
}
wg.Wait()
}
// MustOpenAttrStore returns a new, opened attribute store at a temporary path. Panic on error.
func MustOpenAttrStore() pilosa.AttrStore {
s := NewAttrStore("")
if err := s.Open(); err != nil {
panic(err)
}
return s
}
// Close closes the database and removes the underlying data.
func (s *AttrStore) Close() error {
defer os.RemoveAll(s.Path())
return s.AttrStore.Close()
}

35
test/client.go
View file

@ -1,35 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package test
import (
gohttp "net/http"
"github.com/pilosa/pilosa/http"
)
// Client represents a test wrapper for pilosa.Client.
type Client struct {
*http.InternalClient
}
// MustNewClient returns a new instance of Client. Panic on error.
func MustNewClient(host string, h *gohttp.Client) *Client {
c, err := http.NewInternalClient(host, h)
if err != nil {
panic(err)
}
return &Client{InternalClient: c}
}

2
test/executor.go
View file

@ -20,6 +20,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/inmem"
"github.com/pilosa/pilosa/pql"
)
@ -42,6 +43,7 @@ func NewExecutor(holder *pilosa.Holder, cluster *pilosa.Cluster) *Executor {
e := &Executor{Executor: executor}
e.Holder = holder
e.Cluster = cluster
e.TranslateStore = inmem.NewTranslateStore()
e.Node = cluster.Nodes[0]
return e
}

10
test/frame.go → test/field.go
View file

@ -18,7 +18,6 @@ import (
"io/ioutil"
"os"
"testing"
"time"
"github.com/pilosa/pilosa"
)
@ -75,15 +74,6 @@ func (f *Field) Reopen() error {
return nil
}
// MustSetBit sets a bit on the field. Panic on error.
func (f *Field) MustSetBit(view string, rowID, columnID uint64, t *time.Time) (changed bool) {
changed, err := f.SetBit(view, rowID, columnID, t)
if err != nil {
panic(err)
}
return changed
}
// Ensure field can set its cache
func TestField_SetCacheSize(t *testing.T) {
f := MustOpenField()

4
test/holder.go
View file

@ -142,7 +142,7 @@ func (h *Holder) SetBit(index, field string, rowID, columnID uint64) {
if err != nil {
panic(err)
}
f.SetBit(pilosa.ViewStandard, rowID, columnID, nil)
f.SetBit(rowID, columnID, nil)
}
// ClearBit clears a bit on the given field.
@ -152,7 +152,7 @@ func (h *Holder) ClearBit(index, field string, rowID, columnID uint64) {
if err != nil {
panic(err)
}
f.ClearBit(pilosa.ViewStandard, rowID, columnID, nil)
f.ClearBit(rowID, columnID, nil)
}
// MustSetBits sets columns on a row. Panic on error.

1006
translate.go Normal file
View file

File diff suppressed because it is too large Load diff

565
translate_test.go Normal file
View file

@ -0,0 +1,565 @@
package pilosa_test
import (
"bufio"
"context"
"fmt"
"io/ioutil"
"math/rand"
"os"
"reflect"
"strconv"
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/pilosa/pilosa"
)
func TestTranslateFile_TranslateColumn(t *testing.T) {
s := MustOpenTranslateFile()
defer s.MustClose()
// First translation should start id at zero.
if ids, err := s.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Next translation on the same index should move to one.
if ids, err := s.TranslateColumnsToUint64("IDX0", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{2}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Translation on a different index restarts at 0.
if ids, err := s.TranslateColumnsToUint64("IDX1", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Ensure that string values can be looked up by ID.
if value, err := s.TranslateColumnToString("IDX0", 2); err != nil {
t.Fatal(err)
} else if value != "bar" {
t.Fatalf("unexpected value: %s", value)
}
// Ensure that non-existent values return "".
if value, err := s.TranslateColumnToString("IDX0", 1000); err != nil {
t.Fatal(err)
} else if value != "" {
t.Fatalf("unexpected value: %s", value)
}
// Reopen the store.
if err := s.Reopen(); err != nil {
t.Fatal(err)
}
// Ensure translation is still correct after reopen.
if ids, err := s.TranslateColumnsToUint64("IDX1", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Ensure translation is still correct after reopen.
if value, err := s.TranslateColumnToString("IDX0", 2); err != nil {
t.Fatal(err)
} else if value != "bar" {
t.Fatalf("unexpected value: %s", value)
}
// Next translation on the same index should move to one.
if ids, err := s.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{3}) {
t.Fatalf("unexpected id: %#v", ids)
}
}
func TestTranslateFile_TranslateColumn_Large(t *testing.T) {
s := MustOpenTranslateFile()
defer s.MustClose()
// Generate key/values.
for i := 0; i < 1000000; i += 1000 {
keys := make([]string, 1000)
for j := 0; j < 1000; j++ {
keys[j] = strconv.Itoa(i + j + 1)
}
ids, err := s.TranslateColumnsToUint64("IDX0", keys)
if err != nil {
t.Fatal(err)
}
for j, id := range ids {
if exp := uint64(i + j + 1); id != exp {
t.Fatalf("unexpected id: got=%d, exp=%d", id, exp)
}
}
}
// Verify values can be returned.
for i := 0; i < 1000000; i++ {
exp := strconv.Itoa(i + 1)
if key, err := s.TranslateColumnToString("IDX0", uint64(i+1)); err != nil {
t.Fatal(err)
} else if key != exp {
t.Fatalf("unexpected key: got=%q, exp=%q", key, exp)
}
}
// Reopen and re-verify.
if err := s.Reopen(); err != nil {
t.Fatal(err)
}
for i := 0; i < 1000000; i++ {
exp := strconv.Itoa(i + 1)
if key, err := s.TranslateColumnToString("IDX0", uint64(i+1)); err != nil {
t.Fatal(err)
} else if key != exp {
t.Fatalf("unexpected key: got=%q, exp=%q", key, exp)
}
}
}
func TestTranslateFile_TranslateRow(t *testing.T) {
s := MustOpenTranslateFile()
defer s.MustClose()
// First translation should start id at zero.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Next translation on the same index should move to one.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME0", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{2}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Translation on a different index restarts at 0.
if ids, err := s.TranslateRowsToUint64("IDX1", "FRAME0", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Translation on a different frame restarts at 0.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME1", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Ensure that string values can be looked up by ID.
if value, err := s.TranslateRowToString("IDX0", "FRAME0", 2); err != nil {
t.Fatal(err)
} else if value != "bar" {
t.Fatalf("unexpected value: %s", value)
}
// Ensure that non-existent values return blank.
if value, err := s.TranslateRowToString("IDX0", "FRAME0", 1000); err != nil {
t.Fatal(err)
} else if value != "" {
t.Fatalf("unexpected value: %s", value)
}
// Reopen the store.
if err := s.Reopen(); err != nil {
t.Fatal(err)
}
// Translation on a different frame restarts at 0.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME1", []string{"bar"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{1}) {
t.Fatalf("unexpected id: %#v", ids)
}
// Ensure that string values can be looked up by ID.
if value, err := s.TranslateRowToString("IDX0", "FRAME0", 2); err != nil {
t.Fatal(err)
} else if value != "bar" {
t.Fatalf("unexpected value: %s", value)
}
// Translate new row and increment sequence.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME0", []string{"baz"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{3}) {
t.Fatalf("unexpected id: %#v", ids)
}
}
func TestTranslateFile_TranslateRow_Large(t *testing.T) {
s := MustOpenTranslateFile()
defer s.MustClose()
// Generate key/values.
for i := 0; i < 1000000; i += 1000 {
keys := make([]string, 1000)
for j := 0; j < 1000; j++ {
keys[j] = strconv.Itoa(i + j + 1)
}
ids, err := s.TranslateRowsToUint64("IDX0", "FRAME0", keys)
if err != nil {
t.Fatal(err)
}
for j, id := range ids {
if exp := uint64(i + j + 1); id != exp {
t.Fatalf("unexpected id: got=%d, exp=%d", id, exp)
}
}
}
// Verify values can be returned.
for i := 0; i < 1000000; i++ {
exp := strconv.Itoa(i + 1)
if key, err := s.TranslateRowToString("IDX0", "FRAME0", uint64(i+1)); err != nil {
t.Fatal(err)
} else if key != exp {
t.Fatalf("unexpected key: got=%q, exp=%q", key, exp)
}
}
// Reopen and re-verify.
if err := s.Reopen(); err != nil {
t.Fatal(err)
}
for i := 0; i < 1000000; i++ {
exp := strconv.Itoa(i + 1)
if key, err := s.TranslateRowToString("IDX0", "FRAME0", uint64(i+1)); err != nil {
t.Fatal(err)
} else if key != exp {
t.Fatalf("unexpected key: got=%q, exp=%q", key, exp)
}
}
}
func TestTranslateFile_Reader(t *testing.T) {
t.Run("NoOffset", func(t *testing.T) {
s := MustOpenTranslateFile()
defer s.MustClose()
if _, err := s.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if _, err := s.TranslateRowsToUint64("IDX0", "FRAME0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
rc, err := s.Reader(context.Background(), 0)
if err != nil {
t.Fatal(err)
}
brc := bufio.NewReader(rc)
defer rc.Close()
// Read first entry. Should read 'entry length' (13) plus uvarint(size) (1) = 14b.
var entry pilosa.LogEntry
if n, err := entry.ReadFrom(brc); err != nil {
t.Fatal(err)
} else if n != 14 {
t.Fatalf("unexpected n: %d", n)
} else if diff := cmp.Diff(entry, pilosa.LogEntry{
Type: pilosa.LogEntryTypeInsertColumn,
Index: []byte("IDX0"),
IDs: []uint64{1},
Keys: [][]byte{[]byte("foo")},
Length: 13,
}); diff != "" {
t.Fatal(diff)
}
// Read second entry.
if _, err := entry.ReadFrom(brc); err != nil {
t.Fatal(err)
} else if diff := cmp.Diff(entry, pilosa.LogEntry{
Type: pilosa.LogEntryTypeInsertRow,
Index: []byte("IDX0"),
Frame: []byte("FRAME0"),
IDs: []uint64{1, 2},
Keys: [][]byte{[]byte("bar"), []byte("baz")},
Length: 24,
}); diff != "" {
t.Fatal(diff)
}
// Write new entry.
if _, err := s.TranslateColumnsToUint64("IDX0", []string{"xyz"}); err != nil {
t.Fatal(err)
}
// Read new entry.
if _, err := entry.ReadFrom(brc); err != nil {
t.Fatal(err)
} else if diff := cmp.Diff(entry, pilosa.LogEntry{
Type: pilosa.LogEntryTypeInsertColumn,
Index: []byte("IDX0"),
IDs: []uint64{2},
Keys: [][]byte{[]byte("xyz")},
Length: 13,
}); diff != "" {
t.Fatal(diff)
}
// Close reader and ensure it returns EOF.
if err := rc.Close(); err != nil {
t.Fatal(err)
} else if _, err := entry.ReadFrom(brc); err != pilosa.ErrTranslateStoreReaderClosed {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("WithOffset", func(t *testing.T) {
s := MustOpenTranslateFile()
defer s.MustClose()
if _, err := s.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if _, err := s.TranslateRowsToUint64("IDX0", "FRAME0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
// Start offset after the first entry.
rc, err := s.Reader(context.Background(), 14)
if err != nil {
t.Fatal(err)
}
brc := bufio.NewReader(rc)
defer rc.Close()
// This should be the second entry.
var entry pilosa.LogEntry
if _, err := entry.ReadFrom(brc); err != nil {
t.Fatal(err)
} else if diff := cmp.Diff(entry, pilosa.LogEntry{
Type: pilosa.LogEntryTypeInsertRow,
Index: []byte("IDX0"),
Frame: []byte("FRAME0"),
IDs: []uint64{1, 2},
Keys: [][]byte{[]byte("bar"), []byte("baz")},
Length: 24,
}); diff != "" {
t.Fatal(diff)
}
})
}
func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
// Create a primary store that accepts writes.
primary := MustOpenTranslateFile()
defer primary.MustClose()
// Create a replica that accepts writes from primary.
replica := NewTranslateFile()
replica.PrimaryTranslateStore = primary
if err := replica.Open(); err != nil {
t.Fatal(err)
}
defer replica.MustClose()
// Write to the primary.
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if _, err := primary.TranslateRowsToUint64("IDX0", "FRAME0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica have received writes.
if value, err := replica.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica.TranslateRowToString("IDX0", "FRAME0", 1); err != nil {
return err
} else if value != "bar" {
return fmt.Errorf("unexpected row 1 value: %s", value)
}
if value, err := replica.TranslateRowToString("IDX0", "FRAME0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Disconnect primary store & write more values.
if err := primary.Reopen(); err != nil {
t.Fatal(err)
} else if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica until write appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica.TranslateColumnToString("IDX0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected column 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Disconnect replica store & write more values.
if err := replica.Reopen(); err != nil {
t.Fatal(err)
} else if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foobar"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica until write appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica.TranslateColumnToString("IDX0", 3); err != nil {
return err
} else if value != "foobar" {
return fmt.Errorf("unexpected column 3 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
}
func BenchmarkTranslateFile_TranslateColumnsToUint64(b *testing.B) {
const batchSize = 1000
s := MustOpenTranslateFile()
defer s.MustClose()
// Generate keys before benchmark begins
keySets := make([][]string, b.N/batchSize)
for i := range keySets {
keySets[i] = make([]string, batchSize)
for j, jv := range rand.New(rand.NewSource(0)).Perm(batchSize) {
keySets[i][j] = fmt.Sprintf("%08d%08d", jv, i)
}
}
b.ResetTimer()
for _, keySet := range keySets {
if _, err := s.TranslateColumnsToUint64("IDX0", keySet); err != nil {
b.Fatal(err)
}
}
}
func BenchmarkTranslateFile_TranslateColumnToString(b *testing.B) {
const batchSize = 1000
s := MustOpenTranslateFile()
defer s.MustClose()
// Generate keys before benchmark begins
for i := 0; i < b.N; i += batchSize {
keySet := make([]string, batchSize)
for j, jv := range rand.New(rand.NewSource(0)).Perm(batchSize) {
keySet[j] = fmt.Sprintf("%08d%08d", jv, i)
}
if _, err := s.TranslateColumnsToUint64("IDX0", keySet); err != nil {
b.Fatal(err)
}
}
// Generate random key access.
perm := rand.New(rand.NewSource(0)).Perm(b.N)
b.ResetTimer()
for i := 0; i < b.N; i++ {
if _, err := s.TranslateColumnToString("IDX0", uint64(perm[i])); err != nil {
b.Fatal(err)
}
}
}
type TranslateFile struct {
*pilosa.TranslateFile
}
func NewTranslateFile() *TranslateFile {
f, err := ioutil.TempFile("", "")
if err != nil {
panic(err)
}
f.Close()
s := &TranslateFile{TranslateFile: pilosa.NewTranslateFile()}
s.Path = f.Name()
return s
}
func MustOpenTranslateFile() *TranslateFile {
s := NewTranslateFile()
if err := s.Open(); err != nil {
panic(err)
}
return s
}
func (s *TranslateFile) Close() error {
defer os.Remove(s.Path)
return s.TranslateFile.Close()
}
func (s *TranslateFile) MustClose() {
if err := s.Close(); err != nil {
panic(err)
}
}
// Reopen closes the store and opens a new instance of it for the same path.
func (s *TranslateFile) Reopen() error {
prev := s.TranslateFile
if err := s.TranslateFile.Close(); err != nil {
return err
}
s.TranslateFile = pilosa.NewTranslateFile()
s.Path = prev.Path
s.PrimaryTranslateStore = prev.PrimaryTranslateStore
if err := s.Open(); err != nil {
return err
}
return nil
}
// retryFor executes fn every 100ms until d time passes or until fn return nil.
func retryFor(d time.Duration, fn func() error) (err error) {
timer, ticker := time.NewTimer(d), time.NewTicker(100*time.Millisecond)
defer timer.Stop()
defer ticker.Stop()
for {
if err = fn(); err == nil {
return nil
}
select {
case <-timer.C:
return err
case <-ticker.C:
}
}
}

10
utils_internal_test.go
View file

@ -117,7 +117,7 @@ func (t *ClusterCluster) CreateField(index, field string, opt FieldOptions) erro
return nil
}
func (t *ClusterCluster) SetBit(index, field, view string, rowID, colID uint64, x *time.Time) error {
func (t *ClusterCluster) SetBit(index, field string, rowID, colID uint64, x *time.Time) error {
// Determine which node should receive the SetBit.
c0 := t.Clusters[0] // use the first node's cluster to determine slice location.
slice := colID / SliceWidth
@ -132,7 +132,7 @@ func (t *ClusterCluster) SetBit(index, field, view string, rowID, colID uint64,
if f == nil {
return fmt.Errorf("index/field does not exist: %s/%s", index, field)
}
_, err := f.SetBit(view, rowID, colID, x)
_, err := f.SetBit(rowID, colID, x)
if err != nil {
return err
}
@ -150,11 +150,11 @@ func (t *ClusterCluster) clusterByID(id string) *Cluster {
return nil
}
// AddNode adds a node to the cluster and (potentially) starts a resize job.
func (t *ClusterCluster) AddNode(saveTopology bool) error {
// addNode adds a node to the cluster and (potentially) starts a resize job.
func (t *ClusterCluster) addNode() error {
id := len(t.Clusters)
c, err := t.addCluster(id, saveTopology)
c, err := t.addCluster(id, false)
if err != nil {
return err
}

4
view.go
View file

@ -237,13 +237,13 @@ func (v *View) createFragmentIfNotExists(slice uint64) (*Fragment, error) {
v.maxSlice = slice
// Send the create slice message to all nodes.
err := v.broadcaster.SendAsync(
err := v.broadcaster.SendSync(
&internal.CreateSliceMessage{
Index: v.index,
Slice: slice,
})
if err != nil {
return nil, errors.Wrap(err, "sending message")
return nil, errors.Wrap(err, "sending createslice message")
}
}