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

Merge pull request #1398 from travisturner/disco-schema

Browse source 
Implement Schemator
This commit is contained in:
Travis Turner 2021-02-10 12:01:00 -06:00 • committed by GitHub
commit 88f13d5e8b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
26 changed files with 1133 additions and 473 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

112
api.go
View file

@ -211,37 +211,19 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
return nil, errors.Wrap(err, "validating api method")
}
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
if err := api.server.defaultClient.CreateIndex(ctx, indexName, options); err != nil {
return nil, errors.Wrap(err, "forwarding CreateIndex to coordinator")
}
return api.holder.Index(indexName), nil
// Populate the create index message.
cim := &CreateIndexMessage{
Index: indexName,
CreatedAt: timestamp(),
Meta: &options,
}
// Create index.
index, err := api.holder.CreateIndex(indexName, options)
index, err := api.holder.CreateIndexAndBroadcast(cim)
if err != nil {
return nil, errors.Wrap(err, "creating index")
}
createdAt := timestamp()
index.mu.Lock()
index.createdAt = createdAt
index.mu.Unlock()
// Send the create index message to all nodes.
err = api.server.SendSync(
&CreateIndexMessage{
Index: indexName,
CreatedAt: createdAt,
Meta: &options,
})
if err != nil {
return nil, errors.Wrap(err, "sending CreateIndex message")
}
api.holder.Stats.Count(MetricCreateIndex, 1, 1.0)
return index, nil
}
@ -301,23 +283,10 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
return nil, errors.Wrap(err, "validating api method")
}
// Apply functional options.
fo := FieldOptions{}
for _, opt := range opts {
err := opt(&fo)
if err != nil {
return nil, NewBadRequestError(errors.Wrap(err, "applying option"))
}
}
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
if err := api.server.defaultClient.CreateFieldWithOptions(ctx, indexName, fieldName, fo); err != nil {
return nil, errors.Wrap(err, "forwarding CreateField to coordinator")
}
return api.holder.Field(indexName, fieldName), nil
// Apply and validate functional options.
fo, err := newFieldOptions(opts...)
if err != nil {
return nil, NewBadRequestError(errors.Wrap(err, "applying option"))
}
// Find index.
@ -326,27 +295,20 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
return nil, newNotFoundError(ErrIndexNotFound, indexName)
}
// Populate the create field message.
cfm := &CreateFieldMessage{
Index: indexName,
Field: fieldName,
CreatedAt: timestamp(),
Meta: fo,
}
// Create field.
field, err := index.CreateField(fieldName, opts...)
field, err := index.CreateFieldAndBroadcast(cfm)
if err != nil {
return nil, errors.Wrap(err, "creating field")
}
createdAt := timestamp()
field.mu.Lock()
field.createdAt = createdAt
field.mu.Unlock()
// Send the create field message to all nodes.
err = api.server.SendSync(&CreateFieldMessage{
Index: indexName,
Field: fieldName,
CreatedAt: createdAt,
Meta: &fo,
})
if err != nil {
api.server.logger.Printf("problem sending CreateField message: %s", err)
return nil, errors.Wrap(err, "sending CreateField message")
}
api.holder.Stats.CountWithCustomTags(MetricCreateField, 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)})
return field, nil
}
@ -1020,14 +982,19 @@ func (err MessageProcessingError) Unwrap() error {
// Schema returns information about each index in Pilosa including which fields
// they contain.
func (api *API) Schema(ctx context.Context) ([]*IndexInfo, error) {
func (api *API) Schema(ctx context.Context, withViews bool) ([]*IndexInfo, error) {
if err := api.validate(apiSchema); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
span, _ := tracing.StartSpanFromContext(ctx, "API.Schema")
defer span.Finish()
return api.holder.limitedSchema(), nil
if withViews {
return api.holder.Schema()
}
return api.holder.limitedSchema()
}
// ApplySchema takes the given schema and applies it across the
@ -1044,29 +1011,12 @@ func (api *API) ApplySchema(ctx context.Context, s *Schema, remote bool) error {
return errors.Wrap(err, "validating api method")
}
// set CreatedAt for indexes and fields (if empty), and then apply schema.
for _, index := range s.Indexes {
if index.CreatedAt == 0 {
index.CreatedAt = timestamp()
}
for _, field := range index.Fields {
if field.CreatedAt == 0 {
field.CreatedAt = timestamp()
}
}
err := api.holder.applySchema(s)
if err != nil {
return errors.Wrap(err, "applying schema")
}
if !remote {
nodes := api.cluster.Nodes()
for i, node := range nodes {
err := api.server.defaultClient.PostSchema(ctx, &node.URI, s, true)
if err != nil {
return errors.Wrapf(err, "forwarding post schema to node %d of %d", i+1, len(nodes))
}
}
}
return errors.Wrap(api.holder.applySchema(s), "applying schema")
return nil
}
// Views returns the views in the given field.
@ -1359,7 +1309,7 @@ func (api *API) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts .
return nil
}
// Import bulk imports data into a particular index,field,shard.
// ImportWithTx bulk imports data into a particular index,field,shard.
func (api *API) ImportWithTx(ctx context.Context, qcx *Qcx, req *ImportRequest, opts ...ImportOption) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.Import")
defer span.Finish()

2
api_test.go
View file

@ -269,7 +269,7 @@ func TestAPI_Import(t *testing.T) {
// Relies on the previous test creating an index with TrackExistence and
// adding some data.
t.Run("SchemaHasNoExists", func(t *testing.T) {
schema, err := m1.API.Schema(context.Background())
schema, err := m1.API.Schema(context.Background(), false)
if err != nil {
t.Fatal(err)
}

16
broadcast.go
View file

@ -27,6 +27,17 @@ type Serializer interface {
Unmarshal([]byte, Message) error
}
// NopSerializer represents a Serializer that doesn't do anything.
var NopSerializer Serializer = &nopSerializer{}
type nopSerializer struct{}
// Marshal A no-op implementation of Serializer Marshall method.
func (*nopSerializer) Marshal(Message) ([]byte, error) { return nil, nil }
// Unmarshal A no-op implementation of Serializer Unmarshal method.
func (*nopSerializer) Unmarshal([]byte, Message) error { return nil }
// broadcaster is an interface for broadcasting messages.
type broadcaster interface {
SendSync(Message) error
@ -66,6 +77,7 @@ const (
messageTypeResizeInstructionComplete
messageTypeNodeState
messageTypeRecalculateCaches
messageTypeLoadSchemaMessage
messageTypeNodeEvent
messageTypeNodeStatus
messageTypeTransaction
@ -110,6 +122,8 @@ func getMessage(typ byte) Message {
return &NodeStateMessage{}
case messageTypeRecalculateCaches:
return &RecalculateCaches{}
case messageTypeLoadSchemaMessage:
return &LoadSchemaMessage{}
case messageTypeNodeEvent:
return &NodeEvent{}
case messageTypeNodeStatus:
@ -151,6 +165,8 @@ func getMessageType(m Message) byte {
return messageTypeNodeState
case *RecalculateCaches:
return messageTypeRecalculateCaches
case *LoadSchemaMessage:
return messageTypeLoadSchemaMessage
case *NodeEvent:
return messageTypeNodeEvent
case *NodeStatus:

36
cluster.go
View file

@ -410,11 +410,15 @@ func (c *cluster) generateResizeInstructionOnAdd(addNodeID string) (*ResizeInstr
}
myid := c.disCo.ID()
nodeStatus, err := c.nodeStatus()
if err != nil {
return nil, errors.Wrap(err, "getting node status")
}
return &ResizeInstruction{
Node: c.unprotectedNodeByID(myid),
Sources: fragmentSourcesByNode[myid],
TranslationSources: translationSourcesByNode[myid],
NodeStatus: c.nodeStatus(), // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node.
NodeStatus: nodeStatus, // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node.
ClusterStatus: status,
}, nil
}
@ -594,11 +598,15 @@ func (c *cluster) generateResizeInstructionOnRemove(removeNodeID string) (*Resiz
}
myid := c.disCo.ID()
nodeStatus, err := c.nodeStatus()
if err != nil {
return nil, errors.Wrap(err, "getting node status")
}
return &ResizeInstruction{
Node: toCluster.unprotectedNodeByID(myid),
Sources: fragmentSourcesByNode[myid],
TranslationSources: translationSourcesByNode[myid],
NodeStatus: c.nodeStatus(), // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node.
NodeStatus: nodeStatus, // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node.
ClusterStatus: status,
}, nil
}
@ -610,13 +618,10 @@ func (c *cluster) unprotectedStatus() (*ClusterStatus, error) {
return nil, err
}
// TODO: replace following code by following code,
// after schemator is implemented
// indexes, err := c.holder.Schema()
// if err != nil {
// return nil, errors.Wrap(err, "getting schema")
// }
indexes := c.holder.Schema()
indexes, err := c.holder.Schema()
if err != nil {
return nil, errors.Wrap(err, "getting schema")
}
return &ClusterStatus{
State: string(state),
@ -631,9 +636,6 @@ func (c *cluster) remoteSchema() (*Schema, error) {
continue
}
// TODO: replace following line by:
// ii, err := c.InternalClient.SchemaNode(context.Background(), &n.URI, true)
// after we
ii, err := c.InternalClient.SchemaNode(context.Background(), &n.URI, true)
if err != nil {
return nil, errors.Wrapf(err, "getting schema from %s (%v)", n.ID, n.URI)
@ -1272,10 +1274,14 @@ func (c *cluster) SetNodeState(nodeID string, state string) {}
///////////////////////////////////////////
func (c *cluster) nodeStatus() *NodeStatus {
func (c *cluster) nodeStatus() (*NodeStatus, error) {
indexes, err := c.holder.Schema()
if err != nil {
return nil, errors.Wrap(err, "getting schema")
}
ns := &NodeStatus{
Node: c.Node,
Schema: &Schema{Indexes: c.holder.Schema()},
Schema: &Schema{Indexes: indexes},
}
var availableShards *roaring.Bitmap
for _, idx := range ns.Schema.Indexes {
@ -1294,7 +1300,7 @@ func (c *cluster) nodeStatus() *NodeStatus {
}
ns.Indexes = append(ns.Indexes, is)
}
return ns
return ns, nil
}
// unprotectedPreviousNode returns the node listed before the current node in c.Nodes.

2
cmd/random-query/main.go
View file

@ -73,7 +73,7 @@ type wrapper struct {
}
func (w *wrapper) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error) {
return w.api.Schema(ctx)
return w.api.Schema(ctx, false)
}
func (w *wrapper) Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {

11
dbshard_internal_test.go
View file

@ -73,6 +73,7 @@ func TestShardPerDB_SetBit(t *testing.T) {
func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
tmpdir, err := ioutil.TempDir("", "Test_DBPerShard_GetShardsForIndex_LocalOnly")
panicOn(err)
defer os.RemoveAll(tmpdir)
v2s := NewFieldView2Shards()
stdShardSet := newShardSet()
@ -320,6 +321,7 @@ func makeTxTestDBWithViewsShards(holder *Holder, idx *Index, exp *FieldView2Shar
func Test_DBPerShard_GetFieldView2Shards_map_from_RBF(t *testing.T) {
tmpdir, err := ioutil.TempDir("", "Test_DBPerShard_GetFieldView2Shards_map_from_RBF")
panicOn(err)
defer os.RemoveAll(tmpdir)
cfg := mustHolderConfig()
cfg.StorageConfig.Backend = "rbf"
@ -328,7 +330,14 @@ func Test_DBPerShard_GetFieldView2Shards_map_from_RBF(t *testing.T) {
index := "rick"
field := "f"
idx, err := holder.createIndex(index, IndexOptions{})
cim := &CreateIndexMessage{
Index: index,
CreatedAt: 0,
Meta: &IndexOptions{},
}
idx, err := holder.createIndex(cim, false)
panicOn(err)
exp := NewFieldView2Shards()

55
disco/disco.go
View file

@ -26,6 +26,8 @@ var (
ErrTooManyResults error = fmt.Errorf("too many results")
ErrNoResults error = fmt.Errorf("no results")
ErrKeyDeleted error = fmt.Errorf("key deleted")
ErrIndexExists error = fmt.Errorf("index already exists")
ErrFieldExists error = fmt.Errorf("field already exists")
)
type Peer struct {
@ -86,7 +88,14 @@ type Stator interface {
// for each of its fields.
type Index struct {
Data []byte
Fields map[string][]byte
Fields map[string]*Field
}
// Field is a struct which contains the data encoded for the field as well as
// for each of its views.
type Field struct {
Data []byte
Views map[string][]byte
}
type Schemator interface {
@ -97,6 +106,9 @@ type Schemator interface {
Field(ctx context.Context, index, field string) ([]byte, error)
CreateField(ctx context.Context, index, field string, val []byte) error
DeleteField(ctx context.Context, index, field string) error
View(ctx context.Context, index, field, view string) ([]byte, error)
CreateView(ctx context.Context, index, field, view string, val []byte) error
DeleteView(ctx context.Context, index, field, view string) error
}
type Metadata interface {
@ -233,3 +245,44 @@ func (n *nopSharder) AddShards(ctx context.Context, index, field string, shards
func (n *nopSharder) RemoveShard(ctx context.Context, index, field string, shard uint64) error {
return nil
}
// NopSchemator represents a Schemator that doesn't do anything.
var NopSchemator Schemator = &nopSchemator{}
type nopSchemator struct{}
// Schema is a no-op implementation of the Schemator Schema method.
func (*nopSchemator) Schema(ctx context.Context) (map[string]*Index, error) { return nil, nil }
// Index is a no-op implementation of the Schemator Index method.
func (*nopSchemator) Index(ctx context.Context, name string) ([]byte, error) { return nil, nil }
// CreateIndex is a no-op implementation of the Schemator CreateIndex method.
func (*nopSchemator) CreateIndex(ctx context.Context, name string, val []byte) error { return nil }
// DeleteIndex is a no-op implementation of the Schemator DeleteIndex method.
func (*nopSchemator) DeleteIndex(ctx context.Context, name string) error { return nil }
// Field is a no-op implementation of the Schemator Field method.
func (*nopSchemator) Field(ctx context.Context, index, field string) ([]byte, error) { return nil, nil }
// CreateField is a no-op implementation of the Schemator CreateField method.
func (*nopSchemator) CreateField(ctx context.Context, index, field string, val []byte) error {
return nil
}
// DeleteField is a no-op implementation of the Schemator DeleteField method.
func (*nopSchemator) DeleteField(ctx context.Context, index, field string) error { return nil }
// View is a no-op implementation of the Schemator View method.
func (*nopSchemator) View(ctx context.Context, index, field, view string) ([]byte, error) {
return nil, nil
}
// CreateView is a no-op implementation of the Schemator CreateView method.
func (*nopSchemator) CreateView(ctx context.Context, index, field, view string, val []byte) error {
return nil
}
// DeleteView is a no-op implementation of the Schemator DeleteView method.
func (*nopSchemator) DeleteView(ctx context.Context, index, field, view string) error { return nil }

17
encoding/proto/proto.go
View file

@ -154,6 +154,14 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
}
s.decodeRecalculateCaches(msg, mt)
return nil
case *pilosa.LoadSchemaMessage:
msg := &internal.LoadSchemaMessage{}
err := proto.Unmarshal(buf, msg)
if err != nil {
return errors.Wrap(err, "unmarshaling LoadSchemaMessage")
}
s.decodeLoadSchemaMessage(msg, mt)
return nil
case *pilosa.NodeEvent:
msg := &internal.NodeEventMessage{}
err := proto.Unmarshal(buf, msg)
@ -358,6 +366,8 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message {
return s.encodeNodeStateMessage(mt)
case *pilosa.RecalculateCaches:
return s.encodeRecalculateCaches(mt)
case *pilosa.LoadSchemaMessage:
return s.encodeLoadSchemaMessage(mt)
case *pilosa.NodeEvent:
return s.encodeNodeEventMessage(mt)
case *pilosa.NodeStatus:
@ -855,6 +865,10 @@ func (s Serializer) encodeRecalculateCaches(*pilosa.RecalculateCaches) *internal
return &internal.RecalculateCaches{}
}
func (s Serializer) encodeLoadSchemaMessage(*pilosa.LoadSchemaMessage) *internal.LoadSchemaMessage {
return &internal.LoadSchemaMessage{}
}
func (s Serializer) encodeTranslateKeysRequest(request *pilosa.TranslateKeysRequest) *internal.TranslateKeysRequest {
return &internal.TranslateKeysRequest{
Index: request.Index,
@ -1185,6 +1199,9 @@ func (s Serializer) decodeFieldStatus(pb *internal.FieldStatus, m *pilosa.FieldS
func (s Serializer) decodeRecalculateCaches(pb *internal.RecalculateCaches, m *pilosa.RecalculateCaches) {
}
func (s Serializer) decodeLoadSchemaMessage(pb *internal.LoadSchemaMessage, m *pilosa.LoadSchemaMessage) {
}
func (s Serializer) decodeQueryRequest(pb *internal.QueryRequest, m *pilosa.QueryRequest) {
m.Query = pb.Query
m.Shards = pb.Shards

158
etcd/embed.go
View file

@ -61,9 +61,6 @@ var (
_ disco.Metadator = &Etcd{}
_ disco.Resizer = &Etcd{}
_ disco.Sharder = &Etcd{}
ErrIndexExists = errors.New("index already exists")
ErrFieldExists = errors.New("field already exists")
)
const (
@ -482,36 +479,57 @@ func (e *Etcd) DeleteNode(ctx context.Context, nodeID string) error {
}
func (e *Etcd) Schema(ctx context.Context) (map[string]*disco.Index, error) {
cli, err := e.client()
if err != nil {
return nil, errors.Wrap(err, "Schema: creating client")
}
defer cli.Close()
keys, vals, err := e.getKey(ctx, cli, schemaPrefix)
keys, vals, err := e.getKey(ctx, schemaPrefix)
if err != nil {
return nil, err
}
// The logic in the following for loop assumes that the list of keys is
// ordered such that index comes before field, which comes before view.
// For example:
// /index1
// /index1/field1
// /index1/field1/view1
// /index1/field1/view2
// /index1/field2
// /index2
// /index2/field1
//
m := make(map[string]*disco.Index)
for i, k := range keys {
tokens := strings.Split(strings.Trim(k, "/"), "/")
// token[0] contains the schemaPrefix
// token[1]: index
index := tokens[1]
if _, ok := m[index]; !ok {
m[index] = &disco.Index{
Data: vals[i],
Fields: make(map[string][]byte),
Fields: make(map[string]*disco.Field),
}
continue
}
flds := m[index].Fields
// token[2]: field
if len(tokens) > 2 {
field := tokens[2]
flds[field] = vals[i]
if _, ok := flds[field]; !ok {
flds[field] = &disco.Field{
Data: vals[i],
Views: make(map[string][]byte),
}
continue
}
views := flds[field].Views
// token[3]: view
if len(tokens) > 3 {
view := tokens[3]
views[view] = vals[i]
}
}
}
return m, nil
}
@ -573,46 +591,44 @@ func (e *Etcd) CreateIndex(ctx context.Context, name string, val []byte) error {
}
if !resp.Succeeded {
return ErrIndexExists
return disco.ErrIndexExists
}
return nil
}
func (e *Etcd) Index(ctx context.Context, name string) ([]byte, error) {
cli, err := e.client()
if err != nil {
return nil, errors.Wrap(err, "Index: creating client")
}
defer cli.Close()
return e.getKeyBytes(ctx, cli, schemaPrefix+name)
return e.getKeyBytes(ctx, schemaPrefix+name)
}
func (e *Etcd) DeleteIndex(ctx context.Context, name string) error {
// Delete any fields below the index path.
if err := e.delKey(ctx, schemaPrefix+name+"/", true); err != nil {
return errors.Wrap(err, "deleting index fields")
}
// Delete the index.
return e.delKey(ctx, schemaPrefix+name, false)
}
func (e *Etcd) Field(ctx context.Context, indexName string, name string) ([]byte, error) {
cli, err := e.client()
if err != nil {
return nil, errors.Wrap(err, "GetField: creating client")
return errors.Wrap(err, "DeleteIndex: creating client")
}
defer cli.Close()
key := schemaPrefix + name
// Deleting index and fields in one transaction.
_, err = cli.KV.Txn(ctx).
If(clientv3.Compare(clientv3.Version(key), ">", -1)).
Then(
clientv3.OpDelete(key+"/", clientv3.WithPrefix()), // deleting index fields
clientv3.OpDelete(key), // deleting index
).Commit()
return errors.Wrap(err, "DeleteIndex")
}
func (e *Etcd) Field(ctx context.Context, indexName string, name string) ([]byte, error) {
key := schemaPrefix + indexName + "/" + name
return e.getKeyBytes(ctx, cli, key)
return e.getKeyBytes(ctx, key)
}
func (e *Etcd) CreateField(ctx context.Context, indexName string, name string, val []byte) error {
cli, err := e.client()
if err != nil {
return errors.Wrap(err, "CreateIndex: creating client")
return errors.Wrap(err, "CreateField: creating client")
}
defer cli.Close()
@ -632,14 +648,66 @@ func (e *Etcd) CreateField(ctx context.Context, indexName string, name string, v
}
if !resp.Succeeded {
return ErrFieldExists
return disco.ErrFieldExists
}
return nil
}
func (e *Etcd) DeleteField(ctx context.Context, indexname string, name string) error {
return e.delKey(ctx, schemaPrefix+indexname+"/"+name, false)
cli, err := e.client()
if err != nil {
return errors.Wrap(err, "DeleteField: creating client")
}
defer cli.Close()
key := schemaPrefix + indexname + "/" + name
// Deleting field and views in one transaction.
_, err = cli.KV.Txn(ctx).
If(clientv3.Compare(clientv3.Version(key), ">", -1)).
Then(
clientv3.OpDelete(key+"/", clientv3.WithPrefix()), // deleting field views
clientv3.OpDelete(key), // deleting field
).Commit()
return errors.Wrap(err, "DeleteField")
}
func (e *Etcd) View(ctx context.Context, indexName, fieldName, name string) ([]byte, error) {
key := schemaPrefix + indexName + "/" + fieldName + "/" + name
return e.getKeyBytes(ctx, key)
}
// CreateView differs from CreateIndex and CreateField in that it does not
// return an error if the view already exists. If this logic needs to be
// changed, we likely need to return disco.ErrViewExists.
func (e *Etcd) CreateView(ctx context.Context, indexName, fieldName, name string, val []byte) error {
cli, err := e.client()
if err != nil {
return errors.Wrap(err, "CreateView: creating client")
}
defer cli.Close()
key := schemaPrefix + indexName + "/" + fieldName + "/" + name
// Set up Op to write view value as bytes.
op := clientv3.OpPut(key, "")
op.WithValueBytes(val)
// Check for key existence, and execute Op within a transaction.
_, err = cli.KV.Txn(ctx).
If(clientv3util.KeyMissing(key)).
Then(op).
Commit()
if err != nil {
return errors.Wrap(err, "executing transaction")
}
return nil
}
func (e *Etcd) DeleteView(ctx context.Context, indexName, fieldName, name string) error {
return e.delKey(ctx, schemaPrefix+indexName+"/"+fieldName+"/"+name, false)
}
func (e *Etcd) putKey(ctx context.Context, key, val string, opts ...clientv3.OpOption) error {
@ -656,7 +724,13 @@ func (e *Etcd) putKey(ctx context.Context, key, val string, opts ...clientv3.OpO
return nil
}
func (e *Etcd) getKeyBytes(ctx context.Context, cli *clientv3.Client, key string) ([]byte, error) {
func (e *Etcd) getKeyBytes(ctx context.Context, key string) ([]byte, error) {
cli, err := e.client()
if err != nil {
return nil, errors.Wrap(err, "getKeyBytes: creates a new client")
}
defer cli.Close()
// Get the current value for the key.
resp, err := cli.Get(ctx, key)
if err != nil {
@ -671,7 +745,13 @@ func (e *Etcd) getKeyBytes(ctx context.Context, cli *clientv3.Client, key string
return resp.Kvs[0].Value, nil
}
func (e *Etcd) getKey(ctx context.Context, cli *clientv3.Client, key string) ([]string, [][]byte, error) {
func (e *Etcd) getKey(ctx context.Context, key string) ([]string, [][]byte, error) {
cli, err := e.client()
if err != nil {
return nil, nil, errors.Wrap(err, "getKey: creates a new client")
}
defer cli.Close()
resp, err := cli.KV.Txn(ctx).
If(clientv3.Compare(clientv3.Version(key), ">", -1)).
Then(clientv3.OpGet(key, clientv3.WithPrefix())).
@ -700,9 +780,9 @@ func (e *Etcd) getKey(ctx context.Context, cli *clientv3.Client, key string) ([]
}
func (e *Etcd) delKey(ctx context.Context, key string, withPrefix bool) error {
cli, err := clientv3.NewFromURLs(e.e.Server.Cluster().ClientURLs())
cli, err := e.client()
if err != nil {
return errors.Wrap(err, "delKey")
return errors.Wrap(err, "delKey: creates a new client")
}
defer cli.Close()

1
executor_test.go
View file

@ -651,6 +651,7 @@ func TestExecutor_Execute_Set(t *testing.T) {
})
t.Run("ErrInvalidColValueType", func(t *testing.T) {
hldr.SetBit("i", "f", 1, 0) // creates and Commits a Tx internally.
if err := idx.DeleteField("f"); err != nil {
t.Fatal(err)

64
field.go
View file

@ -31,6 +31,7 @@ import (
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2/disco"
"github.com/pilosa/pilosa/v2/internal"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/roaring"
@ -102,6 +103,8 @@ type Field struct {
broadcaster broadcaster
Stats stats.StatsClient
schemator disco.Schemator
serializer Serializer
// Field options.
options FieldOptions
@ -367,6 +370,8 @@ func newField(holder *Holder, path, index, name string, opts FieldOption) (*Fiel
broadcaster: NopBroadcaster,
Stats: stats.NopStatsClient,
schemator: disco.NopSchemator,
serializer: NopSerializer,
options: *applyDefaultOptions(&fo),
@ -1147,19 +1152,22 @@ func (f *Field) recalculateCaches() {
// createViewIfNotExists returns the named view, creating it if necessary.
// Additionally, a CreateViewMessage is sent to the cluster.
func (f *Field) createViewIfNotExists(name string) (*view, error) {
view, created, err := f.createViewIfNotExistsBase(name)
cvm := &CreateViewMessage{
Index: f.index,
Field: f.name,
View: name,
}
// call this base method to isolate the mu.Lock and ensure we aren't holding
// the lock while calling SendSync below.
view, created, err := f.createViewIfNotExistsBase(cvm)
if err != nil {
return nil, err
}
if created {
// Broadcast view creation to the cluster.
err = f.broadcaster.SendSync(
&CreateViewMessage{
Index: f.index,
Field: f.name,
View: name,
})
err := f.broadcaster.SendSync(cvm)
if err != nil {
return nil, errors.Wrap(err, "sending CreateView message")
}
@ -1169,15 +1177,26 @@ func (f *Field) createViewIfNotExists(name string) (*view, error) {
}
// createViewIfNotExistsBase returns the named view, creating it if necessary.
// The returned bool indicates whether the view was created or not.
func (f *Field) createViewIfNotExistsBase(name string) (*view, bool, error) {
// One purpose of isolating this method from createViewIfNotExists() is that we
// need to enforce the mu.Lock on everything in this method, but we can't be
// holding the lock when broadcasting the CreateViewMessage view
// broadcaster.SendSync(); calling that SendSync() while holding the lock can
// result in a deadlock waiting on the remote node to give up its lock obtained
// by performing the same action. The returned bool indicates whether the view
// was created or not.
func (f *Field) createViewIfNotExistsBase(cvm *CreateViewMessage) (*view, bool, error) {
f.mu.Lock()
defer f.mu.Unlock()
if view := f.viewMap[name]; view != nil {
// Create the view in etcd as the system of record.
if err := f.persistView(context.Background(), cvm); err != nil {
return nil, false, errors.Wrap(err, "persisting view")
}
if view := f.viewMap[cvm.View]; view != nil {
return view, false, nil
}
view := f.newView(f.viewPath(name), name)
view := f.newView(f.viewPath(cvm.View), cvm.View)
if err := view.openEmpty(); err != nil {
return nil, false, errors.Wrap(err, "opening view")
@ -1216,6 +1235,11 @@ func (f *Field) deleteView(name string) error {
delete(f.viewMap, name)
// Delete the view from etcd as the system of record.
if err := f.schemator.DeleteView(context.TODO(), f.index, f.name, name); err != nil {
return errors.Wrapf(err, "deleting view from etcd: %s/%s/%s", f.index, f.name, name)
}
return nil
}
@ -2200,3 +2224,21 @@ func bitDepthInt64(v int64) uint {
func FormatQualifiedFieldName(index, field string) string {
return fmt.Sprintf("%s\x00%s\x00", index, field)
}
// persistView stores the view information in etcd.
func (f *Field) persistView(ctx context.Context, cvm *CreateViewMessage) error {
if cvm.Index == "" {
return ErrIndexRequired
} else if cvm.Field == "" {
return ErrFieldRequired
} else if cvm.View == "" {
return ErrViewRequired
}
if b, err := f.serializer.Marshal(cvm); err != nil {
return errors.Wrap(err, "marshaling")
} else if err := f.schemator.CreateView(ctx, cvm.Index, cvm.Field, cvm.View, b); err != nil {
return errors.Wrapf(err, "writing field to disco: %s/%s/%s", cvm.Index, cvm.Field, cvm.View)
}
return nil
}

8
fragment_internal_test.go
View file

@ -3585,8 +3585,14 @@ func newTestHolder(tb testing.TB) *Holder {
// fragTestMustOpenIndex returns a new, opened index at a temporary path. Panic on error.
func fragTestMustOpenIndex(index string, holder *Holder, opt IndexOptions) *Index {
cim := &CreateIndexMessage{
Index: index,
CreatedAt: 0,
Meta: &opt,
}
holder.mu.Lock()
idx, err := holder.createIndex(index, opt)
idx, err := holder.createIndex(cim, false)
holder.mu.Unlock()
panicOn(err)

390
holder.go
View file

@ -30,6 +30,7 @@ import (
"syscall"
"time"
"github.com/pilosa/pilosa/v2/disco"
"github.com/pilosa/pilosa/v2/logger"
rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg"
"github.com/pilosa/pilosa/v2/roaring"
@ -80,6 +81,8 @@ type Holder struct {
opened lockedChan
broadcaster broadcaster
schemator disco.Schemator
serializer Serializer
NewAttrStore func(string) AttrStore
@ -210,6 +213,8 @@ type HolderConfig struct {
OpenTransactionStore OpenTransactionStoreFunc
OpenIDAllocator OpenIDAllocatorFunc
TranslationSyncer TranslationSyncer
Serializer Serializer
Schemator disco.Schemator
CacheFlushInterval time.Duration
StatsClient stats.StatsClient
NewAttrStore func(string) AttrStore
@ -229,6 +234,8 @@ func DefaultHolderConfig() *HolderConfig {
OpenTransactionStore: OpenInMemTransactionStore,
OpenIDAllocator: func(string) (*idAllocator, error) { return &idAllocator{}, nil },
TranslationSyncer: NopTranslationSyncer,
Serializer: NopSerializer,
Schemator: disco.NopSchemator,
CacheFlushInterval: defaultCacheFlushInterval,
StatsClient: stats.NopStatsClient,
NewAttrStore: newNopAttrStore,
@ -267,6 +274,8 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
OpenTransactionStore: cfg.OpenTransactionStore,
OpenIDAllocator: cfg.OpenIDAllocator,
translationSyncer: cfg.TranslationSyncer,
serializer: cfg.Serializer,
schemator: cfg.Schemator,
Logger: cfg.Logger,
Opts: HolderOpts{StorageBackend: cfg.StorageConfig.Backend, RowcacheOn: cfg.RowcacheOn},
@ -571,7 +580,6 @@ func (h *Holder) Inspect(ctx context.Context, req *InspectRequest) (*HolderInfo,
// Open initializes the root data directory for the holder.
func (h *Holder) Open() error {
h.opening = true
defer func() { h.opening = false }()
@ -853,52 +861,58 @@ func (h *Holder) availableShardsByIndex() map[string]*roaring.Bitmap {
}
// Schema returns schema information for all indexes, fields, and views.
func (h *Holder) Schema() []*IndexInfo {
var a []*IndexInfo
for _, index := range h.Indexes() {
di := &IndexInfo{
Name: index.Name(),
CreatedAt: index.CreatedAt(),
Options: index.Options(),
}
for _, field := range index.Fields() {
fi := &FieldInfo{
Name: field.Name(),
CreatedAt: field.CreatedAt(),
Options: field.Options(),
}
for _, view := range field.views() {
fi.Views = append(fi.Views, &ViewInfo{Name: view.name})
}
sort.Sort(viewInfoSlice(fi.Views))
di.Fields = append(di.Fields, fi)
}
sort.Sort(fieldInfoSlice(di.Fields))
a = append(a, di)
}
sort.Sort(indexInfoSlice(a))
return a
func (h *Holder) Schema() ([]*IndexInfo, error) {
return h.schema(context.TODO(), true)
}
// limitedSchema returns schema information for all indexes and fields.
func (h *Holder) limitedSchema() []*IndexInfo {
func (h *Holder) limitedSchema() ([]*IndexInfo, error) {
return h.schema(context.TODO(), false)
}
func (h *Holder) schema(ctx context.Context, includeViews bool) ([]*IndexInfo, error) {
var a []*IndexInfo
for _, index := range h.Indexes() {
di := &IndexInfo{
Name: index.Name(),
CreatedAt: index.CreatedAt(),
Options: index.Options(),
ShardWidth: ShardWidth,
Fields: make([]*FieldInfo, 0, len(index.Fields())),
schema, err := h.schemator.Schema(ctx)
if err != nil {
return nil, errors.Wrapf(err, "getting schema via schemator")
}
for _, index := range schema {
cim, err := h.decodeCreateIndexMessage(index.Data)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateIndexMessage")
}
for _, field := range index.Fields() {
if strings.HasPrefix(field.name, "_") {
di := &IndexInfo{
Name: cim.Index,
CreatedAt: cim.CreatedAt,
Options: *cim.Meta,
ShardWidth: ShardWidth,
Fields: make([]*FieldInfo, 0, len(index.Fields)),
}
for fieldName, field := range index.Fields {
if fieldName == existenceFieldName {
continue
}
cfm, err := h.decodeCreateFieldMessage(field.Data)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateFieldMessage")
}
fi := &FieldInfo{
Name: field.Name(),
CreatedAt: field.CreatedAt(),
Options: field.Options(),
Name: cfm.Field,
CreatedAt: cfm.CreatedAt,
Options: *cfm.Meta,
}
if includeViews {
for _, viewData := range field.Views {
cvm, err := h.decodeCreateViewMessage(viewData)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateViewMessage")
}
fi.Views = append(fi.Views, &ViewInfo{Name: cvm.View})
}
sort.Sort(viewInfoSlice(fi.Views))
}
di.Fields = append(di.Fields, fi)
}
@ -906,34 +920,26 @@ func (h *Holder) limitedSchema() []*IndexInfo {
a = append(a, di)
}
sort.Sort(indexInfoSlice(a))
return a
return a, nil
}
// applySchema applies an internal Schema to Holder.
func (h *Holder) applySchema(schema *Schema) error {
// Create indexes that don't exist.
// Create indexes.
// We use h.CreateIndex() instead of h.CreateIndexIfNotExists() because we
// want to limit the use of this method for now to only new indexes.
for _, i := range schema.Indexes {
idx, err := h.CreateIndexIfNotExists(i.Name, i.Options)
idx, err := h.CreateIndex(i.Name, i.Options)
if err != nil {
return errors.Wrap(err, "creating index")
}
if i.CreatedAt != 0 {
idx.mu.Lock()
idx.createdAt = i.CreatedAt
idx.mu.Unlock()
}
// Create fields that don't exist.
for _, f := range i.Fields {
fld, err := idx.createFieldIfNotExists(f.Name, &f.Options)
fld, err := idx.CreateFieldIfNotExistsWithOptions(f.Name, &f.Options)
if err != nil {
return errors.Wrap(err, "creating field")
}
if f.CreatedAt != 0 {
fld.mu.Lock()
fld.createdAt = f.CreatedAt
fld.mu.Unlock()
}
// Create views that don't exist.
for _, v := range f.Views {
@ -944,6 +950,12 @@ func (h *Holder) applySchema(schema *Schema) error {
}
}
}
// Send the load schema message to all nodes.
if err := h.broadcaster.SendSync(&LoadSchemaMessage{}); err != nil {
return errors.Wrap(err, "sending LoadSchemaMessage")
}
return nil
}
@ -995,7 +1007,93 @@ func (h *Holder) CreateIndex(name string, opt IndexOptions) (*Index, error) {
if h.Index(name) != nil {
return nil, newConflictError(ErrIndexExists)
}
return h.createIndex(name, opt)
cim := &CreateIndexMessage{
Index: name,
CreatedAt: timestamp(),
Meta: &opt,
}
// Create the index in etcd as the system of record.
if err := h.persistIndex(context.Background(), cim); err != nil {
return nil, errors.Wrap(err, "persisting index")
}
return h.createIndex(cim, false)
}
// LoadSchemaMessage is an internal message used to inform a node to load the
// latest schema from etcd.
type LoadSchemaMessage struct{}
// LoadSchema creates all indexes based on the information stored in schemator.
// It does not return an error if an index already exists. The thinking is that
// this method will load all indexes that don't already exist. We likely want to
// revisit this; for example, we might want to confirm that the createdAt
// timestamps on each of the indexes matches the value in etcd.
func (h *Holder) LoadSchema() error {
h.mu.Lock()
defer h.mu.Unlock()
return h.loadSchema()
}
// LoadIndex creates an index based on the information stored in schemator.
// An error is returned if the index already exists.
func (h *Holder) LoadIndex(name string) (*Index, error) {
h.mu.Lock()
defer h.mu.Unlock()
// Ensure index doesn't already exist.
if h.Index(name) != nil {
return nil, newConflictError(ErrIndexExists)
}
return h.loadIndex(name)
}
// LoadField creates a field based on the information stored in schemator.
// An error is returned if the field already exists.
func (h *Holder) LoadField(index, field string) (*Field, error) {
// Ensure field doesn't already exist.
if h.Field(index, field) != nil {
return nil, newConflictError(ErrFieldExists)
}
h.mu.Lock()
defer h.mu.Unlock()
return h.loadField(index, field)
}
// LoadView creates a view based on the information stored in schemator. Unlike
// index and field, it is not considered an error if the view already exists.
func (h *Holder) LoadView(index, field, view string) (*view, error) {
// If the view already exists, just return with it here.
if v := h.view(index, field, view); v != nil {
return v, nil
}
return h.loadView(index, field, view)
}
// CreateIndexAndBroadcast creates an index locally, then broadcasts the
// creation to other nodes so they can create locally as well. An error is
// returned if the index already exists.
func (h *Holder) CreateIndexAndBroadcast(cim *CreateIndexMessage) (*Index, error) {
h.mu.Lock()
defer h.mu.Unlock()
// Ensure index doesn't already exist.
if h.Index(cim.Index) != nil {
return nil, newConflictError(ErrIndexExists)
}
// Create the index in etcd as the system of record.
if err := h.persistIndex(context.Background(), cim); err != nil {
return nil, errors.Wrap(err, "persisting index")
}
return h.createIndex(cim, true)
}
// CreateIndexIfNotExists returns an index by name.
@ -1004,26 +1102,64 @@ func (h *Holder) CreateIndexIfNotExists(name string, opt IndexOptions) (*Index,
h.mu.Lock()
defer h.mu.Unlock()
// Return index if it exists.
cim := &CreateIndexMessage{
Index: name,
CreatedAt: timestamp(),
Meta: &opt,
}
// Create the index in etcd as the system of record.
err := h.persistIndex(context.Background(), cim)
if err != nil && errors.Cause(err) != disco.ErrIndexExists {
return nil, errors.Wrap(err, "persisting index")
}
if index := h.Index(name); index != nil {
return index, nil
}
return h.createIndex(name, opt)
// It may happen that index is not in memory, but it's already in etcd,
// then we need to create it locally.
return h.createIndex(cim, false)
}
func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) {
if name == "" {
// persistIndex stores the index information in etcd.
func (h *Holder) persistIndex(ctx context.Context, cim *CreateIndexMessage) error {
if cim.Index == "" {
return ErrIndexRequired
}
if err := validateName(cim.Index); err != nil {
return errors.Wrap(err, "validating name")
}
if b, err := h.serializer.Marshal(cim); err != nil {
return errors.Wrap(err, "marshaling")
} else if err := h.schemator.CreateIndex(ctx, cim.Index, b); err != nil {
return errors.Wrapf(err, "writing index to disco: %s", cim.Index)
}
return nil
}
func (h *Holder) createIndex(cim *CreateIndexMessage, broadcast bool) (*Index, error) {
if cim.Index == "" {
return nil, errors.New("index name required")
}
opt := cim.Meta
if opt == nil {
opt = &IndexOptions{}
}
// Otherwise create a new index.
index, err := h.newIndex(h.IndexPath(name), name)
index, err := h.newIndex(h.IndexPath(cim.Index), cim.Index)
if err != nil {
return nil, errors.Wrap(err, "creating")
}
index.keys = opt.Keys
index.trackExistence = opt.TrackExistence
index.createdAt = cim.CreatedAt
if err = index.Open(); err != nil {
return nil, errors.Wrap(err, "opening")
@ -1035,6 +1171,13 @@ func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) {
// Update options.
h.addIndex(index)
if broadcast {
// Send the create index message to all nodes.
if err := h.broadcaster.SendSync(cim); err != nil {
return nil, errors.Wrap(err, "sending CreateIndex message")
}
}
// Since this is a new index, we need to kick off
// its translation sync.
if err := h.translationSyncer.Reset(); err != nil {
@ -1044,6 +1187,97 @@ func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) {
return index, nil
}
func (h *Holder) loadSchema() error {
schema, err := h.schemator.Schema(context.TODO())
if err != nil {
return errors.Wrap(err, "getting schema")
}
// TODO: This is kind of inefficient because we're ignoring the index.Data
// and field.Data values, which contains the index and field information,
// and only using the map key to call loadIndex() and loadField(). These
// make another call to schemator to get the same index and field
// information that we already have in the map. It probably makes sense to
// either copy the parts of the loadIndex and loadField methods here (like
// decodeCreateIndexMessage) or split loadIndex and loadField into smaller
// methods that we could reuse here.
for indexName, index := range schema {
_, err := h.loadIndex(indexName)
if err != nil {
return errors.Wrap(err, "loading index")
}
for fieldName, field := range index.Fields {
_, err := h.loadField(indexName, fieldName)
if err != nil {
return errors.Wrap(err, "loading field")
}
for viewName := range field.Views {
_, err := h.loadView(indexName, fieldName, viewName)
if err != nil {
return errors.Wrap(err, "loading view")
}
}
}
}
return nil
}
func (h *Holder) loadIndex(indexName string) (*Index, error) {
b, err := h.schemator.Index(context.TODO(), indexName)
if err != nil {
return nil, errors.Wrapf(err, "getting index: %s", indexName)
}
cim, err := h.decodeCreateIndexMessage(b)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateIndexMessage")
}
return h.createIndex(cim, false)
}
func (h *Holder) loadField(indexName, fieldName string) (*Field, error) {
b, err := h.schemator.Field(context.TODO(), indexName, fieldName)
if err != nil {
return nil, errors.Wrapf(err, "getting field: %s/%s", indexName, fieldName)
}
// Get index.
idx := h.Index(indexName)
if idx == nil {
return nil, errors.Errorf("local index not found: %s", indexName)
}
cfm, err := h.decodeCreateFieldMessage(b)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateFieldMessage")
}
// TODO: can this take cfm?
return idx.createFieldIfNotExists(fieldName, cfm.Meta)
}
func (h *Holder) loadView(indexName, fieldName, viewName string) (*view, error) {
b, err := h.schemator.View(context.TODO(), indexName, fieldName, viewName)
if err != nil {
return nil, errors.Wrapf(err, "getting view: %s/%s/%s", indexName, fieldName, viewName)
}
// Get field.
fld := h.Field(indexName, fieldName)
if fld == nil {
return nil, errors.Errorf("local field not found: %s/%s", indexName, fieldName)
}
cvm, err := h.decodeCreateViewMessage(b)
if err != nil {
return nil, errors.Wrap(err, "decoding CreateFieldMessage")
}
return fld.createViewIfNotExists(cvm.View)
}
func (h *Holder) newIndex(path, name string) (*Index, error) {
index, err := NewIndex(h, path, name)
if err != nil {
@ -1051,6 +1285,8 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
}
index.Stats = h.Stats.WithTags(fmt.Sprintf("index:%s", index.Name()))
index.broadcaster = h.broadcaster
index.serializer = h.serializer
index.schemator = h.schemator
index.newAttrStore = h.NewAttrStore
index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data"))
index.OpenTranslateStore = h.OpenTranslateStore
@ -1087,6 +1323,11 @@ func (h *Holder) DeleteIndex(name string) error {
// Remove reference.
h.deleteIndex(name)
// Delete the index from etcd as the system of record.
if err := h.schemator.DeleteIndex(context.TODO(), name); err != nil {
return errors.Wrapf(err, "deleting index from etcd: %s", name)
}
// I'm not sure if calling Reset() here is necessary
// since closing the index stops its translation
// sync processes.
@ -1320,8 +1561,13 @@ func (s *holderSyncer) SyncHolder() error {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN)
schema, err := s.Holder.Schema()
if err != nil {
return errors.Wrap(err, "getting schema")
}
// Iterate over schema in sorted order.
for _, di := range s.Holder.Schema() {
for _, di := range schema {
// Verify syncer has not closed.
if s.IsClosing() {
return nil
@ -2047,3 +2293,27 @@ func (h *Holder) HasRoaringData() (has bool, err error) {
}
return
}
func (h *Holder) decodeCreateIndexMessage(b []byte) (*CreateIndexMessage, error) {
var cim CreateIndexMessage
if err := h.serializer.Unmarshal(b, &cim); err != nil {
return nil, errors.Wrap(err, "unmarshaling")
}
return &cim, nil
}
func (h *Holder) decodeCreateFieldMessage(b []byte) (*CreateFieldMessage, error) {
var cfm CreateFieldMessage
if err := h.serializer.Unmarshal(b, &cfm); err != nil {
return nil, errors.Wrap(err, "unmarshaling")
}
return &cfm, nil
}
func (h *Holder) decodeCreateViewMessage(b []byte) (*CreateViewMessage, error) {
var cvm CreateViewMessage
if err := h.serializer.Unmarshal(b, &cvm); err != nil {
return nil, errors.Wrap(err, "unmarshaling")
}
return &cvm, nil
}

12
http/handler.go
View file

@ -234,7 +234,7 @@ func (h *Handler) populateValidators() {
h.validators["PostQuery"] = queryValidationSpecRequired().Optional("shards", "columnAttrs", "excludeRowAttrs", "excludeColumns", "profile")
h.validators["GetInfo"] = queryValidationSpecRequired()
h.validators["RecalculateCaches"] = queryValidationSpecRequired()
h.validators["GetSchema"] = queryValidationSpecRequired()
h.validators["GetSchema"] = queryValidationSpecRequired().Optional("views")
h.validators["PostSchema"] = queryValidationSpecRequired().Optional("remote")
h.validators["GetStatus"] = queryValidationSpecRequired()
h.validators["GetVersion"] = queryValidationSpecRequired()
@ -665,8 +665,11 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
return
}
q := r.URL.Query()
withViews := q.Get("views") == "true"
w.Header().Set("Content-Type", "application/json")
schema, err := h.api.Schema(r.Context())
schema, err := h.api.Schema(r.Context(), withViews)
if err != nil {
h.logger.Printf("getting schema error: %s", err)
}
@ -978,8 +981,11 @@ func (h *Handler) handleGetIndex(w http.ResponseWriter, r *http.Request) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
q := r.URL.Query()
withViews := q.Get("views") == "true"
indexName := mux.Vars(r)["index"]
schema, err := h.api.Schema(r.Context())
schema, err := h.api.Schema(r.Context(), withViews)
if err != nil {
h.logger.Printf("getting schema error: %s", err)
}

158
index.go
View file

@ -26,6 +26,7 @@ import (
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2/disco"
"github.com/pilosa/pilosa/v2/internal"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/stats"
@ -56,6 +57,8 @@ type Index struct {
columnAttrs AttrStore
broadcaster broadcaster
schemator disco.Schemator
serializer Serializer
Stats stats.StatsClient
// Passed to field for foreign-index lookup.
@ -99,6 +102,9 @@ func NewIndex(holder *Holder, path, name string) (*Index, error) {
holder: holder,
trackExistence: true,
schemator: disco.NopSchemator,
serializer: NopSerializer,
translateStores: make(map[int]TranslateStore),
translationSyncer: NopTranslationSyncer,
@ -511,7 +517,44 @@ func (i *Index) CreateField(name string, opts ...FieldOption) (*Field, error) {
return nil, errors.Wrap(err, "applying option")
}
return i.createField(name, fo)
cfm := &CreateFieldMessage{
Index: i.name,
Field: name,
CreatedAt: 0,
Meta: fo,
}
// Create the field in etcd as the system of record.
if err := i.persistField(context.Background(), cfm); err != nil {
return nil, errors.Wrap(err, "persisting field")
}
return i.createField(cfm, false)
}
// CreateFieldAndBroadcast creates a field locally, then broadcasts the
// creation to other nodes so they can create locally as well. An error is
// returned if the field already exists.
func (i *Index) CreateFieldAndBroadcast(cfm *CreateFieldMessage) (*Field, error) {
err := validateName(cfm.Field)
if err != nil {
return nil, errors.Wrap(err, "validating name")
}
i.mu.Lock()
defer i.mu.Unlock()
// Ensure field doesn't already exist.
if i.fields[cfm.Field] != nil {
return nil, newConflictError(ErrFieldExists)
}
// Create the field in etcd as the system of record.
if err := i.persistField(context.Background(), cfm); err != nil {
return nil, errors.Wrap(err, "persisting field")
}
return i.createField(cfm, true)
}
// CreateFieldIfNotExists creates a field with the given options if it doesn't exist.
@ -535,7 +578,81 @@ func (i *Index) CreateFieldIfNotExists(name string, opts ...FieldOption) (*Field
return nil, errors.Wrap(err, "applying option")
}
return i.createField(name, fo)
cfm := &CreateFieldMessage{
Index: i.name,
Field: name,
CreatedAt: 0,
Meta: fo,
}
// Create the field in etcd as the system of record.
if err := i.persistField(context.Background(), cfm); err != nil {
// There is a case where the index is not in memory, but it is in
// persistent storage. In that case, this will return an "index exists"
// error, which in that case should return the index. TODO: We may need
// to allow for that in the future.
return nil, errors.Wrap(err, "persisting field")
}
return i.createField(cfm, false)
}
// CreateFieldIfNotExistsWithOptions is a method which I created because I
// needed the functionality of CreateFieldIfNotExists, but instead of taking
// function options, taking a *FieldOptions struct. TODO: This should
// definintely be refactored so we don't have these virtually equivalent
// methods, but I'm puttin this here for now just to see if it works.
func (i *Index) CreateFieldIfNotExistsWithOptions(name string, opt *FieldOptions) (*Field, error) {
err := validateName(name)
if err != nil {
return nil, errors.Wrap(err, "validating name")
}
i.mu.Lock()
defer i.mu.Unlock()
// Find field in cache first.
if f := i.fields[name]; f != nil {
return f, nil
}
cfm := &CreateFieldMessage{
Index: i.name,
Field: name,
CreatedAt: 0,
Meta: opt,
}
// Create the field in etcd as the system of record.
if err := i.persistField(context.Background(), cfm); err != nil {
// There is a case where the index is not in memory, but it is in
// persistent storage. In that case, this will return an "index exists"
// error, which in that case should return the index. TODO: We may need
// to allow for that in the future.
return nil, errors.Wrap(err, "persisting field")
}
return i.createField(cfm, false)
}
// persistField stores the field information in etcd.
func (i *Index) persistField(ctx context.Context, cfm *CreateFieldMessage) error {
if cfm.Index == "" {
return ErrIndexRequired
} else if cfm.Field == "" {
return ErrFieldRequired
}
if err := validateName(cfm.Field); err != nil {
return errors.Wrap(err, "validating name")
}
if b, err := i.serializer.Marshal(cfm); err != nil {
return errors.Wrap(err, "marshaling")
} else if err := i.schemator.CreateField(ctx, cfm.Index, cfm.Field, b); err != nil {
return errors.Wrapf(err, "writing field to disco: %s/%s", cfm.Index, cfm.Field)
}
return nil
}
func (i *Index) createFieldIfNotExists(name string, opt *FieldOptions) (*Field, error) {
@ -547,21 +664,34 @@ func (i *Index) createFieldIfNotExists(name string, opt *FieldOptions) (*Field,
return f, nil
}
return i.createField(name, opt)
cfm := &CreateFieldMessage{
Index: i.name,
Field: name,
CreatedAt: 0,
Meta: opt,
}
return i.createField(cfm, false)
}
func (i *Index) createField(name string, opt *FieldOptions) (*Field, error) {
if name == "" {
func (i *Index) createField(cfm *CreateFieldMessage, broadcast bool) (*Field, error) {
opt := cfm.Meta
if opt == nil {
opt = &FieldOptions{}
}
if cfm.Field == "" {
return nil, errors.New("field name required")
} else if opt.CacheType != "" && !isValidCacheType(opt.CacheType) {
return nil, ErrInvalidCacheType
}
// Initialize field.
f, err := i.newField(i.fieldPath(name), name)
f, err := i.newField(i.fieldPath(cfm.Field), cfm.Field)
if err != nil {
return nil, errors.Wrap(err, "initializing")
}
f.createdAt = cfm.CreatedAt
// Pass holder through to the field for use in looking
// up a foreign index.
@ -580,11 +710,18 @@ func (i *Index) createField(name string, opt *FieldOptions) (*Field, error) {
}
// Add to index's field lookup.
i.fields[name] = f
i.fields[cfm.Field] = f
// enable Txf to find the index in field_test.go TestField_SetValue
f.idx = i
if broadcast {
// Send the create field message to all nodes.
if err := i.broadcaster.SendSync(cfm); err != nil {
return nil, errors.Wrap(err, "sending CreateField message")
}
}
// Kick off the field's translation sync process.
if err := i.translationSyncer.Reset(); err != nil {
return nil, errors.Wrap(err, "resetting translation syncer")
@ -601,6 +738,8 @@ func (i *Index) newField(path, name string) (*Field, error) {
f.idx = i
f.Stats = i.Stats
f.broadcaster = i.broadcaster
f.schemator = i.schemator
f.serializer = i.serializer
f.rowAttrStore = i.newAttrStore(filepath.Join(f.path, ".data"))
f.OpenTranslateStore = i.OpenTranslateStore
return f, nil
@ -641,6 +780,11 @@ func (i *Index) DeleteField(name string) error {
// Remove reference.
delete(i.fields, name)
// Delete the field from etcd as the system of record.
if err := i.schemator.DeleteField(context.TODO(), i.name, name); err != nil {
return errors.Wrapf(err, "deleting field from etcd: %s/%s", i.name, name)
}
return i.translationSyncer.Reset()
}

327
internal/private.pb.go
View file

@ -2157,6 +2157,45 @@ func (m *RecalculateCaches) XXX_DiscardUnknown() {
var xxx_messageInfo_RecalculateCaches proto.InternalMessageInfo
type LoadSchemaMessage struct {
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *LoadSchemaMessage) Reset() { *m = LoadSchemaMessage{} }
func (m *LoadSchemaMessage) String() string { return proto.CompactTextString(m) }
func (*LoadSchemaMessage) ProtoMessage() {}
func (*LoadSchemaMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_d2a91b51c7bdc125, []int{33}
}
func (m *LoadSchemaMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *LoadSchemaMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_LoadSchemaMessage.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *LoadSchemaMessage) XXX_Merge(src proto.Message) {
xxx_messageInfo_LoadSchemaMessage.Merge(m, src)
}
func (m *LoadSchemaMessage) XXX_Size() int {
return m.Size()
}
func (m *LoadSchemaMessage) XXX_DiscardUnknown() {
xxx_messageInfo_LoadSchemaMessage.DiscardUnknown(m)
}
var xxx_messageInfo_LoadSchemaMessage proto.InternalMessageInfo
type TransactionMessage struct {
Action string `protobuf:"bytes,1,opt,name=Action,proto3" json:"Action,omitempty"`
Transaction *Transaction `protobuf:"bytes,2,opt,name=Transaction,proto3" json:"Transaction,omitempty"`
@ -2169,7 +2208,7 @@ func (m *TransactionMessage) Reset() { *m = TransactionMessage{} }
func (m *TransactionMessage) String() string { return proto.CompactTextString(m) }
func (*TransactionMessage) ProtoMessage() {}
func (*TransactionMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_d2a91b51c7bdc125, []int{33}
return fileDescriptor_d2a91b51c7bdc125, []int{34}
}
func (m *TransactionMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -2228,7 +2267,7 @@ func (m *Transaction) Reset() { *m = Transaction{} }
func (m *Transaction) String() string { return proto.CompactTextString(m) }
func (*Transaction) ProtoMessage() {}
func (*Transaction) Descriptor() ([]byte, []int) {
return fileDescriptor_d2a91b51c7bdc125, []int{34}
return fileDescriptor_d2a91b51c7bdc125, []int{35}
}
func (m *Transaction) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -2309,7 +2348,7 @@ func (m *TransactionStats) Reset() { *m = TransactionStats{} }
func (m *TransactionStats) String() string { return proto.CompactTextString(m) }
func (*TransactionStats) ProtoMessage() {}
func (*TransactionStats) Descriptor() ([]byte, []int) {
return fileDescriptor_d2a91b51c7bdc125, []int{35}
return fileDescriptor_d2a91b51c7bdc125, []int{36}
}
func (m *TransactionStats) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -2348,7 +2387,7 @@ func (m *ResizeAbortMessage) Reset() { *m = ResizeAbortMessage{} }
func (m *ResizeAbortMessage) String() string { return proto.CompactTextString(m) }
func (*ResizeAbortMessage) ProtoMessage() {}
func (*ResizeAbortMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_d2a91b51c7bdc125, []int{36}
return fileDescriptor_d2a91b51c7bdc125, []int{37}
}
func (m *ResizeAbortMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -2389,7 +2428,7 @@ func (m *ResizeNodeMessage) Reset() { *m = ResizeNodeMessage{} }
func (m *ResizeNodeMessage) String() string { return proto.CompactTextString(m) }
func (*ResizeNodeMessage) ProtoMessage() {}
func (*ResizeNodeMessage) Descriptor() ([]byte, []int) {
return fileDescriptor_d2a91b51c7bdc125, []int{37}
return fileDescriptor_d2a91b51c7bdc125, []int{38}
}
func (m *ResizeNodeMessage) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
@ -2467,6 +2506,7 @@ func init() {
proto.RegisterType((*ResizeInstructionComplete)(nil), "internal.ResizeInstructionComplete")
proto.RegisterType((*Topology)(nil), "internal.Topology")
proto.RegisterType((*RecalculateCaches)(nil), "internal.RecalculateCaches")
proto.RegisterType((*LoadSchemaMessage)(nil), "internal.LoadSchemaMessage")
proto.RegisterType((*TransactionMessage)(nil), "internal.TransactionMessage")
proto.RegisterType((*Transaction)(nil), "internal.Transaction")
proto.RegisterType((*TransactionStats)(nil), "internal.TransactionStats")
@ -2477,98 +2517,98 @@ func init() {
func init() { proto.RegisterFile("private.proto", fileDescriptor_d2a91b51c7bdc125) }
var fileDescriptor_d2a91b51c7bdc125 = []byte{
// 1446 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x58, 0xdd, 0x6e, 0x1b, 0x45,
0x14, 0x66, 0xbd, 0x76, 0x6c, 0x1f, 0xc7, 0xa9, 0x33, 0x4d, 0xd3, 0x6d, 0xa8, 0x82, 0x19, 0x10,
0x35, 0x95, 0x1a, 0xaa, 0x16, 0x09, 0x04, 0xaa, 0xd4, 0x24, 0x4e, 0x8b, 0x81, 0xb4, 0xe9, 0x24,
0xed, 0xfd, 0x64, 0x3d, 0x6a, 0x56, 0x59, 0xef, 0xba, 0xfb, 0x93, 0xda, 0x45, 0xe2, 0x16, 0x04,
0x57, 0x08, 0x2e, 0xb8, 0xe4, 0x3d, 0x78, 0x01, 0x2e, 0x79, 0x04, 0x54, 0x9e, 0x80, 0x37, 0x40,
0x73, 0x66, 0x66, 0x77, 0xed, 0x38, 0x75, 0x68, 0xb9, 0xdb, 0xf3, 0xff, 0x9d, 0x9f, 0x39, 0x33,
0x36, 0x34, 0x87, 0x91, 0x77, 0xc2, 0x13, 0xb1, 0x31, 0x8c, 0xc2, 0x24, 0x24, 0x35, 0x2f, 0x48,
0x44, 0x14, 0x70, 0x7f, 0x6d, 0x71, 0x98, 0x1e, 0xfa, 0x9e, 0xab, 0xf8, 0xf4, 0x3e, 0xd4, 0x7b,
0x41, 0x5f, 0x8c, 0x76, 0x45, 0xc2, 0x09, 0x81, 0xf2, 0x57, 0x62, 0x1c, 0x3b, 0x76, 0xdb, 0xea,
0xd4, 0x18, 0x7e, 0x93, 0x0f, 0x60, 0xe9, 0x20, 0xe2, 0xee, 0xf1, 0xce, 0xc8, 0x8b, 0x13, 0x11,
0xb8, 0xc2, 0x29, 0xa3, 0x74, 0x8a, 0x4b, 0x7f, 0xb3, 0x61, 0xf1, 0x9e, 0x27, 0xfc, 0xfe, 0xc3,
0x61, 0xe2, 0x85, 0x41, 0x2c, 0x9d, 0x1d, 0x8c, 0x87, 0xc2, 0xa9, 0xb5, 0xad, 0x4e, 0x9d, 0xe1,
0x37, 0xb9, 0x0a, 0xf5, 0x6d, 0xee, 0x1e, 0x09, 0x14, 0xd8, 0x28, 0xc8, 0x19, 0x99, 0x74, 0xdf,
0x7b, 0xa1, 0xa2, 0x34, 0x59, 0xce, 0x20, 0x6d, 0x68, 0x1c, 0x78, 0x03, 0xf1, 0x28, 0xe5, 0x41,
0x92, 0x0e, 0x9c, 0x0a, 0x5a, 0x17, 0x59, 0x64, 0x15, 0x16, 0x1e, 0xfa, 0xfd, 0x5d, 0x2f, 0x70,
0xea, 0x6d, 0xab, 0x63, 0x33, 0x4d, 0x19, 0x3e, 0x1f, 0x39, 0x90, 0xf3, 0xf9, 0x28, 0x4b, 0xb7,
0x31, 0x99, 0xee, 0x83, 0x70, 0x3f, 0xe1, 0x41, 0x9f, 0x47, 0xfd, 0x27, 0x9e, 0x78, 0xee, 0x2c,
0xaa, 0x74, 0x27, 0xb9, 0xd2, 0x76, 0x8b, 0xc7, 0xc2, 0x69, 0xa2, 0x47, 0xfc, 0x26, 0x6b, 0x50,
0xdb, 0xf2, 0x92, 0xae, 0x18, 0x26, 0x47, 0xce, 0x52, 0xdb, 0xea, 0x94, 0x59, 0x46, 0x93, 0x15,
0xa8, 0xec, 0xbb, 0xdc, 0x17, 0xce, 0x05, 0x34, 0x50, 0x04, 0xa1, 0xb0, 0x78, 0x2f, 0x8c, 0x84,
0xf7, 0x34, 0xc0, 0x26, 0x38, 0x2d, 0x4c, 0x6a, 0x82, 0x47, 0xde, 0x03, 0x5b, 0xa6, 0xb4, 0xdc,
0xb6, 0x3a, 0x8d, 0x5b, 0xcb, 0x1b, 0xa6, 0x8f, 0x1b, 0x5d, 0xe1, 0x7a, 0x03, 0xee, 0x33, 0x29,
0x45, 0x25, 0x3e, 0x72, 0xc8, 0xd9, 0x4a, 0x7c, 0x44, 0x29, 0x2c, 0xf5, 0x06, 0xc3, 0x30, 0x4a,
0x98, 0x88, 0x87, 0x61, 0x10, 0x0b, 0xd2, 0x02, 0x7b, 0x27, 0x8a, 0x1c, 0x0b, 0xc3, 0xca, 0x4f,
0xfa, 0x2d, 0xb4, 0xb6, 0xfc, 0xd0, 0x3d, 0xee, 0xf2, 0x84, 0x33, 0xf1, 0x2c, 0x15, 0x71, 0x22,
0xb1, 0x2b, 0x78, 0x4a, 0x4f, 0x11, 0x92, 0x8b, 0xfd, 0x76, 0x4a, 0x8a, 0x8b, 0x84, 0xac, 0x0b,
0x56, 0x4d, 0xb5, 0x07, 0xbf, 0x31, 0xf7, 0x23, 0x1e, 0xf5, 0xb1, 0xa7, 0x65, 0xa6, 0x08, 0xc9,
0xc5, 0x48, 0x38, 0x07, 0x65, 0xa6, 0x08, 0xda, 0x83, 0xe5, 0x42, 0x7c, 0x0d, 0x73, 0x15, 0x16,
0x58, 0xf8, 0xbc, 0xd7, 0x8d, 0x1d, 0xab, 0x6d, 0x77, 0xca, 0x4c, 0x53, 0x38, 0x30, 0xa1, 0x9f,
0x0e, 0x02, 0x29, 0x2a, 0xa1, 0x28, 0x67, 0xd0, 0x2b, 0x50, 0xc1, 0xe9, 0x91, 0x59, 0xe6, 0xb6,
0xf2, 0x93, 0x7e, 0x67, 0x41, 0x7d, 0x97, 0x8f, 0x10, 0x48, 0x4c, 0xee, 0x40, 0xcd, 0xf4, 0x16,
0x95, 0x1a, 0xb7, 0xde, 0xcd, 0x2b, 0x98, 0xa9, 0x6d, 0x18, 0x9d, 0x9d, 0x20, 0x89, 0xc6, 0x2c,
0x33, 0x59, 0xfb, 0x1c, 0x9a, 0x13, 0x22, 0x19, 0xef, 0x58, 0x8c, 0x4d, 0x55, 0x8f, 0xc5, 0x58,
0xe6, 0x7a, 0xc2, 0xfd, 0x54, 0x60, 0xad, 0xca, 0x4c, 0x11, 0x9f, 0x95, 0x3e, 0xb5, 0xe8, 0x13,
0x20, 0xdb, 0x91, 0xe0, 0x89, 0xc0, 0x20, 0xbb, 0x22, 0x8e, 0xf9, 0x53, 0x31, 0xaf, 0xe2, 0x76,
0xb1, 0xe2, 0x59, 0x75, 0x4b, 0x85, 0xea, 0xd2, 0xeb, 0x40, 0xba, 0xc2, 0x17, 0x89, 0xd0, 0xa7,
0xfb, 0x15, 0x7e, 0xe9, 0x33, 0x83, 0x61, 0xbe, 0x2e, 0xb9, 0x06, 0x65, 0xb9, 0x2a, 0x30, 0x58,
0xe3, 0xd6, 0xc5, 0xbc, 0x4e, 0xd9, 0x16, 0x61, 0xa8, 0x80, 0xbd, 0x41, 0xa7, 0xfd, 0xcd, 0x04,
0x01, 0xdb, 0x2c, 0x67, 0xd0, 0x1f, 0x2c, 0x13, 0x13, 0x93, 0x38, 0x67, 0xde, 0x13, 0x93, 0x76,
0x5d, 0x23, 0xb1, 0x11, 0xc9, 0x6a, 0x8e, 0xa4, 0xb8, 0x85, 0x66, 0x81, 0x29, 0x4f, 0x83, 0xb9,
0x6b, 0x6a, 0xf5, 0xba, 0x58, 0xa8, 0x0b, 0x6f, 0x2b, 0x0f, 0x9b, 0x27, 0xdc, 0xf3, 0xf9, 0xa1,
0xff, 0x9f, 0xda, 0x39, 0x91, 0x96, 0x03, 0x55, 0xb4, 0xed, 0x75, 0xf5, 0xc1, 0x30, 0x24, 0xfd,
0x06, 0xf2, 0x33, 0xf6, 0x80, 0x0f, 0x84, 0xf6, 0x86, 0xdf, 0x59, 0x35, 0x4a, 0xe7, 0xa8, 0xc6,
0x0a, 0x54, 0xe4, 0xb9, 0x94, 0x7b, 0xde, 0x96, 0x81, 0x91, 0x98, 0x53, 0xa3, 0xdb, 0xb0, 0xb0,
0xef, 0x1e, 0x89, 0x01, 0x27, 0x1f, 0x42, 0x15, 0xf1, 0x8b, 0x58, 0x1f, 0x96, 0x0b, 0x53, 0x43,
0xc0, 0x8c, 0x9c, 0xfe, 0x64, 0xe9, 0xc4, 0x67, 0x42, 0x9e, 0x08, 0x58, 0x9a, 0x0a, 0x48, 0x6e,
0x40, 0x55, 0xa3, 0xc6, 0x5d, 0x72, 0xc6, 0xac, 0x19, 0x1d, 0x72, 0x0d, 0x16, 0x30, 0xd3, 0xd8,
0x29, 0x4f, 0x83, 0x42, 0x3e, 0xd3, 0x62, 0xba, 0x03, 0xf6, 0x63, 0xd6, 0x93, 0x2b, 0x05, 0xf3,
0x31, 0x90, 0x34, 0x25, 0x81, 0x7e, 0x11, 0xc6, 0x89, 0xee, 0x09, 0x7e, 0x4b, 0xde, 0x5e, 0x18,
0xa9, 0x29, 0x6e, 0x32, 0xfc, 0xa6, 0xbf, 0x58, 0x50, 0x7e, 0x10, 0xf6, 0x05, 0x59, 0x82, 0x52,
0xaf, 0xab, 0x9d, 0x94, 0x7a, 0x5d, 0xf2, 0x0e, 0xfa, 0xd7, 0x7d, 0x68, 0xe6, 0x28, 0x1e, 0xb3,
0x1e, 0xc3, 0xc8, 0x57, 0xa1, 0xde, 0x8b, 0xf7, 0x22, 0x6f, 0xc0, 0xa3, 0xb1, 0xbe, 0x69, 0x73,
0x06, 0x9e, 0xe6, 0x84, 0x27, 0xea, 0xfe, 0xab, 0x33, 0x45, 0x90, 0x6b, 0x50, 0xbd, 0xcf, 0xf6,
0xb6, 0xa5, 0xe3, 0xca, 0x2c, 0xc7, 0x46, 0x4a, 0xef, 0x42, 0x4b, 0xa2, 0x42, 0x2b, 0x33, 0x7d,
0xab, 0xb0, 0x20, 0x79, 0x19, 0x4a, 0x4d, 0xe5, 0xa1, 0x4a, 0x85, 0x50, 0xf4, 0x6b, 0xe5, 0x61,
0xe7, 0x44, 0x04, 0x49, 0x61, 0x7e, 0x91, 0x46, 0x07, 0x4d, 0xa6, 0x08, 0x42, 0x55, 0x05, 0x74,
0xaa, 0x4b, 0x39, 0x22, 0xc9, 0x65, 0x28, 0xa3, 0x3f, 0x5a, 0x00, 0x06, 0x50, 0x1a, 0x67, 0x26,
0xd6, 0xd9, 0x26, 0xa4, 0x63, 0x26, 0x4d, 0x9f, 0xec, 0x56, 0xae, 0xa5, 0xf8, 0xcc, 0x4c, 0xe2,
0x47, 0xf9, 0x24, 0xaa, 0xa6, 0x5f, 0x9a, 0x1a, 0x11, 0x15, 0x35, 0x9f, 0xc7, 0x00, 0x1a, 0x05,
0xfe, 0xcc, 0xa1, 0xbc, 0x91, 0xcd, 0x51, 0x69, 0xda, 0x25, 0xf2, 0xb5, 0x4b, 0xad, 0x34, 0x67,
0xcb, 0x79, 0xd0, 0x28, 0x18, 0xcd, 0x8c, 0xd7, 0x81, 0x0b, 0x93, 0x3b, 0xc3, 0x5c, 0x64, 0xd3,
0xec, 0x39, 0xa1, 0x7e, 0xb6, 0xa0, 0xb9, 0xed, 0xa7, 0x71, 0x22, 0x22, 0x1d, 0x4d, 0xea, 0x2b,
0x46, 0xd6, 0xf9, 0x9c, 0x31, 0xbb, 0xf9, 0xe4, 0x7d, 0xa8, 0xc8, 0x1e, 0xa8, 0xcd, 0x70, 0xba,
0x41, 0x4a, 0x58, 0xe8, 0x50, 0xf9, 0xd5, 0x1d, 0xa2, 0x4f, 0xa0, 0xb6, 0xb5, 0xdf, 0xbb, 0x1f,
0x85, 0xe9, 0x70, 0x66, 0xf6, 0xe6, 0x8d, 0x58, 0x2a, 0xbc, 0x11, 0x5b, 0xea, 0xbd, 0xa3, 0x32,
0xc4, 0xc7, 0x4d, 0x4b, 0x3d, 0x6e, 0xca, 0x9a, 0xc3, 0x47, 0x74, 0x1f, 0x96, 0x55, 0xea, 0x72,
0x75, 0xbd, 0xce, 0x96, 0x35, 0xcf, 0x14, 0x3b, 0x7f, 0xa6, 0x48, 0xa7, 0x6a, 0x89, 0xff, 0x9f,
0x4e, 0xff, 0x29, 0xc1, 0x32, 0x13, 0xb1, 0xf7, 0x42, 0xf4, 0x82, 0x38, 0x89, 0x52, 0x57, 0xae,
0x2b, 0x69, 0xff, 0x65, 0x78, 0xa8, 0xfb, 0x62, 0x33, 0x45, 0x9c, 0xe7, 0x40, 0x91, 0x0e, 0x54,
0x8b, 0xbb, 0xe3, 0xb4, 0x9a, 0x11, 0x93, 0x9b, 0x50, 0xdd, 0x0f, 0xd3, 0xc8, 0xcd, 0x4e, 0x47,
0xe1, 0x52, 0x50, 0x88, 0x94, 0x98, 0x19, 0x35, 0xf2, 0x08, 0xc8, 0x41, 0xc4, 0x83, 0xd8, 0xe7,
0x12, 0xa4, 0x31, 0xae, 0x4d, 0xbf, 0x88, 0x0a, 0x3a, 0x13, 0x7e, 0x66, 0x18, 0x93, 0x8f, 0x8b,
0xc7, 0xdf, 0xa9, 0x22, 0xe2, 0x95, 0x49, 0xc4, 0xfa, 0x44, 0x15, 0xd7, 0xc4, 0x9d, 0xa9, 0x59,
0x76, 0x16, 0xd0, 0xf0, 0x72, 0x6e, 0x38, 0x21, 0x66, 0x93, 0xda, 0xf4, 0x7b, 0x0b, 0x16, 0x8b,
0xc8, 0xce, 0xb5, 0x76, 0xb2, 0x46, 0x97, 0xe6, 0x3f, 0xb9, 0x4c, 0xa3, 0xcb, 0xb3, 0x1e, 0xb9,
0x95, 0xe2, 0x33, 0x2c, 0x85, 0xcb, 0x67, 0x94, 0xeb, 0x0d, 0x40, 0xb5, 0xa1, 0xb1, 0xc7, 0xa3,
0xc4, 0x93, 0x2e, 0xf5, 0x33, 0xa1, 0xc2, 0x8a, 0x2c, 0x7a, 0x0c, 0x57, 0x4e, 0x0d, 0xdd, 0x76,
0x38, 0x18, 0xca, 0xe9, 0x7e, 0x83, 0xe1, 0x93, 0xf7, 0x40, 0x14, 0x85, 0x91, 0xa9, 0x06, 0x12,
0x74, 0x0b, 0x6a, 0x07, 0xe1, 0x30, 0xf4, 0xc3, 0xa7, 0xe3, 0x39, 0x4b, 0xc7, 0x81, 0xaa, 0xba,
0x7b, 0xd4, 0x92, 0xab, 0x33, 0x43, 0xd2, 0x8b, 0xf2, 0x94, 0xb8, 0xdc, 0x77, 0x53, 0x9f, 0x27,
0x02, 0x9f, 0xed, 0x31, 0x15, 0x7a, 0x1e, 0x39, 0xe2, 0x2f, 0x5c, 0x67, 0x9b, 0xc8, 0x30, 0xd7,
0x99, 0xa2, 0xc8, 0x27, 0xd0, 0x28, 0x68, 0xeb, 0x3c, 0x2e, 0x4d, 0x8d, 0xad, 0x12, 0xb2, 0xa2,
0x26, 0xfd, 0xdd, 0x9a, 0xb0, 0x3c, 0x75, 0xa3, 0xeb, 0x80, 0x27, 0xaa, 0x36, 0x35, 0xa6, 0x29,
0x99, 0xeb, 0xce, 0xc8, 0xf5, 0xd3, 0x58, 0x8a, 0xf4, 0x45, 0x9e, 0x31, 0x64, 0xae, 0xf2, 0xb7,
0x69, 0x98, 0x9a, 0xc7, 0x94, 0x21, 0xe5, 0xcf, 0xc4, 0xae, 0xe0, 0x7d, 0xdf, 0x0b, 0x04, 0x0e,
0x8b, 0xcd, 0x32, 0x9a, 0xdc, 0x54, 0x6b, 0xd9, 0x4c, 0xfc, 0xda, 0x4c, 0xf8, 0xa8, 0xa1, 0x56,
0x76, 0x4c, 0x09, 0xb4, 0xa6, 0x45, 0x74, 0x05, 0x88, 0x6a, 0xff, 0xe6, 0x61, 0x18, 0x99, 0x5b,
0x9c, 0x6e, 0x9b, 0x4d, 0x24, 0x8b, 0x3e, 0xef, 0x71, 0x90, 0x57, 0xb9, 0x54, 0xac, 0xf2, 0x56,
0xeb, 0x8f, 0x97, 0xeb, 0xd6, 0x9f, 0x2f, 0xd7, 0xad, 0xbf, 0x5e, 0xae, 0x5b, 0xbf, 0xfe, 0xbd,
0xfe, 0xd6, 0xe1, 0x02, 0xfe, 0x91, 0x70, 0xfb, 0xdf, 0x00, 0x00, 0x00, 0xff, 0xff, 0x59, 0x5b,
0x70, 0x29, 0x71, 0x10, 0x00, 0x00,
// 1456 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x58, 0xdd, 0x6e, 0x1b, 0xc5,
0x17, 0xff, 0xaf, 0xd7, 0x8e, 0xed, 0xe3, 0x38, 0x75, 0xa6, 0x69, 0xba, 0xcd, 0xbf, 0x0a, 0x66,
0x40, 0xd4, 0x54, 0x6a, 0xa8, 0x5a, 0x24, 0x10, 0xa8, 0x52, 0x93, 0x38, 0x2d, 0x86, 0xa6, 0x4d,
0x27, 0x69, 0xef, 0x27, 0xeb, 0x51, 0xb3, 0xca, 0x7a, 0xd7, 0xdd, 0x8f, 0xd4, 0x2e, 0x12, 0xb7,
0x20, 0xb8, 0x42, 0x70, 0xc1, 0x25, 0xef, 0xc1, 0x0b, 0x70, 0xc9, 0x23, 0xa0, 0xf2, 0x04, 0xbc,
0x01, 0x9a, 0x33, 0x33, 0xbb, 0x6b, 0xc7, 0xa9, 0x43, 0xcb, 0xdd, 0x9e, 0xef, 0xdf, 0xf9, 0x98,
0x33, 0x63, 0x43, 0x73, 0x18, 0x79, 0x27, 0x3c, 0x11, 0x1b, 0xc3, 0x28, 0x4c, 0x42, 0x52, 0xf3,
0x82, 0x44, 0x44, 0x01, 0xf7, 0xd7, 0x16, 0x87, 0xe9, 0xa1, 0xef, 0xb9, 0x8a, 0x4f, 0xef, 0x43,
0xbd, 0x17, 0xf4, 0xc5, 0x68, 0x57, 0x24, 0x9c, 0x10, 0x28, 0x7f, 0x25, 0xc6, 0xb1, 0x63, 0xb7,
0xad, 0x4e, 0x8d, 0xe1, 0x37, 0xf9, 0x00, 0x96, 0x0e, 0x22, 0xee, 0x1e, 0xef, 0x8c, 0xbc, 0x38,
0x11, 0x81, 0x2b, 0x9c, 0x32, 0x4a, 0xa7, 0xb8, 0xf4, 0x57, 0x1b, 0x16, 0xef, 0x79, 0xc2, 0xef,
0x3f, 0x1a, 0x26, 0x5e, 0x18, 0xc4, 0xd2, 0xd9, 0xc1, 0x78, 0x28, 0x9c, 0x5a, 0xdb, 0xea, 0xd4,
0x19, 0x7e, 0x93, 0xab, 0x50, 0xdf, 0xe6, 0xee, 0x91, 0x40, 0x81, 0x8d, 0x82, 0x9c, 0x91, 0x49,
0xf7, 0xbd, 0x97, 0x2a, 0x4a, 0x93, 0xe5, 0x0c, 0xd2, 0x86, 0xc6, 0x81, 0x37, 0x10, 0x8f, 0x53,
0x1e, 0x24, 0xe9, 0xc0, 0xa9, 0xa0, 0x75, 0x91, 0x45, 0x56, 0x61, 0xe1, 0x91, 0xdf, 0xdf, 0xf5,
0x02, 0xa7, 0xde, 0xb6, 0x3a, 0x36, 0xd3, 0x94, 0xe1, 0xf3, 0x91, 0x03, 0x39, 0x9f, 0x8f, 0xb2,
0x74, 0x1b, 0x93, 0xe9, 0x3e, 0x0c, 0xf7, 0x13, 0x1e, 0xf4, 0x79, 0xd4, 0x7f, 0xea, 0x89, 0x17,
0xce, 0xa2, 0x4a, 0x77, 0x92, 0x2b, 0x6d, 0xb7, 0x78, 0x2c, 0x9c, 0x26, 0x7a, 0xc4, 0x6f, 0xb2,
0x06, 0xb5, 0x2d, 0x2f, 0xe9, 0x8a, 0x61, 0x72, 0xe4, 0x2c, 0xb5, 0xad, 0x4e, 0x99, 0x65, 0x34,
0x59, 0x81, 0xca, 0xbe, 0xcb, 0x7d, 0xe1, 0x5c, 0x40, 0x03, 0x45, 0x10, 0x0a, 0x8b, 0xf7, 0xc2,
0x48, 0x78, 0xcf, 0x02, 0x6c, 0x82, 0xd3, 0xc2, 0xa4, 0x26, 0x78, 0xe4, 0x3d, 0xb0, 0x65, 0x4a,
0xcb, 0x6d, 0xab, 0xd3, 0xb8, 0xb5, 0xbc, 0x61, 0xfa, 0xb8, 0xd1, 0x15, 0xae, 0x37, 0xe0, 0x3e,
0x93, 0x52, 0x54, 0xe2, 0x23, 0x87, 0x9c, 0xad, 0xc4, 0x47, 0x94, 0xc2, 0x52, 0x6f, 0x30, 0x0c,
0xa3, 0x84, 0x89, 0x78, 0x18, 0x06, 0xb1, 0x20, 0x2d, 0xb0, 0x77, 0xa2, 0xc8, 0xb1, 0x30, 0xac,
0xfc, 0xa4, 0xdf, 0x40, 0x6b, 0xcb, 0x0f, 0xdd, 0xe3, 0x2e, 0x4f, 0x38, 0x13, 0xcf, 0x53, 0x11,
0x27, 0x12, 0xbb, 0x82, 0xa7, 0xf4, 0x14, 0x21, 0xb9, 0xd8, 0x6f, 0xa7, 0xa4, 0xb8, 0x48, 0xc8,
0xba, 0x60, 0xd5, 0x54, 0x7b, 0xf0, 0x1b, 0x73, 0x3f, 0xe2, 0x51, 0x1f, 0x7b, 0x5a, 0x66, 0x8a,
0x90, 0x5c, 0x8c, 0x84, 0x73, 0x50, 0x66, 0x8a, 0xa0, 0x3d, 0x58, 0x2e, 0xc4, 0xd7, 0x30, 0x57,
0x61, 0x81, 0x85, 0x2f, 0x7a, 0xdd, 0xd8, 0xb1, 0xda, 0x76, 0xa7, 0xcc, 0x34, 0x85, 0x03, 0x13,
0xfa, 0xe9, 0x20, 0x90, 0xa2, 0x12, 0x8a, 0x72, 0x06, 0xbd, 0x02, 0x15, 0x9c, 0x1e, 0x99, 0x65,
0x6e, 0x2b, 0x3f, 0xe9, 0xb7, 0x16, 0xd4, 0x77, 0xf9, 0x08, 0x81, 0xc4, 0xe4, 0x0e, 0xd4, 0x4c,
0x6f, 0x51, 0xa9, 0x71, 0xeb, 0xdd, 0xbc, 0x82, 0x99, 0xda, 0x86, 0xd1, 0xd9, 0x09, 0x92, 0x68,
0xcc, 0x32, 0x93, 0xb5, 0xcf, 0xa1, 0x39, 0x21, 0x92, 0xf1, 0x8e, 0xc5, 0xd8, 0x54, 0xf5, 0x58,
0x8c, 0x65, 0xae, 0x27, 0xdc, 0x4f, 0x05, 0xd6, 0xaa, 0xcc, 0x14, 0xf1, 0x59, 0xe9, 0x53, 0x8b,
0x3e, 0x05, 0xb2, 0x1d, 0x09, 0x9e, 0x08, 0x0c, 0xb2, 0x2b, 0xe2, 0x98, 0x3f, 0x13, 0xf3, 0x2a,
0x6e, 0x17, 0x2b, 0x9e, 0x55, 0xb7, 0x54, 0xa8, 0x2e, 0xbd, 0x0e, 0xa4, 0x2b, 0x7c, 0x91, 0x08,
0x7d, 0xba, 0x5f, 0xe3, 0x97, 0x3e, 0x37, 0x18, 0xe6, 0xeb, 0x92, 0x6b, 0x50, 0x96, 0xab, 0x02,
0x83, 0x35, 0x6e, 0x5d, 0xcc, 0xeb, 0x94, 0x6d, 0x11, 0x86, 0x0a, 0xd8, 0x1b, 0x74, 0xda, 0xdf,
0x4c, 0x10, 0xb0, 0xcd, 0x72, 0x06, 0xfd, 0xde, 0x32, 0x31, 0x31, 0x89, 0x73, 0xe6, 0x3d, 0x31,
0x69, 0xd7, 0x35, 0x12, 0x1b, 0x91, 0xac, 0xe6, 0x48, 0x8a, 0x5b, 0x68, 0x16, 0x98, 0xf2, 0x34,
0x98, 0xbb, 0xa6, 0x56, 0x6f, 0x8a, 0x85, 0xba, 0xf0, 0x7f, 0xe5, 0x61, 0xf3, 0x84, 0x7b, 0x3e,
0x3f, 0xf4, 0xff, 0x55, 0x3b, 0x27, 0xd2, 0x72, 0xa0, 0x8a, 0xb6, 0xbd, 0xae, 0x3e, 0x18, 0x86,
0xa4, 0x5f, 0x43, 0x7e, 0xc6, 0x1e, 0xf2, 0x81, 0xd0, 0xde, 0xf0, 0x3b, 0xab, 0x46, 0xe9, 0x1c,
0xd5, 0x58, 0x81, 0x8a, 0x3c, 0x97, 0x72, 0xcf, 0xdb, 0x32, 0x30, 0x12, 0x73, 0x6a, 0x74, 0x1b,
0x16, 0xf6, 0xdd, 0x23, 0x31, 0xe0, 0xe4, 0x43, 0xa8, 0x22, 0x7e, 0x11, 0xeb, 0xc3, 0x72, 0x61,
0x6a, 0x08, 0x98, 0x91, 0xd3, 0x1f, 0x2d, 0x9d, 0xf8, 0x4c, 0xc8, 0x13, 0x01, 0x4b, 0x53, 0x01,
0xc9, 0x0d, 0xa8, 0x6a, 0xd4, 0xb8, 0x4b, 0xce, 0x98, 0x35, 0xa3, 0x43, 0xae, 0xc1, 0x02, 0x66,
0x1a, 0x3b, 0xe5, 0x69, 0x50, 0xc8, 0x67, 0x5a, 0x4c, 0x77, 0xc0, 0x7e, 0xc2, 0x7a, 0x72, 0xa5,
0x60, 0x3e, 0x06, 0x92, 0xa6, 0x24, 0xd0, 0x2f, 0xc2, 0x38, 0xd1, 0x3d, 0xc1, 0x6f, 0xc9, 0xdb,
0x0b, 0x23, 0x35, 0xc5, 0x4d, 0x86, 0xdf, 0xf4, 0x67, 0x0b, 0xca, 0x0f, 0xc3, 0xbe, 0x20, 0x4b,
0x50, 0xea, 0x75, 0xb5, 0x93, 0x52, 0xaf, 0x4b, 0xde, 0x41, 0xff, 0xba, 0x0f, 0xcd, 0x1c, 0xc5,
0x13, 0xd6, 0x63, 0x18, 0xf9, 0x2a, 0xd4, 0x7b, 0xf1, 0x5e, 0xe4, 0x0d, 0x78, 0x34, 0xd6, 0x37,
0x6d, 0xce, 0xc0, 0xd3, 0x9c, 0xf0, 0x44, 0xdd, 0x7f, 0x75, 0xa6, 0x08, 0x72, 0x0d, 0xaa, 0xf7,
0xd9, 0xde, 0xb6, 0x74, 0x5c, 0x99, 0xe5, 0xd8, 0x48, 0xe9, 0x5d, 0x68, 0x49, 0x54, 0x68, 0x65,
0xa6, 0x6f, 0x15, 0x16, 0x24, 0x2f, 0x43, 0xa9, 0xa9, 0x3c, 0x54, 0xa9, 0x10, 0x8a, 0x3e, 0x50,
0x1e, 0x76, 0x4e, 0x44, 0x90, 0x14, 0xe6, 0x17, 0x69, 0x74, 0xd0, 0x64, 0x8a, 0x20, 0x54, 0x55,
0x40, 0xa7, 0xba, 0x94, 0x23, 0x92, 0x5c, 0x86, 0x32, 0xfa, 0x83, 0x05, 0x60, 0x00, 0xa5, 0x71,
0x66, 0x62, 0x9d, 0x6d, 0x42, 0x3a, 0x66, 0xd2, 0xf4, 0xc9, 0x6e, 0xe5, 0x5a, 0x8a, 0xcf, 0xcc,
0x24, 0x7e, 0x94, 0x4f, 0xa2, 0x6a, 0xfa, 0xa5, 0xa9, 0x11, 0x51, 0x51, 0xf3, 0x79, 0x0c, 0xa0,
0x51, 0xe0, 0xcf, 0x1c, 0xca, 0x1b, 0xd9, 0x1c, 0x95, 0xa6, 0x5d, 0x22, 0x5f, 0xbb, 0xd4, 0x4a,
0x73, 0xb6, 0x9c, 0x07, 0x8d, 0x82, 0xd1, 0xcc, 0x78, 0x1d, 0xb8, 0x30, 0xb9, 0x33, 0xcc, 0x45,
0x36, 0xcd, 0x9e, 0x13, 0xea, 0x27, 0x0b, 0x9a, 0xdb, 0x7e, 0x1a, 0x27, 0x22, 0xd2, 0xd1, 0xa4,
0xbe, 0x62, 0x64, 0x9d, 0xcf, 0x19, 0xb3, 0x9b, 0x4f, 0xde, 0x87, 0x8a, 0xec, 0x81, 0xda, 0x0c,
0xa7, 0x1b, 0xa4, 0x84, 0x85, 0x0e, 0x95, 0x5f, 0xdf, 0x21, 0xfa, 0x14, 0x6a, 0x5b, 0xfb, 0xbd,
0xfb, 0x51, 0x98, 0x0e, 0x67, 0x66, 0x6f, 0xde, 0x88, 0xa5, 0xc2, 0x1b, 0xb1, 0xa5, 0xde, 0x3b,
0x2a, 0x43, 0x7c, 0xdc, 0xb4, 0xd4, 0xe3, 0xa6, 0xac, 0x39, 0x7c, 0x44, 0xf7, 0x61, 0x59, 0xa5,
0x2e, 0x57, 0xd7, 0x9b, 0x6c, 0x59, 0xf3, 0x4c, 0xb1, 0xf3, 0x67, 0x8a, 0x74, 0xaa, 0x96, 0xf8,
0x7f, 0xe9, 0xf4, 0xef, 0x12, 0x2c, 0x33, 0x11, 0x7b, 0x2f, 0x45, 0x2f, 0x88, 0x93, 0x28, 0x75,
0xe5, 0xba, 0x92, 0xf6, 0x5f, 0x86, 0x87, 0xba, 0x2f, 0x36, 0x53, 0xc4, 0x79, 0x0e, 0x14, 0xe9,
0x40, 0xb5, 0xb8, 0x3b, 0x4e, 0xab, 0x19, 0x31, 0xb9, 0x09, 0xd5, 0xfd, 0x30, 0x8d, 0xdc, 0xec,
0x74, 0x14, 0x2e, 0x05, 0x85, 0x48, 0x89, 0x99, 0x51, 0x23, 0x8f, 0x81, 0x1c, 0x44, 0x3c, 0x88,
0x7d, 0x2e, 0x41, 0x1a, 0xe3, 0xda, 0xf4, 0x8b, 0xa8, 0xa0, 0x33, 0xe1, 0x67, 0x86, 0x31, 0xf9,
0xb8, 0x78, 0xfc, 0x9d, 0x2a, 0x22, 0x5e, 0x99, 0x44, 0xac, 0x4f, 0x54, 0x71, 0x4d, 0xdc, 0x99,
0x9a, 0x65, 0x67, 0x01, 0x0d, 0x2f, 0xe7, 0x86, 0x13, 0x62, 0x36, 0xa9, 0x4d, 0xbf, 0xb3, 0x60,
0xb1, 0x88, 0xec, 0x5c, 0x6b, 0x27, 0x6b, 0x74, 0x69, 0xfe, 0x93, 0xcb, 0x34, 0xba, 0x3c, 0xeb,
0x91, 0x5b, 0x29, 0x3e, 0xc3, 0x52, 0xb8, 0x7c, 0x46, 0xb9, 0xde, 0x02, 0x54, 0x1b, 0x1a, 0x7b,
0x3c, 0x4a, 0x3c, 0xe9, 0x52, 0x3f, 0x13, 0x2a, 0xac, 0xc8, 0xa2, 0xc7, 0x70, 0xe5, 0xd4, 0xd0,
0x6d, 0x87, 0x83, 0xa1, 0x9c, 0xee, 0xb7, 0x18, 0x3e, 0x79, 0x0f, 0x44, 0x51, 0x18, 0x99, 0x6a,
0x20, 0x41, 0xb7, 0xa0, 0x76, 0x10, 0x0e, 0x43, 0x3f, 0x7c, 0x36, 0x9e, 0xb3, 0x74, 0x1c, 0xa8,
0xaa, 0xbb, 0x47, 0x2d, 0xb9, 0x3a, 0x33, 0x24, 0xbd, 0x28, 0x4f, 0x89, 0xcb, 0x7d, 0x37, 0xf5,
0x79, 0x22, 0xf0, 0xd9, 0x8e, 0xcc, 0x07, 0x21, 0xef, 0xab, 0x5d, 0xa2, 0x0f, 0x24, 0x15, 0x7a,
0x48, 0x39, 0x26, 0x55, 0xb8, 0xe3, 0x36, 0x91, 0x61, 0xee, 0x38, 0x45, 0x91, 0x4f, 0xa0, 0x51,
0xd0, 0xd6, 0xc9, 0x5d, 0x9a, 0x9a, 0x65, 0x25, 0x64, 0x45, 0x4d, 0xfa, 0x9b, 0x35, 0x61, 0x79,
0xea, 0x9a, 0xd7, 0x01, 0x4f, 0x54, 0xc1, 0x6a, 0x4c, 0x53, 0xb2, 0x00, 0x3b, 0x23, 0xd7, 0x4f,
0x63, 0x29, 0xd2, 0xb7, 0x7b, 0xc6, 0x90, 0x05, 0x90, 0x3f, 0x58, 0xc3, 0xd4, 0xbc, 0xb0, 0x0c,
0x29, 0x7f, 0x3b, 0x76, 0x05, 0xef, 0xfb, 0x5e, 0x20, 0x70, 0x82, 0x6c, 0x96, 0xd1, 0xe4, 0xa6,
0xda, 0xd5, 0xe6, 0x18, 0xac, 0xcd, 0x84, 0x8f, 0x1a, 0x6a, 0x8f, 0xc7, 0x94, 0x40, 0x6b, 0x5a,
0x44, 0x57, 0x80, 0xa8, 0x99, 0xd8, 0x3c, 0x0c, 0x23, 0x73, 0xb5, 0xd3, 0x6d, 0xb3, 0x9e, 0x64,
0x27, 0xe6, 0xbd, 0x18, 0xf2, 0x2a, 0x97, 0x8a, 0x55, 0xde, 0x6a, 0xfd, 0xfe, 0x6a, 0xdd, 0xfa,
0xe3, 0xd5, 0xba, 0xf5, 0xe7, 0xab, 0x75, 0xeb, 0x97, 0xbf, 0xd6, 0xff, 0x77, 0xb8, 0x80, 0xff,
0x2e, 0xdc, 0xfe, 0x27, 0x00, 0x00, 0xff, 0xff, 0x17, 0x40, 0x19, 0xfb, 0x86, 0x10, 0x00, 0x00,
}
func (m *IndexMeta) Marshal() (dAtA []byte, err error) {
@ -4379,6 +4419,33 @@ func (m *RecalculateCaches) MarshalToSizedBuffer(dAtA []byte) (int, error) {
return len(dAtA) - i, nil
}
func (m *LoadSchemaMessage) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *LoadSchemaMessage) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *LoadSchemaMessage) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
return len(dAtA) - i, nil
}
func (m *TransactionMessage) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
@ -5432,6 +5499,18 @@ func (m *RecalculateCaches) Size() (n int) {
return n
}
func (m *LoadSchemaMessage) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
func (m *TransactionMessage) Size() (n int) {
if m == nil {
return 0
@ -10683,6 +10762,60 @@ func (m *RecalculateCaches) Unmarshal(dAtA []byte) error {
}
return nil
}
func (m *LoadSchemaMessage) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPrivate
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: LoadSchemaMessage: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: LoadSchemaMessage: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
default:
iNdEx = preIndex
skippy, err := skipPrivate(dAtA[iNdEx:])
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPrivate
}
if (iNdEx + skippy) < 0 {
return ErrInvalidLengthPrivate
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func (m *TransactionMessage) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0

2
internal/private.proto
View file

@ -208,6 +208,8 @@ message Topology {
message RecalculateCaches {}
message LoadSchemaMessage {}
message TransactionMessage {
string Action = 1;
Transaction Transaction = 2;

2
pilosa.go
View file

@ -53,6 +53,8 @@ var (
ErrInvalidBetweenValue = errors.New("invalid value for between operation")
ErrDecimalOutOfRange = errors.New("decimal value out of range")
ErrViewRequired = errors.New("view required")
ErrViewExists = errors.New("view already exists")
ErrInvalidView = errors.New("invalid view")
ErrInvalidCacheType = errors.New("invalid cache type")

46
server.go
View file

@ -415,12 +415,14 @@ func NewServer(opts ...ServerOption) (*Server, error) {
metricInterval: 0,
diagnosticInterval: 0,
disCo: disco.NopDisCo,
stator: disco.NopStator,
metadator: disco.NopMetadator,
resizer: disco.NopResizer,
noder: topology.NewEmptyLocalNoder(),
sharder: disco.NopSharder,
disCo: disco.NopDisCo,
stator: disco.NopStator,
metadator: disco.NopMetadator,
resizer: disco.NopResizer,
noder: topology.NewEmptyLocalNoder(),
sharder: disco.NopSharder,
schemator: disco.NopSchemator,
serializer: NopSerializer,
confirmDownRetries: defaultConfirmDownRetries,
confirmDownSleep: defaultConfirmDownSleep,
@ -488,6 +490,8 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.cluster.confirmDownRetries = s.confirmDownRetries
s.cluster.confirmDownSleep = s.confirmDownSleep
s.holder.broadcaster = s
s.holder.schemator = s.schemator
s.holder.serializer = s.serializer
return s, nil
}
@ -778,14 +782,9 @@ func (s *Server) receiveMessage(m Message) error {
}
case *CreateIndexMessage:
opt := obj.Meta
idx, err := s.holder.CreateIndex(obj.Index, *opt)
if err != nil {
if _, err := s.holder.LoadIndex(obj.Index); err != nil {
return err
}
idx.mu.Lock()
idx.createdAt = obj.CreatedAt
idx.mu.Unlock()
case *DeleteIndexMessage:
if err := s.holder.DeleteIndex(obj.Index); err != nil {
@ -793,18 +792,9 @@ func (s *Server) receiveMessage(m Message) error {
}
case *CreateFieldMessage:
idx := s.holder.Index(obj.Index)
if idx == nil {
return fmt.Errorf("local index not found: %s", obj.Index)
}
opt := obj.Meta
fld, err := idx.createFieldIfNotExists(obj.Field, opt)
if err != nil {
if _, err := s.holder.LoadField(obj.Index, obj.Field); err != nil {
return err
}
fld.mu.Lock()
fld.createdAt = obj.CreatedAt
fld.mu.Unlock()
case *DeleteFieldMessage:
idx := s.holder.Index(obj.Index)
@ -819,11 +809,7 @@ func (s *Server) receiveMessage(m Message) error {
}
case *CreateViewMessage:
f := s.holder.Field(obj.Index, obj.Field)
if f == nil {
return fmt.Errorf("local field not found: %s", obj.Field)
}
if _, _, err := f.createViewIfNotExistsBase(obj.View); err != nil {
if _, err := s.holder.LoadView(obj.Index, obj.Field, obj.View); err != nil {
return err
}
@ -868,6 +854,12 @@ func (s *Server) receiveMessage(m Message) error {
case *RecalculateCaches:
s.holder.recalculateCaches()
case *LoadSchemaMessage:
err := s.holder.LoadSchema()
if err != nil {
return errors.Wrapf(err, "handling load schema message: %v", obj)
}
case *NodeStatus:
s.handleRemoteStatus(obj)

155
server/cluster_test.go
View file

@ -211,7 +211,10 @@ func TestClusterResize_AddNode(t *testing.T) {
t.Run("ContinuousShards", func(t *testing.T) {
// Configure node0
m0 := test.MustRunCluster(t, 1).GetNode(0)
c := test.MustRunCluster(t, 2)
defer c.Close()
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
@ -240,20 +243,7 @@ func TestClusterResize_AddNode(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := test.NewCommandNode(t)
if err := port.GetListeners(func(lsns []*net.TCPListener) error {
portsCfg := test.GenPortsConfig(test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
return m1.Start()
}, 3, 10); err != nil {
t.Fatalf("starting second main: %v", err)
}
m1 := c.GetNode(1)
defer m1.Close()
state0, err0 := m0.API.State()
@ -268,9 +258,14 @@ func TestClusterResize_AddNode(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
t.Run("OneShard", func(t *testing.T) {
// Configure node0
m0 := test.MustRunCluster(t, 1).GetNode(0)
c := test.MustRunCluster(t, 2)
defer c.Close()
// Configure node0
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
@ -296,20 +291,7 @@ func TestClusterResize_AddNode(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := test.NewCommandNode(t)
if err := port.GetListeners(func(lsns []*net.TCPListener) error {
portsCfg := test.GenPortsConfig(test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
return m1.Start()
}, 3, 10); err != nil {
t.Fatalf("starting second main: %v", err)
}
m1 := c.GetNode(1)
defer m1.Close()
state0, err0 := m0.API.State()
@ -324,11 +306,14 @@ func TestClusterResize_AddNode(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
t.Run("SkippedShard", func(t *testing.T) {
// same reason as the ContinuousShards test above.
c := test.MustRunCluster(t, 2)
defer c.Close()
// Configure node0
m0 := test.MustRunCluster(t, 1).GetNode(0)
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
@ -358,21 +343,7 @@ func TestClusterResize_AddNode(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := test.NewCommandNode(t)
if err := port.GetListeners(func(lsns []*net.TCPListener) error {
portsCfg := test.GenPortsConfig(test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
return m1.Start()
}, 3, 10); err != nil {
t.Fatalf("starting second main: %v", err)
}
m1 := c.GetNode(1)
defer m1.Close()
state0, err0 := m0.API.State()
@ -394,8 +365,11 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
skipTestUnderBlueGreenWithRoaring(t)
t.Run("WithIndex", func(t *testing.T) {
c := test.MustRunCluster(t, 2)
defer c.Close()
// Configure node0
m0 := test.MustRunCluster(t, 1).GetNode(0)
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
@ -415,18 +389,7 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
}()
// Configure node1
m1 := test.NewCommandNode(t)
if err := port.GetListeners(func(lsns []*net.TCPListener) error {
portsCfg := test.GenPortsConfig(test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
return m1.Start()
}, 3, 10); err != nil {
t.Fatalf("starting second main: %v", err)
}
m1 := c.GetNode(1)
defer m1.Close()
state0, err0 := m0.API.State()
@ -441,9 +404,13 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
t.Fatalf("error from index creation: %v", err)
}
})
t.Run("ContinuousShards", func(t *testing.T) {
c := test.MustRunCluster(t, 2)
defer c.Close()
// Configure node0
m0 := test.MustRunCluster(t, 1).GetNode(0)
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
@ -473,23 +440,7 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := test.NewCommandNode(t)
if err := port.GetListeners(func(lsns []*net.TCPListener) error {
portsCfg := test.GenPortsConfig(test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
return m1.Start()
}, 3, 10); err != nil {
t.Fatalf("starting second main: %v", err)
}
errc := make(chan error, 1)
go func() {
_, err := m0.API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{})
errc <- err
}()
m1 := c.GetNode(1)
defer m1.Close()
state0, err0 := m0.API.State()
@ -504,10 +455,13 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
t.Run("SkippedShard", func(t *testing.T) {
c := test.MustRunCluster(t, 2)
defer c.Close()
// Configure node0
m0 := test.MustRunCluster(t, 1).GetNode(0)
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
@ -537,24 +491,7 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := test.NewCommandNode(t)
if err := port.GetListeners(func(lsns []*net.TCPListener) error {
portsCfg := test.GenPortsConfig(test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
errc := make(chan error, 1)
go func() {
_, err := m0.API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{})
errc <- err
}()
return m1.Start()
}, 3, 10); err != nil {
t.Fatalf("starting second main: %v", err)
}
m1 := c.GetNode(1)
defer m1.Close()
state0, err0 := m0.API.State()
@ -569,9 +506,13 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
m1.QueryExpect(t, "i", "", `Row(f=1)`, exp)
})
t.Run("WithIndexKeys", func(t *testing.T) {
c := test.MustRunCluster(t, 2)
defer c.Close()
// Configure node0
m0 := test.MustRunCluster(t, 1).GetNode(0)
m0 := c.GetNode(0)
defer m0.Close()
// Create a client for each node.
@ -599,24 +540,8 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
m0.QueryExpect(t, "i", "", `Row(f=1)`, exp)
// Configure node1
m1 := test.NewCommandNode(t)
if err := port.GetListeners(func(lsns []*net.TCPListener) error {
portsCfg := test.GenPortsConfig(test.NewPorts(lsns))
m1.Config.Etcd = portsCfg[0].Etcd
m1.Config.Name = portsCfg[0].Name
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
m1.Config.BindGRPC = portsCfg[0].BindGRPC
errc := make(chan error, 1)
go func() {
_, err := m0.API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{})
errc <- err
}()
return m1.Start()
}, 3, 10); err != nil {
t.Fatalf("starting second main: %v", err)
}
m1 := c.GetNode(1)
defer m1.Close()
state0, err0 := m0.API.State()
state1, err1 := m1.API.State()

8
server/grpc.go
View file

@ -316,7 +316,7 @@ func (h *GRPCHandler) CreateIndex(ctx context.Context, req *pb.CreateIndexReques
// GetIndex returns a single Index given a name
func (h *GRPCHandler) GetIndex(ctx context.Context, req *pb.GetIndexRequest) (*pb.GetIndexResponse, error) {
schema, err := h.api.Schema(ctx)
schema, err := h.api.Schema(ctx, false)
if err != nil {
return nil, errToStatusError(err)
}
@ -331,7 +331,7 @@ func (h *GRPCHandler) GetIndex(ctx context.Context, req *pb.GetIndexRequest) (*p
// GetIndexes returns a list of all Indexes
func (h *GRPCHandler) GetIndexes(ctx context.Context, req *pb.GetIndexesRequest) (*pb.GetIndexesResponse, error) {
schema, err := h.api.Schema(ctx)
schema, err := h.api.Schema(ctx, false)
if err != nil {
return nil, errToStatusError(err)
}
@ -381,7 +381,7 @@ func (h *VDSMGRPCHandler) GetVDS(ctx context.Context, req *vdsm_pb.GetVDSRequest
case *vdsm_pb.GetVDSRequest_Id:
return nil, status.Error(codes.InvalidArgument, "VDS IDs are no longer supported")
case *vdsm_pb.GetVDSRequest_Name:
schema, err := h.api.Schema(ctx)
schema, err := h.api.Schema(ctx, false)
if err != nil {
return nil, errToStatusError(err)
}
@ -399,7 +399,7 @@ func (h *VDSMGRPCHandler) GetVDS(ctx context.Context, req *vdsm_pb.GetVDSRequest
// GetVDSs returns a list of all VDSs
func (h *VDSMGRPCHandler) GetVDSs(ctx context.Context, req *vdsm_pb.GetVDSsRequest) (*vdsm_pb.GetVDSsResponse, error) {
schema, err := h.api.Schema(ctx)
schema, err := h.api.Schema(ctx, false)
if err != nil {
return nil, errToStatusError(err)
}

10
server/grpc_test.go
View file

@ -1035,7 +1035,7 @@ func TestCRUDIndexes(t *testing.T) {
t.Fatal(err)
}
schema, err := m.API.Schema(ctx)
schema, err := m.API.Schema(ctx, false)
if err != nil {
t.Fatal("Getting schema error", err)
}
@ -1058,7 +1058,7 @@ func TestCRUDIndexes(t *testing.T) {
t.Fatal(err)
}
schema, err = m.API.Schema(ctx)
schema, err = m.API.Schema(ctx, false)
if err != nil {
t.Fatal("Getting schema error", err)
}
@ -1069,7 +1069,7 @@ func TestCRUDIndexes(t *testing.T) {
_ = m.API.DeleteIndex(ctx, "testindex1")
schema, err = m.API.Schema(ctx)
schema, err = m.API.Schema(ctx, false)
if err != nil {
t.Fatal("Getting schema error", err)
}
@ -1094,7 +1094,7 @@ func TestCRUDIndexes(t *testing.T) {
// Check errors for CreateIndex: create index with no name
_, err = gh.CreateIndex(ctx, &pb.CreateIndexRequest{Name: ""})
errStatus, _ = status.FromError(err)
if errStatus.Code() != codes.Unknown {
if errStatus.Code() != codes.FailedPrecondition {
t.Fatalf("Error code should be codes.Unknown, but is %v", errStatus.Code())
}
@ -1183,7 +1183,7 @@ func TestCRUDIndexes(t *testing.T) {
t.Fatal(err)
}
schema, err := m.API.Schema(ctx)
schema, err := m.API.Schema(ctx, false)
if err != nil {
t.Fatal("Getting schema error", err)
}

2
server/handler_test.go
View file

@ -226,7 +226,7 @@ func TestHandler_Endpoints(t *testing.T) {
})
t.Run("Import", func(t *testing.T) {
indexInfo, err := cmd.API.Schema(context.Background())
indexInfo, err := cmd.API.Schema(context.Background(), false)
if err != nil {
t.Fatalf("getting schema: %v", err)
}

2
server/server_test.go
View file

@ -1250,7 +1250,7 @@ func TestClusterCreatedAtRace(t *testing.T) {
schemas := make([]*pilosa.IndexInfo, len(cluster.Nodes))
for i, cmd := range cluster.Nodes {
s, err := cmd.API.Schema(context.Background())
s, err := cmd.API.Schema(context.Background(), false)
if err != nil {
t.Fatalf("getting schema: %v", err)
}

2
sql/show.go
View file

@ -54,7 +54,7 @@ func (s *ShowHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.ToR
}
func (s *ShowHandler) execShowTables(ctx context.Context, showStmt *sqlparser.Show) (pproto.ToRowser, error) {
indexInfo, err := s.api.Schema(ctx)
indexInfo, err := s.api.Schema(ctx, false)
if err != nil {
return nil, errors.Wrap(err, "getting schema")
}

8
view_internal_test.go
View file

@ -37,7 +37,13 @@ func mustOpenView(tb testing.TB, index, field, name string) *view {
h := NewHolder(path, nil)
// h needs an *Index so we can call h.Index() and get Index.Txf, in TestView_DeleteFragment
idx, err := h.createIndex(index, IndexOptions{})
cim := &CreateIndexMessage{
Index: index,
CreatedAt: 0,
Meta: &IndexOptions{},
}
idx, err := h.createIndex(cim, false)
testhook.Cleanup(tb, func() {
h.Close()
})