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

Merge branch 'master' into 1492-ae-and-resize

Browse source
This commit is contained in:
Matt Jaffee 2018-08-20 09:13:35 -05:00
commit e16c43c3bc
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
28 changed files with 1259 additions and 243 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
Gopkg.lock generated
View file

@ -154,10 +154,10 @@
revision = "ef8a98b0bbce4a65b5aa4c368430a80ddc533168"
[[projects]]
branch = "master"
name = "github.com/hashicorp/memberlist"
packages = ["."]
revision = "2288bf30e9c8d7b5f6549bf62e07120d72fd4b6c"
revision = "ce8abaa0c60c2d6bee7219f5ddf500e0a1457b28"
version = "v0.1.0"
[[projects]]
name = "github.com/inconshreveable/mousetrap"
@ -324,6 +324,6 @@
[solve-meta]
analyzer-name = "dep"
analyzer-version = 1
inputs-digest = "6ae38dc521f55a89507ec21c575e0b65505de9514d1db6255a1b9b99fcf19ab3"
inputs-digest = "8290156ce8b4066c46ab83d743f4c81df0a17e148415bb1ee8409a51ac4c3ba4"
solver-name = "gps-cdcl"
solver-version = 1

5
Gopkg.toml
View file

@ -17,11 +17,6 @@
# Only one of "branch", "version" or "revision" can be specified.
branch = "master"
[[constraint]]
name = "github.com/hashicorp/memberlist"
branch = "master"
[[constraint]]
name = "github.com/gorilla/handlers"
version = "=1.3.0"

53
api.go
View file

@ -136,9 +136,9 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
}
// Translate column attributes, if necessary.
if api.server.translateFile != nil {
if api.holder.translateFile != nil {
for _, col := range resp.ColumnAttrSets {
v, err := api.server.translateFile.TranslateColumnToString(req.Index, col.ID)
v, err := api.holder.translateFile.TranslateColumnToString(req.Index, col.ID)
if err != nil {
return resp, err
}
@ -391,7 +391,7 @@ func (api *API) FragmentBlockData(_ context.Context, body io.Reader) ([]byte, er
}
// Retrieve fragment from holder.
f := api.holder.fragment(req.Index, req.Field, viewStandard, req.Shard)
f := api.holder.fragment(req.Index, req.Field, req.View, req.Shard)
if f == nil {
return nil, ErrFragmentNotFound
}
@ -409,13 +409,13 @@ func (api *API) FragmentBlockData(_ context.Context, body io.Reader) ([]byte, er
}
// FragmentBlocks returns the checksums and block ids for all blocks in the specified fragment.
func (api *API) FragmentBlocks(_ context.Context, indexName string, fieldName string, shard uint64) ([]FragmentBlock, error) {
func (api *API) FragmentBlocks(_ context.Context, indexName, fieldName, viewName string, shard uint64) ([]FragmentBlock, error) {
if err := api.validate(apiFragmentBlocks); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve fragment from holder.
f := api.holder.fragment(indexName, fieldName, viewStandard, shard)
f := api.holder.fragment(indexName, fieldName, viewName, shard)
if f == nil {
return nil, ErrFragmentNotFound
}
@ -610,11 +610,36 @@ func (api *API) Import(_ context.Context, req *ImportRequest) error {
return errors.Wrap(err, "validating api method")
}
index := api.holder.Index(req.Index)
if index == nil {
return newNotFoundError(ErrIndexNotFound)
}
field, err := api.indexField(req.Index, req.Field, req.Shard)
if err != nil {
return errors.Wrap(err, "getting field")
}
// Translate row keys.
if field.keys() {
if len(req.RowIDs) != 0 {
return errors.New("row ids cannot be used because field uses string keys")
}
if req.RowIDs, err = api.holder.translateFile.TranslateRowsToUint64(index.Name(), field.Name(), req.RowKeys); err != nil {
return errors.Wrap(err, "translating rows")
}
}
// Translate column keys.
if index.Keys() {
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil {
return errors.Wrap(err, "translating columns")
}
}
// Convert timestamps to time.Time.
timestamps := make([]*time.Time, len(req.Timestamps))
for i, ts := range req.Timestamps {
@ -639,10 +664,26 @@ func (api *API) ImportValue(_ context.Context, req *ImportValueRequest) error {
return errors.Wrap(err, "validating api method")
}
index := api.holder.Index(req.Index)
if index == nil {
return newNotFoundError(ErrIndexNotFound)
}
field, err := api.indexField(req.Index, req.Field, req.Shard)
if err != nil {
return errors.Wrap(err, "getting field")
}
// Translate column keys.
if index.Keys() {
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = api.holder.translateFile.TranslateColumnsToUint64(index.Name(), req.ColumnKeys); err != nil {
return errors.Wrap(err, "translating columns")
}
}
// Import into fragment.
err = field.importValue(req.ColumnIDs, req.Values)
if err != nil {
@ -764,7 +805,7 @@ func (api *API) ResizeAbort() error {
// 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)
rc, err := api.holder.translateFile.Reader(ctx, offset)
if err != nil {
return nil, errors.Wrap(err, "read from translate store")
}

21
client.go
View file

@ -18,8 +18,9 @@ type Bit struct {
// FieldValue represents the value for a column within a
// range-encoded field.
type FieldValue struct {
ColumnID uint64
Value int64
ColumnID uint64
ColumnKey string
Value int64
}
// InternalClient should be implemented by any struct that enables any transport between nodes
@ -33,6 +34,7 @@ type InternalClient interface {
Schema(ctx context.Context) ([]*IndexInfo, error)
CreateIndex(ctx context.Context, index string, opt IndexOptions) error
FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error)
Nodes(ctx context.Context) ([]*Node, error)
Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error)
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
Import(ctx context.Context, index, field string, shard uint64, bits []Bit) error
@ -40,10 +42,11 @@ type InternalClient interface {
EnsureIndex(ctx context.Context, name string, options IndexOptions) error
EnsureField(ctx context.Context, indexName string, fieldName string) error
ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue) error
ImportValueK(ctx context.Context, index, field string, vals []FieldValue) error
ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error
CreateField(ctx context.Context, index, field string) error
FragmentBlocks(ctx context.Context, uri *URI, index, field string, shard uint64) ([]FragmentBlock, error)
BlockData(ctx context.Context, uri *URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error)
FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error)
BlockData(ctx context.Context, uri *URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error)
ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
SendMessage(ctx context.Context, uri *URI, msg []byte) error
@ -88,6 +91,9 @@ func (n nopInternalClient) CreateIndex(ctx context.Context, index string, opt In
func (n nopInternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error) {
return nil, nil
}
func (n nopInternalClient) Nodes(ctx context.Context) ([]*Node, error) {
return nil, nil
}
func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
@ -109,14 +115,17 @@ func (n nopInternalClient) EnsureField(ctx context.Context, indexName string, fi
func (n nopInternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue) error {
return nil
}
func (n nopInternalClient) ImportValueK(ctx context.Context, index, field string, vals []FieldValue) error {
return nil
}
func (n nopInternalClient) ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error {
return nil
}
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, shard uint64) ([]FragmentBlock, error) {
func (n nopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error) {
return nil, nil
}
func (n nopInternalClient) BlockData(ctx context.Context, uri *URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) {
func (n nopInternalClient) BlockData(ctx context.Context, uri *URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) {
return nil, nil, nil
}
func (n nopInternalClient) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {

39
cluster.go
View file

@ -169,7 +169,7 @@ type nodeAction struct {
type cluster struct { // nolint: maligned
id string
Node *Node
nodes []*Node // TODO phase this out?
nodes []*Node
// Hashing algorithm used to assign partitions to nodes.
Hasher Hasher
@ -489,12 +489,7 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error {
c.Topology.mu.Unlock()
c.logger.Printf("received state %s (%s)", state, nodeID)
// Set cluster state to NORMAL.
if c.haveTopologyAgreement() && c.allNodesReady() {
return c.unprotectedSetStateAndBroadcast(ClusterStateNormal)
}
return nil
return c.unprotectedSetStateAndBroadcast(c.determineClusterState())
}
// determineClusterState is unprotected.
@ -983,7 +978,6 @@ func (c *cluster) allNodesReady() (ret bool) {
}
func (c *cluster) handleNodeAction(nodeAction nodeAction) error {
c.mu.Lock()
j, err := c.unprotectedGenerateResizeJob(nodeAction)
c.mu.Unlock()
@ -1007,7 +1001,7 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error {
c.logger.Printf("wait for jobResult")
jobResult := <-j.result
// Make sure j.Run() didn't return an error.
// Make sure j.run() didn't return an error.
if eg.Wait() != nil {
return errors.Wrap(err, "running job")
}
@ -1616,11 +1610,6 @@ func (c *cluster) considerTopology() error {
return fmt.Errorf("coordinator %s is not in topology: %v", c.Node.ID, c.Topology.nodeIDs)
}
// If local node is the only thing in .topology, continue.
//if len(c.Topology.NodeIDs) == 1 {
// return nil
//}
// Keep the cluster in state "STARTING" until hearing from all nodes.
// Topology contains 2+ hosts.
return nil
@ -1832,6 +1821,10 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
}
}
// If the cluster membership has changed, reset the primary for
// translate store replication.
c.holder.setPrimaryTranslateStore(c.unprotectedPreviousNode())
c.unprotectedSetState(cs.State)
c.markAsJoined()
@ -1839,6 +1832,24 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
return nil
}
// unprotectedPreviousNode returns the node listed before the current node in c.Nodes.
// If there is only one node in the cluster, returns nil.
// If the current node is the first node in the list, returns the last node.
func (c *cluster) unprotectedPreviousNode() *Node {
if len(c.nodes) <= 1 {
return nil
}
pos := c.nodePositionByID(c.Node.ID)
if pos == -1 {
return nil
} else if pos == 0 {
return c.nodes[len(c.nodes)-1]
} else {
return c.nodes[pos-1]
}
}
// setStatic is unprotected, but only called before the cluster has been started
// (and therefore not concurrently).
func (c *cluster) setStatic(hosts []string) error {

54
cluster_internal_test.go
View file

@ -450,6 +450,60 @@ func TestCluster_Nodes(t *testing.T) {
})
}
func TestCluster_PreviousNode(t *testing.T) {
node0 := &Node{ID: "node0"}
node1 := &Node{ID: "node1"}
node2 := &Node{ID: "node2"}
t.Run("OneNode", func(t *testing.T) {
c := newCluster()
c.addNodeBasicSorted(node0)
c.Node = node0
if prev := c.unprotectedPreviousNode(); prev != nil {
t.Errorf("expected: nil, but got: %v", prev)
}
})
t.Run("TwoNode", func(t *testing.T) {
c := newCluster()
c.addNodeBasicSorted(node0)
c.addNodeBasicSorted(node1)
c.Node = node0
if prev := c.unprotectedPreviousNode(); prev != node1 {
t.Errorf("expected: node1, but got: %v", prev)
}
c.Node = node1
if prev := c.unprotectedPreviousNode(); prev != node0 {
t.Errorf("expected: node0, but got: %v", prev)
}
})
t.Run("ThreeNode", func(t *testing.T) {
c := newCluster()
c.addNodeBasicSorted(node0)
c.addNodeBasicSorted(node1)
c.addNodeBasicSorted(node2)
c.Node = node0
if prev := c.unprotectedPreviousNode(); prev != node2 {
t.Errorf("expected: node2, but got: %v", prev)
}
c.Node = node1
if prev := c.unprotectedPreviousNode(); prev != node0 {
t.Errorf("expected: node0, but got: %v", prev)
}
c.Node = node2
if prev := c.unprotectedPreviousNode(); prev != node1 {
t.Errorf("expected: node1, but got: %v", prev)
}
})
}
// NEXT: move this test to internal and unexport IsCoordinator
func TestCluster_Coordinator(t *testing.T) {
uri1 := NewTestURIFromHostPort("node1", 0)

2
cmd/import.go
View file

@ -51,7 +51,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM.
flags.StringVarP(&Importer.Host, "host", "", "localhost:10101", "host:port of Pilosa.")
flags.StringVarP(&Importer.Index, "index", "i", "", "Pilosa index to import into.")
flags.StringVarP(&Importer.Field, "field", "f", "", "Field to import into.")
flags.BoolVar(&Importer.StringKeys, "string-keys", false, "Treat payload as string keys.")
flags.BoolVar(&Importer.StringKeys, "string-keys", false, "REMOVED (key type is now determined by index/field configuration): Treat payload as string keys.")
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.")

183
ctl/import.go
View file

@ -46,7 +46,8 @@ type ImportCommand struct { // nolint: maligned
// CreateSchema ensures the schema exists before import
CreateSchema bool
// Indicates that the payload should be treated as string keys.
// REMOVED: Indicates that the payload should be treated as string keys.
// TODO: remove this in a future release
StringKeys bool `json:"StringKeys"`
// Filenames to import from.
@ -79,6 +80,11 @@ func NewImportCommand(stdin io.Reader, stdout, stderr io.Writer) *ImportCommand
func (cmd *ImportCommand) Run(ctx context.Context) error {
logger := log.New(cmd.Stderr, "", log.LstdFlags)
// REMOVED: warning that --string-keys flag has been deprecated.
if cmd.StringKeys {
logger.Printf("REMOVED: The string-keys flag is no longer used.")
}
// Validate arguments.
// Index and field are validated early before the files are parsed.
if cmd.Index == "" {
@ -108,20 +114,26 @@ func (cmd *ImportCommand) Run(ctx context.Context) error {
if err != nil {
return errors.Wrap(err, "getting schema")
}
var useColumnKeys, useRowKeys bool
for _, index := range schema {
if index.Name == cmd.Index {
useColumnKeys = index.Options.Keys
for _, field := range index.Fields {
if field.Name == cmd.Field {
useRowKeys = field.Options.Keys
fieldType = field.Options.Type
break
}
}
break
}
}
// Import each path and import by shard.
for _, path := range cmd.Paths {
logger.Printf("parsing: %s", path)
if err := cmd.importPath(ctx, fieldType, path); err != nil {
if err := cmd.importPath(ctx, fieldType, useColumnKeys, useRowKeys, path); err != nil {
return err
}
}
@ -142,21 +154,16 @@ func (cmd *ImportCommand) ensureSchema(ctx context.Context) error {
}
// importPath parses a path into bits and imports it to the server.
func (cmd *ImportCommand) importPath(ctx context.Context, fieldType, path string) error {
func (cmd *ImportCommand) importPath(ctx context.Context, fieldType string, useColumnKeys, useRowKeys bool, path string) error {
// If fieldType is `int`, treat the import data as values to be range-encoded.
if fieldType == pilosa.FieldTypeInt {
return cmd.bufferValues(ctx, path)
} else {
if cmd.StringKeys {
return cmd.bufferBitsK(ctx, path)
} else {
return cmd.bufferBits(ctx, path)
}
return cmd.bufferValues(ctx, useColumnKeys, path)
}
return cmd.bufferBits(ctx, useColumnKeys, useRowKeys, path)
}
// bufferBits buffers slices of bits to be imported as a batch.
func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error {
func (cmd *ImportCommand) bufferBits(ctx context.Context, useColumnKeys, useRowKeys bool, path string) error {
a := make([]pilosa.Bit, 0, cmd.BufferSize)
var r *csv.Reader
@ -198,18 +205,22 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error {
var bit pilosa.Bit
// Parse row id.
rowID, err := strconv.ParseUint(record[0], 10, 64)
if err != nil {
return fmt.Errorf("invalid row id on row %d: %q", rnum, record[0])
if useRowKeys {
bit.RowKey = record[0]
} else {
if bit.RowID, err = strconv.ParseUint(record[0], 10, 64); err != nil {
return fmt.Errorf("invalid row id on row %d: %q", rnum, record[0])
}
}
bit.RowID = rowID
// Parse column id.
columnID, err := strconv.ParseUint(record[1], 10, 64)
if err != nil {
return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1])
if useColumnKeys {
bit.ColumnKey = record[1]
} else {
if bit.ColumnID, err = strconv.ParseUint(record[1], 10, 64); err != nil {
return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1])
}
}
bit.ColumnID = columnID
// Parse time, if exists.
if len(record) > 2 && record[2] != "" {
@ -224,7 +235,7 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error {
// If we've reached the buffer size then import bits.
if len(a) == cmd.BufferSize {
if err := cmd.importBits(ctx, a); err != nil {
if err := cmd.importBits(ctx, useColumnKeys, useRowKeys, a); err != nil {
return err
}
a = a[:0]
@ -232,13 +243,22 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error {
}
// If there are still bits in the buffer then flush them.
return cmd.importBits(ctx, a)
return cmd.importBits(ctx, useColumnKeys, useRowKeys, a)
}
// importBits sends batches of bits to the server.
func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) error {
func (cmd *ImportCommand) importBits(ctx context.Context, useColumnKeys, useRowKeys bool, bits []pilosa.Bit) error {
logger := log.New(cmd.Stderr, "", log.LstdFlags)
// If keys are used, all bits are sent to the primary translate store (i.e. coordinator).
if useColumnKeys || useRowKeys {
logger.Printf("importing keys: n=%d", len(bits))
if err := cmd.client.ImportK(ctx, cmd.Index, cmd.Field, bits); err != nil {
return errors.Wrap(err, "importing keys")
}
return nil
}
// Group bits by shard.
logger.Printf("grouping %d bits", len(bits))
bitsByShard := http.Bits(bits).GroupByShard()
@ -258,100 +278,8 @@ func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) err
return nil
}
// bufferBitsK buffers slices of keys to be imported as a batch.
func (cmd *ImportCommand) bufferBitsK(ctx context.Context, path string) error {
a := make([]pilosa.Bit, 0, cmd.BufferSize)
var r *csv.Reader
if path != "-" {
// Open file for reading.
f, err := os.Open(path)
if err != nil {
return errors.Wrap(err, "opening file")
}
defer f.Close()
// Read rows as bits.
r = csv.NewReader(f)
} else {
r = csv.NewReader(cmd.Stdin)
}
r.FieldsPerRecord = -1
rnum := 0
for {
rnum++
// Read CSV row.
record, err := r.Read()
if err == io.EOF {
break
} else if err != nil {
return errors.Wrap(err, "reading")
}
// Ignore blank rows.
if record[0] == "" {
continue
} else if len(record) < 2 {
return fmt.Errorf("bad column count on row %d: col=%d", rnum, len(record))
}
var bit pilosa.Bit
// Parse row key.
if record[0] == "" {
return fmt.Errorf("invalid row key on row %d: %q", rnum, record[0])
}
bit.RowKey = record[0]
// Parse column key.
if record[1] == "" {
return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1])
}
bit.ColumnKey = record[1]
// Parse time, if exists.
if len(record) > 2 && record[2] != "" {
t, err := time.Parse(pilosa.TimeFormat, record[2])
if err != nil {
return fmt.Errorf("invalid timestamp on row %d: %q", rnum, record[2])
}
bit.Timestamp = t.UnixNano()
}
a = append(a, bit)
// If we've reached the buffer size then import bits.
if len(a) == cmd.BufferSize {
if err := cmd.importBitsK(ctx, a); err != nil {
return err
}
a = a[:0]
}
}
// If there are still bitKs in the buffer then flush them.
return cmd.importBitsK(ctx, a)
}
// importBitsK sends batches of bitKs to the server.
func (cmd *ImportCommand) importBitsK(ctx context.Context, bits []pilosa.Bit) error {
logger := log.New(cmd.Stderr, "", log.LstdFlags)
// TODO: does it help to sort the rowKeys?
logger.Printf("importing keys: n=%d", len(bits))
if err := cmd.client.ImportK(ctx, cmd.Index, cmd.Field, bits); err != nil {
return errors.Wrap(err, "importing keys")
}
return nil
}
// bufferValues buffers slices of FieldValues to be imported as a batch.
func (cmd *ImportCommand) bufferValues(ctx context.Context, path string) error {
func (cmd *ImportCommand) bufferValues(ctx context.Context, useColumnKeys bool, path string) error {
a := make([]pilosa.FieldValue, 0, cmd.BufferSize)
var r *csv.Reader
@ -393,11 +321,13 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, path string) error {
var val pilosa.FieldValue
// Parse column id.
columnID, err := strconv.ParseUint(record[0], 10, 64)
if err != nil {
return fmt.Errorf("invalid column id on row %d: %q", rnum, record[0])
if useColumnKeys {
val.ColumnKey = record[0]
} else {
if val.ColumnID, err = strconv.ParseUint(record[0], 10, 64); err != nil {
return fmt.Errorf("invalid column id on row %d: %q", rnum, record[0])
}
}
val.ColumnID = columnID
// Parse FieldValue.
value, err := strconv.ParseInt(record[1], 10, 64)
@ -410,7 +340,7 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, path string) error {
// If we've reached the buffer size then import FieldValues.
if len(a) == cmd.BufferSize {
if err := cmd.importValues(ctx, a); err != nil {
if err := cmd.importValues(ctx, useColumnKeys, a); err != nil {
return err
}
a = a[:0]
@ -418,13 +348,22 @@ func (cmd *ImportCommand) bufferValues(ctx context.Context, path string) error {
}
// If there are still values in the buffer then flush them.
return cmd.importValues(ctx, a)
return cmd.importValues(ctx, useColumnKeys, a)
}
// importValues sends batches of FieldValues to the server.
func (cmd *ImportCommand) importValues(ctx context.Context, vals []pilosa.FieldValue) error {
func (cmd *ImportCommand) importValues(ctx context.Context, useColumnKeys bool, vals []pilosa.FieldValue) error {
logger := log.New(cmd.Stderr, "", log.LstdFlags)
// If keys are used, all values are sent to the primary translate store (i.e. coordinator).
if useColumnKeys {
logger.Printf("importing keyed values: n=%d", len(vals))
if err := cmd.client.ImportValueK(ctx, cmd.Index, cmd.Field, vals); err != nil {
return errors.Wrap(err, "importing keys")
}
return nil
}
// Group vals by shard.
logger.Printf("grouping %d vals", len(vals))
valsByShard := http.FieldValues(vals).GroupByShard()

54
ctl/import_test.go
View file

@ -101,6 +101,60 @@ func TestImportCommand_RunValue(t *testing.T) {
}
}
// Ensure that import with keys runs.
func TestImportCommand_RunKeys(t *testing.T) {
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)
cm := NewImportCommand(stdin, stdout, stderr)
file, err := ioutil.TempFile("", "import-key.csv")
file.Write([]byte("foo1,bar2\nfoo3,bar4\nfoo5,bar6"))
ctx := context.Background()
if err != nil {
t.Fatal(err)
}
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.API.Node().URI.HostPort()
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader(`{"options":{"keys": true}}`)))
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i/field/f", strings.NewReader(`{"options":{"keys": true}}`)))
cm.Index = "i"
cm.Field = "f"
cm.Paths = []string{file.Name()}
err = cm.Run(ctx)
if err != nil {
t.Fatalf("Import Run with keys doesn't work: %s", err)
}
}
// Ensure that integer import with keys runs.
func TestImportCommand_RunValueKeys(t *testing.T) {
buf := bytes.Buffer{}
stdin, stdout, stderr := GetIO(buf)
cm := NewImportCommand(stdin, stdout, stderr)
file, err := ioutil.TempFile("", "import-key.csv")
file.Write([]byte("foo1,2\nfoo3,4\nfoo5,6"))
ctx := context.Background()
if err != nil {
t.Fatal(err)
}
cmd := test.MustRunCluster(t, 1)[0]
cm.Host = cmd.API.Node().URI.HostPort()
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader(`{"options":{"keys": true}}`)))
http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i/field/f", strings.NewReader(`{"options":{"type": "int", "min": 0, "max": 100}}`)))
cm.Index = "i"
cm.Field = "f"
cm.Paths = []string{file.Name()}
err = cm.Run(ctx)
if err != nil {
t.Fatalf("Import Run with keys doesn't work: %s", err)
}
}
func TestImportCommand_InvalidFile(t *testing.T) {
cmd := test.MustRunCluster(t, 1)[0]

2
ctl/server.go
View file

@ -44,7 +44,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Duration that will trigger log and stat messages for slow queries.")
// Translation
flags.StringVarP(&srv.Config.Translation.PrimaryURL, "translation.primary-url", "", srv.Config.Translation.PrimaryURL, "URL for primary translation node for replication.")
flags.StringVarP(&srv.Config.Translation.PrimaryURL, "translation.primary-url", "", srv.Config.Translation.PrimaryURL, "DEPRECATED: URL for primary translation node for replication.")
// Gossip
flags.StringVarP(&srv.Config.Gossip.Port, "gossip.port", "", srv.Config.Gossip.Port, "Port to which pilosa should bind for internal state sharing.")

1
encoding/proto/proto.go
View file

@ -890,6 +890,7 @@ func decodeRow(pr *internal.Row) *pilosa.Row {
r := pilosa.NewRow()
r.Attrs = decodeAttrs(pr.Attrs)
r.Keys = pr.Keys
for _, v := range pr.Columns {
r.SetBit(v)
}

4
fragment.go
View file

@ -1837,7 +1837,7 @@ func (s *fragmentSyncer) syncFragment() error {
}
// Retrieve remote blocks.
blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), &node.URI, s.Fragment.index, s.Fragment.field, s.Fragment.shard)
blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), &node.URI, s.Fragment.index, s.Fragment.field, s.Fragment.view, s.Fragment.shard)
if err != nil && err != ErrFragmentNotFound {
return errors.Wrap(err, "getting blocks")
}
@ -1916,7 +1916,7 @@ func (s *fragmentSyncer) syncBlock(id int) error {
uris = append(uris, uri)
// Only sync the standard block.
rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(context.Background(), &node.URI, f.index, f.field, f.shard, id)
rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(context.Background(), &node.URI, f.index, f.field, f.view, f.shard, id)
if err != nil {
return errors.Wrap(err, "getting block")
}

24
holder.go
View file

@ -46,6 +46,10 @@ type Holder struct {
// Indexes by name.
indexes map[string]*Index
// Key/ID translation
translateFile *TranslateFile
NewPrimaryTranslateStore func(interface{}) TranslateStore
// opened channel is closed once Open() completes.
opened chan struct{}
@ -77,6 +81,9 @@ func NewHolder() *Holder {
opened: make(chan struct{}),
translateFile: NewTranslateFile(),
NewPrimaryTranslateStore: newNopTranslateStore,
broadcaster: NopBroadcaster,
Stats: NopStatsClient,
@ -160,6 +167,13 @@ func (h *Holder) Close() error {
return errors.Wrap(err, "closing index")
}
}
if h.translateFile != nil {
if err := h.translateFile.Close(); err != nil {
return err
}
}
return nil
}
@ -560,6 +574,14 @@ func (h *Holder) logStartup() error {
return nil
}
func (h *Holder) setPrimaryTranslateStore(node *Node) {
var nodeID string
if node != nil {
nodeID = node.ID
}
h.translateFile.SetPrimaryStore(nodeID, h.NewPrimaryTranslateStore(node))
}
// holderSyncer is an active anti-entropy tool that compares the local holder
// with a remote holder based on block checksums and resolves differences.
type holderSyncer struct {
@ -638,7 +660,7 @@ func (s *holderSyncer) SyncHolder() error {
// Sync fragment if own it.
if err := s.syncFragment(di.Name, fi.Name, vi.Name, shard); err != nil {
return fmt.Errorf("fragment sync error: index=%s, field=%s, shard=%d, err=%s", di.Name, fi.Name, shard, err)
return fmt.Errorf("fragment sync error: index=%s, field=%s, view=%s, shard=%d, err=%s", di.Name, fi.Name, vi.Name, shard, err)
}
}
}

54
holder_test.go
View file

@ -22,6 +22,7 @@ import (
"reflect"
"strings"
"testing"
"time"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/test"
@ -362,3 +363,56 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
}
}
}
// Ensure holder can sync time quantum views with a remote holder.
func TestHolderSyncer_TimeQuantum(t *testing.T) {
c := test.MustNewCluster(t, 2)
c[0].Config.Cluster.ReplicaN = 2
c[0].Config.AntiEntropy.Interval = 0
c[1].Config.Cluster.ReplicaN = 2
c[1].Config.AntiEntropy.Interval = 0
err := c.Start()
if err != nil {
t.Fatalf("starting cluster: %v", err)
}
defer c.Close()
quantum := "D"
_, err = c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
if err != nil {
t.Fatalf("creating index i: %v", err)
}
_, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeTime(pilosa.TimeQuantum(quantum)))
if err != nil {
t.Fatalf("creating field f: %v", err)
}
hldr0 := &test.Holder{Holder: c[0].Server.Holder()}
hldr1 := &test.Holder{Holder: c[1].Server.Holder()}
// Set data on the local holder.
t1 := time.Date(2018, 8, 1, 12, 30, 0, 0, time.UTC)
t2 := time.Date(2018, 8, 2, 12, 30, 0, 0, time.UTC)
hldr0.SetBitTime("i", "f", 0, 1, &t1)
hldr0.SetBitTime("i", "f", 0, 2, &t2)
err = c[0].Server.SyncData()
if err != nil {
t.Fatalf("syncing node 0: %v", err)
}
err = c[1].Server.SyncData()
if err != nil {
t.Fatalf("syncing node 1: %v", err)
}
// Verify data is the same on both nodes.
for i, hldr := range []*test.Holder{hldr0, hldr1} {
if a := hldr.RowTime("i", "f", 0, t1, quantum).Columns(); !reflect.DeepEqual(a, []uint64{1}) {
t.Errorf("unexpected columns(%d/0): %+v", i, a)
}
if a := hldr.RowTime("i", "f", 0, t2, quantum).Columns(); !reflect.DeepEqual(a, []uint64{2}) {
t.Errorf("unexpected columns(%d/0): %+v", i, a)
}
}
}

189
http/client.go
View file

@ -207,6 +207,37 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
return a, nil
}
// Nodes returns a list of all nodes.
func (c *InternalClient) Nodes(ctx context.Context) ([]*pilosa.Node, error) {
// Execute request against the host.
u := uriPathToURL(c.defaultURI, "/internal/nodes")
// Build request.
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("Accept", "application/json")
// Execute request.
resp, err := c.httpClient.Do(req.WithContext(ctx))
if err != nil {
return nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
var a []*pilosa.Node
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("http: status=%d", resp.StatusCode)
} else if err := json.NewDecoder(resp.Body).Decode(&a); err != nil {
return nil, fmt.Errorf("json decode: %s", err)
}
return a, nil
}
// Query executes query against the index.
func (c *InternalClient) Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
return c.QueryNode(ctx, c.defaultURI, index, queryRequest)
@ -291,26 +322,42 @@ func (c *InternalClient) Import(ctx context.Context, index, field string, shard
return nil
}
func getCoordinatorNode(nodes []*pilosa.Node) *pilosa.Node {
for _, node := range nodes {
if node.IsCoordinator {
return node
}
}
return nil
}
// ImportK bulk imports bits specified by string keys to a host.
func (c *InternalClient) ImportK(ctx context.Context, index, field string, columns []pilosa.Bit) error {
func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits []pilosa.Bit) error {
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
return pilosa.ErrFieldRequired
}
buf, err := c.marshalImportPayloadK(index, field, columns)
buf, err := c.marshalImportPayload(index, field, 0, bits)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
}
node := &pilosa.Node{
URI: *c.defaultURI,
// Get the coordinator node; all bits are sent to the
// primary translate store (i.e. coordinator).
nodes, err := c.Nodes(ctx)
if err != nil {
return fmt.Errorf("getting nodes: %s", err)
}
coord := getCoordinatorNode(nodes)
if coord == nil {
return fmt.Errorf("could not find the coordinator node")
}
// Import to node.
if err := c.importNode(ctx, node, index, field, buf); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", node.URI, err)
if err := c.importNode(ctx, coord, index, field, buf); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err)
}
return nil
@ -336,7 +383,9 @@ func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fiel
func (c *InternalClient) marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit) ([]byte, error) {
// Separate row and column IDs to reduce allocations.
rowIDs := Bits(bits).RowIDs()
rowKeys := Bits(bits).RowKeys()
columnIDs := Bits(bits).ColumnIDs()
columnKeys := Bits(bits).ColumnKeys()
timestamps := Bits(bits).Timestamps()
// Marshal data to protobuf.
@ -345,27 +394,8 @@ func (c *InternalClient) marshalImportPayload(index, field string, shard uint64,
Field: field,
Shard: shard,
RowIDs: rowIDs,
ColumnIDs: columnIDs,
Timestamps: timestamps,
})
if err != nil {
return nil, fmt.Errorf("marshal import request: %s", err)
}
return buf, nil
}
// marshalImportPayloadK marshalls the import parameters into a protobuf byte slice.
func (c *InternalClient) marshalImportPayloadK(index, field string, bits []pilosa.Bit) ([]byte, error) {
// Separate row and column IDs to reduce allocations.
rowKeys := Bits(bits).RowKeys()
columnKeys := Bits(bits).ColumnKeys()
timestamps := Bits(bits).Timestamps()
// Marshal data to protobuf.
buf, err := c.serializer.Marshal(&pilosa.ImportRequest{
Index: index,
Field: field,
RowKeys: rowKeys,
ColumnIDs: columnIDs,
ColumnKeys: columnKeys,
Timestamps: timestamps,
})
@ -443,19 +473,53 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s
return nil
}
// ImportValueK bulk imports keyed field values to a host.
func (c *InternalClient) ImportValueK(ctx context.Context, index, field string, vals []pilosa.FieldValue) error {
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
return pilosa.ErrFieldRequired
}
buf, err := c.marshalImportValuePayload(index, field, 0, vals)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
}
// Get the coordinator node; all bits are sent to the
// primary translate store (i.e. coordinator).
nodes, err := c.Nodes(ctx)
if err != nil {
return fmt.Errorf("getting nodes: %s", err)
}
coord := getCoordinatorNode(nodes)
if coord == nil {
return fmt.Errorf("could not find the coordinator node")
}
// Import to node.
if err := c.importNode(ctx, coord, index, field, buf); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", coord.URI, err)
}
return nil
}
// marshalImportValuePayload marshalls the import parameters into a protobuf byte slice.
func (c *InternalClient) marshalImportValuePayload(index, field string, shard uint64, vals []pilosa.FieldValue) ([]byte, error) {
// Separate row and column IDs to reduce allocations.
columnIDs := FieldValues(vals).ColumnIDs()
columnKeys := FieldValues(vals).ColumnKeys()
values := FieldValues(vals).Values()
// Marshal data to protobuf.
buf, err := c.serializer.Marshal(&pilosa.ImportValueRequest{
Index: index,
Field: field,
Shard: shard,
ColumnIDs: columnIDs,
Values: values,
Index: index,
Field: field,
Shard: shard,
ColumnIDs: columnIDs,
ColumnKeys: columnKeys,
Values: values,
})
if err != nil {
return nil, fmt.Errorf("marshal import request: %s", err)
@ -624,7 +688,7 @@ func (c *InternalClient) CreateField(ctx context.Context, index, field string) e
// FragmentBlocks returns a list of block checksums for a fragment on a host.
// Only returns blocks which contain data.
func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64) ([]pilosa.FragmentBlock, error) {
func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64) ([]pilosa.FragmentBlock, error) {
if uri == nil {
uri = c.defaultURI
}
@ -632,6 +696,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in
u.RawQuery = url.Values{
"index": {index},
"field": {field},
"view": {view},
"shard": {strconv.FormatUint(shard, 10)},
}.Encode()
@ -669,13 +734,14 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in
}
// BlockData returns row/column id pairs for a block.
func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) {
func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) {
if uri == nil {
panic("need to pass a URI to BlockData")
}
buf, err := c.serializer.Marshal(&pilosa.BlockDataRequest{
Index: index,
Field: field,
View: view,
Shard: shard,
Block: uint64(block),
})
@ -858,8 +924,31 @@ func (p Bits) Less(i, j int) bool {
return p[i].RowID < p[j].RowID
}
// HasRowKeys returns true if any values use a row key.
func (p Bits) HasRowKeys() bool {
for i := range p {
if p[i].RowKey != "" {
return true
}
}
return false
}
// HasColumnKeys returns true if any values use a column key.
func (p Bits) HasColumnKeys() bool {
for i := range p {
if p[i].ColumnKey != "" {
return true
}
}
return false
}
// RowIDs returns a slice of all the row IDs.
func (p Bits) RowIDs() []uint64 {
if p.HasRowKeys() {
return nil
}
other := make([]uint64, len(p))
for i := range p {
other[i] = p[i].RowID
@ -869,6 +958,9 @@ func (p Bits) RowIDs() []uint64 {
// ColumnIDs returns a slice of all the column IDs.
func (p Bits) ColumnIDs() []uint64 {
if p.HasColumnKeys() {
return nil
}
other := make([]uint64, len(p))
for i := range p {
other[i] = p[i].ColumnID
@ -878,6 +970,9 @@ func (p Bits) ColumnIDs() []uint64 {
// RowKeys returns a slice of all the row keys.
func (p Bits) RowKeys() []string {
if !p.HasRowKeys() {
return nil
}
other := make([]string, len(p))
for i := range p {
other[i] = p[i].RowKey
@ -887,6 +982,9 @@ func (p Bits) RowKeys() []string {
// ColumnKeys returns a slice of all the column keys.
func (p Bits) ColumnKeys() []string {
if !p.HasColumnKeys() {
return nil
}
other := make([]string, len(p))
for i := range p {
other[i] = p[i].ColumnKey
@ -929,8 +1027,21 @@ func (p FieldValues) Less(i, j int) bool {
return p[i].ColumnID < p[j].ColumnID
}
// HasColumnKeys returns true if any values use a column key.
func (p FieldValues) HasColumnKeys() bool {
for i := range p {
if p[i].ColumnKey != "" {
return true
}
}
return false
}
// ColumnIDs returns a slice of all the column IDs.
func (p FieldValues) ColumnIDs() []uint64 {
if p.HasColumnKeys() {
return nil
}
other := make([]uint64, len(p))
for i := range p {
other[i] = p[i].ColumnID
@ -938,6 +1049,18 @@ func (p FieldValues) ColumnIDs() []uint64 {
return other
}
// ColumnKeys returns a slice of all the column keys.
func (p FieldValues) ColumnKeys() []string {
if !p.HasColumnKeys() {
return nil
}
other := make([]string, len(p))
for i := range p {
other[i] = p[i].ColumnKey
}
return other
}
// Values returns a slice of all the values.
func (p FieldValues) Values() []int64 {
other := make([]int64, len(p))

235
http/client_test.go
View file

@ -20,6 +20,7 @@ import (
gohttp "net/http"
"reflect"
"testing"
"time"
"github.com/davecgh/go-spew/spew"
"github.com/pilosa/pilosa"
@ -128,11 +129,6 @@ func TestClient_MultiNode(t *testing.T) {
Remote: false,
}
_, err = client[0].Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
result, err := client[0].Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
@ -202,6 +198,231 @@ func TestClient_Import(t *testing.T) {
}
}
// Ensure client can bulk import data.
func TestClient_ImportKeys(t *testing.T) {
t.Run("SingleNode", func(t *testing.T) {
cmd := test.MustRunCluster(t, 1)[0]
host := cmd.URL()
cmd.MustCreateIndex(t, "keyed", pilosa.IndexOptions{Keys: true})
cmd.MustCreateIndex(t, "unkeyed", pilosa.IndexOptions{Keys: false})
cmd.MustCreateField(t, "keyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
cmd.MustCreateField(t, "keyed", "unkeyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000))
cmd.MustCreateField(t, "unkeyed", "keyedf", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
t.Run("Import keyed,keyed", func(t *testing.T) {
if err := c.Import(context.Background(), "keyed", "keyedf", 0, []pilosa.Bit{
{RowKey: "green", ColumnKey: "eve"},
{RowKey: "green", ColumnKey: "alice"},
{RowKey: "green", ColumnKey: "bob"},
{RowKey: "blue", ColumnKey: "eve"},
{RowKey: "blue", ColumnKey: "alice"},
{RowKey: "purple", ColumnKey: "eve"},
}); err != nil {
t.Fatal(err)
}
cmd.MustRecalculateCaches(t)
resp := cmd.MustQuery(t, &pilosa.QueryRequest{
Index: "keyed",
Query: "TopN(keyedf)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{Key: "green", Count: 3},
{Key: "blue", Count: 2},
{Key: "purple", Count: 1},
}) {
t.Fatalf("unexpected topn result: %v", pairs)
}
})
t.Run("Import keyed,unkeyedf", func(t *testing.T) {
if err := c.Import(context.Background(), "keyed", "unkeyedf", 0, []pilosa.Bit{
{RowID: 1, ColumnKey: "eve"},
{RowID: 1, ColumnKey: "alice"},
{RowID: 1, ColumnKey: "bob"},
{RowID: 2, ColumnKey: "eve"},
{RowID: 2, ColumnKey: "alice"},
{RowID: 3, ColumnKey: "eve"},
}); err != nil {
t.Fatal(err)
}
cmd.MustRecalculateCaches(t)
resp := cmd.MustQuery(t, &pilosa.QueryRequest{
Index: "keyed",
Query: "TopN(unkeyedf)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{ID: 1, Count: 3},
{ID: 2, Count: 2},
{ID: 3, Count: 1},
}) {
t.Fatalf("unexpected topn result: %v", pairs)
}
})
t.Run("Import unkeyed,keyed", func(t *testing.T) {
if err := c.Import(context.Background(), "unkeyed", "keyedf", 0, []pilosa.Bit{
{RowKey: "green", ColumnID: 1},
{RowKey: "green", ColumnID: 2},
{RowKey: "green", ColumnID: 3},
{RowKey: "blue", ColumnID: 1},
{RowKey: "blue", ColumnID: 2},
{RowKey: "purple", ColumnID: 1},
}); err != nil {
t.Fatal(err)
}
cmd.MustRecalculateCaches(t)
resp := cmd.MustQuery(t, &pilosa.QueryRequest{
Index: "unkeyed",
Query: "TopN(keyedf)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{Key: "green", Count: 3},
{Key: "blue", Count: 2},
{Key: "purple", Count: 1},
}) {
t.Fatalf("unexpected topn result: %v", pairs)
}
})
})
t.Run("MultiNode", func(t *testing.T) {
cluster := test.MustRunCluster(t, 2)
cmd0 := cluster[0]
cmd1 := cluster[1]
host0 := cmd0.URL()
host1 := cmd1.URL()
cmd0.MustCreateIndex(t, "keyed", pilosa.IndexOptions{Keys: true})
cmd0.MustCreateField(t, "keyed", "keyedf0", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
cmd0.MustCreateField(t, "keyed", "keyedf1", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000), pilosa.OptFieldKeys())
// Send import request.
c0 := MustNewClient(host0, http.GetHTTPClient(nil))
c1 := MustNewClient(host1, http.GetHTTPClient(nil))
// Import to node0.
t.Run("Import node0", func(t *testing.T) {
if err := c0.ImportK(context.Background(), "keyed", "keyedf0", []pilosa.Bit{
{RowKey: "green", ColumnKey: "eve"},
{RowKey: "green", ColumnKey: "alice"},
{RowKey: "green", ColumnKey: "bob"},
{RowKey: "blue", ColumnKey: "eve"},
{RowKey: "blue", ColumnKey: "alice"},
{RowKey: "purple", ColumnKey: "eve"},
}); err != nil {
t.Fatal(err)
}
cmd0.MustRecalculateCaches(t)
resp := cmd0.MustQuery(t, &pilosa.QueryRequest{
Index: "keyed",
Query: "TopN(keyedf0)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{Key: "green", Count: 3},
{Key: "blue", Count: 2},
{Key: "purple", Count: 1},
}) {
t.Fatalf("unexpected topn result: %v", pairs)
}
})
// Import to node1 (ensure import is routed to coordinator for translation).
t.Run("Import node1", func(t *testing.T) {
if err := c1.ImportK(context.Background(), "keyed", "keyedf1", []pilosa.Bit{
{RowKey: "green", ColumnKey: "eve"},
{RowKey: "green", ColumnKey: "alice"},
{RowKey: "green", ColumnKey: "bob"},
{RowKey: "blue", ColumnKey: "eve"},
{RowKey: "blue", ColumnKey: "alice"},
{RowKey: "purple", ColumnKey: "eve"},
}); err != nil {
t.Fatal(err)
}
// Wait for translation replication.
time.Sleep(500 * time.Millisecond)
cmd1.MustRecalculateCaches(t)
resp := cmd1.MustQuery(t, &pilosa.QueryRequest{
Index: "keyed",
Query: "TopN(keyedf1)",
})
if pairs, ok := resp.Results[0].([]pilosa.Pair); !ok {
t.Fatalf("unexpected response type %T", resp.Results[0])
} else if !reflect.DeepEqual(pairs, []pilosa.Pair{
{Key: "green", Count: 3},
{Key: "blue", Count: 2},
{Key: "purple", Count: 1},
}) {
t.Fatalf("unexpected topn result: %v", pairs)
}
})
})
t.Run("IntegerFieldSingleNode", func(t *testing.T) {
cmd := test.MustRunCluster(t, 1)[0]
host := cmd.URL()
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
fldName := "f"
// Load bitmap into cache to ensure cache gets updated.
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{Keys: true})
field, err := index.CreateFieldIfNotExists(fldName, pilosa.OptFieldTypeInt(-100, 100))
if err != nil {
t.Fatal(err)
}
// Send import request.
c := MustNewClient(host, http.GetHTTPClient(nil))
if err := c.ImportValue(context.Background(), "i", "f", 0, []pilosa.FieldValue{
{ColumnKey: "col1", Value: -10},
{ColumnKey: "col2", Value: 20},
{ColumnKey: "col3", Value: 40},
}); err != nil {
t.Fatal(err)
}
// Verify Sum.
sum, cnt, err := field.Sum(nil, fldName)
if err != nil {
t.Fatal(err)
}
if sum != 50 || cnt != 3 {
t.Fatalf("unexpected values: got sum=%v, count=%v; expected sum=50, cnt=3", sum, cnt)
}
// Verify Range
queryRequest := &pilosa.QueryRequest{
Query: fmt.Sprintf(`Range(%s>10)`, fldName),
Remote: false,
}
result, err := c.Query(context.Background(), "i", queryRequest)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(result.Results[0].(*pilosa.Row).Keys, []string{"col2", "col3"}) {
t.Fatalf("unexpected column keys: %s", spew.Sdump(result))
}
})
}
// Ensure client can bulk import value data.
func TestClient_ImportValue(t *testing.T) {
cmd := test.MustRunCluster(t, 1)[0]
@ -281,7 +502,7 @@ func TestClient_FragmentBlocks(t *testing.T) {
// Set a bit on a different shard.
hldr.SetBit("i", "f", 0, 1)
c := MustNewClient(cmd.URL(), http.GetHTTPClient(nil))
blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", 0)
blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", "standard", 0)
if err != nil {
t.Fatal(err)
} else if len(blocks) != 2 {
@ -293,7 +514,7 @@ func TestClient_FragmentBlocks(t *testing.T) {
}
// Verify data matches local blocks.
if a, err := cmd.API.FragmentBlocks(context.Background(), "i", "f", 0); err != nil {
if a, err := cmd.API.FragmentBlocks(context.Background(), "i", "f", "standard", 0); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(a, blocks) {
t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks))

21
http/handler.go
View file

@ -176,7 +176,7 @@ func (h *Handler) populateValidators() {
h.validators["GetExport"] = queryValidationSpecRequired("index", "field", "shard")
h.validators["GetFragmentData"] = queryValidationSpecRequired("index", "field", "shard")
h.validators["PostFragmentData"] = queryValidationSpecRequired("index", "field", "shard")
h.validators["GetFragmentBlocks"] = queryValidationSpecRequired("index", "field", "shard")
h.validators["GetFragmentBlocks"] = queryValidationSpecRequired("index", "field", "view", "shard")
}
func (h *Handler) queryArgValidator(next http.Handler) http.Handler {
@ -232,6 +232,7 @@ func newRouter(handler *Handler) *mux.Router {
router.HandleFunc("/internal/fragment/nodes", handler.handleGetFragmentNodes).Methods("GET").Name("GetFragmentNodes")
router.HandleFunc("/internal/index/{index}/attr/diff", handler.handlePostIndexAttrDiff).Methods("POST")
router.HandleFunc("/internal/index/{index}/field/{field}/attr/diff", handler.handlePostFieldAttrDiff).Methods("POST")
router.HandleFunc("/internal/nodes", handler.handleGetNodes).Methods("GET").Name("GetNodes")
router.HandleFunc("/internal/shards/max", handler.handleGetShardsMax).Methods("GET") // TODO: deprecate, but it's being used by the client
router.HandleFunc("/internal/translate/data", handler.handleGetTranslateData).Methods("GET")
@ -1062,6 +1063,22 @@ func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request)
}
}
// handleGetNodes handles /internal/nodes requests.
func (h *Handler) handleGetNodes(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
// Retrieve all nodes.
nodes := h.api.Hosts(r.Context())
// Write to response.
if err := json.NewEncoder(w).Encode(nodes); err != nil {
h.logger.Printf("json write error: %s", err)
}
}
// handleGetFragmentBlockData handles GET /internal/fragment/block/data requests.
func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Request) {
buf, err := h.api.FragmentBlockData(r.Context(), r.Body)
@ -1096,7 +1113,7 @@ func (h *Handler) handleGetFragmentBlocks(w http.ResponseWriter, r *http.Request
return
}
blocks, err := h.api.FragmentBlocks(r.Context(), q.Get("index"), q.Get("field"), shard)
blocks, err := h.api.FragmentBlocks(r.Context(), q.Get("index"), q.Get("field"), q.Get("view"), shard)
if err != nil {
if errors.Cause(err) == pilosa.ErrFragmentNotFound {
http.Error(w, err.Error(), http.StatusNotFound)

31
http/translator.go
View file

@ -6,11 +6,13 @@ import (
"fmt"
"io"
"io/ioutil"
"log"
"net/http"
"net/url"
"strconv"
"github.com/pilosa/pilosa"
"github.com/pkg/errors"
)
// Ensure implementation implements inteface.
@ -19,12 +21,31 @@ var _ pilosa.TranslateStore = (*translateStore)(nil)
// translateStore represents an implementation of pilosa.TranslateStore that
// communicates over HTTP. This is used with the TranslateHandler.
type translateStore struct {
URL string
node *pilosa.Node
}
// NewTranslateStore returns a new instance of TranslateStore.
func NewTranslateStore(rawurl string) *translateStore {
return &translateStore{URL: rawurl}
// NewTranslateStore returns a new instance of TranslateStore based on node.
// DEPRECATED: Providing a string url to this function is being deprecated. Instead,
// provide a *pilosa.Node.
func NewTranslateStore(node interface{}) pilosa.TranslateStore {
var n *pilosa.Node
switch v := node.(type) {
case string:
log.Printf("WARNING: providing a string url to NewTranslateStore() has been deprecated.")
if uri, err := pilosa.NewURIFromAddress(v); err != nil {
log.Println(errors.Wrap(err, "creating uri"))
} else {
n = &pilosa.Node{
ID: v,
URI: *uri,
}
}
case *pilosa.Node:
n = v
default:
log.Printf("WARNING: a *pilosa.Node is the only type supported by NewTranslateStore().")
}
return &translateStore{node: n}
}
// TranslateColumnsToUint64 is not currently implemented.
@ -50,7 +71,7 @@ func (s *translateStore) TranslateRowToString(index, frame string, values uint64
// Reader returns a reader that can stream data from a remote store.
func (s *translateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
// Generate remote URL.
u, err := url.Parse(s.URL)
u, err := url.Parse(s.node.URI.String())
if err != nil {
return nil, err
}

31
server.go
View file

@ -49,7 +49,6 @@ type Server struct { // nolint: maligned
// Internal
holder *Holder
cluster *cluster
translateFile *TranslateFile
diagnostics *diagnosticsCollector
executor *executor
hosts []string
@ -70,8 +69,6 @@ type Server struct { // nolint: maligned
isCoordinator bool
syncer holderSyncer
primaryTranslateStore TranslateStore
defaultClient InternalClient
dataDir string
}
@ -163,9 +160,17 @@ func OptServerInternalClient(c InternalClient) ServerOption {
}
}
// DEPRECATED
func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption {
return func(s *Server) error {
s.primaryTranslateStore = store
s.logger.Printf("DEPRECATED: OptServerPrimaryTranslateStore")
return nil
}
}
func OptServerPrimaryTranslateStoreFunc(tf func(interface{}) TranslateStore) ServerOption {
return func(s *Server) error {
s.holder.NewPrimaryTranslateStore = tf
return nil
}
}
@ -265,6 +270,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
}
s.holder.Path = path
s.holder.translateFile.Path = filepath.Join(path, ".keys")
s.holder.Logger = s.logger
s.holder.Stats.SetLogger(s.logger)
@ -272,11 +278,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.cluster.logger = s.logger
s.cluster.holder = s.holder
// Initialize translation database.
s.translateFile = NewTranslateFile()
s.translateFile.Path = filepath.Join(path, ".keys")
s.translateFile.PrimaryTranslateStore = s.primaryTranslateStore
// Get or create NodeID.
s.nodeID = s.loadNodeID()
if s.isCoordinator {
@ -303,7 +304,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.executor.Holder = s.holder
s.executor.Node = node
s.executor.Cluster = s.cluster
s.executor.TranslateStore = s.translateFile
s.executor.TranslateStore = s.holder.translateFile
s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
s.cluster.broadcaster = s
s.cluster.maxWritesPerRequest = s.maxWritesPerRequest
@ -328,7 +329,7 @@ func (s *Server) Open() error {
}
// Initialize id-key storage.
if err := s.translateFile.Open(); err != nil {
if err := s.holder.translateFile.Open(); err != nil {
return err
}
@ -374,7 +375,6 @@ func (s *Server) Close() error {
s.wg.Wait()
var errh error
var errt error
var errc error
if s.cluster != nil {
errc = s.cluster.close()
@ -382,17 +382,12 @@ func (s *Server) Close() error {
if s.holder != nil {
errh = s.holder.Close()
}
if s.translateFile != nil {
errt = s.translateFile.Close()
}
// prefer to return holder error over translateFile error over cluster
// prefer to return holder error over cluster
// error. This order is somewhat arbitrary. It would be better if we had
// some way to combine all the errors, but probably not important enough to
// warrant the extra complexity.
if errh != nil {
return errors.Wrap(errh, "closing holder")
} else if errt != nil {
return errors.Wrap(errt, "closing translateFile")
}
return errors.Wrap(errc, "closing cluster")
}

2
server/config.go
View file

@ -71,7 +71,7 @@ type Config struct {
// Gossip config is based around memberlist.Config.
Gossip gossip.Config `toml:"gossip"`
// Translation config supports translation store replication.
// DEPRECATED: Translation config supports translation store replication.
Translation struct {
PrimaryURL string `toml:"primary-url"`
} `toml:"translation"`

7
server/server.go
View file

@ -250,10 +250,9 @@ func (m *Command) SetupServer() error {
c := http.GetHTTPClient(TLSConfig)
// Setup connection to primary store if this is a replica.
var primaryTranslateStore pilosa.TranslateStore
// Primary store configuration is handled automatically now.
if m.Config.Translation.PrimaryURL != "" {
primaryTranslateStore = http.NewTranslateStore(m.Config.Translation.PrimaryURL)
m.logger.Printf("DEPRECATED: The primary-url configuration option is no longer used.")
}
// Set Coordinator.
@ -278,7 +277,7 @@ func (m *Command) SetupServer() error {
pilosa.OptServerStatsClient(statsClient),
pilosa.OptServerURI(uri),
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore),
pilosa.OptServerPrimaryTranslateStoreFunc(http.NewTranslateStore),
pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts),
pilosa.OptServerSerializer(proto.Serializer{}),
coordinatorOpt,

4
server/server_test.go
View file

@ -489,6 +489,10 @@ func TestClusteringNodesReplica2(t *testing.T) {
t.Fatalf("restarting node 2: %v", err)
}
if cluster[0].API.State() != pilosa.ClusterStateDegraded {
t.Fatalf("expected state to be DEGRADED, but got %s", cluster[0].API.State())
}
// Create new main with the same config.
config = cluster[1].Command.Config
// config.Bind = cluster[1].API.Node().URI.HostPort()

9
test/holder.go
View file

@ -113,14 +113,19 @@ func (h *Holder) RowTime(index, field string, rowID uint64, t time.Time, quantum
return row
}
// SetBit clears a bit on the given field.
// SetBit sets a bit on the given field.
func (h *Holder) SetBit(index, field string, rowID, columnID uint64) {
h.SetBitTime(index, field, rowID, columnID, nil)
}
// SetBitTime sets a bit with timestamp on the given field.
func (h *Holder) SetBitTime(index, field string, rowID, columnID uint64, t *time.Time) {
idx := h.MustCreateIndexIfNotExists(index, pilosa.IndexOptions{})
f, err := idx.CreateFieldIfNotExists(field, pilosa.OptFieldTypeDefault())
if err != nil {
panic(err)
}
_, err = f.SetBit(rowID, columnID, nil)
_, err = f.SetBit(rowID, columnID, t)
if err != nil {
panic(err)
}

42
test/pilosa.go
View file

@ -16,6 +16,7 @@ package test
import (
"bytes"
"context"
"fmt"
"io"
"io/ioutil"
@ -27,6 +28,7 @@ import (
"testing"
"time"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
@ -125,6 +127,45 @@ func (m *Command) Reopen() error {
return m.Start()
}
// MustCreateIndex uses this command's API to create an index and fails the test
// if there is an error.
func (m *Command) MustCreateIndex(t *testing.T, name string, opts pilosa.IndexOptions) *pilosa.Index {
idx, err := m.API.CreateIndex(context.Background(), name, opts)
if err != nil {
t.Fatalf("creating index: %v with options: %v, err: %v", name, opts, err)
}
return idx
}
// MustCreateField uses this command's API to create the field. The index must
// already exist - it fails the test if there is an error.
func (m *Command) MustCreateField(t *testing.T, index, field string, opts ...pilosa.FieldOption) *pilosa.Field {
f, err := m.API.CreateField(context.Background(), index, field, opts...)
if err != nil {
t.Fatalf("creating field: %s in index: %s err: %v", field, index, err)
}
return f
}
// MustQuery uses this command's API to execute the given query request, failing
// if Query returns a non-nil error, otherwise returning the QueryResponse.
func (m *Command) MustQuery(t *testing.T, req *pilosa.QueryRequest) pilosa.QueryResponse {
resp, err := m.API.Query(context.Background(), req)
if err != nil {
t.Fatalf("making query: %v, err: %v", req, err)
}
return resp
}
// MustRecalculateCaches calls RecalculateCaches on the command's API, and fails
// if there is an error.
func (m *Command) MustRecalculateCaches(t *testing.T) {
err := m.API.RecalculateCaches(context.Background())
if err != nil {
t.Fatalf("recalcluating caches: %v", err)
}
}
// URL returns the base URL string for accessing the running program.
func (m *Command) URL() string { return m.API.Node().URI.String() }
@ -146,6 +187,7 @@ func (m *Command) Query(index, rawQuery, query string) (string, error) {
return resp.Body, nil
}
// RecalculateCaches is deprecated. Use MustRecalculateCaches.
func (m *Command) RecalculateCaches() error {
resp := MustDo("POST", fmt.Sprintf("%s/recalculate-caches", m.URL()), "")
if resp.StatusCode != 204 {

131
translate.go
View file

@ -8,6 +8,7 @@ import (
"errors"
"fmt"
"io"
"io/ioutil"
"log"
"os"
"path/filepath"
@ -71,6 +72,10 @@ type TranslateFile struct {
// If non-nil, data is streamed from a primary and this is a read-only store.
PrimaryTranslateStore TranslateStore
primaryID string // unique ID used to identify the primary store
replicationClosing chan struct{}
primaryStoreEvents chan primaryStoreEvent
repWG sync.WaitGroup
// Delay after attempting to connect to a primary that the store will retry.
replicationRetryInterval time.Duration
@ -86,6 +91,9 @@ func NewTranslateFile() *TranslateFile {
mapSize: defaultMapSize,
replicationClosing: make(chan struct{}),
primaryStoreEvents: make(chan primaryStoreEvent),
replicationRetryInterval: defaultReplicationRetryInterval,
}
}
@ -109,10 +117,65 @@ func (s *TranslateFile) Open() (err error) {
return err
}
// Stream from primary, if available.
// Listen to primaryStoreEvents channel.
s.wg.Add(1)
go func() { defer s.wg.Done(); s.monitorPrimaryStoreEvents() }()
return nil
}
// primaryStoreEvent is used to set/change the primary translate store.
// It contains a TranslateStore along with an associated string ID which
// is used to determine whether the primary needs to be changed from the
// current value.
type primaryStoreEvent struct {
id string
ts TranslateStore
}
// SetPrimaryStore sets the translate files's primary translate store.
// The id value is used to determine whether the primary needs to be changed
// from the current value (i.e. calling this multiple times with the same
// input values will no-op on all subsequent calls).
func (s *TranslateFile) SetPrimaryStore(id string, ts TranslateStore) {
go func() {
s.primaryStoreEvents <- primaryStoreEvent{
id: id,
ts: ts,
}
}()
}
// handlePrimaryStoreEvent changes the PrimaryTranslateStore
// used for replication by TranslateFile.
func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error {
s.mu.Lock()
defer s.mu.Unlock()
if ev.id == s.primaryID {
return nil
}
// Stop translate store replication.
log.Printf("stop monitor replication")
close(s.replicationClosing)
s.repWG.Wait()
// Set the primary node for translate store replication.
log.Printf("set primary translate store to %s", ev.id)
s.primaryID = ev.id
if ev.id == "" {
s.PrimaryTranslateStore = nil
} else {
s.PrimaryTranslateStore = ev.ts
}
// Start translate store replication. Stream from primary, if available.
log.Printf("start monitor replication")
if s.PrimaryTranslateStore != nil {
s.wg.Add(1)
go func() { defer s.wg.Done(); s.monitorReplication() }()
s.replicationClosing = make(chan struct{})
s.repWG.Add(1)
go func() { defer s.repWG.Done(); s.monitorReplication() }()
}
return nil
@ -259,7 +322,13 @@ func (s *TranslateFile) replayEntries() error {
func (s *TranslateFile) monitorReplication() {
// Create context that will cancel on close.
ctx, cancel := context.WithCancel(context.Background())
go func() { <-s.closing; cancel() }()
go func() {
select {
case <-s.closing:
case <-s.replicationClosing:
}
cancel()
}()
// Keep attempting to replicate until the store closes.
for {
@ -270,12 +339,31 @@ func (s *TranslateFile) monitorReplication() {
select {
case <-s.closing:
return
case <-s.replicationClosing:
return
case <-time.After(s.replicationRetryInterval):
log.Printf("pilosa: reconnecting to primary replica")
}
}
}
// monitorPrimaryStoreEvents is executed in a separate goroutine and listens for changes
// to the primary store assignment.
func (s *TranslateFile) monitorPrimaryStoreEvents() {
log.Printf("monitor primary store events")
// Keep handling events until the store closes.
for {
select {
case <-s.closing:
return
case ev := <-s.primaryStoreEvents:
if err := s.handlePrimaryStoreEvent(ev); err != nil {
log.Printf("handle primary store event")
}
}
}
}
func (s *TranslateFile) replicate(ctx context.Context) error {
off := s.size()
@ -989,3 +1077,38 @@ func uVarintSize(x uint64) (i int) {
}
return i + 1
}
// nopTStore represents a TranslateStore that doesn't do anything.
var nopTStore TranslateStore = nopTranslateStore{}
// newNopTranslateStore returns a translate store which does nothing. It returns a global
// object to avoid unnecessary allocations.
func newNopTranslateStore(interface{}) TranslateStore { return nopTStore }
// nopTranslateStore represents a no-op implementation of the TranslateStore interface.
type nopTranslateStore struct{}
// TranslateColumnsToUint64 is a no-op implementation of the TranslateStore TranslateColumnsToUint64 method.
func (s nopTranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
return []uint64{}, nil
}
// TranslateColumnToString is a no-op implementation of the TranslateStore TranslateColumnToString method.
func (s nopTranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
return "", nil
}
// TranslateRowsToUint64 is a no-op implementation of the TranslateStore TranslateRowsToUint64 method.
func (s nopTranslateStore) TranslateRowsToUint64(index, field string, values []string) ([]uint64, error) {
return []uint64{}, nil
}
// TranslateRowToString is a no-op implementation of the TranslateStore TranslateRowToString method.
func (s nopTranslateStore) TranslateRowToString(index, field string, values uint64) (string, error) {
return "", nil
}
// Reader is a no-op implementation of the TranslateStore Reader method.
func (s nopTranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
return ioutil.NopCloser(bytes.NewReader(nil)), nil
}

293
translate_test.go
View file

@ -383,7 +383,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
// Create a replica that accepts writes from primary.
replica := NewTranslateFile()
replica.PrimaryTranslateStore = primary
replica.SetPrimaryStore("primary", primary)
if err := replica.Open(); err != nil {
t.Fatal(err)
}
@ -398,7 +398,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
// Attempt to read replica until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica have received writes.
// Verify that replica has received writes.
if value, err := replica.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
@ -429,7 +429,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
t.Fatal(err)
}
// Attempt to read replica until write appear.
// Attempt to read replica until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica.TranslateColumnToString("IDX0", 2); err != nil {
return err
@ -448,7 +448,7 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
t.Fatal(err)
}
// Attempt to read replica until write appear.
// Attempt to read replica until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica.TranslateColumnToString("IDX0", 3); err != nil {
return err
@ -461,6 +461,289 @@ func TestTranslateFile_PrimaryTranslateStore(t *testing.T) {
}
}
func TestTranslateFile_ReassignPrimaryTranslateStore(t *testing.T) {
t.Run("AddNode", func(t *testing.T) {
// Create a primary store that accepts writes.
primary := MustOpenTranslateFile()
defer primary.MustClose()
// Create replica1 that accepts writes from primary.
replica1 := NewTranslateFile()
replica1.SetPrimaryStore("primary", primary)
if err := replica1.Open(); err != nil {
t.Fatal(err)
}
defer replica1.MustClose()
// Write to the primary.
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica1 until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica1 has received writes.
if value, err := replica1.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica1.TranslateRowToString("IDX0", "FIELD0", 1); err != nil {
return err
} else if value != "bar" {
return fmt.Errorf("unexpected row 1 value: %s", value)
}
if value, err := replica1.TranslateRowToString("IDX0", "FIELD0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Create replica2 that accepts writes from primary,
// and change replica1's primary to be replica2.
// From: P <- R1
// To: P <- R2 <- R1
// Momentarily, replica1 should be ahead of replica2, so we will see log
// messages like "translate store reader past file size: sz=0 off=39"
// But eventually it should get in sync and new writes will be available
// on replica1.
replica2 := NewTranslateFile()
replica2.SetPrimaryStore("primary", primary)
replica1.SetPrimaryStore("replica2", replica2)
if err := replica2.Open(); err != nil {
t.Fatal(err)
}
defer replica2.MustClose()
// Attempt to read replica2 until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica2 have received writes.
if value, err := replica2.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 1); err != nil {
return err
} else if value != "bar" {
return fmt.Errorf("unexpected row 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Add more data to the primary and ensure that replica1 receives the
// data (via replica2).
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica1 until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica1.TranslateColumnToString("IDX0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected column 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
})
t.Run("RemoveNode", func(t *testing.T) {
// Create a primary store that accepts writes.
primary := MustOpenTranslateFile()
defer primary.MustClose()
// Create two replicas that accepts writes from the primary
// in a daisy-chain configuration.
// P <- R1 <- R2
// Create replica1.
replica1 := NewTranslateFile()
replica1.SetPrimaryStore("primary", primary)
if err := replica1.Open(); err != nil {
t.Fatal(err)
}
defer replica1.MustClose()
// Create replica2.
replica2 := NewTranslateFile()
replica2.SetPrimaryStore("replica1", replica1)
if err := replica2.Open(); err != nil {
t.Fatal(err)
}
defer replica2.MustClose()
// Write to the primary.
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica2 until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica2 has received writes.
if value, err := replica2.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 1); err != nil {
return err
} else if value != "bar" {
return fmt.Errorf("unexpected row 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Remove replica1 from the replication chain.
// From: P <- R1 <- R2
// To: P <- R2
replica1.SetPrimaryStore("", nil)
// Add more data to the primary and ensure that replica2 receives the
// data (after replica1 is removed).
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil {
t.Fatal(err)
}
// Set replica2 to replicate from primary.
replica2.SetPrimaryStore("primary", primary)
// Attempt to read replica2 until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := replica2.TranslateColumnToString("IDX0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected column 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
})
t.Run("ChangePrimaryWriter", func(t *testing.T) {
// Create a primary store that accepts writes.
primary := MustOpenTranslateFile()
defer primary.MustClose()
// Create two replicas that accepts writes from the primary
// in a daisy-chain configuration.
// P <- R1 <- R2
// Create replica1.
replica1 := NewTranslateFile()
replica1.SetPrimaryStore("primary", primary)
if err := replica1.Open(); err != nil {
t.Fatal(err)
}
defer replica1.MustClose()
// Create replica2.
replica2 := NewTranslateFile()
replica2.SetPrimaryStore("replica1", replica1)
if err := replica2.Open(); err != nil {
t.Fatal(err)
}
defer replica2.MustClose()
// Write to the primary.
if _, err := primary.TranslateColumnsToUint64("IDX0", []string{"foo"}); err != nil {
t.Fatal(err)
} else if _, err := primary.TranslateRowsToUint64("IDX0", "FIELD0", []string{"bar", "baz"}); err != nil {
t.Fatal(err)
}
// Attempt to read replica2 until writes appear.
if err := retryFor(2*time.Second, func() error {
// Verify that replica2 has received writes.
if value, err := replica2.TranslateColumnToString("IDX0", 1); err != nil {
return err
} else if value != "foo" {
return fmt.Errorf("unexpected column 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 1); err != nil {
return err
} else if value != "bar" {
return fmt.Errorf("unexpected row 1 value: %s", value)
}
if value, err := replica2.TranslateRowToString("IDX0", "FIELD0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected row 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
// Change replica1 to be the primary writer.
// From: P <- R1 <- R2
// To: R1 <- R2 <- P
replica1.SetPrimaryStore("", nil)
// SetPrimaryStore is asynchronous, so we need to wait before writing new data.
time.Sleep(200 * time.Millisecond)
// Add more data to the primary (now replica1) and ensure that primary (now a read-only replica)
// receives the data.
if _, err := replica1.TranslateColumnsToUint64("IDX0", []string{"baz"}); err != nil {
t.Fatal(err)
}
// Set primary to replicate from replica2.
primary.SetPrimaryStore("replica2", replica2)
// Attempt to read primary until writes appear.
if err := retryFor(2*time.Second, func() error {
if value, err := primary.TranslateColumnToString("IDX0", 2); err != nil {
return err
} else if value != "baz" {
return fmt.Errorf("unexpected column 2 value: %s", value)
}
return nil
}); err != nil {
t.Fatal(err)
}
})
}
func BenchmarkTranslateFile_TranslateColumnsToUint64(b *testing.B) {
const batchSize = 1000
@ -567,7 +850,7 @@ func (s *TranslateFile) Reopen() error {
s.TranslateFile = pilosa.NewTranslateFile()
s.lock.Unlock()
s.Path = prev.Path
s.PrimaryTranslateStore = prev.PrimaryTranslateStore
s.SetPrimaryStore("restored-primary", prev.PrimaryTranslateStore)
return s.Open()
}

3
uri.go
View file

@ -173,6 +173,9 @@ func parseAddress(address string) (uri *URI, err error) {
if err != nil {
return nil, errors.New("converting port string to int")
}
if port > 65535 {
return nil, errors.New("port must be in range 0 - 65535")
}
}
uri = &URI{
Scheme: scheme,

2
uri_internal_test.go
View file

@ -172,5 +172,5 @@ func validFixture() []uriItem {
}
func invalidFixture() []string {
return []string{"foo:bar", "http://foo:", "foo:", ":bar", "http://pilosa.com:129999999999999999999999993", "fd42:4201:f86b:7e09:216:3eff:fefa:ed80"}
return []string{"foo:bar", "http://foo:", "foo:", ":bar", "http://pilosa.com:129999999999999999999999993", "fd42:4201:f86b:7e09:216:3eff:fefa:ed80", ":65536"}
}