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 branch 'develop' into clearbit-notime

Browse source
This commit is contained in:
tgruben 2018-06-28 14:12:29 -05:00 • committed by GitHub
commit a81e01b019
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
30 changed files with 527 additions and 267 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
api.go
View file

@ -242,11 +242,20 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
}
// CreateField makes the named field in the named index with the given options.
func (api *API) CreateField(ctx context.Context, indexName string, fieldName string, options FieldOptions) (*Field, error) {
// This method currently only takes a single functional option, but that may be
// changed in the future to support multiple options.
func (api *API) CreateField(ctx context.Context, indexName string, fieldName string, opts FieldOption) (*Field, error) {
if err := api.validate(apiCreateField); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
// Apply functional option.
fo := FieldOptions{}
err := opts(&fo)
if err != nil {
return nil, errors.Wrap(err, "applying option")
}
// Find index.
index := api.Holder.Index(indexName)
if index == nil {
@ -254,7 +263,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
}
// Create field.
field, err := index.CreateField(fieldName, options)
field, err := index.CreateField(fieldName, fo)
if err != nil {
return nil, errors.Wrap(err, "creating field")
}
@ -264,7 +273,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
&internal.CreateFieldMessage{
Index: indexName,
Field: fieldName,
Meta: options.Encode(),
Meta: fo.Encode(),
})
if err != nil {
api.server.logger.Printf("problem sending CreateField message: %s", err)

8
client.go
View file

@ -41,10 +41,10 @@ type InternalClient interface {
Import(ctx context.Context, index, field string, slice uint64, bits []Bit) error
ImportK(ctx context.Context, index, field string, bits []Bit) error
EnsureIndex(ctx context.Context, name string, options IndexOptions) error
EnsureField(ctx context.Context, indexName string, fieldName string, options FieldOptions) error
EnsureField(ctx context.Context, indexName string, fieldName string) error
ImportValue(ctx context.Context, index, field string, slice uint64, vals []FieldValue) error
ExportCSV(ctx context.Context, index, field string, slice uint64, w io.Writer) error
CreateField(ctx context.Context, index, field string, opt FieldOptions) error
CreateField(ctx context.Context, index, field string) error
FragmentBlocks(ctx context.Context, uri *URI, index, field string, slice uint64) ([]FragmentBlock, error)
BlockData(ctx context.Context, uri *URI, index, field string, slice uint64, block int) ([]uint64, []uint64, error)
ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
@ -108,7 +108,7 @@ func (n *NopInternalClient) ImportK(ctx context.Context, index, field string, bi
func (n *NopInternalClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error {
return nil
}
func (n *NopInternalClient) EnsureField(ctx context.Context, indexName string, fieldName string, options FieldOptions) error {
func (n *NopInternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error {
return nil
}
func (n *NopInternalClient) ImportValue(ctx context.Context, index, field string, slice uint64, vals []FieldValue) error {
@ -117,7 +117,7 @@ func (n *NopInternalClient) ImportValue(ctx context.Context, index, field string
func (n *NopInternalClient) ExportCSV(ctx context.Context, index, field string, slice uint64, w io.Writer) error {
return nil
}
func (n *NopInternalClient) CreateField(ctx context.Context, index, field string, opt FieldOptions) error {
func (n *NopInternalClient) CreateField(ctx context.Context, index, field string) error {
return nil
}
func (n *NopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field string, slice uint64) ([]FragmentBlock, error) {

7
cmd/import.go
View file

@ -20,7 +20,6 @@ import (
"github.com/spf13/cobra"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/ctl"
)
@ -59,9 +58,9 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM.
flags.IntVarP(&Importer.BufferSize, "buffer-size", "s", 10000000, "Number of bits to buffer/sort before importing.")
flags.BoolVarP(&Importer.Sort, "sort", "", false, "Enables sorting before import.")
flags.BoolVarP(&Importer.CreateSchema, "create", "e", false, "Create the schema if it does not exist before import.")
flags.Var(&Importer.FieldOptions.TimeQuantum, "field-time-quantum", "Time quantum for the field")
flags.StringVar(&Importer.FieldOptions.CacheType, "field-cache-type", pilosa.CacheTypeRanked, "Cache type for the field; valid values: none, lru, ranked")
flags.Uint32Var(&Importer.FieldOptions.CacheSize, "field-cache-size", 50000, "Cache size for the field")
//flags.Var(&Importer.FieldOptions.TimeQuantum, "field-time-quantum", "Time quantum for the field")
//flags.StringVar(&Importer.FieldOptions.CacheType, "field-cache-type", pilosa.CacheTypeRanked, "Cache type for the field; valid values: none, lru, ranked")
//flags.Uint32Var(&Importer.FieldOptions.CacheSize, "field-cache-size", 50000, "Cache size for the field")
ctl.SetTLSConfig(flags, &Importer.TLS.CertificatePath, &Importer.TLS.CertificateKeyPath, &Importer.TLS.SkipVerify)
return importCmd

2
ctl/export_test.go
View file

@ -44,7 +44,7 @@ func TestExportCommand_Validation(t *testing.T) {
}
func TestExportCommand_Run(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)

3
ctl/import.go
View file

@ -42,7 +42,6 @@ type ImportCommand struct {
// Options for index & field to be created if they don't exist
IndexOptions pilosa.IndexOptions
FieldOptions pilosa.FieldOptions
// CreateSchema ensures the schema exists before import
CreateSchema bool
@ -135,7 +134,7 @@ func (cmd *ImportCommand) ensureSchema(ctx context.Context) error {
if err != nil {
return fmt.Errorf("Error Creating Index: %s", err)
}
err = cmd.Client.EnsureField(ctx, cmd.Index, cmd.Field, cmd.FieldOptions)
err = cmd.Client.EnsureField(ctx, cmd.Index, cmd.Field)
if err != nil {
return fmt.Errorf("Error Creating Field: %s", err)
}

8
ctl/import_test.go
View file

@ -61,7 +61,7 @@ func TestImportCommand_Run(t *testing.T) {
t.Fatal(err)
}
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.Server.URI.HostPort()
cm.Index = "i"
@ -86,7 +86,7 @@ func TestImportCommand_RunValue(t *testing.T) {
t.Fatal(err)
}
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.Server.URI.HostPort()
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader("")))
@ -102,7 +102,7 @@ func TestImportCommand_RunValue(t *testing.T) {
}
func TestImportCommand_InvalidFile(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)
@ -176,7 +176,7 @@ func GetIO(buf bytes.Buffer) (io.Reader, io.Writer, io.Writer) {
}
func TestImportCommand_BugOverwriteValue(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)

4
executor_test.go
View file

@ -267,7 +267,7 @@ func TestExecutor_Execute_Count(t *testing.T) {
// Ensure a set query can be executed.
func TestExecutor_Execute_SetBit(t *testing.T) {
t.Run("ID", func(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
hldr.SetBit("i", "f", 1, 0)
@ -312,7 +312,7 @@ func TestExecutor_Execute_SetBit(t *testing.T) {
})
t.Run("Keys", func(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{Keys: true})

123
field.go
View file

@ -15,6 +15,7 @@
package pilosa
import (
"encoding/json"
"fmt"
"io/ioutil"
"os"
@ -34,13 +35,14 @@ import (
const (
DefaultFieldType = FieldTypeSet
defaultCacheType = CacheTypeRanked
DefaultCacheType = CacheTypeRanked
// Default ranked field cache
defaultCacheSize = 50000
bitsPerWord = 32 << (^uint(0) >> 63) // either 32 or 64
maxInt = 1<<(bitsPerWord-1) - 1 // either 1<<31 - 1 or 1<<63 - 1
)
// Field types.
@ -73,19 +75,52 @@ type Field struct {
Logger Logger
}
// FieldOption is a functional option type for pilosa.Fielde.
type FieldOption func(f *Field) error
// FieldOption is a functional option type for pilosa.FieldOptions.
type FieldOption func(fo *FieldOptions) error
// TODO: break these out into separate Options (not a FieldOptions object)
func OptFieldFieldOptions(o FieldOptions) FieldOption {
return func(f *Field) error {
f.options = o
func OptFieldTypeSet(cacheType string, cacheSize uint32) FieldOption {
return func(fo *FieldOptions) error {
if fo.Type != "" {
return errors.Errorf("field type is already set to: %s", fo.Type)
}
fo.Type = FieldTypeSet
fo.CacheType = cacheType
fo.CacheSize = cacheSize
return nil
}
}
func OptFieldTypeInt(min, max int64) FieldOption {
return func(fo *FieldOptions) error {
if fo.Type != "" {
return errors.Errorf("field type is already set to: %s", fo.Type)
}
if min > max {
return ErrInvalidBSIGroupRange
}
fo.Type = FieldTypeInt
fo.Min = min
fo.Max = max
return nil
}
}
func OptFieldTypeTime(timeQuantum TimeQuantum) FieldOption {
return func(fo *FieldOptions) error {
if fo.Type != "" {
return errors.Errorf("field type is already set to: %s", fo.Type)
}
if !timeQuantum.Valid() {
return ErrInvalidTimeQuantum
}
fo.Type = FieldTypeTime
fo.TimeQuantum = timeQuantum
return nil
}
}
// NewField returns a new instance of field.
func NewField(path, index, name string, opts ...FieldOption) (*Field, error) {
func NewField(path, index, name string, options FieldOptions) (*Field, error) {
err := validateName(name)
if err != nil {
return nil, err
@ -103,22 +138,10 @@ func NewField(path, index, name string, opts ...FieldOption) (*Field, error) {
broadcaster: NopBroadcaster,
Stats: NopStatsClient,
options: FieldOptions{
Type: DefaultFieldType,
CacheType: defaultCacheType,
CacheSize: defaultCacheSize,
},
options: applyDefaultOptions(options),
Logger: NopLogger,
}
for _, opt := range opts {
err := opt(f)
if err != nil {
return nil, errors.Wrap(err, "applying option")
}
}
return f, nil
}
@ -1096,23 +1119,17 @@ type FieldOptions struct {
Keys bool `json:"keys,omitempty"`
}
// Validate ensures that FieldOption values are valid.
func (o *FieldOptions) Validate() error {
switch o.Type {
case FieldTypeSet, "":
// TODO: cacheType, cacheSize validation
case FieldTypeInt:
if o.Min > o.Max {
return ErrInvalidBSIGroupRange
// applyDefaultOptions returns a new FieldOptions object
// with default values if o does not contain a valid type.
func applyDefaultOptions(o FieldOptions) FieldOptions {
if o.Type == "" {
return FieldOptions{
Type: DefaultFieldType,
CacheType: DefaultCacheType,
CacheSize: DefaultCacheSize,
}
case FieldTypeTime:
if o.TimeQuantum == "" || !o.TimeQuantum.Valid() {
return ErrInvalidTimeQuantum
}
default:
return errors.New("invalid field type")
}
return nil
return o
}
// Encode converts o into its internal representation.
@ -1150,6 +1167,40 @@ func decodeFieldOptions(options *internal.FieldOptions) *FieldOptions {
}
}
func (o *FieldOptions) MarshalJSON() ([]byte, error) {
switch o.Type {
case FieldTypeSet:
return json.Marshal(struct {
Type string `json:"type"`
CacheType string `json:"cacheType"`
CacheSize uint32 `json:"cacheSize"`
}{
o.Type,
o.CacheType,
o.CacheSize,
})
case FieldTypeInt:
return json.Marshal(struct {
Type string `json:"type"`
Min int64 `json:"min"`
Max int64 `json:"max"`
}{
o.Type,
o.Min,
o.Max,
})
case FieldTypeTime:
return json.Marshal(struct {
Type string `json:"type"`
TimeQuantum TimeQuantum `json:"timeQuantum"`
}{
o.Type,
o.TimeQuantum,
})
}
return nil, errors.New("invalid field type")
}
// List of bsiGroup types.
const (
bsiGroupTypeInt = "int"

15
field_test.go
View file

@ -24,7 +24,7 @@ import (
// Ensure field can open and retrieve a view.
func TestField_CreateViewIfNotExists(t *testing.T) {
f := test.MustOpenField()
f := test.MustOpenField(pilosa.FieldOptions{})
defer f.Close()
// Create view.
@ -50,10 +50,7 @@ func TestField_CreateViewIfNotExists(t *testing.T) {
// Ensure field can set its time quantum.
func TestField_SetTimeQuantum(t *testing.T) {
fo := pilosa.FieldOptions{
Type: "time",
}
f := test.MustOpenField(pilosa.OptFieldFieldOptions(fo))
f := test.MustOpenField(pilosa.FieldOptions{Type: pilosa.FieldTypeTime})
defer f.Close()
// Set & retrieve time quantum.
@ -208,7 +205,7 @@ func TestField_NameRestriction(t *testing.T) {
if err != nil {
panic(err)
}
field, err := pilosa.NewField(path, "i", ".meta")
field, err := pilosa.NewField(path, "i", ".meta", pilosa.FieldOptions{})
if field != nil {
t.Fatalf("unexpected field name %s", err)
}
@ -240,13 +237,13 @@ func TestField_NameValidation(t *testing.T) {
panic(err)
}
for _, name := range validFieldNames {
_, err := pilosa.NewField(path, "i", name)
_, err := pilosa.NewField(path, "i", name, pilosa.FieldOptions{})
if err != nil {
t.Fatalf("unexpected field name: %s %s", name, err)
}
}
for _, name := range invalidFieldNames {
_, err := pilosa.NewField(path, "i", name)
_, err := pilosa.NewField(path, "i", name, pilosa.FieldOptions{})
if err == nil {
t.Fatalf("expected error on field name: %s", name)
}
@ -255,7 +252,7 @@ func TestField_NameValidation(t *testing.T) {
// Ensure field can open and retrieve a view.
func TestField_DeleteView(t *testing.T) {
f := test.MustOpenField()
f := test.MustOpenField(pilosa.FieldOptions{})
defer f.Close()
viewName := pilosa.ViewStandard + "_v"

4
fragment.go
View file

@ -117,8 +117,8 @@ func NewFragment(path, index, field, view string, slice uint64) *Fragment {
field: field,
view: view,
slice: slice,
CacheType: defaultCacheType,
CacheSize: defaultCacheSize,
CacheType: DefaultCacheType,
CacheSize: DefaultCacheSize,
Logger: NopLogger,
MaxOpN: defaultFragmentMaxOpN,

2
fragment_internal_test.go
View file

@ -1245,7 +1245,7 @@ func mustOpenFragment(index, field, view string, slice uint64, cacheType string)
file.Close()
if cacheType == "" {
cacheType = defaultCacheType
cacheType = DefaultCacheType
}
f := NewFragment(file.Name(), index, field, view, slice)

9
http/client.go
View file

@ -334,8 +334,8 @@ func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options p
return err
}
func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fieldName string, options pilosa.FieldOptions) error {
err := c.CreateField(ctx, indexName, fieldName, options)
func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error {
err := c.CreateField(ctx, indexName, fieldName)
if err == nil || err == pilosa.ErrFieldExists {
return nil
}
@ -620,14 +620,15 @@ func (c *InternalClient) backupSliceNode(ctx context.Context, index, field strin
}
// CreateField creates a new field on the server.
func (c *InternalClient) CreateField(ctx context.Context, index, field string, opt pilosa.FieldOptions) error {
func (c *InternalClient) CreateField(ctx context.Context, index, field string) error {
if index == "" {
return pilosa.ErrIndexRequired
}
// TODO: remove buf completely? (depends on whether importer needs to create specific field types)
// Encode query request.
buf, err := json.Marshal(&postFieldRequest{
Options: opt,
//Options: opt,
})
if err != nil {
return errors.Wrap(err, "marshaling")

6
http/client_test.go
View file

@ -219,7 +219,7 @@ func TestClient_MultiNode(t *testing.T) {
// Ensure client can bulk import data.
func TestClient_Import(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
host := cmd.URL()
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
@ -249,7 +249,7 @@ func TestClient_Import(t *testing.T) {
// Ensure client can bulk import value data.
func TestClient_ImportValue(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
host := cmd.URL()
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
@ -321,7 +321,7 @@ func TestClient_ImportValue(t *testing.T) {
// Ensure client can retrieve a list of all checksums for blocks in a fragment.
func TestClient_FragmentBlocks(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}

144
http/handler.go
View file

@ -457,6 +457,17 @@ func (p *postIndexRequest) UnmarshalJSON(b []byte) error {
return nil
}
func getValidOptions(option interface{}) []string {
validOptions := []string{}
val := reflect.ValueOf(option)
for i := 0; i < val.Type().NumField(); i++ {
jsonTag := val.Type().Field(i).Tag.Get("json")
s := strings.Split(jsonTag, ",")
validOptions = append(validOptions, s[0])
}
return validOptions
}
// Raise errors for any unknown key
func validateOptions(data map[string]interface{}, validIndexOptions []string) error {
for k, v := range data {
@ -597,7 +608,9 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) {
// Decode request.
var req postFieldRequest
err := json.NewDecoder(r.Body).Decode(&req)
dec := json.NewDecoder(r.Body)
dec.DisallowUnknownFields()
err := dec.Decode(&req)
if err == io.EOF {
// If no data was provided (EOF), we still create the field
// with default values.
@ -605,7 +618,25 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
_, err = h.API.CreateField(r.Context(), indexName, fieldName, req.Options)
// Validate field options.
if err := req.Options.validate(); err != nil {
http.Error(w, err.Error(), http.StatusNotAcceptable)
return
}
// Convert json options into functional options.
var fos pilosa.FieldOption
switch req.Options.Type {
case pilosa.FieldTypeSet:
fos = pilosa.OptFieldTypeSet(*req.Options.CacheType, *req.Options.CacheSize)
case pilosa.FieldTypeInt:
fos = pilosa.OptFieldTypeInt(*req.Options.Min, *req.Options.Max)
case pilosa.FieldTypeTime:
fos = pilosa.OptFieldTypeTime(*req.Options.TimeQuantum)
}
_, err = h.API.CreateField(r.Context(), indexName, fieldName, fos)
if err != nil {
switch errors.Cause(err) {
case pilosa.ErrIndexNotFound:
@ -623,51 +654,80 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) {
}
}
type _postFieldRequest postFieldRequest
// Custom Unmarshal JSON to validate request body when creating a new field. If there's new FieldOptions,
// adding it to validFieldOptions to make sure the new option is validated, otherwise the request will be failed
func (p *postFieldRequest) UnmarshalJSON(b []byte) error {
// m is an overflow map used to capture additional, unexpected keys.
m := make(map[string]interface{})
if err := json.Unmarshal(b, &m); err != nil {
return errors.Wrap(err, "unmarshaling unexpected keys")
}
validFieldOptions := getValidOptions(pilosa.FieldOptions{})
err := validateOptions(m, validFieldOptions)
if err != nil {
return err
}
// Unmarshal expected values.
var _p _postFieldRequest
if err := json.Unmarshal(b, &_p); err != nil {
return errors.Wrap(err, "unmarshalling expected keys")
}
p.Options = _p.Options
return nil
}
func getValidOptions(option interface{}) []string {
validOptions := []string{}
val := reflect.ValueOf(option)
for i := 0; i < val.Type().NumField(); i++ {
jsonTag := val.Type().Field(i).Tag.Get("json")
s := strings.Split(jsonTag, ",")
validOptions = append(validOptions, s[0])
}
return validOptions
}
type postFieldRequest struct {
Options pilosa.FieldOptions `json:"options"`
Options fieldOptions `json:"options"`
}
type postFieldResponse struct{}
// fieldOptions tracks pilosa.FieldOptions. It is made up of pointers to values,
// and used for input validation.
type fieldOptions struct {
Type string `json:"type,omitempty"`
CacheType *string `json:"cacheType,omitempty"`
CacheSize *uint32 `json:"cacheSize,omitempty"`
Min *int64 `json:"min,omitempty"`
Max *int64 `json:"max,omitempty"`
TimeQuantum *pilosa.TimeQuantum `json:"timeQuantum,omitempty"`
Keys *bool `json:"keys,omitempty"`
}
func (o *fieldOptions) validate() error {
// Pointers to default values.
defaultCacheType := pilosa.DefaultCacheType
defaultCacheSize := uint32(pilosa.DefaultCacheSize)
switch o.Type {
case pilosa.FieldTypeSet, "":
// Because FieldTypeSet is the default, its arguments are
// not required. Instead, the defaults are applied whenever
// a value does not exist.
if o.Type == "" {
o.Type = pilosa.FieldTypeSet
}
if o.CacheType == nil {
o.CacheType = &defaultCacheType
}
if o.CacheSize == nil {
o.CacheSize = &defaultCacheSize
}
if o.Min != nil {
return errors.New("min does not apply to field type set")
} else if o.Max != nil {
return errors.New("max does not apply to field type set")
} else if o.TimeQuantum != nil {
return errors.New("timeQuantum does not apply to field type set")
}
case pilosa.FieldTypeInt:
if o.CacheType != nil {
return errors.New("cacheType does not apply to field type int")
} else if o.CacheSize != nil {
return errors.New("cacheSize does not apply to field type int")
} else if o.Min == nil {
return errors.New("min is required for field type int")
} else if o.Max == nil {
return errors.New("max is required for field type int")
} else if o.TimeQuantum != nil {
return errors.New("timeQuantum does not apply to field type int")
}
case pilosa.FieldTypeTime:
if o.CacheType != nil {
return errors.New("cacheType does not apply to field type time")
} else if o.CacheSize != nil {
return errors.New("cacheSize does not apply to field type time")
} else if o.Min != nil {
return errors.New("min does not apply to field type time")
} else if o.Max != nil {
return errors.New("max does not apply to field type time")
} else if o.TimeQuantum == nil {
return errors.New("timeQuantum is required for field type time")
}
default:
return errors.Errorf("invalid field type: %s", o.Type)
}
return nil
}
// handleDeleteField handles DELETE /field request.
func (h *Handler) handleDeleteField(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {

111
http/handler_internal_test.go
View file

@ -15,6 +15,7 @@
package http
import (
"bytes"
"encoding/json"
"reflect"
"testing"
@ -59,31 +60,119 @@ func TestPostIndexRequestUnmarshalJSON(t *testing.T) {
// Test custom UnmarshalJSON for postFieldRequest object
func TestPostFieldRequestUnmarshalJSON(t *testing.T) {
foo := "foo"
tests := []struct {
json string
expected postFieldRequest
err string
}{
{json: `{"options": {}}`, expected: postFieldRequest{Options: pilosa.FieldOptions{}}},
{json: `{"options": 4}`, err: "options is not map[string]interface{}"},
{json: `{"option": {}}`, err: "Unknown key: option:map[]"},
{json: `{"options": {"badKey": "test"}}`, err: "Unknown key: badKey:test"},
{json: `{"options": {"inverseEnabled": true}}`, err: "Unknown key: inverseEnabled:true"},
{json: `{"options": {"cacheType": "type"}}`, expected: postFieldRequest{Options: pilosa.FieldOptions{CacheType: "type"}}},
{json: `{"options": {"inverse": true, "cacheType": "type"}}`, err: "Unknown key: inverse:true"},
{json: `{"options": {}}`, expected: postFieldRequest{}},
{json: `{"options": 4}`, err: "json: cannot unmarshal number into Go struct field postFieldRequest.options of type http.fieldOptions"},
{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"`},
{json: `{"options": {"cacheType": "foo"}}`, expected: postFieldRequest{Options: fieldOptions{CacheType: &foo}}},
{json: `{"options": {"inverse": true, "cacheType": "foo"}}`, err: `json: unknown field "inverse"`},
}
for _, test := range tests {
for i, test := range tests {
actual := &postFieldRequest{}
err := json.Unmarshal([]byte(test.json), actual)
dec := json.NewDecoder(bytes.NewReader([]byte(test.json)))
dec.DisallowUnknownFields()
err := dec.Decode(actual)
if err != nil {
if test.err == "" || test.err != err.Error() {
t.Errorf("expected error: %v, but got result: %v", test.err, err)
t.Errorf("test %d: expected error: %v, but got result: %v", i, test.err, err)
}
}
if test.err == "" {
if !reflect.DeepEqual(*actual, test.expected) {
t.Errorf("expected: %v, but got: %v", test.expected, *actual)
t.Errorf("test %d: expected: %v, but got: %v", i, test.expected, *actual)
}
}
}
}
func stringPtr(s string) *string {
return &s
}
func int64Ptr(i int64) *int64 {
return &i
}
// Test fieldOption validation.
func TestFieldOptionValidation(t *testing.T) {
timeQuantum := pilosa.TimeQuantum("YMD")
defaultCacheSize := uint32(pilosa.DefaultCacheSize)
tests := []struct {
json string
expected postFieldRequest
err string
}{
// FieldType: Set
{json: `{"options": {}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeSet,
CacheType: stringPtr(pilosa.DefaultCacheType),
CacheSize: &defaultCacheSize,
}}},
{json: `{"options": {"type": "set"}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeSet,
CacheType: stringPtr(pilosa.DefaultCacheType),
CacheSize: &defaultCacheSize,
}}},
{json: `{"options": {"type": "set", "cacheType": "lru"}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeSet,
CacheType: stringPtr("lru"),
CacheSize: &defaultCacheSize,
}}},
{json: `{"options": {"type": "set", "min": 0}}`, err: "min does not apply to field type set"},
{json: `{"options": {"type": "set", "max": 100}}`, err: "max does not apply to field type set"},
{json: `{"options": {"type": "set", "timeQuantum": "YMD"}}`, err: "timeQuantum does not apply to field type set"},
// FieldType: Int
{json: `{"options": {"type": "int"}}`, err: "min is required for field type int"},
{json: `{"options": {"type": "int", "min": 0}}`, err: "max is required for field type int"},
{json: `{"options": {"type": "int", "min": 0, "max": 1000}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeInt,
Min: int64Ptr(0),
Max: int64Ptr(1000),
}}},
{json: `{"options": {"type": "int", "min": 0, "max": 1000, "cacheType": "ranked"}}`, err: "cacheType does not apply to field type int"},
{json: `{"options": {"type": "int", "min": 0, "max": 1000, "cacheSize": 1000}}`, err: "cacheSize does not apply to field type int"},
{json: `{"options": {"type": "int", "min": 0, "max": 1000, "timeQuantum": "YMD"}}`, err: "timeQuantum does not apply to field type int"},
// FieldType: Time
{json: `{"options": {"type": "time"}}`, err: "timeQuantum is required for field type time"},
{json: `{"options": {"type": "time", "timeQuantum": "YMD"}}`, expected: postFieldRequest{Options: fieldOptions{
Type: pilosa.FieldTypeTime,
TimeQuantum: &timeQuantum,
}}},
{json: `{"options": {"type": "time", "timeQuantum": "YMD", "min": 0}}`, err: "min does not apply to field type time"},
{json: `{"options": {"type": "time", "timeQuantum": "YMD", "max": 1000}}`, err: "max does not apply to field type time"},
{json: `{"options": {"type": "time", "timeQuantum": "YMD", "cacheType": "ranked"}}`, err: "cacheType does not apply to field type time"},
{json: `{"options": {"type": "time", "timeQuantum": "YMD", "cacheSize": 1000}}`, err: "cacheSize does not apply to field type time"},
}
for i, test := range tests {
actual := &postFieldRequest{}
dec := json.NewDecoder(bytes.NewReader([]byte(test.json)))
dec.DisallowUnknownFields()
err := dec.Decode(actual)
if err != nil {
t.Errorf("test %d: %v", i, err)
}
// Validate field options.
if err := actual.Options.validate(); err != nil {
if test.err == "" || test.err != err.Error() {
t.Errorf("test %d: expected error: %v, but got result: %v", i, test.err, err)
}
}
if test.err == "" {
if !reflect.DeepEqual(*actual, test.expected) {
t.Errorf("test %d: expected: %v, but got: %v", i, test.expected, *actual)
}
}

51
http/translator_test.go
View file

@ -15,6 +15,17 @@ import (
"github.com/pilosa/pilosa/test"
)
func newMockReadCloser() *mock.ReadCloser {
return &mock.ReadCloser{
ReadFunc: func(p []byte) (int, error) {
return 0, io.EOF
},
CloseFunc: func() error {
return nil
},
}
}
func TestTranslateStore_Reader(t *testing.T) {
// Ensure client can connect and stream the translate store data.
t.Run("OK", func(t *testing.T) {
@ -37,9 +48,9 @@ func TestTranslateStore_Reader(t *testing.T) {
return 0, nil
}
}
var closeInvoked bool
closeInvoked := make(chan struct{})
mrc.CloseFunc = func() error {
closeInvoked = true
close(closeInvoked)
return nil
}
@ -55,19 +66,11 @@ func TestTranslateStore_Reader(t *testing.T) {
}
return &mrc, nil
}
mrc2 := mock.ReadCloser{
ReadFunc: func(p []byte) (int, error) {
return 0, io.EOF
},
CloseFunc: func() error {
return nil
},
}
return &mrc2, nil
return newMockReadCloser(), nil
}
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
@ -85,8 +88,11 @@ func TestTranslateStore_Reader(t *testing.T) {
t.Fatal(err)
}
if !closeInvoked {
select {
case <-time.NewTimer(time.Millisecond * 100).C:
t.Fatal("expected server close")
case <-closeInvoked:
return
}
})
@ -100,19 +106,22 @@ func TestTranslateStore_Reader(t *testing.T) {
<-done
return 0, io.EOF
}
var closeInvoked bool
closeInvoked := make(chan struct{})
mrc.CloseFunc = func() error {
closeInvoked = true
close(closeInvoked)
return nil
}
var translateStore mock.TranslateStore
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
return &mrc, nil
}
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
defer close(done)
@ -126,9 +135,11 @@ func TestTranslateStore_Reader(t *testing.T) {
// Cancel the context and check if server is closed.
cancel()
time.Sleep(100 * time.Millisecond)
if !closeInvoked {
t.Fatal("expected server-side close")
select {
case <-time.NewTimer(time.Millisecond * 100).C:
t.Fatal("expected server close")
case <-closeInvoked:
return
}
})
})
@ -141,7 +152,7 @@ func TestTranslateStore_Reader(t *testing.T) {
}
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
_, err := http.NewTranslateStore(main.Server.URI.String()).Reader(context.Background(), 0)

7
index.go
View file

@ -301,11 +301,6 @@ func (i *Index) createField(name string, opt FieldOptions) (*Field, error) {
return nil, ErrInvalidCacheType
}
// Validate options.
if err := opt.Validate(); err != nil {
return nil, errors.Wrap(err, "validating options")
}
// Initialize field.
f, err := i.newField(i.FieldPath(name), name)
if err != nil {
@ -335,7 +330,7 @@ func (i *Index) createField(name string, opt FieldOptions) (*Field, error) {
}
func (i *Index) newField(path, name string) (*Field, error) {
f, err := NewField(path, i.name, name)
f, err := NewField(path, i.name, name, FieldOptions{}) // TODO: NewField should be un-exported along with FieldOptions
if err != nil {
return nil, err
}

9
mock/mock.go
View file

@ -1,8 +1,11 @@
package mock
import "sync"
type ReadCloser struct {
ReadFunc func(p []byte) (int, error)
CloseFunc func() error
once sync.Once
}
func (rc *ReadCloser) Read(p []byte) (int, error) {
@ -10,5 +13,9 @@ func (rc *ReadCloser) Read(p []byte) (int, error) {
}
func (rc *ReadCloser) Close() error {
return rc.CloseFunc()
var err error = nil
rc.once.Do(func() {
err = rc.CloseFunc()
})
return err
}

40
server/cluster_test.go
View file

@ -31,7 +31,7 @@ import (
// Ensure program can send/receive broadcast messages.
func TestMain_SendReceiveMessage(t *testing.T) {
ms := test.MustRunMainWithCluster(t, 2)
ms := test.MustRunCluster(t, 2)
m0, m1 := ms[0], ms[1]
defer m0.Close()
defer m1.Close()
@ -48,7 +48,7 @@ func TestMain_SendReceiveMessage(t *testing.T) {
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil {
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
@ -116,7 +116,7 @@ func TestMain_SendReceiveMessage(t *testing.T) {
// Ensure that an empty node comes up in a NORMAL state.
func TestClusterResize_EmptyNode(t *testing.T) {
m0 := test.MustRunMain()
m0 := test.MustRunCommand()
defer m0.Close()
if m0.API.State() != pilosa.ClusterStateNormal {
@ -126,7 +126,7 @@ func TestClusterResize_EmptyNode(t *testing.T) {
// Ensure that a cluster of empty nodes comes up in a NORMAL state.
func TestClusterResize_EmptyNodes(t *testing.T) {
clus := test.MustRunMainWithCluster(t, 2)
clus := test.MustRunCluster(t, 2)
defer clus[0].Close()
defer clus[1].Close()
@ -140,7 +140,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
// Ensure that adding a node correctly resizes the cluster.
func TestClusterResize_AddNode(t *testing.T) {
t.Run("NoData", func(t *testing.T) {
clus := test.MustRunMainWithCluster(t, 2)
clus := test.MustRunCluster(t, 2)
if !checkClusterState(clus[0], pilosa.ClusterStateNormal, 1000) {
t.Fatalf("unexpected node0 cluster state: %s", clus[0].API.State())
@ -150,7 +150,7 @@ func TestClusterResize_AddNode(t *testing.T) {
})
t.Run("WithIndex", func(t *testing.T) {
// Configure node0
m0 := test.MustRunMainWithCluster(t, 1)[0]
m0 := test.MustRunCluster(t, 1)[0]
defer m0.Close()
seed := m0.GossipAddress()
@ -161,12 +161,12 @@ func TestClusterResize_AddNode(t *testing.T) {
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil {
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
// Configure node1
m1 := test.NewMainWithCluster(false)
m1 := test.NewCommandNode(false)
m1.Config.Gossip.Port = "0"
m1.Config.Gossip.Seeds = []string{seed}
err := m1.Start()
@ -183,7 +183,7 @@ func TestClusterResize_AddNode(t *testing.T) {
})
t.Run("ContinuousSlices", func(t *testing.T) {
// Configure node0
m0 := test.MustRunMainWithCluster(t, 1)[0]
m0 := test.MustRunCluster(t, 1)[0]
defer m0.Close()
seed := m0.GossipAddress()
@ -194,7 +194,7 @@ func TestClusterResize_AddNode(t *testing.T) {
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil {
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
@ -207,7 +207,7 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := test.NewMainWithCluster(false)
m1 := test.NewCommandNode(false)
m1.Config.Gossip.Port = "0"
m1.Config.Gossip.Seeds = []string{seed}
err := m1.Start()
@ -224,7 +224,7 @@ func TestClusterResize_AddNode(t *testing.T) {
})
t.Run("SkippedSlice", func(t *testing.T) {
// Configure node0
m0 := test.MustRunMainWithCluster(t, 1)[0]
m0 := test.MustRunCluster(t, 1)[0]
defer m0.Close()
seed := m0.GossipAddress()
@ -235,7 +235,7 @@ func TestClusterResize_AddNode(t *testing.T) {
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil {
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
@ -248,7 +248,7 @@ func TestClusterResize_AddNode(t *testing.T) {
}
// Configure node1
m1 := test.NewMainWithCluster(false)
m1 := test.NewCommandNode(false)
m1.Config.Gossip.Port = "0"
m1.Config.Gossip.Seeds = []string{seed}
err := m1.Start()
@ -269,7 +269,7 @@ func TestClusterResize_AddNode(t *testing.T) {
func TestCluster_GossipMembership(t *testing.T) {
t.Run("Node0Down", func(t *testing.T) {
// Configure node0
m0 := test.MustRunMainWithCluster(t, 1)[0]
m0 := test.MustRunCluster(t, 1)[0]
defer m0.Close()
seed := m0.GossipAddress()
@ -277,7 +277,7 @@ func TestCluster_GossipMembership(t *testing.T) {
var eg errgroup.Group
// Configure node1
m1 := test.NewMainWithCluster(false)
m1 := test.NewCommandNode(false)
defer m1.Close()
eg.Go(func() error {
m1.Config.Gossip.Port = "0"
@ -291,7 +291,7 @@ func TestCluster_GossipMembership(t *testing.T) {
})
// Configure node1
m2 := test.NewMainWithCluster(false)
m2 := test.NewCommandNode(false)
defer m2.Close()
eg.Go(func() error {
m2.Config.Gossip.Port = "0"
@ -324,7 +324,7 @@ func TestCluster_GossipMembership(t *testing.T) {
}
func TestClusterResize_RemoveNode(t *testing.T) {
cluster := test.MustRunMainWithCluster(t, 3)
cluster := test.MustRunCluster(t, 3)
m0 := cluster[0]
m1 := cluster[1]
@ -382,7 +382,7 @@ func TestClusterResize_RemoveNode(t *testing.T) {
// Create indexes and fields on one node.
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil {
} else if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal(err)
}
@ -410,7 +410,7 @@ func TestClusterResize_RemoveNode(t *testing.T) {
// checkClusterState polls a given cluster for its state until it
// receives a matching state. It polls up to n times before returning.
func checkClusterState(m *test.Main, state string, n int) bool {
func checkClusterState(m *test.Command, state string, n int) bool {
for i := 0; i < n; i++ {
if m.API.State() == state {
return true

4
server/handler_test.go
View file

@ -37,7 +37,7 @@ import (
// Ensure the handler returns "not found" for invalid paths.
func TestHandler_Endpoints(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
h := cmd.Handler.(*http.Handler).Handler
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
@ -566,7 +566,7 @@ func TestHandler_Endpoints(t *testing.T) {
t.Fatalf("CORS preflight status should be 405, but is %v", result.StatusCode)
}
clus := test.MustRunMainWithCluster(t, 1, []server.CommandOption{test.OptAllowedOrigins([]string{"http://test/"})})
clus := test.MustRunCluster(t, 1, []server.CommandOption{test.OptAllowedOrigins([]string{"http://test/"})})
w = httptest.NewRecorder()
h := clus[0].Handler.(*http.Handler).Handler
h.ServeHTTP(w, req)

21
server/server_test.go
View file

@ -40,7 +40,7 @@ func TestMain_Set_Quick(t *testing.T) {
}
if err := quick.Check(func(cmds []SetCommand) bool {
m := test.MustRunMain()
m := test.MustRunCommand()
defer m.Close()
// Create client.
@ -54,7 +54,7 @@ func TestMain_Set_Quick(t *testing.T) {
if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
}
if err := client.CreateField(context.Background(), "i", cmd.Field, pilosa.FieldOptions{}); err != nil && err != pilosa.ErrFieldExists {
if err := client.CreateField(context.Background(), "i", cmd.Field); err != nil && err != pilosa.ErrFieldExists {
t.Fatal(err)
}
if _, err := m.Query("i", "", fmt.Sprintf(`Set(%d, %s=%d)`, cmd.ColumnID, cmd.Field, cmd.ID)); err != nil {
@ -116,18 +116,18 @@ func TestMain_Set_Quick(t *testing.T) {
// Ensure program can set row attributes and retrieve them.
func TestMain_SetRowAttrs(t *testing.T) {
m := test.MustRunMain()
m := test.MustRunCommand()
defer m.Close()
// Create fields.
client := m.Client()
if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client.CreateField(context.Background(), "i", "x", pilosa.FieldOptions{}); err != nil {
} else if err := client.CreateField(context.Background(), "i", "x"); err != nil {
t.Fatal(err)
} else if err := client.CreateField(context.Background(), "i", "z", pilosa.FieldOptions{}); err != nil {
} else if err := client.CreateField(context.Background(), "i", "z"); err != nil {
t.Fatal(err)
} else if err := client.CreateField(context.Background(), "i", "neg", pilosa.FieldOptions{}); err != nil {
} else if err := client.CreateField(context.Background(), "i", "neg"); err != nil {
t.Fatal(err)
}
@ -193,14 +193,14 @@ func TestMain_SetRowAttrs(t *testing.T) {
// Ensure program can set column attributes and retrieve them.
func TestMain_SetColumnAttrs(t *testing.T) {
m := test.MustRunMain()
m := test.MustRunCommand()
defer m.Close()
// Create fields.
client := m.Client()
if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal(err)
} else if err := client.CreateField(context.Background(), "i", "x", pilosa.FieldOptions{}); err != nil {
} else if err := client.CreateField(context.Background(), "i", "x"); err != nil {
t.Fatal(err)
}
@ -264,16 +264,17 @@ func tempMkdir(t *testing.T) string {
func TestMain_RecalculateHashes(t *testing.T) {
const clusterSize = 5
cluster := test.MustRunMainWithCluster(t, clusterSize)
cluster := test.MustRunCluster(t, clusterSize)
// Create the schema.
client0 := cluster[0].Client()
if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
t.Fatal("create index:", err)
}
if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{CacheType: "ranked"}); err != nil {
if err := client0.CreateField(context.Background(), "i", "f"); err != nil {
t.Fatal("create field:", err)
}
return
// Set some columns
data := []string{}

5
server_test.go
View file

@ -28,13 +28,14 @@ import (
// pilosa.Server was not having its remoteClient field set by an option and so
// it was using a nil client in monitorAntiEntropy.
func TestMonitorAntiEntropy(t *testing.T) {
cluster := test.MustRunMainWithCluster(t, 3, []server.CommandOption{test.OptAntiEntropyInterval(time.Millisecond * 20)})
cluster := test.MustRunCluster(t, 3, []server.CommandOption{test.OptAntiEntropyInterval(time.Millisecond * 20)})
client := cluster[1].Client()
err := client.CreateIndex(context.Background(), "balh", pilosa.IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
err = client.CreateField(context.Background(), "balh", "fralh", pilosa.FieldOptions{})
err = client.CreateField(context.Background(), "balh", "fralh")
if err != nil {
t.Fatalf("creating field: %v", err)
}

2
stats_test.go
View file

@ -209,7 +209,7 @@ func TestStatsCount_SetProfileAttrs(t *testing.T) {
}
func TestStatsCount_APICalls(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
cmd := test.MustRunCluster(t, 1)[0]
h := cmd.Handler.(*http.Handler).Handler
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}

12
test/field.go
View file

@ -28,12 +28,12 @@ type Field struct {
}
// NewField returns a new instance of Field d/0.
func NewField(opt ...pilosa.FieldOption) *Field {
func NewField(options pilosa.FieldOptions) *Field {
path, err := ioutil.TempDir("", "pilosa-field-")
if err != nil {
panic(err)
}
field, err := pilosa.NewField(path, "i", "f", opt...)
field, err := pilosa.NewField(path, "i", "f", options)
if err != nil {
panic(err)
}
@ -41,8 +41,8 @@ func NewField(opt ...pilosa.FieldOption) *Field {
}
// MustOpenField returns a new, opened field at a temporary path. Panic on error.
func MustOpenField(opt ...pilosa.FieldOption) *Field {
f := NewField(opt...)
func MustOpenField(options pilosa.FieldOptions) *Field {
f := NewField(options)
if err := f.Open(); err != nil {
panic(err)
}
@ -63,7 +63,7 @@ func (f *Field) Reopen() error {
}
path, index, name := f.Path(), f.Index(), f.Name()
f.Field, err = pilosa.NewField(path, index, name)
f.Field, err = pilosa.NewField(path, index, name, pilosa.FieldOptions{})
if err != nil {
return err
}
@ -76,7 +76,7 @@ func (f *Field) Reopen() error {
// Ensure field can set its cache
func TestField_SetCacheSize(t *testing.T) {
f := MustOpenField()
f := MustOpenField(pilosa.FieldOptions{})
defer f.Close()
cacheSize := uint32(100)

2
test/holder.go
View file

@ -92,7 +92,7 @@ func (h *Holder) MustCreateFieldIfNotExists(index, field string) *Field {
// MustCreateRankedFragmentIfNotExists returns a given fragment with a ranked cache. Panic on error.
func (h *Holder) MustCreateRankedFragmentIfNotExists(index, field, view string, slice uint64) *Fragment {
idx := h.MustCreateIndexIfNotExists(index, pilosa.IndexOptions{})
f, err := idx.CreateFieldIfNotExists(field, pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked})
f, err := idx.CreateFieldIfNotExists(field, pilosa.FieldOptions{})
if err != nil {
panic(err)
}

168
test/pilosa.go
View file

@ -32,8 +32,8 @@ import (
)
////////////////////////////////////////////////////////////////////////////////////
// Main represents a test wrapper for server.Command.
type Main struct {
// Command represents a test wrapper for server.Command.
type Command struct {
*server.Command
commandOptions []server.CommandOption
@ -57,21 +57,14 @@ func OptAllowedOrigins(origins []string) server.CommandOption {
}
}
// GossipAddress returns the address on which gossip is listening after a Main
// has been setup. Useful to pass as a seed to other nodes when creating and
// testing clusters.
func (m *Main) GossipAddress() string {
return m.GossipTransport().URI.String()
}
// NewMain returns a new instance of Main with a temporary data directory and random port.
func NewMain(opts ...server.CommandOption) *Main {
// NewCommand returns a new instance of Main with a temporary data directory and random port.
func NewCommand(opts ...server.CommandOption) *Command {
path, err := ioutil.TempDir("", "pilosa-")
if err != nil {
panic(err)
}
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr, opts...), commandOptions: opts}
m := &Command{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr, opts...), commandOptions: opts}
m.Config.DataDir = path
m.Config.Bind = "http://localhost:0"
m.Config.Cluster.Disabled = true
@ -92,58 +85,17 @@ func NewMain(opts ...server.CommandOption) *Main {
return m
}
// NewMainWithCluster returns a new instance of Main with clustering enabled.
func NewMainWithCluster(isCoordinator bool, opts ...server.CommandOption) *Main {
m := NewMain(opts...)
// NewCommandNode returns a new instance of Command with clustering enabled.
func NewCommandNode(isCoordinator bool, opts ...server.CommandOption) *Command {
m := NewCommand(opts...)
m.Config.Cluster.Disabled = false
m.Config.Cluster.Coordinator = isCoordinator
return m
}
// MustRunMainWithCluster ruturns a running array of *Main where
// all nodes are joined via memberlist (i.e. clustering enabled).
func MustRunMainWithCluster(t *testing.T, size int, opts ...[]server.CommandOption) []*Main {
ma, err := runMainWithCluster(size, opts...)
if err != nil {
t.Fatalf("new main array with cluster: %v", err)
}
return ma
}
// runMainWithCluster runs an array of *Main where all nodes are
// joined via memberlist (i.e. clustering enabled).
func runMainWithCluster(size int, opts ...[]server.CommandOption) ([]*Main, error) {
if size == 0 {
return nil, errors.New("cluster must contain at least one node")
}
if len(opts) != size && len(opts) != 0 && len(opts) != 1 {
return nil, errors.New("Slice of CommandOptions must be of length 0, 1, or equal to the number of cluster nodes")
}
mains := make([]*Main, size)
var gossipSeeds = make([]string, size)
for i := 0; i < size; i++ {
var commandOpts []server.CommandOption
if len(opts) > 0 {
commandOpts = opts[i%len(opts)]
}
m := NewMainWithCluster(i == 0, commandOpts...)
m.Config.Gossip.Port = "0"
m.Config.Gossip.Seeds = gossipSeeds[:i]
if err := m.Start(); err != nil {
return nil, errors.Wrapf(err, "Starting server %d", i)
}
gossipSeeds[i] = m.GossipTransport().URI.String()
mains[i] = m
}
return mains, nil
}
// MustRunMain returns a new, running Main. Panic on error.
func MustRunMain() *Main {
m := NewMain()
// MustRunCommand returns a new, running Main. Panic on error.
func MustRunCommand() *Command {
m := NewCommand()
m.Config.Metric.Diagnostics = false // Disable diagnostics.
if err := m.Start(); err != nil {
panic(err)
@ -151,14 +103,21 @@ func MustRunMain() *Main {
return m
}
// GossipAddress returns the address on which gossip is listening after a Main
// has been setup. Useful to pass as a seed to other nodes when creating and
// testing clusters.
func (m *Command) GossipAddress() string {
return m.GossipTransport().URI.String()
}
// Close closes the program and removes the underlying data directory.
func (m *Main) Close() error {
func (m *Command) Close() error {
defer os.RemoveAll(m.Config.DataDir)
return m.Command.Close()
}
// Reopen closes the program and reopens it.
func (m *Main) Reopen() error {
func (m *Command) Reopen() error {
if err := m.Command.Close(); err != nil {
return err
}
@ -180,10 +139,10 @@ func (m *Main) Reopen() error {
}
// URL returns the base URL string for accessing the running program.
func (m *Main) URL() string { return m.Server.URI.String() }
func (m *Command) URL() string { return m.Server.URI.String() }
// Client returns a client to connect to the program.
func (m *Main) Client() *http.InternalClient {
func (m *Command) Client() *http.InternalClient {
client, err := http.NewInternalClient(m.Server.URI.HostPort(), http.GetHTTPClient(nil))
if err != nil {
panic(err)
@ -192,7 +151,7 @@ func (m *Main) Client() *http.InternalClient {
}
// Query executes a query against the program through the HTTP API.
func (m *Main) Query(index, rawQuery, query string) (string, error) {
func (m *Command) Query(index, rawQuery, query string) (string, error) {
resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/query?", index)+rawQuery, query)
if resp.StatusCode != gohttp.StatusOK {
return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
@ -200,7 +159,7 @@ func (m *Main) Query(index, rawQuery, query string) (string, error) {
return resp.Body, nil
}
func (m *Main) RecalculateCaches() error {
func (m *Command) RecalculateCaches() error {
resp := MustDo("POST", fmt.Sprintf("%s/recalculate-caches", m.URL()), "")
if resp.StatusCode != 204 {
return fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body)
@ -208,6 +167,85 @@ func (m *Main) RecalculateCaches() error {
return nil
}
// Cluster represents a Pilosa cluster (multiple Command instances)
type Cluster []*Command
// Start runs a Cluster
func (c Cluster) Start() error {
var gossipSeeds = make([]string, len(c))
for i, cc := range c {
cc.Config.Gossip.Port = "0"
cc.Config.Gossip.Seeds = gossipSeeds[:i]
if err := cc.Start(); err != nil {
return errors.Wrapf(err, "starting server %d", i)
}
gossipSeeds[i] = cc.GossipAddress()
}
return nil
}
// Stop stops a Cluster
func (c Cluster) Close() error {
for i, cc := range c {
if err := cc.Close(); err != nil {
return errors.Wrapf(err, "stopping server %d", i)
}
}
return nil
}
// MustNewCluster creates a new cluster
func MustNewCluster(t *testing.T, size int, opts ...[]server.CommandOption) Cluster {
c, err := newCluster(size, opts...)
if err != nil {
t.Fatalf("new cluster: %v", err)
}
return c
}
// newCluster creates a new cluster
func newCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
if size == 0 {
return nil, errors.New("cluster must contain at least one node")
}
if len(opts) != size && len(opts) != 0 && len(opts) != 1 {
return nil, errors.New("Slice of CommandOptions must be of length 0, 1, or equal to the number of cluster nodes")
}
cluster := make(Cluster, size)
for i := 0; i < size; i++ {
var commandOpts []server.CommandOption
if len(opts) > 0 {
commandOpts = opts[i%len(opts)]
}
m := NewCommandNode(i == 0, commandOpts...)
cluster[i] = m
}
return cluster, nil
}
// runCluster creates and starts a new cluster
func runCluster(size int, opts ...[]server.CommandOption) (Cluster, error) {
cluster, err := newCluster(size, opts...)
if err != nil {
return nil, errors.Wrap(err, "new cluster")
}
if err = cluster.Start(); err != nil {
return nil, errors.Wrap(err, "starting cluster")
}
return cluster, nil
}
// MustRunCluster creates and starts a new cluster
func MustRunCluster(t *testing.T, size int, opts ...[]server.CommandOption) Cluster {
c, err := runCluster(size, opts...)
if err != nil {
t.Fatalf("run cluster: %v", err)
}
return c
}
////////////////////////////////////////////////////////////////////////////////////
// MustDo executes http.Do() with an http.NewRequest(). Panic on error.

4
test/pilosa_test.go
View file

@ -27,7 +27,7 @@ import (
func TestNewCluster(t *testing.T) {
numNodes := 3
cluster := test.MustRunMainWithCluster(t, numNodes)
cluster := test.MustRunCluster(t, numNodes)
coordinator := getCoordinator(cluster[0])
for i := 1; i < numNodes; i++ {
@ -78,7 +78,7 @@ func TestNewCluster(t *testing.T) {
}
}
func getCoordinator(m *test.Main) string {
func getCoordinator(m *test.Command) string {
hosts := m.API.Hosts(context.Background())
for _, host := range hosts {
if host.IsCoordinator {

4
translate.go
View file

@ -306,11 +306,13 @@ func (s *TranslateFile) replicate(ctx context.Context) error {
} else if err != nil {
return err
}
s.mu.Lock()
// Write to local store.
if err := s.appendEntry(&entry); err != nil {
s.mu.Unlock()
return err
}
s.mu.Unlock()
}
}

2
view.go
View file

@ -72,7 +72,7 @@ func NewView(path, index, field, name string, cacheSize uint32) *View {
name: name,
cacheSize: cacheSize,
cacheType: defaultCacheType,
cacheType: DefaultCacheType,
fragments: make(map[uint64]*Fragment),
broadcaster: NopBroadcaster,

2
view_internal_test.go
View file

@ -26,7 +26,7 @@ func mustOpenView(index, field, name string) *View {
panic(err)
}
v := NewView(path, index, field, name, defaultCacheSize)
v := NewView(path, index, field, name, DefaultCacheSize)
if err := v.open(); err != nil {
panic(err)
}