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 input-definition

Browse source
This commit is contained in:
Michael Baird 2017-07-18 14:51:15 -05:00 • committed by GitHub
commit 1569c932ff
15 changed files with 479 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
CONTRIBUTING.md
View file

@ -15,6 +15,8 @@ Before you start working on new features, you should [open a new issue][1] to le
you're doing before you start working, otherwise you run the risk of duplicating effort. This also
gives others an opportunity to provide input for your feature.
If you want to help but you aren't sure where to start, check out our [github label for low-effort issues][6].
- Fork the [Pilosa repository][2] and then clone your fork:
```shell
@ -57,3 +59,4 @@ gives others an opportunity to provide input for your feature.
[3]: https://github.com/pilosa/pilosa/compare/
[4]: https://github.com/pilosa/general/blob/master/proposal.md
[5]: https://github.com/pilosa/pilosa/issues
[6]: https://github.com/pilosa/pilosa/issues?q=is%3Aopen+is%3Aissue+label%3Anewcomer

57
attr.go
View file

@ -39,21 +39,56 @@ const (
AttrTypeFloat = 4
)
// AttrCache represents a cache for attributes.
type AttrCache struct {
mu sync.RWMutex
attrs map[uint64]map[string]interface{}
}
// Get returns the cached attributes for a given id.
func (c *AttrCache) Get(id uint64) map[string]interface{} {
c.mu.RLock()
defer c.mu.RUnlock()
attrs := c.attrs[id]
if attrs == nil {
return nil
}
// Make a copy for safety
ret := make(map[string]interface{})
for k, v := range attrs {
ret[k] = v
}
return ret
}
// Set updates the cached attributes for a given id.
func (c *AttrCache) Set(id uint64, attrs map[string]interface{}) {
c.mu.Lock()
defer c.mu.Unlock()
c.attrs[id] = attrs
}
// AttrStore represents a storage layer for attributes.
type AttrStore struct {
mu sync.RWMutex
path string
db *bolt.DB
mu sync.RWMutex
path string
db *bolt.DB
attrCache *AttrCache
}
// in-memory cache
attrs map[uint64]map[string]interface{}
// NewAttrCache returns a new instance of AttrCache.
func NewAttrCache() *AttrCache {
return &AttrCache{
attrs: make(map[uint64]map[string]interface{}),
}
}
// NewAttrStore returns a new instance of AttrStore.
func NewAttrStore(path string) *AttrStore {
return &AttrStore{
path: path,
attrs: make(map[uint64]map[string]interface{}),
path: path,
attrCache: NewAttrCache(),
}
}
@ -96,7 +131,7 @@ func (s *AttrStore) Attrs(id uint64) (m map[string]interface{}, err error) {
defer s.mu.RUnlock()
// Check cache for map.
if m = s.attrs[id]; m != nil {
if m = s.attrCache.Get(id); m != nil {
return m, nil
}
@ -112,7 +147,7 @@ func (s *AttrStore) Attrs(id uint64) (m map[string]interface{}, err error) {
}
// Add to cache.
s.attrs[id] = m
s.attrCache.Set(id, m)
return
}
@ -149,7 +184,7 @@ func (s *AttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
}
// Swap attributes map in cache.
s.attrs[id] = attr
s.attrCache.Set(id, attr)
return nil
}
@ -184,7 +219,7 @@ func (s *AttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error {
// Swap attributes map in cache.
for id, attr := range attrs {
s.attrs[id] = attr
s.attrCache.Set(id, attr)
}
return nil

16
cmd/server_test.go
View file

@ -52,18 +52,23 @@ func TestServerConfig(t *testing.T) {
[cluster]
poll-interval = "45s"
type = "http"
replicas = 2
hosts = [
"localhost:19444",
]
internal-hosts = [
"localhost:19500",
"localhost:19501",
]
`,
validation: func() error {
v := validator{}
v.Check(cmd.Server.Config.DataDir, actualDataDir)
v.Check(cmd.Server.Config.Host, "example.com:10111")
v.Check(cmd.Server.Config.Bind, "example.com:10111")
v.Check(cmd.Server.Config.Cluster.ReplicaN, 2)
v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:10111", "example.com:10110"})
v.Check(cmd.Server.Config.Cluster.PollingInterval, pilosa.Duration(time.Second*182))
v.Check(cmd.Server.Config.Cluster.PollInterval, pilosa.Duration(time.Second*182))
return v.Error()
},
},
@ -75,9 +80,14 @@ func TestServerConfig(t *testing.T) {
bind = "localhost:0"
data-dir = "` + actualDataDir + `"
[cluster]
type = "http"
hosts = [
"localhost:19444",
]
internal-hosts = [
"localhost:19500",
"localhost:19501",
]
[plugins]
path = "/var/sloth"
`,
@ -113,7 +123,7 @@ func TestServerConfig(t *testing.T) {
validation: func() error {
v := validator{}
v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:19444"})
v.Check(cmd.Server.Config.Cluster.PollingInterval, pilosa.Duration(time.Minute*2))
v.Check(cmd.Server.Config.Cluster.PollInterval, pilosa.Duration(time.Minute*2))
v.Check(cmd.Server.Config.AntiEntropy.Interval, pilosa.Duration(time.Minute*11))
v.Check(cmd.Server.CPUProfile, profFile.Name())
v.Check(cmd.Server.CPUTime, time.Minute)

77
config.go
View file

@ -14,7 +14,17 @@
package pilosa
import "time"
import (
"time"
)
// Cluster types.
const (
ClusterNone = ""
ClusterStatic = "static"
ClusterHTTP = "http"
ClusterGossip = "gossip"
)
const (
// DefaultHost is the default hostname to use.
@ -24,7 +34,7 @@ const (
DefaultPort = "10101"
// DefaultClusterType sets the node intercommunication method.
DefaultClusterType = "static"
DefaultClusterType = ClusterStatic
// DefaultInternalPort the port the nodes intercommunicate on.
DefaultInternalPort = "14000"
@ -36,20 +46,23 @@ const (
DefaultMaxWritesPerRequest = 5000
)
// ClusterTypes set of cluster types.
var ClusterTypes = []string{ClusterNone, ClusterStatic, ClusterHTTP, ClusterGossip}
// Config represents the configuration for the command.
type Config struct {
DataDir string `toml:"data-dir"`
Host string `toml:"host"`
Bind string `toml:"bind"`
Cluster struct {
ReplicaN int `toml:"replicas"`
Type string `toml:"type"`
Hosts []string `toml:"hosts"`
InternalHosts []string `toml:"internal-hosts"`
PollingInterval Duration `toml:"polling-interval"`
InternalPort string `toml:"internal-port"`
GossipSeed string `toml:"gossip-seed"`
LongQueryTime Duration `toml:"long-query-time"`
ReplicaN int `toml:"replicas"`
Type string `toml:"type"`
Hosts []string `toml:"hosts"`
InternalHosts []string `toml:"internal-hosts"`
PollInterval Duration `toml:"poll-interval"`
InternalPort string `toml:"internal-port"`
GossipSeed string `toml:"gossip-seed"`
LongQueryTime Duration `toml:"long-query-time"`
} `toml:"cluster"`
Plugins struct {
@ -67,21 +80,21 @@ type Config struct {
LogPath string `toml:"log-path"`
Metric struct {
Service string `toml:"service"`
Host string `toml:"host"`
PollingInterval Duration `toml:"interval"`
} `toml:"metrics"`
Service string `toml:"service"`
Host string `toml:"host"`
PollInterval Duration `toml:"poll-interval"`
} `toml:"metric"`
}
// NewConfig returns an instance of Config with default options.
func NewConfig() *Config {
c := &Config{
Host: DefaultHost + ":" + DefaultPort,
Bind: DefaultHost + ":" + DefaultPort,
MaxWritesPerRequest: DefaultMaxWritesPerRequest,
}
c.Cluster.ReplicaN = DefaultReplicaN
c.Cluster.Type = DefaultClusterType
c.Cluster.PollingInterval = Duration(DefaultPollingInterval)
c.Cluster.PollInterval = Duration(DefaultPollingInterval)
c.Cluster.Hosts = []string{}
c.Cluster.InternalHosts = []string{}
c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval)
@ -89,6 +102,35 @@ func NewConfig() *Config {
return c
}
// Validate that all configuration permutations are compatible with each other.
func (c *Config) Validate() error {
if !StringInSlice(c.Cluster.Type, ClusterTypes) {
return ErrConfigClusterTypeInvalid
}
if len(c.Cluster.Hosts) > 1 && !(c.Cluster.Type == ClusterHTTP || c.Cluster.Type == ClusterGossip) {
return ErrConfigClusterTypeMissing
}
if c.Cluster.Type == ClusterHTTP || c.Cluster.Type == ClusterGossip {
if c.Cluster.ReplicaN > len(c.Cluster.Hosts) {
return ErrConfigReplicaNInvalid
}
if len(c.Cluster.Hosts) != len(c.Cluster.InternalHosts) {
return ErrConfigHostsMismatch
}
if !foundItem(c.Cluster.Hosts, c.Bind) {
return ErrConfigHostsMissing
}
if !ContainsSubstring(c.Cluster.InternalPort, c.Cluster.InternalHosts) {
return ErrConfigBroadcastPort
}
}
if c.Cluster.Type == ClusterGossip && !StringInSlice(c.Cluster.GossipSeed, c.Cluster.InternalHosts) {
return ErrConfigGossipSeed
}
return nil
}
// Duration is a TOML wrapper type for time.Duration.
type Duration time.Duration
@ -111,6 +153,7 @@ func (d Duration) MarshalText() (text []byte, err error) {
return []byte(d.String()), nil
}
// MarshalTOML write duration into valid TOML.
func (d Duration) MarshalTOML() ([]byte, error) {
return []byte(d.String()), nil
}

88
config_test.go Normal file
View file

@ -0,0 +1,88 @@
package pilosa_test
import (
"reflect"
"testing"
"time"
"github.com/pilosa/pilosa"
)
func Test_NewConfig(t *testing.T) {
c := pilosa.NewConfig()
c.Cluster.Hosts = []string{c.Bind, "localhost:10102"}
if err := c.Validate(); err != pilosa.ErrConfigClusterTypeMissing {
t.Fatal(err)
}
c.Cluster.Type = "test"
if err := c.Validate(); err != pilosa.ErrConfigClusterTypeInvalid {
t.Fatal(err)
}
c.Cluster.Type = pilosa.ClusterHTTP
if err := c.Validate(); err != pilosa.ErrConfigHostsMismatch {
t.Fatal(err)
}
c.Cluster.InternalPort = pilosa.DefaultInternalPort
c.Cluster.InternalHosts = []string{"localhost:14004", "localhost:14001"}
if err := c.Validate(); err != pilosa.ErrConfigBroadcastPort {
t.Fatal(err)
}
c.Cluster.InternalHosts = []string{"localhost:14000", "localhost:14001"}
c.Bind = "localhost:1"
// Check for bind addres in cluster hosts
if err := c.Validate(); err != pilosa.ErrConfigHostsMissing {
t.Fatal(err)
}
c.Bind = "localhost:10101"
c.Cluster.ReplicaN = 3
if err := c.Validate(); err != pilosa.ErrConfigReplicaNInvalid {
t.Fatal(err)
}
c.Cluster.ReplicaN = 2
c.Cluster.Type = pilosa.ClusterGossip
c.Cluster.GossipSeed = "localhost:10101"
if err := c.Validate(); err != pilosa.ErrConfigGossipSeed {
t.Fatal(err)
}
c.Cluster.GossipSeed = "localhost:14000"
if err := c.Validate(); err != nil {
t.Fatal(err)
}
}
func TestDuration(t *testing.T) {
d := pilosa.Duration(time.Second * 182)
if d.String() != "3m2s" {
t.Fatalf("Unexpected time Duration %s", d)
}
b := []byte{51, 109, 50, 115}
v, _ := d.MarshalText()
if !reflect.DeepEqual(b, v) {
t.Fatalf("Unexpected marshalled value %v", v)
}
v, _ = d.MarshalTOML()
if !reflect.DeepEqual(b, v) {
t.Fatalf("Unexpected marshalled value %v", v)
}
err := d.UnmarshalText([]byte("5"))
if err.Error() != "time: missing unit in duration 5" {
t.Fatalf("expected time: missing unit in duration: %s", err)
}
err = d.UnmarshalText([]byte("3m2s"))
v, _ = d.MarshalText()
if !reflect.DeepEqual(b, v) {
t.Fatalf("Unexpected marshalled value %v", v)
}
}

7
ctl/server.go
View file

@ -25,12 +25,13 @@ import (
func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags := cmd.Flags()
flags.StringVarP(&srv.Config.DataDir, "data-dir", "d", "~/.pilosa", "Directory to store pilosa data files.")
flags.StringVarP(&srv.Config.Host, "bind", "b", ":10101", "Default URI on which pilosa should listen.")
flags.StringVarP(&srv.Config.Bind, "bind", "b", ":10101", "Default URI on which pilosa should listen.")
flags.IntVarP(&srv.Config.MaxWritesPerRequest, "max-writes-per-request", "", srv.Config.MaxWritesPerRequest, "Number of write commands per request.")
flags.IntVarP(&srv.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number of hosts each piece of data should be stored on.")
flags.StringSliceVarP(&srv.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.")
flags.StringSliceVarP(&srv.Config.Cluster.InternalHosts, "cluster.internal-hosts", "", []string{}, "Comma separated list of hosts in cluster used for internal communication.")
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.PollingInterval), "cluster.poll-interval", "", time.Minute, "Polling interval for cluster.") // TODO what actually is this?
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.PollInterval), "cluster.poll-interval", "", time.Minute, "Polling interval for cluster.") // TODO what actually is this?
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Long Query Time.")
flags.StringVarP(&srv.Config.Plugins.Path, "plugins.path", "", "", "Path to plugin directory.")
flags.StringVar(&srv.Config.LogPath, "log-path", "", "Log path")
flags.DurationVarP((*time.Duration)(&srv.Config.AntiEntropy.Interval), "anti-entropy.interval", "", time.Minute*10, "Interval at which to run anti-entropy routine.")
@ -41,5 +42,5 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.StringVarP(&srv.Config.Cluster.InternalPort, "cluster.internal-port", "", "", "Port to which pilosa should bind for internal state sharing.")
flags.StringVarP(&srv.Config.Metric.Service, "metric.service", "", "nop", "Default URI on which pilosa should listen.")
flags.StringVarP(&srv.Config.Metric.Host, "metric.host", "", "", "Default URI to send metrics.")
flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollingInterval), "metric.poll-interval", "", time.Minute*0, "Polling interval metrics.")
flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", "", time.Minute*0, "Polling interval metrics.")
}

15
docs/configuration.md
View file

@ -144,6 +144,19 @@ Any flag that has a value that is a comma separated list on the command line bec
data-dir = "~/.pilosa"
```
#### Gossip Seed
* Description: When using the gossip [Cluster Type]({{< ref "#cluster-type" >}}), this specifies which internal host should be used to initialize membership in the cluster. Typcially this can be the address of any available host in the cluster. For example, when starting a three-node cluster made up of `node0`, `node1`, and `node2`, the `gossip-seed` for all three nodes can be configured to be the address of `node0`.
* Flag: `--gossip-seed="localhost:11101"`
* Env: `PILOSA_GOSSIP_SEED="localhost:11101"`
* Config:
```toml
[cluster]
type = "gossip"
gossip-seed = "localhost:11101"
```
#### Profile CPU
* Description: If this is set to a path, collect a cpu profile and store it there.
@ -199,4 +212,4 @@ Any flag that has a value that is a comma separated list on the command line bec
```toml
[metric]
poll-interval = "0m15s"
```
```

7
docs/query-language.md
View file

@ -341,6 +341,13 @@ have the attribute specified by `field` with one of the values specified in
**Result Type:** array of key/count objects
**Caveats:**
* Performing a TopN() query on a frame with cache type ranked will return the top bitmaps sorted by count in descending order.
* Frames with cache type lru will maintain an LRU (Least Recently Used) cache, thus a TopN() query on this type of frame will return bitmaps sorted in order of most recently set bit.
* The frame's cache size determines the number of sorted bitmaps to maintain in the cache for purposes of TopN() queries. There is a tradeoff between performance and accuracy; increasing the cache size will improve accuracy of results at the cost of performance.
* Once full, the cache will truncate the set of bitmaps according to the frame option CacheSize. Bitmaps that straddle the limit and have the same count will be truncated in no particular order.
* The TopN() query's attribute filter is applied to the existing sorted cache of bitmaps. Bitmaps that fall outside of the sorted cache range, even if they would normally pass the filter, are ignored.
**Examples:**
```

73
executor.go
View file

@ -168,6 +168,8 @@ func (e *Executor) executeCall(ctx context.Context, index string, c *pql.Call, s
return e.executeCount(ctx, index, c, slices, opt)
case "SetBit":
return e.executeSetBit(ctx, index, c, opt)
case "SetFieldValue":
return nil, e.executeSetFieldValue(ctx, index, c, opt)
case "SetRowAttrs":
return nil, e.executeSetRowAttrs(ctx, index, c, opt)
case "SetColumnAttrs":
@ -839,6 +841,77 @@ func (e *Executor) executeSetBitView(ctx context.Context, index string, c *pql.C
return ret, nil
}
// executeSetFieldValue executes a SetFieldValue() call.
func (e *Executor) executeSetFieldValue(ctx context.Context, index string, c *pql.Call, opt *ExecOptions) error {
frameName, ok := c.Args["frame"].(string)
if !ok {
return errors.New("SetFieldValue() frame required")
}
// Retrieve column label.
idx := e.Holder.Index(index)
if idx == nil {
return ErrIndexNotFound
}
columnLabel := idx.ColumnLabel()
// Retrieve frame.
frame := e.Holder.Frame(index, frameName)
if frame == nil {
return ErrFrameNotFound
}
// Parse labels.
columnID, ok, err := c.UintArg(columnLabel)
if err != nil {
return fmt.Errorf("reading SetFieldValue() column: %v", err)
} else if !ok {
return fmt.Errorf("SetFieldValue() column field '%v' required", columnLabel)
}
// Copy args and remove reserved fields.
args := pql.CopyArgs(c.Args)
delete(args, "frame")
delete(args, columnLabel)
// Set values.
for name, value := range args {
switch value := value.(type) {
case int64:
if _, err := frame.SetFieldValue(columnID, name, value); err != nil {
return err
}
default:
return ErrInvalidFieldValueType
}
}
frame.Stats.Count("SetFieldValue", 1, 1.0)
// Do not forward call if this is already being forwarded.
if opt.Remote {
return nil
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.Nodes).FilterHost(e.Host)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
_, err := e.exec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, opt)
resp <- err
}(node)
}
// Return first error.
for range nodes {
if err := <-resp; err != nil {
return err
}
}
return nil
}
// executeSetRowAttrs executes a SetRowAttrs() call.
func (e *Executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.Call, opt *ExecOptions) error {
frameName, ok := c.Args["frame"].(string)

97
executor_test.go
View file

@ -234,6 +234,103 @@ func TestExecutor_Execute_SetBit(t *testing.T) {
}
}
// Ensure a SetFieldValue() query can be executed.
func TestExecutor_Execute_SetFieldValue(t *testing.T) {
t.Run("OK", func(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
// Create frames.
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{})
if _, err := index.CreateFrameIfNotExists("f", pilosa.FrameOptions{
RangeEnabled: true,
Fields: []*pilosa.Field{
{Name: "field0", Type: pilosa.FieldTypeInt, Min: 0, Max: 50},
{Name: "field1", Type: pilosa.FieldTypeInt, Min: 1, Max: 2},
},
}); err != nil {
t.Fatal(err)
} else if _, err := index.CreateFrameIfNotExists("xxx", pilosa.FrameOptions{}); err != nil {
t.Fatal(err)
}
// Set field values.
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(columnID=10, frame=f, field0=25, field1=2)`), nil, nil); err != nil {
t.Fatal(err)
} else if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(columnID=100, frame=f, field0=10)`), nil, nil); err != nil {
t.Fatal(err)
}
f := hldr.Frame("i", "f")
if value, exists, err := f.FieldValue(10, "field0"); err != nil {
t.Fatal(err)
} else if !exists {
t.Fatal("expected value to exist")
} else if value != 25 {
t.Fatal("unexpected value: %v", value)
}
if value, exists, err := f.FieldValue(10, "field1"); err != nil {
t.Fatal(err)
} else if !exists {
t.Fatal("expected value to exist")
} else if value != 2 {
t.Fatal("unexpected value: %v", value)
}
if value, exists, err := f.FieldValue(100, "field0"); err != nil {
t.Fatal(err)
} else if !exists {
t.Fatal("expected value to exist")
} else if value != 10 {
t.Fatal("unexpected value: %v", value)
}
})
t.Run("", func(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{})
if _, err := index.CreateFrameIfNotExists("f", pilosa.FrameOptions{
RangeEnabled: true,
Fields: []*pilosa.Field{
{Name: "field0", Type: pilosa.FieldTypeInt, Min: 0, Max: 100},
},
}); err != nil {
t.Fatal(err)
}
t.Run("ErrFrameRequired", func(t *testing.T) {
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(columnID=10, field0=100)`), nil, nil); err == nil || err.Error() != `SetFieldValue() frame required` {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ErrColumnFieldRequired", func(t *testing.T) {
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(invalid_column_name=10, frame=f, field0=100)`), nil, nil); err == nil || err.Error() != `SetFieldValue() column field 'columnID' required` {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ErrColumnFieldValue", func(t *testing.T) {
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(invalid_column_name="bad_column", frame=f, field0=100)`), nil, nil); err == nil || err.Error() != `SetFieldValue() column field 'columnID' required` {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ErrInvalidFieldValueType", func(t *testing.T) {
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(columnID=10, frame=f, field0="hello")`), nil, nil); err == nil || err.Error() != `invalid field value type` {
t.Fatalf("unexpected error: %s", err)
}
})
})
}
// Ensure a SetRowAttrs() query can be executed.
func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
hldr := test.MustOpenHolder()

5
fragment_test.go
View file

@ -176,7 +176,8 @@ func TestFragment_SetFieldValue(t *testing.T) {
if err := quick.Check(func(bitDepth uint, columnN uint64, values []uint64) bool {
// Limit bit depth & maximum values.
bitDepth = (bitDepth % 62) + 1
columnN = (columnN % 100) + 1
columnN = (columnN % 99) + 1
for i := range values {
values[i] = values[i] % (1 << bitDepth)
}
@ -497,7 +498,7 @@ func TestFragment_Checksum(t *testing.T) {
// Ensure new checksum is different.
if chksum := f.Checksum(); bytes.Equal(chksum, orig) {
t.Fatalf("expected checksum to change: %x", chksum, orig)
t.Fatalf("expected checksum to change: %x - %x", chksum, orig)
}
}

30
pilosa.go
View file

@ -17,6 +17,7 @@ package pilosa
import (
"errors"
"regexp"
"strings"
"github.com/pilosa/pilosa/internal"
)
@ -53,6 +54,7 @@ var (
ErrInverseRangeNotAllowed = errors.New("inverse range not allowed")
ErrRangeCacheNotAllowed = errors.New("range cache not allowed")
ErrFrameFieldsNotAllowed = errors.New("frame fields not allowed")
ErrInvalidFieldValueType = errors.New("invalid field value type")
ErrFieldValueTooLow = errors.New("field value too low")
ErrFieldValueTooHigh = errors.New("field value too high")
@ -66,6 +68,14 @@ var (
ErrFragmentNotFound = errors.New("fragment not found")
ErrQueryRequired = errors.New("query required")
ErrTooManyWrites = errors.New("too many write commands")
ErrConfigClusterTypeInvalid = errors.New("invalid cluster type")
ErrConfigClusterTypeMissing = errors.New("missing cluster type")
ErrConfigHostsMissing = errors.New("missing bind address in cluster hosts")
ErrConfigBroadcastPort = errors.New("internal-port not found in internal-hosts")
ErrConfigHostsMismatch = errors.New("hosts and internal-hosts length mismatch")
ErrConfigReplicaNInvalid = errors.New("replica number must be <= hosts")
ErrConfigGossipSeed = errors.New("invalid gossip seed")
)
// Regular expression to validate index and frame names.
@ -142,3 +152,23 @@ func ValidateLabel(label string) error {
}
return nil
}
// StringInSlice checks is substring a is in the slice
func StringInSlice(a string, list []string) bool {
for _, b := range list {
if b == a {
return true
}
}
return false
}
// ContainsSubstring checks is substring a is contained in the slice
func ContainsSubstring(a string, list []string) bool {
for _, b := range list {
if strings.Contains(b, a) {
return true
}
}
return false
}

24
pilosa_test.go
View file

@ -54,3 +54,27 @@ func TestValidateLabelInvalid(t *testing.T) {
}
}
}
func TestStringInSlice(t *testing.T) {
list := []string{"localhost:10101", "localhost:10102", "localhost:10103"}
substr := "localhost:10101"
if !pilosa.StringInSlice(substr, list) {
t.Fatalf("Expected substring %s in %v", substr, list)
}
substr = "10101"
if pilosa.StringInSlice(substr, list) {
t.Fatalf("Expected substring %s not in %v", substr, list)
}
}
func TestContainsSubstring(t *testing.T) {
list := []string{"localhost:10101", "localhost:10102", "localhost:10103"}
substr := "10101"
if !pilosa.ContainsSubstring(substr, list) {
t.Fatalf("Expected substring %s contained in %v", substr, list)
}
substr = "4000"
if pilosa.ContainsSubstring(substr, list) {
t.Fatalf("Expected substring %s in not contained in %v", substr, list)
}
}

22
server/server.go
View file

@ -107,7 +107,11 @@ func (m *Command) Run(args ...string) (err error) {
// SetupServer use the cluster configuration to setup this server
func (m *Command) SetupServer() error {
var err error
err := m.Config.Validate()
if err != nil {
return err
}
cluster := pilosa.NewCluster()
cluster.ReplicaN = m.Config.Cluster.ReplicaN
@ -131,7 +135,7 @@ func (m *Command) SetupServer() error {
// Configure holder.
m.Server.Logger().Printf("Using data from: %s\n", m.Config.DataDir)
m.Server.Holder.Path = m.Config.DataDir
m.Server.MetricInterval = time.Duration(m.Config.Metric.PollingInterval)
m.Server.MetricInterval = time.Duration(m.Config.Metric.PollInterval)
m.Server.Holder.Stats, err = NewStatsClient(m.Config.Metric.Service, m.Config.Metric.Host)
if err != nil {
return err
@ -142,7 +146,7 @@ func (m *Command) SetupServer() error {
// Copy configuration flags.
m.Server.MaxWritesPerRequest = m.Config.MaxWritesPerRequest
m.Server.Host, err = normalizeHost(m.Config.Host)
m.Server.Host, err = normalizeHost(m.Config.Bind)
if err != nil {
return err
}
@ -154,7 +158,7 @@ func (m *Command) SetupServer() error {
}
switch m.Config.Cluster.Type {
case "http":
case pilosa.ClusterHTTP:
m.Server.Broadcaster = httpbroadcast.NewHTTPBroadcaster(m.Server, internalPortStr)
m.Server.BroadcastReceiver = httpbroadcast.NewHTTPBroadcastReceiver(internalPortStr, m.Server.LogOutput)
m.Server.Cluster.NodeSet = httpbroadcast.NewHTTPNodeSet()
@ -162,7 +166,7 @@ func (m *Command) SetupServer() error {
if err != nil {
return err
}
case "gossip":
case pilosa.ClusterGossip:
gossipPort, err := strconv.Atoi(internalPortStr)
if err != nil {
return err
@ -172,15 +176,15 @@ func (m *Command) SetupServer() error {
gossipSeed = m.Config.Cluster.GossipSeed
}
// get the host portion of addr to use for binding
gossipHost, _, err := net.SplitHostPort(m.Config.Host)
gossipHost, _, err := net.SplitHostPort(m.Config.Bind)
if err != nil {
gossipHost = m.Config.Host
gossipHost = m.Config.Bind
}
gossipNodeSet := gossip.NewGossipNodeSet(m.Config.Host, gossipHost, gossipPort, gossipSeed, m.Server)
gossipNodeSet := gossip.NewGossipNodeSet(m.Config.Bind, gossipHost, gossipPort, gossipSeed, m.Server)
m.Server.Cluster.NodeSet = gossipNodeSet
m.Server.Broadcaster = gossipNodeSet
m.Server.BroadcastReceiver = gossipNodeSet
case "static", "":
case pilosa.ClusterStatic, pilosa.ClusterNone:
m.Server.Broadcaster = pilosa.NopBroadcaster
m.Server.Cluster.NodeSet = pilosa.NewStaticNodeSet()
m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver

8
server/server_test.go
View file

@ -346,10 +346,10 @@ func TestMain_FrameRestore(t *testing.T) {
// Ensure the host can be parsed.
func TestConfig_Parse_Host(t *testing.T) {
if c, err := ParseConfig(`host = "local"`); err != nil {
if c, err := ParseConfig(`bind = "local"`); err != nil {
t.Fatal(err)
} else if c.Host != "local" {
t.Fatalf("unexpected host: %s", c.Host)
} else if c.Bind != "local" {
t.Fatalf("unexpected host: %s", c.Bind)
}
}
@ -606,7 +606,7 @@ func NewMain() *Main {
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)}
m.Server.Network = *test.Network
m.Config.DataDir = path
m.Config.Host = "localhost:0"
m.Config.Bind = "localhost:0"
m.Command.Stdin = &m.Stdin
m.Command.Stdout = &m.Stdout
m.Command.Stderr = &m.Stderr