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

Merge branch 'master' into 1513-metalinter-megacheck

Browse source
This commit is contained in:
Cody Soyland 2018-08-02 10:29:47 -05:00 • committed by GitHub
commit dd7b5f517b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
20 changed files with 1373 additions and 1184 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
api.go
View file

@ -135,9 +135,9 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
}
// Translate column attributes, if necessary.
if api.server.primaryTranslateStore != nil {
if api.server.translateFile != nil {
for _, col := range resp.ColumnAttrSets {
v, err := api.server.primaryTranslateStore.TranslateColumnToString(req.Index, col.ID)
v, err := api.server.translateFile.TranslateColumnToString(req.Index, col.ID)
if err != nil {
return resp, err
}
@ -756,46 +756,17 @@ func (api *API) ResizeAbort() error {
return errors.Wrap(err, "complete current job")
}
// translateStoreBufferSize is the buffer size used for streaming data.
const translateStoreBufferSize = 65536
func (api *API) GetTranslateData(ctx context.Context, w io.WriteCloser, offset int64) error {
rc, err := api.server.primaryTranslateStore.Reader(ctx, offset)
// GetTranslateData provides a reader for key translation logs starting at offset.
func (api *API) GetTranslateData(ctx context.Context, offset int64) (io.ReadCloser, error) {
rc, err := api.server.translateFile.Reader(ctx, offset)
if err != nil {
return errors.Wrap(err, "read from translate store")
return nil, errors.Wrap(err, "read from translate store")
}
// Ensure reader is closed when the client disconnects.
go func() { <-ctx.Done(); rc.Close() }()
go func() {
defer rc.Close()
defer w.Close()
buf := make([]byte, translateStoreBufferSize)
// Copy from reader to client until store or client disconnect.
for {
// Read from store.
n, err := rc.Read(buf)
if err == io.EOF {
return
} else if err != nil {
api.server.logger.Printf("api: translate store read error: %s", err)
return
} else if n == 0 {
continue
}
// Write to response & flush.
if _, err := w.Write(buf[:n]); err != nil {
api.server.logger.Printf("api: translate store response write error: %s", err)
return
}
}
}()
return nil
return rc, nil
}
// State returns the cluster state which is usually "NORMAL", but could be

4
docs/query-language.md
View file

@ -419,7 +419,7 @@ Count(Row(stargazer=1))
**Spec:**
```
TopN([ROW_CALL], <FIELD>, [n=UINT],
TopN(<FIELD>, [ROW_CALL], [n=UINT],
[attrName=<ATTR_NAME>, attrValues=<[]ATTR_VALUE>])
```
@ -468,7 +468,7 @@ TopN(stargazer, n=2)
Filter based on an existing row:
```request
TopN(Row(language=1), stargazer, n=2)
TopN(stargazer, Row(language=1), n=2)
```
```response
{"results":[[{"id":1240,"count":35},{"id":7508,"count":32}]]}

26
executor.go
View file

@ -101,9 +101,12 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
}
// 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
// No need to translate a remote call.
if !opt.Remote {
for i := range q.Calls {
if err := e.translateCall(index, idx, q.Calls[i]); err != nil {
return nil, err
}
}
}
@ -113,10 +116,13 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
}
// 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
// No need to translate a remote call.
if !opt.Remote {
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
@ -1256,7 +1262,7 @@ func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, cal
rowID, ok, err := c.UintArg("_" + rowLabel)
if err != nil {
return nil, fmt.Errorf("reading SetRowAttrs() row: %v", rowLabel)
return nil, errors.Wrap(err, "reading SetRowAttrs() row")
} else if !ok {
return nil, fmt.Errorf("SetRowAttrs row field '%v' required", rowLabel)
}
@ -1550,6 +1556,10 @@ func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
colKey = "_" + columnLabel
fieldName, _ = c.FieldArg()
rowKey = fieldName
} else if c.Name == "SetRowAttrs" {
// Positional args in new PQL syntax require special handling here.
rowKey = "_" + rowLabel
fieldName = callArgString(c, "_field")
} else {
colKey = "col"
fieldName = callArgString(c, "field")

64
executor_test.go
View file

@ -471,29 +471,53 @@ func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
t.Fatal(err)
} else if _, err := index.CreateFieldIfNotExists("xxx", pilosa.OptFieldTypeDefault()); err != nil {
t.Fatal(err)
}
// Set two attrs on f/10.
// Also set attrs on other bitmaps and fields to test isolation.
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(f, 10, foo="bar")`}); err != nil {
t.Fatal(err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(f, 200, YYY=1)`}); err != nil {
t.Fatal(err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(xxx, 10, YYY=1)`}); err != nil {
t.Fatal(err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(f, 10, baz=123, bat=true)`}); err != nil {
} else if _, err := index.CreateFieldIfNotExists("kf", pilosa.OptFieldTypeDefault(), pilosa.OptFieldKeys()); err != nil {
t.Fatal(err)
}
f := hldr.Field("i", "f")
if m, err := f.RowAttrStore().Attrs(10); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(m, map[string]interface{}{"foo": "bar", "baz": int64(123), "bat": true}) {
t.Fatalf("unexpected bitmap attr: %#v", m)
}
t.Run("rowID", func(t *testing.T) {
// Set two attrs on f/10.
// Also set attrs on other bitmaps and fields to test isolation.
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(f, 10, foo="bar")`}); err != nil {
t.Fatal(err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(f, 200, YYY=1)`}); err != nil {
t.Fatal(err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(xxx, 10, YYY=1)`}); err != nil {
t.Fatal(err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(f, 10, baz=123, bat=true)`}); err != nil {
t.Fatal(err)
}
f := hldr.Field("i", "f")
if m, err := f.RowAttrStore().Attrs(10); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(m, map[string]interface{}{"foo": "bar", "baz": int64(123), "bat": true}) {
t.Fatalf("unexpected bitmap attr: %#v", m)
}
})
t.Run("rowKey", func(t *testing.T) {
// Set two attrs on f/10.
// Also set attrs on other bitmaps and fields to test isolation.
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(kf, "row10", foo="bar")`}); err != nil {
t.Fatal(err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(kf, "row200", YYY=1)`}); err != nil {
t.Fatal(err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetRowAttrs(kf, "row10", baz=123, bat=true)`}); err != nil {
t.Fatal(err)
}
if result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(kf="row10")`}); err != nil {
t.Fatal(err)
} else if attrs := result.Results[0].(*pilosa.Row).Attrs; !reflect.DeepEqual(attrs, map[string]interface{}{"foo": "bar", "baz": int64(123), "bat": true}) {
t.Fatalf("unexpected attrs: %+v", attrs)
}
})
}
// Ensure a TopN() query can be executed.

3
holder.go
View file

@ -248,8 +248,7 @@ func (h *Holder) limitedSchema() []*IndexInfo {
func (h *Holder) applySchema(schema *Schema) error {
// Create indexes that don't exist.
for _, index := range schema.Indexes {
opt := IndexOptions{}
idx, err := h.CreateIndexIfNotExists(index.Name, opt)
idx, err := h.CreateIndexIfNotExists(index.Name, index.Options)
if err != nil {
return errors.Wrap(err, "creating index")
}

31
http/handler.go
View file

@ -1306,14 +1306,14 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques
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)
pipeR, pipeW := io.Pipe()
err := h.api.GetTranslateData(r.Context(), pipeW, offset)
rdr, err := h.api.GetTranslateData(r.Context(), offset)
if err != nil {
if errors.Cause(err) == pilosa.ErrNotImplemented {
http.Error(w, err.Error(), http.StatusNotImplemented)
@ -1329,7 +1329,28 @@ func (h *Handler) handleGetTranslateData(w http.ResponseWriter, r *http.Request)
w.Flush()
}
io.Copy(w, pipeR)
// Copy from reader to client until store or client disconnect.
buf := make([]byte, translateStoreBufferSize)
for {
// Read from store.
n, err := rdr.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 {

5
http/handler_internal_test.go
View file

@ -18,6 +18,7 @@ import (
"bytes"
"encoding/json"
"reflect"
"strings"
"testing"
"github.com/pilosa/pilosa"
@ -68,7 +69,7 @@ func TestPostFieldRequestUnmarshalJSON(t *testing.T) {
err string
}{
{json: `{"options": {}}`, expected: postFieldRequest{}},
{json: `{"options": 4}`, err: "json: cannot unmarshal number into Go struct field postFieldRequest.options of type http.fieldOptions"},
{json: `{"options": 4}`, err: "json: cannot unmarshal number"},
{json: `{"option": {}}`, err: `json: unknown field "option"`},
{json: `{"options": {"badKey": "test"}}`, err: `json: unknown field "badKey"`},
{json: `{"options": {"inverseEnabled": true}}`, err: `json: unknown field "inverseEnabled"`},
@ -81,7 +82,7 @@ func TestPostFieldRequestUnmarshalJSON(t *testing.T) {
dec.DisallowUnknownFields()
err := dec.Decode(actual)
if err != nil {
if test.err == "" || test.err != err.Error() {
if test.err == "" || !strings.HasPrefix(err.Error(), test.err) {
t.Errorf("test %d: expected error: %v, but got result: %v", i, test.err, err)
}
}

2
http/translator.go
View file

@ -16,7 +16,7 @@ import (
// Ensure implementation implements inteface.
var _ pilosa.TranslateStore = (*translateStore)(nil)
// translateStore represents an implementation of translateStore that
// translateStore represents an implementation of pilosa.TranslateStore that
// communicates over HTTP. This is used with the TranslateHandler.
type translateStore struct {
URL string

94
http/translator_test.go
View file

@ -2,9 +2,9 @@ package http_test
import (
"context"
"fmt"
"io"
"io/ioutil"
gohttp "net/http"
"testing"
"time"
@ -30,74 +30,53 @@ 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
}
}
closeInvoked := make(chan struct{})
mrc.CloseFunc = func() error {
close(closeInvoked)
return nil
// This test is currently flawed, breaking intermittently with message:
// "translator_test.go:65: unexpected EOF"
t.Skip()
primary := test.MustRunCluster(t, 1)[0]
hldr := test.Holder{Holder: primary.Server.Holder()}
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{Keys: true})
_, err := index.CreateField("f", pilosa.OptFieldTypeDefault())
if err != nil {
t.Fatal(err)
}
// Setup handler on test server.
var translateStore mock.TranslateStore
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
// Check context to make sure this is the call we are looking for.
// (Something else calls ReaderFunc on server startup)
if ctx.Value(gohttp.ServerContextKey) != nil {
if off != 100 {
t.Fatalf("unexpected off: %d", off)
}
return &mrc, nil
}
return newMockReadCloser(), nil
// Set data on the primary node.
if _, err := primary.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `` +
fmt.Sprintf("Set(%s, f=%d)\n", `"foo"`, 10) +
fmt.Sprintf("Set(%s, f=%d)\n", `"bar"`, 10) +
fmt.Sprintf("Set(%s, f=%d)\n", `"baz"`, 10),
}); err != nil {
t.Fatal(err)
}
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
// Connect to server and stream all available data.
store := http.NewTranslateStore(main.URL())
store := http.NewTranslateStore(primary.URL())
// Wait to ensure writes make it to translate store
time.Sleep(500 * time.Millisecond)
rc, err := store.Reader(context.Background(), 11) // offset=11 skips the first entry: \n\x01\x01i\x00\x01\x01\x03foo
// Close the primary to disconnect reader.
primary.Close()
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` {
} else if string(data) != "\n\x01\x01i\x00\x01\x02\x03bar\n\x01\x01i\x00\x01\x03\x03baz" {
t.Fatalf("unexpected data: %q", data)
} else if err := rc.Close(); err != nil {
t.Fatal(err)
}
select {
case <-time.NewTimer(time.Millisecond * 100).C:
t.Fatal("expected server close")
case <-closeInvoked:
return
}
})
// Ensure server closes store reader if client disconnects.
t.Run("ClientDisconnect", func(t *testing.T) {
t.Skip() // can't mock server from http package
// Setup mock so that Read() hangs.
done := make(chan struct{})
@ -121,14 +100,14 @@ func TestTranslateStore_Reader(t *testing.T) {
}
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0]
primary := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
defer primary.Close()
defer close(done)
// Connect to server and begin streaming.
ctx, cancel := context.WithCancel(context.Background())
store := http.NewTranslateStore(main.URL())
store := http.NewTranslateStore(primary.URL())
if _, err := store.Reader(ctx, 0); err != nil {
t.Fatal(err)
}
@ -146,16 +125,17 @@ func TestTranslateStore_Reader(t *testing.T) {
// Ensure client is notified if the server doesn't support streaming replication.
t.Run("ErrNotImplemented", func(t *testing.T) {
t.Skip() // can't mock server from http package
var translateStore mock.TranslateStore
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
return nil, pilosa.ErrNotImplemented
}
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
primary := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0]
defer primary.Close()
_, err := http.NewTranslateStore(main.URL()).Reader(context.Background(), 0)
_, err := http.NewTranslateStore(primary.URL()).Reader(context.Background(), 0)
if err != pilosa.ErrNotImplemented {
t.Fatalf("unexpected error: %s", err)
}

12
index.go
View file

@ -292,7 +292,7 @@ func (i *Index) CreateField(name string, opts ...FieldOption) (*Field, error) {
}
// CreateFieldIfNotExists creates a field with the given options if it doesn't exist.
func (i *Index) CreateFieldIfNotExists(name string, opts FieldOption) (*Field, error) {
func (i *Index) CreateFieldIfNotExists(name string, opts ...FieldOption) (*Field, error) {
i.mu.Lock()
defer i.mu.Unlock()
@ -301,11 +301,13 @@ func (i *Index) CreateFieldIfNotExists(name string, opts FieldOption) (*Field, e
return f, nil
}
// Apply functional option.
// Apply functional options.
fo := FieldOptions{}
err := opts(&fo)
if err != nil {
return nil, errors.Wrap(err, "applying option")
for _, opt := range opts {
err := opt(&fo)
if err != nil {
return nil, errors.Wrap(err, "applying option")
}
}
return i.createField(name, fo)

12
mock/translator.go
View file

@ -12,8 +12,8 @@ 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)
TranslateRowsToUint64Func func(index, field string, values []string) ([]uint64, error)
TranslateRowToStringFunc func(index, field string, values uint64) (string, error)
ReaderFunc func(ctx context.Context, off int64) (io.ReadCloser, error)
}
@ -25,12 +25,12 @@ func (s TranslateStore) TranslateColumnToString(index string, values uint64) (st
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) TranslateRowsToUint64(index, field string, values []string) ([]uint64, error) {
return s.TranslateRowsToUint64Func(index, field, values)
}
func (s TranslateStore) TranslateRowToString(index, frame string, value uint64) (string, error) {
return s.TranslateRowToStringFunc(index, frame, value)
func (s TranslateStore) TranslateRowToString(index, field string, value uint64) (string, error) {
return s.TranslateRowToStringFunc(index, field, value)
}
func (s TranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {

6
pql/pql.peg
View file

@ -7,7 +7,7 @@ type PQL Peg {
Calls <- sp (Call sp)* !.
Call <- 'Set' {p.startCall("Set")} open col comma args (comma timestamp)? close {p.endCall()}
/ 'SetRowAttrs' {p.startCall("SetRowAttrs")} open posfield comma uintrow comma args close {p.endCall()}
/ 'SetRowAttrs' {p.startCall("SetRowAttrs")} open posfield comma row comma args close {p.endCall()}
/ 'SetColumnAttrs' {p.startCall("SetColumnAttrs")} open col comma args close {p.endCall()}
/ 'Clear' {p.startCall("Clear")} open col comma args close {p.endCall()}
/ 'TopN' {p.startCall("TopN")} open posfield (comma allargs)? close {p.endCall()}
@ -60,6 +60,10 @@ col <- ( <uint> {p.addPosNum("_col", buffer[begin:end])}
/ '\'' <singlequotedstring> '\'' {p.addPosStr("_col", buffer[begin:end])}
/ '"' <doublequotedstring> '"' {p.addPosStr("_col", buffer[begin:end])}
)
row <- ( <uint> {p.addPosNum("_row", buffer[begin:end])}
/ '\'' <singlequotedstring> '\'' {p.addPosStr("_row", buffer[begin:end])}
/ '"' <doublequotedstring> '"' {p.addPosStr("_row", buffer[begin:end])}
)
open <- '(' sp
close <- ')' sp

2040
pql/pql.peg.go
View file

File diff suppressed because it is too large Load diff

58
pql/pqlpeg_test.go
View file

@ -160,6 +160,14 @@ func TestPEGWorking(t *testing.T) {
name: "SetRowAttrs2args",
input: "SetRowAttrs(blah, 9, a=47, b=bval)",
ncalls: 1},
{
name: "SetRowAttrsWithRowKeySingleQuote",
input: "SetRowAttrs(blah, 'rowKey', a=47)",
ncalls: 1},
{
name: "SetRowAttrsWithRowKeyDoubleQuote",
input: `SetRowAttrs(blah, "rowKey", a=47)`,
ncalls: 1},
{
name: "SetColumnAttrs",
input: "SetColumnAttrs(9, a=47)",
@ -168,6 +176,14 @@ func TestPEGWorking(t *testing.T) {
name: "SetColumnAttrs2args",
input: "SetColumnAttrs(9, a=47, b=bval)",
ncalls: 1},
{
name: "SetColumnAttrsWithColKeySingleQuote",
input: "SetColumnAttrs('colKey', a=47)",
ncalls: 1},
{
name: "SetColumnAttrsWithColKeyDoubleQuote",
input: `SetColumnAttrs("colKey", a=47)`,
ncalls: 1},
{
name: "Clear",
input: "Clear(1, a=53)",
@ -330,6 +346,28 @@ func TestPQLDeepEquality(t *testing.T) {
"_row": int64(9),
},
}},
{
name: "SetRowAttrsWithRowKeySingleQuote",
call: "SetRowAttrs(myfield, 'rowKey', z=4)",
exp: &Call{
Name: "SetRowAttrs",
Args: map[string]interface{}{
"z": int64(4),
"_field": "myfield",
"_row": "rowKey",
},
}},
{
name: "SetRowAttrsWithRowKeyDoubleQuote",
call: `SetRowAttrs(myfield, "rowKey", z=4)`,
exp: &Call{
Name: "SetRowAttrs",
Args: map[string]interface{}{
"z": int64(4),
"_field": "myfield",
"_row": "rowKey",
},
}},
{
name: "SetColumnAttrs",
call: "SetColumnAttrs(9, z=4)",
@ -340,6 +378,26 @@ func TestPQLDeepEquality(t *testing.T) {
"_col": int64(9),
},
}},
{
name: "SetColumnAttrsWithColKeySingleQuote",
call: "SetColumnAttrs('colKey', z=4)",
exp: &Call{
Name: "SetColumnAttrs",
Args: map[string]interface{}{
"z": int64(4),
"_col": "colKey",
},
}},
{
name: "SetColumnAttrsWithColKeyDoubleQuote",
call: `SetColumnAttrs("colKey", z=4)`,
exp: &Call{
Name: "SetColumnAttrs",
Args: map[string]interface{}{
"z": int64(4),
"_col": "colKey",
},
}},
{
name: "Clear",
call: "Clear(1, a=7)",

4
server.go
View file

@ -457,8 +457,8 @@ func (s *Server) receiveMessage(m Message) error {
}
idx.setRemoteMaxShard(obj.Shard)
case *CreateIndexMessage:
opt := IndexOptions{}
_, err := s.holder.CreateIndex(obj.Index, opt)
opt := obj.Meta
_, err := s.holder.CreateIndex(obj.Index, *opt)
if err != nil {
return err
}

2
server/config.go
View file

@ -74,7 +74,7 @@ type Config struct {
// Translation config supports translation store replication.
Translation struct {
PrimaryURL string `toml:"primary-url"`
}
} `toml:"translation"`
AntiEntropy struct {
Interval toml.Duration `toml:"interval"`

51
server/handler_test.go
View file

@ -24,6 +24,7 @@ import (
"reflect"
"strings"
"testing"
"time"
gohttp "net/http"
@ -655,6 +656,56 @@ func TestHandler_Endpoints(t *testing.T) {
})
}
func TestClusterTranslator(t *testing.T) {
cluster := make(test.Cluster, 2)
cluster[0] = test.NewCommandNode(true)
cluster[0].Config.Gossip.Port = "0"
cluster[0].Start()
httpTranslateStore := http.NewTranslateStore(cluster[0].URL())
cluster[1] = test.NewCommandNode(false,
server.OptCommandServerOptions(
pilosa.OptServerPrimaryTranslateStore(httpTranslateStore),
),
)
cluster[1].Config.Gossip.Port = "0"
cluster[1].Config.Gossip.Seeds = []string{cluster[0].GossipAddress()}
cluster[1].Start()
test.MustDo("POST", cluster[0].URL()+"/index/i0", "{\"options\": {\"keys\": true}}")
test.MustDo("POST", cluster[0].URL()+"/index/i0/field/f0", "{\"options\": {\"keys\": true}}")
test.MustDo("POST", cluster[0].URL()+"/index/i0/query", "Set(\"foo\", f0=\"bar\")")
// wait for key to replicate to second node
time.Sleep(500 * time.Millisecond)
result0 := test.MustDo("POST", cluster[0].URL()+"/index/i0/query", "Row(f0=\"bar\")").Body
result1 := test.MustDo("POST", cluster[1].URL()+"/index/i0/query", "Row(f0=\"bar\")").Body
if result0 != result1 {
t.Fatalf("`%s` != `%s`", result0, result1)
}
for _, i := range []string{result0, result1} {
var resp map[string]interface{}
err := json.Unmarshal([]byte(i), &resp)
if err != nil {
t.Fatalf("json unmarshal error: %s", err)
}
if results, ok := resp["results"].([]interface{}); ok {
if result, ok := results[0].(map[string]interface{}); ok {
if keys, ok := result["keys"].([]interface{}); ok {
if key, ok := keys[0].(string); ok {
if key != "foo" {
t.Fatalf("Key is %s but should be 'foo'", key)
}
}
}
}
}
}
}
func mustJSONDecode(t *testing.T, r io.Reader) (ret map[string]interface{}) {
dec := json.NewDecoder(r)
err := dec.Decode(&ret)

4
test/index.go
View file

@ -83,8 +83,8 @@ func (i *Index) CreateField(name string, opts ...pilosa.FieldOption) (*Field, er
}
// CreateFieldIfNotExists creates a field with the given options if it doesn't exist.
func (i *Index) CreateFieldIfNotExists(name string, opts pilosa.FieldOption) (*Field, error) {
f, err := i.Index.CreateFieldIfNotExists(name, opts)
func (i *Index) CreateFieldIfNotExists(name string, opts ...pilosa.FieldOption) (*Field, error) {
f, err := i.Index.CreateFieldIfNotExists(name, opts...)
if err != nil {
return nil, err
}

54
translate.go
View file

@ -31,7 +31,7 @@ var (
ErrTranslateStoreClosed = errors.New("pilosa: translate store closed")
ErrTranslateStoreReaderClosed = errors.New("pilosa: translate store reader closed")
ErrReplicationNotSupported = errors.New("pilosa: replication not supported")
ErrTranslateStoreReadOnly = errors.New("pilosa: operation not supported, translate store read only")
ErrTranslateStoreReadOnly = errors.New("pilosa: translate store could not find or create key, translate store read only")
)
// TranslateStore is the storage for translation string-to-uint64 values.
@ -39,8 +39,8 @@ type TranslateStore interface {
TranslateColumnsToUint64(index string, values []string) ([]uint64, error)
TranslateColumnToString(index string, values uint64) (string, error)
TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error)
TranslateRowToString(index, frame string, values uint64) (string, error)
TranslateRowsToUint64(index, field string, values []string) ([]uint64, error)
TranslateRowToString(index, field string, values uint64) (string, error)
// Returns a reader from the given offset of the raw data file.
// The returned reader must be closed by the caller when done.
@ -64,7 +64,7 @@ type TranslateFile struct {
closing chan struct{}
cols map[string]*index
rows map[frameKey]*index
rows map[fieldKey]*index
Path string
mapSize int
@ -82,7 +82,7 @@ func NewTranslateFile() *TranslateFile {
writeNotify: make(chan struct{}),
closing: make(chan struct{}),
cols: make(map[string]*index),
rows: make(map[frameKey]*index),
rows: make(map[fieldKey]*index),
mapSize: defaultMapSize,
@ -201,7 +201,7 @@ func (s *TranslateFile) applyEntry(entry *LogEntry, offset int64) error {
idx = s.col(string(entry.Index))
case LogEntryTypeInsertRow:
idx = s.row(string(entry.Index), string(entry.Frame))
idx = s.row(string(entry.Index), string(entry.Field))
default:
return fmt.Errorf("enterprise.TranslateFile.applyEntry(): unknown log entry type: 0x%20x", entry.Type)
@ -318,11 +318,11 @@ func (s *TranslateFile) col(index string) *index {
return idx
}
func (s *TranslateFile) row(index, frame string) *index {
idx := s.rows[frameKey{index, frame}]
func (s *TranslateFile) row(index, field string) *index {
idx := s.rows[fieldKey{index, field}]
if idx == nil {
idx = newIndex(s.data)
s.rows[frameKey{index, frame}] = idx
s.rows[fieldKey{index, field}] = idx
}
return idx
}
@ -433,8 +433,8 @@ func (s *TranslateFile) TranslateColumnToString(index string, value uint64) (str
return "", nil
}
func (s *TranslateFile) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
key := frameKey{index, frame}
func (s *TranslateFile) TranslateRowsToUint64(index, field string, values []string) ([]uint64, error) {
key := fieldKey{index, field}
ret := make([]uint64, len(values))
@ -495,7 +495,7 @@ func (s *TranslateFile) TranslateRowsToUint64(index, frame string, values []stri
entry := &LogEntry{
Type: LogEntryTypeInsertRow,
Index: []byte(index),
Frame: []byte(frame),
Field: []byte(field),
IDs: make([]uint64, 0, len(values)),
Keys: make([][]byte, 0, len(values)),
}
@ -524,9 +524,9 @@ func (s *TranslateFile) TranslateRowsToUint64(index, frame string, values []stri
return ret, nil
}
func (s *TranslateFile) TranslateRowToString(index, frame string, id uint64) (string, error) {
func (s *TranslateFile) TranslateRowToString(index, field string, id uint64) (string, error) {
s.mu.RLock()
if idx := s.rows[frameKey{index, frame}]; idx != nil {
if idx := s.rows[fieldKey{index, field}]; idx != nil {
if ret, ok := idx.keyByID(id); ok {
s.mu.RUnlock()
return string(ret), nil
@ -548,7 +548,7 @@ func (s *TranslateFile) Reader(ctx context.Context, offset int64) (io.ReadCloser
type LogEntry struct {
Type uint8
Index []byte
Frame []byte
Field []byte
IDs []uint64
Keys [][]byte
@ -558,12 +558,12 @@ type LogEntry struct {
Length uint64
}
// headerSize returns the number of bytes required for size, type, index, frame, & pair count.
// headerSize returns the number of bytes required for size, type, index, field, & pair count.
func (e *LogEntry) headerSize() int64 {
sz := uVarintSize(e.Length) + // total entry length
1 + // type
uVarintSize(uint64(len(e.Index))) + len(e.Index) + // Index length and data
uVarintSize(uint64(len(e.Frame))) + len(e.Frame) + // Frame length and data
uVarintSize(uint64(len(e.Field))) + len(e.Field) + // Field length and data
uVarintSize(uint64(len(e.IDs))) // ID/Key pair count
return int64(sz)
}
@ -604,14 +604,14 @@ func (e *LogEntry) ReadFrom(r io.Reader) (_ int64, err error) {
}
}
// Read frame name.
// Read field name.
if sz, err := binary.ReadUvarint(br); err != nil {
return n64, err
} else if sz == 0 {
e.Frame = nil
e.Field = nil
} else {
e.Frame = make([]byte, sz)
if _, err := io.ReadFull(r, e.Frame); err != nil {
e.Field = make([]byte, sz)
if _, err := io.ReadFull(r, e.Field); err != nil {
return n64, err
}
}
@ -663,11 +663,11 @@ func (e *LogEntry) WriteTo(w io.Writer) (_ int64, err error) {
return 0, err
}
// Write frame name.
sz = binary.PutUvarint(b, uint64(len(e.Frame)))
// Write field name.
sz = binary.PutUvarint(b, uint64(len(e.Field)))
if _, err := buf.Write(b[:sz]); err != nil {
return 0, err
} else if _, err := buf.Write(e.Frame); err != nil {
} else if _, err := buf.Write(e.Field); err != nil {
return 0, err
}
@ -722,9 +722,9 @@ func validLogEntriesLen(p []byte) (n int) {
}
}
type frameKey struct {
type fieldKey struct {
index string
frame string
field string
}
const defaultLoadFactor = 90
@ -904,7 +904,7 @@ type translateFileReader struct {
closing chan struct{}
}
// newTranslateFileReader returns a new instance of TranslateFileReader.
// newTranslateFileReader returns a new instance of translateFileReader.
func newTranslateFileReader(ctx context.Context, store *TranslateFile, offset int64) *translateFileReader {
return &translateFileReader{
ctx: ctx,

42
translate_test.go
View file

@ -136,42 +136,42 @@ func TestTranslateFile_TranslateRow(t *testing.T) {
defer s.MustClose()
// First translation should start id at zero.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME0", []string{"foo"}); err != nil {
if ids, err := s.TranslateRowsToUint64("IDX0", "FIELD0", []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 {
if ids, err := s.TranslateRowsToUint64("IDX0", "FIELD0", []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 {
if ids, err := s.TranslateRowsToUint64("IDX1", "FIELD0", []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 {
// Translation on a different field restarts at 0.
if ids, err := s.TranslateRowsToUint64("IDX0", "FIELD1", []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 {
if value, err := s.TranslateRowToString("IDX0", "FIELD0", 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 {
if value, err := s.TranslateRowToString("IDX0", "FIELD0", 1000); err != nil {
t.Fatal(err)
} else if value != "" {
t.Fatalf("unexpected value: %s", value)
@ -182,22 +182,22 @@ func TestTranslateFile_TranslateRow(t *testing.T) {
t.Fatal(err)
}
// Translation on a different frame restarts at 0.
if ids, err := s.TranslateRowsToUint64("IDX0", "FRAME1", []string{"bar"}); err != nil {
// Translation on a different field restarts at 0.
if ids, err := s.TranslateRowsToUint64("IDX0", "FIELD1", []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 {
if value, err := s.TranslateRowToString("IDX0", "FIELD0", 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 {
if ids, err := s.TranslateRowsToUint64("IDX0", "FIELD0", []string{"baz"}); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, []uint64{3}) {
t.Fatalf("unexpected id: %#v", ids)
@ -215,7 +215,7 @@ func TestTranslateFile_TranslateRow_Large(t *testing.T) {
keys[j] = strconv.Itoa(i + j + 1)
}
ids, err := s.TranslateRowsToUint64("IDX0", "FRAME0", keys)
ids, err := s.TranslateRowsToUint64("IDX0", "FIELD0", keys)
if err != nil {
t.Fatal(err)
}
@ -230,7 +230,7 @@ func TestTranslateFile_TranslateRow_Large(t *testing.T) {
// 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 {
if key, err := s.TranslateRowToString("IDX0", "FIELD0", uint64(i+1)); err != nil {
t.Fatal(err)
} else if key != exp {
t.Fatalf("unexpected key: got=%q, exp=%q", key, exp)
@ -243,7 +243,7 @@ func TestTranslateFile_TranslateRow_Large(t *testing.T) {
}
for i := 0; i < 1000000; i++ {
exp := strconv.Itoa(i + 1)
if key, err := s.TranslateRowToString("IDX0", "FRAME0", uint64(i+1)); err != nil {
if key, err := s.TranslateRowToString("IDX0", "FIELD0", uint64(i+1)); err != nil {
t.Fatal(err)
} else if key != exp {
t.Fatalf("unexpected key: got=%q, exp=%q", key, exp)
@ -257,7 +257,7 @@ func TestTranslateFile_Reader(t *testing.T) {
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 {
} else if _, err := s.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
@ -290,7 +290,7 @@ func TestTranslateFile_Reader(t *testing.T) {
} else if diff := cmp.Diff(entry, pilosa.LogEntry{
Type: pilosa.LogEntryTypeInsertRow,
Index: []byte("IDX0"),
Frame: []byte("FRAME0"),
Field: []byte("FIELD0"),
IDs: []uint64{1, 2},
Keys: [][]byte{[]byte("bar"), []byte("baz")},
Length: 24,
@ -329,7 +329,7 @@ func TestTranslateFile_Reader(t *testing.T) {
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 {
} else if _, err := s.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
@ -348,7 +348,7 @@ func TestTranslateFile_Reader(t *testing.T) {
} else if diff := cmp.Diff(entry, pilosa.LogEntry{
Type: pilosa.LogEntryTypeInsertRow,
Index: []byte("IDX0"),
Frame: []byte("FRAME0"),
Field: []byte("FIELD0"),
IDs: []uint64{1, 2},
Keys: [][]byte{[]byte("bar"), []byte("baz")},
Length: 24,
@ -392,7 +392,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
// 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 {
} else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
@ -405,13 +405,13 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica.TranslateRowToString("IDX0", "FRAME0", 1); err != nil {
if value, err := replica.TranslateRowToString("IDX0", "FIELD0", 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 {
if value, err := replica.TranslateRowToString("IDX0", "FIELD0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)