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

Merge pull request #1337 from benbjohnson/translator

Browse source 
ID-Key Translation
This commit is contained in:
Ben Johnson 2018-06-18 08:26:27 -06:00 • committed by GitHub
commit c9ff679efc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
28 changed files with 2829 additions and 213 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

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.")

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

14
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()
@ -1038,6 +1049,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 +1087,7 @@ func encodeFieldOptions(o *FieldOptions) *internal.FieldOptions {
Min: o.Min,
Max: o.Max,
TimeQuantum: string(o.TimeQuantum),
Keys: o.Keys,
}
}
@ -1089,6 +1102,7 @@ func decodeFieldOptions(options *internal.FieldOptions) *FieldOptions {
Min: options.Min,
Max: options.Max,
TimeQuantum: TimeQuantum(options.TimeQuantum),
Keys: options.Keys,
}
}

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

52
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
}
@ -1184,6 +1186,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{}

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

@ -40,6 +40,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))
@ -100,6 +117,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))

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 {

31
server.go
View file

@ -52,10 +52,11 @@ 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
// External
handler Handler
@ -75,6 +76,8 @@ type Server struct {
diagnosticInterval time.Duration
maxWritesPerRequest int
primaryTranslateStore TranslateStore
defaultClient InternalClient
dataDir string
}
@ -169,6 +172,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
@ -241,6 +251,14 @@ 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))
@ -259,8 +277,10 @@ func NewServer(opts ...ServerOption) (*Server, error) {
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
}
@ -354,6 +374,9 @@ func (s *Server) Close() error {
if s.Holder != nil {
s.Holder.Close()
}
if s.TranslateFile != nil {
s.TranslateFile.Close()
}
return nil
}

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

7
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,7 @@ func (m *Command) SetupServer() error {
pilosa.OptServerListener(ln),
pilosa.OptServerURI(uri),
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore),
)
return errors.Wrap(err, "new server")

10
statik/statik.go Normal file
View file

File diff suppressed because one or more lines are too long

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
}

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