diff --git a/api.go b/api.go index 2ce6d2607..929611681 100644 --- a/api.go +++ b/api.go @@ -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() diff --git a/api_test.go b/api_test.go index 936e7b8c4..c512131f3 100644 --- a/api_test.go +++ b/api_test.go @@ -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) } diff --git a/broadcast.go b/broadcast.go index 915108a56..141c43e36 100644 --- a/broadcast.go +++ b/broadcast.go @@ -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: diff --git a/cluster.go b/cluster.go index 3b46a3bb8..222e5c4c4 100644 --- a/cluster.go +++ b/cluster.go @@ -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. diff --git a/cmd/random-query/main.go b/cmd/random-query/main.go index c271f3e26..ead3ccd9d 100644 --- a/cmd/random-query/main.go +++ b/cmd/random-query/main.go @@ -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) { diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index 509333f5b..00241d756 100644 --- a/dbshard_internal_test.go +++ b/dbshard_internal_test.go @@ -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() diff --git a/disco/disco.go b/disco/disco.go index 03443d384..0725d2d82 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -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 } diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 7785a6f26..7226dce7f 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -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 diff --git a/etcd/embed.go b/etcd/embed.go index e671442cd..b50f74a1d 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -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() diff --git a/executor_test.go b/executor_test.go index 9f1d7728b..11b15fcbe 100644 --- a/executor_test.go +++ b/executor_test.go @@ -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) diff --git a/field.go b/field.go index e5936503d..37a38a249 100644 --- a/field.go +++ b/field.go @@ -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 +} diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 093b5e7aa..9b784cf0b 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -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) diff --git a/holder.go b/holder.go index 607707968..f3c7999e6 100644 --- a/holder.go +++ b/holder.go @@ -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 +} diff --git a/http/handler.go b/http/handler.go index 77fccd7e5..2865a6807 100644 --- a/http/handler.go +++ b/http/handler.go @@ -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) } diff --git a/index.go b/index.go index c81419c46..83d5d8c44 100644 --- a/index.go +++ b/index.go @@ -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() } diff --git a/internal/private.pb.go b/internal/private.pb.go index 08fe39847..742211a43 100644 --- a/internal/private.pb.go +++ b/internal/private.pb.go @@ -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 diff --git a/internal/private.proto b/internal/private.proto index 7a29abbe2..8cc195fdf 100644 --- a/internal/private.proto +++ b/internal/private.proto @@ -208,6 +208,8 @@ message Topology { message RecalculateCaches {} +message LoadSchemaMessage {} + message TransactionMessage { string Action = 1; Transaction Transaction = 2; diff --git a/pilosa.go b/pilosa.go index c98f886e0..dbd4e6567 100644 --- a/pilosa.go +++ b/pilosa.go @@ -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") diff --git a/server.go b/server.go index bbcc453e1..2b455085c 100644 --- a/server.go +++ b/server.go @@ -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) diff --git a/server/cluster_test.go b/server/cluster_test.go index b5047ad72..0239cbbc8 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -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() diff --git a/server/grpc.go b/server/grpc.go index 40bcc07d0..cdfe8108d 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -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) } diff --git a/server/grpc_test.go b/server/grpc_test.go index d24c2d6e4..683881ba7 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -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) } diff --git a/server/handler_test.go b/server/handler_test.go index 346ddc1b1..a14a04952 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -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) } diff --git a/server/server_test.go b/server/server_test.go index 04c2da30c..4fcc7f253 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -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) } diff --git a/sql/show.go b/sql/show.go index c574ae4a5..0bac67d46 100644 --- a/sql/show.go +++ b/sql/show.go @@ -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") } diff --git a/view_internal_test.go b/view_internal_test.go index 2b62ed04b..b895122f4 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -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() })