diff --git a/api.go b/api.go index 63d249c16..6de319dbc 100644 --- a/api.go +++ b/api.go @@ -692,6 +692,30 @@ func (api *API) FragmentData(ctx context.Context, indexName, fieldName, viewName return f, nil } +// TranslateData returns all translation data in the specified partition. +func (api *API) TranslateData(ctx context.Context, indexName string, partition int) (io.WriterTo, error) { + span, _ := tracing.StartSpanFromContext(ctx, "API.TranslateData") + defer span.Finish() + + if err := api.validate(apiTranslateData); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + + // Retrieve index from holder. + idx := api.holder.Index(indexName) + if idx == nil { + return nil, ErrIndexNotFound + } + + // Retrieve translatestore from holder. + store := idx.TranslateStore(partition) + if store == nil { + return nil, ErrTranslateStoreNotFound + } + + return store, nil +} + // Hosts returns a list of the hosts in the cluster including their ID, // URL, and which is the coordinator. func (api *API) Hosts(ctx context.Context) []*Node { @@ -1586,6 +1610,7 @@ const ( apiFragmentBlockData apiFragmentBlocks apiFragmentData + apiTranslateData apiField apiFieldAttrDiff //apiHosts // not implemented @@ -1616,8 +1641,9 @@ var methodsCommon = map[apiMethod]struct{}{ } var methodsResizing = map[apiMethod]struct{}{ - apiFragmentData: {}, - apiResizeAbort: {}, + apiFragmentData: {}, + apiTranslateData: {}, + apiResizeAbort: {}, } var methodsNormal = map[apiMethod]struct{}{ diff --git a/apimethod_string.go b/apimethod_string.go index 309648a62..d8217b035 100644 --- a/apimethod_string.go +++ b/apimethod_string.go @@ -19,25 +19,26 @@ func _() { _ = x[apiFragmentBlockData-8] _ = x[apiFragmentBlocks-9] _ = x[apiFragmentData-10] - _ = x[apiField-11] - _ = x[apiFieldAttrDiff-12] - _ = x[apiImport-13] - _ = x[apiImportValue-14] - _ = x[apiIndex-15] - _ = x[apiIndexAttrDiff-16] - _ = x[apiQuery-17] - _ = x[apiRecalculateCaches-18] - _ = x[apiRemoveNode-19] - _ = x[apiResizeAbort-20] - _ = x[apiSetCoordinator-21] - _ = x[apiShardNodes-22] - _ = x[apiViews-23] - _ = x[apiApplySchema-24] + _ = x[apiTranslateData-11] + _ = x[apiField-12] + _ = x[apiFieldAttrDiff-13] + _ = x[apiImport-14] + _ = x[apiImportValue-15] + _ = x[apiIndex-16] + _ = x[apiIndexAttrDiff-17] + _ = x[apiQuery-18] + _ = x[apiRecalculateCaches-19] + _ = x[apiRemoveNode-20] + _ = x[apiResizeAbort-21] + _ = x[apiSetCoordinator-22] + _ = x[apiShardNodes-23] + _ = x[apiViews-24] + _ = x[apiApplySchema-25] } -const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiViewsapiApplySchema" +const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiViewsapiApplySchema" -var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 181, 197, 206, 220, 228, 244, 252, 272, 285, 299, 316, 329, 337, 351} +var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 197, 213, 222, 236, 244, 260, 268, 288, 301, 315, 332, 345, 353, 367} func (i apiMethod) String() string { if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) { diff --git a/boltdb/translate.go b/boltdb/translate.go index a607d494f..cb8088674 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -17,6 +17,7 @@ package boltdb import ( "bytes" "context" + "io" "os" "path/filepath" "sync" @@ -33,6 +34,11 @@ var ( ErrTranslateStoreClosed = errors.New("boltdb: translate store closing") ) +const ( + // snapshotExt is the file extension used for an in-process snapshot. + snapshotExt = ".snapshotting" +) + // OpenTranslateStore opens and initializes a boltdb translation store. func OpenTranslateStore(path, index, field string, partitionID, partitionN int) (pilosa.TranslateStore, error) { s := NewTranslateStore(index, field, partitionID, partitionN) @@ -343,6 +349,53 @@ func (s *TranslateStore) MaxID() (max uint64, err error) { return max, nil } +// WriteTo writes the contents of the store to the writer. +func (s *TranslateStore) WriteTo(w io.Writer) (int64, error) { + tx, err := s.db.Begin(false) + if err != nil { + return 0, err + } + defer func() { _ = tx.Rollback() }() + return tx.WriteTo(w) +} + +// ReadFrom reads the content and overwrites the existing store. +func (s *TranslateStore) ReadFrom(r io.Reader) (n int64, err error) { + // Close store. + if err := s.Close(); err != nil { + return 0, errors.Wrap(err, "closing store") + } + + // Create a temporary file to snapshot to. + snapshotPath := s.Path + snapshotExt + file, err := os.Create(snapshotPath) + if err != nil { + return n, errors.Wrap(err, "creating snapshot file") + } + + // Write payload to snapshot. + if n, err = io.Copy(file, r); err != nil { + file.Close() + return n, errors.Wrap(err, "snapshot write to") + } + + // we close the file here so we don't still have it open when trying + // to open it in a moment. + file.Close() + + // Move snapshot to data file location. + if err := os.Rename(snapshotPath, s.Path); err != nil { + return n, errors.Wrap(err, "renaming snapshot") + } + + // Re-open the store. + if err := s.Open(); err != nil { + return n, errors.Wrap(err, "re-opening store") + } + + return n, nil +} + // MaxID returns the highest id in the store. func maxID(tx *bolt.Tx) uint64 { if key, _ := tx.Bucket([]byte("ids")).Cursor().Last(); key != nil { diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index d3fb1b6e8..9ddb68907 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -14,9 +14,12 @@ package boltdb_test import ( + "bytes" "context" + "fmt" "io/ioutil" "os" + "reflect" "testing" "time" @@ -322,6 +325,70 @@ func MustNewTranslateStore() *boltdb.TranslateStore { return s } +func TestTranslateStore_ReadWrite(t *testing.T) { + t.Run("WriteTo_ReadFrom", func(t *testing.T) { + s := MustOpenNewTranslateStore() + defer MustCloseTranslateStore(s) + + batch0 := []string{} + for i := 0; i < 100; i++ { + batch0 = append(batch0, fmt.Sprintf("key%d", i)) + } + batch1 := []string{} + for i := 100; i < 200; i++ { + batch1 = append(batch1, fmt.Sprintf("key%d", i)) + } + + // Populate the store with the keys in batch0. + if _, err := s.TranslateKeys(batch0); err != nil { + t.Fatal(err) + } + + // Put the contents of the store into a buffer. + buf := bytes.NewBuffer(nil) + expN := int64(32768) + + // After this, the buffer should contain batch0. + if n, err := s.WriteTo(buf); err != nil { + t.Fatalf("writing to buffer: %s", err) + } else if n != expN { + t.Fatalf("expected buffer size: %d, but got: %d", expN, n) + } + + // Populate the store with the keys in batch1. + if _, err := s.TranslateKeys(batch1); err != nil { + t.Fatal(err) + } + + expIDs := []uint64{247463987, 247464087} + + // Check the IDs for a key from each batch. + if ids, err := s.TranslateKeys([]string{"key50", "key150"}); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(expIDs, ids) { + t.Fatalf("first expected ids: %v, but got: %v", expIDs, ids) + } + + // Reset the contents of the store with the data in the buffer. + if n, err := s.ReadFrom(buf); err != nil { + t.Fatalf("reading from buffer: %s", err) + } else if n != expN { + t.Fatalf("expected buffer size: %d, but got: %d", expN, n) + } + + // This time, we expect the second key to be different because + // we overwrote the store, and then just set that key. + expIDs = []uint64{247463987, 247464037} + + // Check the IDs for a key from each batch. + if ids, err := s.TranslateKeys([]string{"key50", "key150"}); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(expIDs, ids) { + t.Fatalf("last expected ids: %v, but got: %v", expIDs, ids) + } + }) +} + // MustOpenNewTranslateStore returns a new, opened TranslateStore. func MustOpenNewTranslateStore() *boltdb.TranslateStore { s := MustNewTranslateStore() diff --git a/client.go b/client.go index ffeb19b07..a1b808a77 100644 --- a/client.go +++ b/client.go @@ -70,6 +70,7 @@ type InternalClient interface { RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) SendMessage(ctx context.Context, uri *URI, msg []byte) error RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri URI) (io.ReadCloser, error) + RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error) ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error ImportColumnAttrs(ctx context.Context, uri *URI, index string, req *ImportColumnAttrsRequest) error } @@ -200,3 +201,6 @@ func (n nopInternalClient) SendMessage(ctx context.Context, uri *URI, msg []byte func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri URI) (io.ReadCloser, error) { return nil, nil } +func (n nopInternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error) { + return nil, nil +} diff --git a/cluster.go b/cluster.go index afd8dbd00..54f8795ab 100644 --- a/cluster.go +++ b/cluster.go @@ -864,6 +864,66 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour return m, nil } +// translationNodes returns a list of translationResizeNodes - for each node +// in the `to` cluster - required to move from cluster `c` to cluster `to`. unprotected. +// Because the parition scheme for every index is the same, this is used as a template +// to create index-specific `TranslationResizeSource`s. +func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResizeNode, error) { + m := make(map[string][]*translationResizeNode) + + // Determine if a node is being added or removed. + action, diffNodeID, err := c.diff(to) + if err != nil { + return nil, errors.Wrap(err, "diffing") + } + + // Initialize the map with all the nodes in `to`. + for _, n := range to.nodes { + m[n.ID] = nil + } + + for pid := 0; pid < c.partitionN; pid++ { + fNodes := c.partitionNodes(pid) + tNodes := to.partitionNodes(pid) + + // For `to` cluster, we include all nodes containing a + // replica for the partition. The source for each replica + // will be the primary in the `from` cluster. For the `from` + // cluster, we only need the first node, unless that node is + // being removed, then we use the second node. If no second + // node exists in that case, then we have to raise an error + // indicating that not enough replicas exist to support + // the resize. + if len(tNodes) > 0 { + var foundPrimary bool + for i := range fNodes { + if action == resizeJobActionRemove && fNodes[i].ID == diffNodeID { + continue + } + // We only need to add the source if the nodes differ; + // in other words if the primary partition is on the + // same node, it doesn't need to retrieve it. + for n := range tNodes { + if tNodes[n].ID != fNodes[i].ID { + m[tNodes[n].ID] = append(m[tNodes[n].ID], + &translationResizeNode{ + node: fNodes[i], + partitionID: pid, + }) + } + } + foundPrimary = true + break + } + if !foundPrimary { + return nil, ErrResizeNoReplicas + } + } + } + + return m, nil +} + // shardPartition returns the partition that a shard belongs to. func (c *cluster) shardPartition(index string, shard uint64) int { return shardPartition(index, shard, c.partitionN) @@ -1271,38 +1331,82 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (* toCluster.addNodeBasicSorted(nodeAction.node) } - // multiIndex is a map of sources initialized with all the nodes in toCluster. - multiIndex := make(map[string][]*ResizeSource) + indexes := c.holder.Indexes() + // fragmentSourcesByNode is a map of Node.ID to sources of fragment data. + // It is initialized with all the nodes in toCluster. + fragmentSourcesByNode := make(map[string][]*ResizeSource) for _, n := range toCluster.nodes { - multiIndex[n.ID] = nil + fragmentSourcesByNode[n.ID] = nil } - // Add to multiIndex the instructions for each index. - for _, idx := range c.holder.Indexes() { + // Add to fragmentSourcesByNode the instructions for each index. + for _, idx := range indexes { fragSources, err := c.fragSources(toCluster, idx) if err != nil { return nil, errors.Wrap(err, "getting sources") } - for id, sources := range fragSources { - multiIndex[id] = append(multiIndex[id], sources...) + for nodeid, sources := range fragSources { + fragmentSourcesByNode[nodeid] = append(fragmentSourcesByNode[nodeid], sources...) } } - for id, sources := range multiIndex { + // translationSourcesByNode is a map of Node.ID to sources of partitioned + // key translation data for indexes. + // It is initialized with all the nodes in toCluster. + translationSourcesByNode := make(map[string][]*TranslationResizeSource) + for _, n := range toCluster.nodes { + translationSourcesByNode[n.ID] = nil + } + + if len(indexes) > 0 { + // Add to translationSourcesByNode the instructions for the cluster. + translationNodes, err := c.translationNodes(toCluster) + if err != nil { + return nil, errors.Wrap(err, "getting translation sources") + } + + // Create a list of TranslationResizeSource for each index, + // using translationNodes as a template. + translationSources := make(map[string][]*TranslationResizeSource) + for _, idx := range indexes { + // Only include indexes with keys. + if !idx.Keys() { + continue + } + indexName := idx.Name() + for node, resizeNodes := range translationNodes { + for i := range resizeNodes { + translationSources[node] = append(translationSources[node], + &TranslationResizeSource{ + Node: resizeNodes[i].node, + Index: indexName, + PartitionID: resizeNodes[i].partitionID, + }) + } + } + } + + for nodeid, sources := range translationSources { + translationSourcesByNode[nodeid] = sources + } + } + + for _, node := range toCluster.nodes { // If a host doesn't need to request data, mark it as complete. - if len(sources) == 0 { - j.IDs[id] = true + if len(fragmentSourcesByNode[node.ID]) == 0 && len(translationSourcesByNode[node.ID]) == 0 { + j.IDs[node.ID] = true continue } instr := &ResizeInstruction{ - JobID: j.ID, - Node: toCluster.unprotectedNodeByID(id), - Coordinator: c.unprotectedCoordinatorNode(), - Sources: sources, - NodeStatus: c.nodeStatus(), // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node. - ClusterStatus: c.unprotectedStatus(), + JobID: j.ID, + Node: toCluster.unprotectedNodeByID(node.ID), + Coordinator: c.unprotectedCoordinatorNode(), + Sources: fragmentSourcesByNode[node.ID], + TranslationSources: translationSourcesByNode[node.ID], + NodeStatus: c.nodeStatus(), // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node. + ClusterStatus: c.unprotectedStatus(), } j.Instructions = append(j.Instructions, instr) } @@ -1386,9 +1490,8 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { // Request each source file in ResizeSources. for _, src := range instr.Sources { - c.logger.Printf("get shard %d for index %s from host %s", src.Shard, src.Index, src.Node.URI) - srcURI := src.Node.URI + c.logger.Printf("get shard %d for index %s from host %s", src.Shard, src.Index, srcURI) // Retrieve field. f := c.holder.Field(src.Index, src.Field) @@ -1409,7 +1512,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { } // Stream shard from remote node. - c.logger.Printf("retrieve shard %d for index %s from host %s", src.Shard, src.Index, src.Node.URI) + c.logger.Printf("retrieve shard %d for index %s from host %s", src.Shard, src.Index, srcURI) rd, err := c.InternalClient.RetrieveShardFromURI(ctx, src.Index, src.Field, src.View, src.Shard, srcURI) if err != nil { // For now it is an acceptable error if the fragment is not found @@ -1423,7 +1526,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { } return errors.Wrap(err, "retrieving shard") } else if rd == nil { - return fmt.Errorf("shard %v doesn't exist on host: %s", src.Shard, src.Node.URI) + return fmt.Errorf("shard %v doesn't exist on host: %s", src.Shard, srcURI) } // Write to local field and always close reader. @@ -1435,6 +1538,37 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { return errors.Wrap(err, "copying remote shard") } } + + // Request each translation source file in TranslationResizeSources. + for _, src := range instr.TranslationSources { + srcURI := src.Node.URI + + idx := c.holder.Index(src.Index) + if idx == nil { + return ErrIndexNotFound + } + + // Retrieve partition from remote node. + c.logger.Printf("retrieve translate partition %d for index %s from host %s", src.PartitionID, src.Index, srcURI) + rd, err := c.InternalClient.RetrieveTranslatePartitionFromURI(ctx, src.Index, src.PartitionID, srcURI) + if err != nil { + return errors.Wrap(err, "retrieving translate partition") + } else if rd == nil { + return fmt.Errorf("partition %d doesn't exist on host: %s", src.PartitionID, src.Node.URI) + } + + // Write to local store and always close reader. + if err := func() error { + defer rd.Close() + // Get the translate store for this index/partition. + store := idx.TranslateStore(src.PartitionID) + _, err = store.ReadFrom(rd) + return errors.Wrap(err, "reading from reader") + }(); err != nil { + return errors.Wrap(err, "copying remote partition") + } + } + return nil }(); err != nil { complete.Error = err.Error() @@ -2291,12 +2425,13 @@ type ClusterStatus struct { // ResizeInstruction contains the instruction provided to a node // during a cluster resize operation. type ResizeInstruction struct { - JobID int64 - Node *Node - Coordinator *Node - Sources []*ResizeSource - NodeStatus *NodeStatus - ClusterStatus *ClusterStatus + JobID int64 + Node *Node + Coordinator *Node + Sources []*ResizeSource + TranslationSources []*TranslationResizeSource + NodeStatus *NodeStatus + ClusterStatus *ClusterStatus } // ResizeSource is the source of data for a node acting on a @@ -2309,6 +2444,21 @@ type ResizeSource struct { Shard uint64 `protobuf:"varint,5,opt,name=Shard,proto3" json:"Shard,omitempty"` } +// TranslationResizeSource is the source of translation data for +// a node acting on a ResizeInstruction. +type TranslationResizeSource struct { + Node *Node + Index string + PartitionID int +} + +// translateResizeNode holds the node/partition pairs used +// to create a TranslationResizeSource for each index. +type translationResizeNode struct { + node *Node + partitionID int +} + // Schema contains information about indexes and their configuration. type Schema struct { Indexes []*IndexInfo diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 47a73d0ed..eca95de2f 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -517,12 +517,13 @@ func encodeQueryResponse(m *pilosa.QueryResponse) *internal.QueryResponse { func encodeResizeInstruction(m *pilosa.ResizeInstruction) *internal.ResizeInstruction { return &internal.ResizeInstruction{ - JobID: m.JobID, - Node: encodeNode(m.Node), - Coordinator: encodeNode(m.Coordinator), - Sources: encodeResizeSources(m.Sources), - NodeStatus: encodeNodeStatus(m.NodeStatus), - ClusterStatus: encodeClusterStatus(m.ClusterStatus), + JobID: m.JobID, + Node: encodeNode(m.Node), + Coordinator: encodeNode(m.Coordinator), + Sources: encodeResizeSources(m.Sources), + TranslationSources: encodeTranslationResizeSources(m.TranslationSources), + NodeStatus: encodeNodeStatus(m.NodeStatus), + ClusterStatus: encodeClusterStatus(m.ClusterStatus), } } @@ -544,6 +545,22 @@ func encodeResizeSource(m *pilosa.ResizeSource) *internal.ResizeSource { } } +func encodeTranslationResizeSources(srcs []*pilosa.TranslationResizeSource) []*internal.TranslationResizeSource { + new := make([]*internal.TranslationResizeSource, 0, len(srcs)) + for _, src := range srcs { + new = append(new, encodeTranslationResizeSource(src)) + } + return new +} + +func encodeTranslationResizeSource(m *pilosa.TranslationResizeSource) *internal.TranslationResizeSource { + return &internal.TranslationResizeSource{ + Node: encodeNode(m.Node), + Index: m.Index, + PartitionID: int32(m.PartitionID), + } +} + func encodeSchema(m *pilosa.Schema) *internal.Schema { return &internal.Schema{ Indexes: encodeIndexInfos(m.Indexes), @@ -560,8 +577,9 @@ func encodeIndexInfos(idxs []*pilosa.IndexInfo) []*internal.Index { func encodeIndexInfo(idx *pilosa.IndexInfo) *internal.Index { return &internal.Index{ - Name: idx.Name, - Fields: encodeFieldInfos(idx.Fields), + Name: idx.Name, + Options: encodeIndexMeta(&idx.Options), + Fields: encodeFieldInfos(idx.Fields), } } @@ -819,6 +837,8 @@ func decodeResizeInstruction(ri *internal.ResizeInstruction, m *pilosa.ResizeIns decodeNode(ri.Coordinator, m.Coordinator) m.Sources = make([]*pilosa.ResizeSource, len(ri.Sources)) decodeResizeSources(ri.Sources, m.Sources) + m.TranslationSources = make([]*pilosa.TranslationResizeSource, len(ri.TranslationSources)) + decodeTranslationResizeSources(ri.TranslationSources, m.TranslationSources) m.NodeStatus = &pilosa.NodeStatus{} decodeNodeStatus(ri.NodeStatus, m.NodeStatus) m.ClusterStatus = &pilosa.ClusterStatus{} @@ -841,6 +861,20 @@ func decodeResizeSource(rs *internal.ResizeSource, m *pilosa.ResizeSource) { m.Shard = rs.Shard } +func decodeTranslationResizeSources(srcs []*internal.TranslationResizeSource, m []*pilosa.TranslationResizeSource) { + for i := range srcs { + m[i] = &pilosa.TranslationResizeSource{} + decodeTranslationResizeSource(srcs[i], m[i]) + } +} + +func decodeTranslationResizeSource(rs *internal.TranslationResizeSource, m *pilosa.TranslationResizeSource) { + m.Node = &pilosa.Node{} + decodeNode(rs.Node, m.Node) + m.Index = rs.Index + m.PartitionID = int(rs.PartitionID) +} + func decodeSchema(s *internal.Schema, m *pilosa.Schema) { m.Indexes = make([]*pilosa.IndexInfo, len(s.Indexes)) decodeIndexes(s.Indexes, m.Indexes) @@ -855,6 +889,8 @@ func decodeIndexes(idxs []*internal.Index, m []*pilosa.IndexInfo) { func decodeIndex(idx *internal.Index, m *pilosa.IndexInfo) { m.Name = idx.Name + m.Options = pilosa.IndexOptions{} + decodeIndexMeta(idx.Options, &m.Options) m.Fields = make([]*pilosa.FieldInfo, len(idx.Fields)) decodeFields(idx.Fields, m.Fields) } diff --git a/go.mod b/go.mod index e8f70176e..1676bf805 100644 --- a/go.mod +++ b/go.mod @@ -37,7 +37,7 @@ require ( github.com/uber/jaeger-lib v2.2.0+incompatible // indirect go.uber.org/atomic v1.4.0 // indirect golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734 // indirect - golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6 + golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6 // indirect golang.org/x/sync v0.0.0-20190423024810-112230192c58 golang.org/x/sys v0.0.0-20190429190828-d89cdac9e872 // indirect golang.org/x/text v0.3.2 // indirect diff --git a/go.sum b/go.sum index f932bb2c1..85e0347f6 100644 --- a/go.sum +++ b/go.sum @@ -105,7 +105,6 @@ github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I= -github.com/pilosa/pilosa v1.4.0 h1:nqHNIK4nDslFnem3yDp9R+6TgLdlkY9WdJD88Z83T8U= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= diff --git a/holder.go b/holder.go index 48796da8e..987b8939d 100644 --- a/holder.go +++ b/holder.go @@ -348,7 +348,10 @@ func (h *Holder) availableShardsByIndex() map[string]*roaring.Bitmap { func (h *Holder) Schema() []*IndexInfo { var a []*IndexInfo for _, index := range h.Indexes() { - di := &IndexInfo{Name: index.Name()} + di := &IndexInfo{ + Name: index.Name(), + Options: index.Options(), + } for _, field := range index.Fields() { fi := &FieldInfo{Name: field.Name(), Options: field.Options()} for _, view := range field.views() { diff --git a/http/client.go b/http/client.go index 507d1d632..c688e5565 100644 --- a/http/client.go +++ b/http/client.go @@ -1469,3 +1469,40 @@ func nodePathToURL(node *pilosa.Node, path string) url.URL { Path: path, } } + +// RetrieveTranslatePartitionFromURI returns a ReadCloser which contains the data of the +// specified translate partition from the specified node. Caller *must* close the returned +// ReadCloser or risk leaking goroutines/tcp connections. +func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pilosa.URI) (io.ReadCloser, error) { + span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveTranslatePartitionFromURI") + defer span.Finish() + + node := &pilosa.Node{ + URI: uri, + } + + u := nodePathToURL(node, "/internal/translate/data") + u.RawQuery = url.Values{ + "index": {index}, + "partition": {strconv.FormatInt(int64(partition), 10)}, + }.Encode() + + // Build request. + req, err := http.NewRequest("GET", u.String(), nil) + if err != nil { + return nil, errors.Wrap(err, "creating request") + } + + req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) + + // Execute request. + resp, err := c.executeRequest(req.WithContext(ctx)) + if err != nil { + if resp != nil && resp.StatusCode == http.StatusNotFound { + return nil, pilosa.ErrFragmentNotFound + } + return nil, err + } + + return resp.Body, nil +} diff --git a/http/handler.go b/http/handler.go index 8b5d764ad..5d11c55f1 100644 --- a/http/handler.go +++ b/http/handler.go @@ -182,7 +182,7 @@ func (h *Handler) populateValidators() { h.validators["GetIndex"] = queryValidationSpecRequired() h.validators["PostIndex"] = queryValidationSpecRequired() h.validators["DeleteIndex"] = queryValidationSpecRequired() - h.validators["GetTranslateData"] = queryValidationSpecRequired("offset") + h.validators["GetTranslateData"] = queryValidationSpecRequired("index", "partition") h.validators["PostTranslateKeys"] = queryValidationSpecRequired() h.validators["PostField"] = queryValidationSpecRequired() h.validators["DeleteField"] = queryValidationSpecRequired() @@ -312,6 +312,7 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/internal/fragment/data", handler.handleGetFragmentData).Methods("GET").Name("GetFragmentData") router.HandleFunc("/internal/fragment/nodes", handler.handleGetFragmentNodes).Methods("GET").Name("GetFragmentNodes") router.HandleFunc("/internal/index/{index}/attr/diff", handler.handlePostIndexAttrDiff).Methods("POST").Name("PostIndexAttrDiff") + router.HandleFunc("/internal/translate/data", handler.handleGetTranslateData).Methods("GET").Name("GetTranslateData") router.HandleFunc("/internal/translate/data", handler.handlePostTranslateData).Methods("POST").Name("PostTranslateData") router.HandleFunc("/internal/translate/keys", handler.handlePostTranslateKeys).Methods("POST").Name("PostTranslateKeys") router.HandleFunc("/internal/translate/ids", handler.handlePostTranslateIDs).Methods("POST").Name("PostTranslateIDs") @@ -1394,6 +1395,27 @@ func (h *Handler) handleGetFragmentData(w http.ResponseWriter, r *http.Request) } } +// handleGetTranslateData handles GET /internal/translate/data requests. +func (h *Handler) handleGetTranslateData(w http.ResponseWriter, r *http.Request) { + // Read partition parameter. + q := r.URL.Query() + partition, err := strconv.ParseUint(q.Get("partition"), 10, 32) + if err != nil { + http.Error(w, "partition required", http.StatusBadRequest) + return + } + // Retrieve partition data from holder. + p, err := h.api.TranslateData(r.Context(), q.Get("index"), int(partition)) + if err != nil { + http.Error(w, err.Error(), http.StatusNotFound) + return + } + // Stream translate partition to response body. + if _, err := p.WriteTo(w); err != nil { + h.logger.Printf("error streaming translation data: %s", err) + } +} + // handleGetVersion handles /version requests. func (h *Handler) handleGetVersion(w http.ResponseWriter, r *http.Request) { if !validHeaderAcceptJSON(r.Header) { diff --git a/index.go b/index.go index e0ec7fc3b..9c53083fd 100644 --- a/index.go +++ b/index.go @@ -264,7 +264,8 @@ func (i *Index) openExistenceField() error { // loadMeta reads meta data for the index, if any. func (i *Index) loadMeta() error { - var pb internal.IndexMeta + // TrackExistence is by default true + pb := &internal.IndexMeta{TrackExistence: true} // Read data from meta file. buf, err := ioutil.ReadFile(filepath.Join(i.path, ".meta")) @@ -273,14 +274,18 @@ func (i *Index) loadMeta() error { } else if err != nil { return errors.Wrap(err, "reading") } else { - if err := proto.Unmarshal(buf, &pb); err != nil { + if err := proto.Unmarshal(buf, pb); err != nil { return errors.Wrap(err, "unmarshalling") } } // Copy metadata fields. - i.keys = pb.Keys - i.trackExistence = pb.TrackExistence + if pb == nil { + i.trackExistence = true + } else { + i.trackExistence = pb.TrackExistence + } + i.keys = pb.GetKeys() return nil } diff --git a/internal/private.pb.go b/internal/private.pb.go index 401b92786..4ca901b58 100644 --- a/internal/private.pb.go +++ b/internal/private.pb.go @@ -999,11 +999,12 @@ func (m *Schema) GetIndexes() []*Index { } type Index struct { - Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"` - Fields []*Field `protobuf:"bytes,4,rep,name=Fields,proto3" json:"Fields,omitempty"` - XXX_NoUnkeyedLiteral struct{} `json:"-"` - XXX_unrecognized []byte `json:"-"` - XXX_sizecache int32 `json:"-"` + Name string `protobuf:"bytes,1,opt,name=Name,proto3" json:"Name,omitempty"` + Options *IndexMeta `protobuf:"bytes,5,opt,name=Options,proto3" json:"Options,omitempty"` + Fields []*Field `protobuf:"bytes,4,rep,name=Fields,proto3" json:"Fields,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` } func (m *Index) Reset() { *m = Index{} } @@ -1046,6 +1047,13 @@ func (m *Index) GetName() string { return "" } +func (m *Index) GetOptions() *IndexMeta { + if m != nil { + return m.Options + } + return nil +} + func (m *Index) GetFields() []*Field { if m != nil { return m.Fields @@ -1731,15 +1739,16 @@ func (m *DeleteViewMessage) GetView() string { } type ResizeInstruction struct { - JobID int64 `protobuf:"varint,1,opt,name=JobID,proto3" json:"JobID,omitempty"` - Node *Node `protobuf:"bytes,2,opt,name=Node,proto3" json:"Node,omitempty"` - Coordinator *Node `protobuf:"bytes,3,opt,name=Coordinator,proto3" json:"Coordinator,omitempty"` - Sources []*ResizeSource `protobuf:"bytes,4,rep,name=Sources,proto3" json:"Sources,omitempty"` - NodeStatus *NodeStatus `protobuf:"bytes,7,opt,name=NodeStatus,proto3" json:"NodeStatus,omitempty"` - ClusterStatus *ClusterStatus `protobuf:"bytes,6,opt,name=ClusterStatus,proto3" json:"ClusterStatus,omitempty"` - XXX_NoUnkeyedLiteral struct{} `json:"-"` - XXX_unrecognized []byte `json:"-"` - XXX_sizecache int32 `json:"-"` + JobID int64 `protobuf:"varint,1,opt,name=JobID,proto3" json:"JobID,omitempty"` + Node *Node `protobuf:"bytes,2,opt,name=Node,proto3" json:"Node,omitempty"` + Coordinator *Node `protobuf:"bytes,3,opt,name=Coordinator,proto3" json:"Coordinator,omitempty"` + Sources []*ResizeSource `protobuf:"bytes,4,rep,name=Sources,proto3" json:"Sources,omitempty"` + TranslationSources []*TranslationResizeSource `protobuf:"bytes,8,rep,name=TranslationSources,proto3" json:"TranslationSources,omitempty"` + NodeStatus *NodeStatus `protobuf:"bytes,7,opt,name=NodeStatus,proto3" json:"NodeStatus,omitempty"` + ClusterStatus *ClusterStatus `protobuf:"bytes,6,opt,name=ClusterStatus,proto3" json:"ClusterStatus,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` } func (m *ResizeInstruction) Reset() { *m = ResizeInstruction{} } @@ -1803,6 +1812,13 @@ func (m *ResizeInstruction) GetSources() []*ResizeSource { return nil } +func (m *ResizeInstruction) GetTranslationSources() []*TranslationResizeSource { + if m != nil { + return m.TranslationSources + } + return nil +} + func (m *ResizeInstruction) GetNodeStatus() *NodeStatus { if m != nil { return m.NodeStatus @@ -1896,6 +1912,69 @@ func (m *ResizeSource) GetShard() uint64 { return 0 } +type TranslationResizeSource struct { + Node *Node `protobuf:"bytes,1,opt,name=Node,proto3" json:"Node,omitempty"` + Index string `protobuf:"bytes,2,opt,name=Index,proto3" json:"Index,omitempty"` + PartitionID int32 `protobuf:"varint,3,opt,name=PartitionID,proto3" json:"PartitionID,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *TranslationResizeSource) Reset() { *m = TranslationResizeSource{} } +func (m *TranslationResizeSource) String() string { return proto.CompactTextString(m) } +func (*TranslationResizeSource) ProtoMessage() {} +func (*TranslationResizeSource) Descriptor() ([]byte, []int) { + return fileDescriptor_d2a91b51c7bdc125, []int{30} +} +func (m *TranslationResizeSource) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *TranslationResizeSource) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_TranslationResizeSource.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 *TranslationResizeSource) XXX_Merge(src proto.Message) { + xxx_messageInfo_TranslationResizeSource.Merge(m, src) +} +func (m *TranslationResizeSource) XXX_Size() int { + return m.Size() +} +func (m *TranslationResizeSource) XXX_DiscardUnknown() { + xxx_messageInfo_TranslationResizeSource.DiscardUnknown(m) +} + +var xxx_messageInfo_TranslationResizeSource proto.InternalMessageInfo + +func (m *TranslationResizeSource) GetNode() *Node { + if m != nil { + return m.Node + } + return nil +} + +func (m *TranslationResizeSource) GetIndex() string { + if m != nil { + return m.Index + } + return "" +} + +func (m *TranslationResizeSource) GetPartitionID() int32 { + if m != nil { + return m.PartitionID + } + return 0 +} + type ResizeInstructionComplete struct { JobID int64 `protobuf:"varint,1,opt,name=JobID,proto3" json:"JobID,omitempty"` Node *Node `protobuf:"bytes,2,opt,name=Node,proto3" json:"Node,omitempty"` @@ -1909,7 +1988,7 @@ func (m *ResizeInstructionComplete) Reset() { *m = ResizeInstructionComp func (m *ResizeInstructionComplete) String() string { return proto.CompactTextString(m) } func (*ResizeInstructionComplete) ProtoMessage() {} func (*ResizeInstructionComplete) Descriptor() ([]byte, []int) { - return fileDescriptor_d2a91b51c7bdc125, []int{30} + return fileDescriptor_d2a91b51c7bdc125, []int{31} } func (m *ResizeInstructionComplete) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -1970,7 +2049,7 @@ func (m *SetCoordinatorMessage) Reset() { *m = SetCoordinatorMessage{} } func (m *SetCoordinatorMessage) String() string { return proto.CompactTextString(m) } func (*SetCoordinatorMessage) ProtoMessage() {} func (*SetCoordinatorMessage) Descriptor() ([]byte, []int) { - return fileDescriptor_d2a91b51c7bdc125, []int{31} + return fileDescriptor_d2a91b51c7bdc125, []int{32} } func (m *SetCoordinatorMessage) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -2017,7 +2096,7 @@ func (m *UpdateCoordinatorMessage) Reset() { *m = UpdateCoordinatorMessa func (m *UpdateCoordinatorMessage) String() string { return proto.CompactTextString(m) } func (*UpdateCoordinatorMessage) ProtoMessage() {} func (*UpdateCoordinatorMessage) Descriptor() ([]byte, []int) { - return fileDescriptor_d2a91b51c7bdc125, []int{32} + return fileDescriptor_d2a91b51c7bdc125, []int{33} } func (m *UpdateCoordinatorMessage) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -2065,7 +2144,7 @@ func (m *Topology) Reset() { *m = Topology{} } func (m *Topology) String() string { return proto.CompactTextString(m) } func (*Topology) ProtoMessage() {} func (*Topology) Descriptor() ([]byte, []int) { - return fileDescriptor_d2a91b51c7bdc125, []int{33} + return fileDescriptor_d2a91b51c7bdc125, []int{34} } func (m *Topology) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -2118,7 +2197,7 @@ func (m *RecalculateCaches) Reset() { *m = RecalculateCaches{} } func (m *RecalculateCaches) String() string { return proto.CompactTextString(m) } func (*RecalculateCaches) ProtoMessage() {} func (*RecalculateCaches) Descriptor() ([]byte, []int) { - return fileDescriptor_d2a91b51c7bdc125, []int{34} + return fileDescriptor_d2a91b51c7bdc125, []int{35} } func (m *RecalculateCaches) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -2179,6 +2258,7 @@ func init() { proto.RegisterType((*DeleteViewMessage)(nil), "internal.DeleteViewMessage") proto.RegisterType((*ResizeInstruction)(nil), "internal.ResizeInstruction") proto.RegisterType((*ResizeSource)(nil), "internal.ResizeSource") + proto.RegisterType((*TranslationResizeSource)(nil), "internal.TranslationResizeSource") proto.RegisterType((*ResizeInstructionComplete)(nil), "internal.ResizeInstructionComplete") proto.RegisterType((*SetCoordinatorMessage)(nil), "internal.SetCoordinatorMessage") proto.RegisterType((*UpdateCoordinatorMessage)(nil), "internal.UpdateCoordinatorMessage") @@ -2189,85 +2269,88 @@ func init() { func init() { proto.RegisterFile("private.proto", fileDescriptor_d2a91b51c7bdc125) } var fileDescriptor_d2a91b51c7bdc125 = []byte{ - // 1233 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xdd, 0x72, 0xdb, 0x44, - 0x14, 0x46, 0x92, 0xe3, 0x9f, 0xe3, 0x38, 0x75, 0xb6, 0x3f, 0xa8, 0x85, 0x09, 0x66, 0xe9, 0x50, - 0xd3, 0x19, 0x42, 0xa7, 0x85, 0x19, 0xfe, 0x3a, 0x53, 0x1c, 0xa7, 0x45, 0x94, 0x84, 0xb2, 0x4e, - 0x73, 0xc7, 0xc5, 0x46, 0xde, 0x49, 0x34, 0x91, 0x25, 0x23, 0xad, 0x93, 0xb8, 0x17, 0xdc, 0xc2, - 0x0c, 0x2f, 0xc0, 0x13, 0xf0, 0x2c, 0x5c, 0xf2, 0x08, 0x4c, 0x78, 0x01, 0x1e, 0x81, 0xd9, 0xb3, - 0xab, 0x1f, 0x3b, 0x0e, 0x09, 0x81, 0xbb, 0x3d, 0xe7, 0xec, 0x39, 0xe7, 0x3b, 0xbf, 0x2b, 0x41, - 0x6b, 0x9c, 0x04, 0x47, 0x5c, 0x8a, 0xf5, 0x71, 0x12, 0xcb, 0x98, 0xd4, 0x83, 0x48, 0x8a, 0x24, - 0xe2, 0x21, 0x7d, 0x06, 0x0d, 0x2f, 0x1a, 0x8a, 0x93, 0x2d, 0x21, 0x39, 0x21, 0x50, 0x79, 0x2e, - 0xa6, 0xa9, 0xeb, 0x74, 0xac, 0x6e, 0x9d, 0xe1, 0x99, 0xbc, 0x0b, 0x2b, 0x3b, 0x09, 0xf7, 0x0f, - 0x37, 0x4f, 0x82, 0x54, 0x8a, 0xc8, 0x17, 0x6e, 0x05, 0xa5, 0x73, 0x5c, 0xfa, 0x97, 0x0d, 0xcb, - 0x4f, 0x03, 0x11, 0x0e, 0xbf, 0x19, 0xcb, 0x20, 0x8e, 0x52, 0x65, 0x6c, 0x67, 0x3a, 0x16, 0x6e, - 0xbd, 0x63, 0x75, 0x1b, 0x0c, 0xcf, 0xe4, 0x4d, 0x68, 0x6c, 0x70, 0xff, 0x40, 0xa0, 0xc0, 0x41, - 0x41, 0xc1, 0xc8, 0xa5, 0x83, 0xe0, 0x95, 0xf6, 0xd2, 0x62, 0x05, 0x83, 0x74, 0xa0, 0xb9, 0x13, - 0x8c, 0xc4, 0xb7, 0x13, 0x1e, 0xc9, 0xc9, 0xc8, 0x5d, 0x42, 0xed, 0x32, 0x2b, 0x87, 0xdf, 0x9c, - 0x85, 0xbf, 0x1d, 0x0f, 0x24, 0x8f, 0x86, 0x3c, 0x19, 0xee, 0x06, 0xe2, 0xd8, 0x5d, 0xd6, 0xf0, - 0x67, 0xb9, 0x4a, 0xb7, 0xc7, 0x53, 0xe1, 0xb6, 0x3a, 0x56, 0xd7, 0x61, 0x78, 0x26, 0x77, 0xa0, - 0xde, 0x0b, 0x64, 0x5f, 0x8c, 0xe5, 0x81, 0xbb, 0xd2, 0xb1, 0xba, 0x15, 0x96, 0xd3, 0xe4, 0x06, - 0x2c, 0x0d, 0x7c, 0x1e, 0x0a, 0xf7, 0x1a, 0x2a, 0x68, 0x82, 0x50, 0x58, 0x7e, 0x1a, 0x27, 0x22, - 0xd8, 0x8f, 0x30, 0xa9, 0x6e, 0x1b, 0x41, 0xce, 0xf0, 0xc8, 0x3b, 0xe0, 0x6c, 0x05, 0x91, 0xbb, - 0xda, 0xb1, 0xba, 0xcd, 0x87, 0xab, 0xeb, 0x59, 0x25, 0xd6, 0xfb, 0xc2, 0x0f, 0x46, 0x3c, 0x64, - 0x4a, 0x8a, 0x97, 0xf8, 0x89, 0x4b, 0xce, 0xbf, 0xc4, 0x4f, 0xe8, 0x47, 0x50, 0x33, 0xb4, 0x82, - 0xb3, 0xcb, 0xc3, 0x89, 0x70, 0x2d, 0x0d, 0x07, 0x89, 0x02, 0xa4, 0x5d, 0x02, 0x49, 0x29, 0xac, - 0x78, 0xa3, 0x71, 0x9c, 0x48, 0x26, 0xd2, 0x71, 0x1c, 0xa5, 0x82, 0xb4, 0xc1, 0xd9, 0x4c, 0x12, - 0xd4, 0x6d, 0x30, 0x75, 0xa4, 0x3f, 0x40, 0xbb, 0x17, 0xc6, 0xfe, 0x61, 0x9f, 0x4b, 0xce, 0xc4, - 0xf7, 0x13, 0x91, 0x4a, 0x65, 0x4d, 0x47, 0xa5, 0xef, 0x69, 0x42, 0x71, 0xb1, 0xec, 0xe8, 0xa3, - 0xc1, 0x34, 0xa1, 0xd2, 0x89, 0xc9, 0xd6, 0x55, 0xc2, 0x33, 0xa2, 0x39, 0xe0, 0xc9, 0x10, 0x4b, - 0x5b, 0x61, 0x9a, 0x50, 0x5c, 0xf4, 0x84, 0xed, 0x50, 0x61, 0x9a, 0xa0, 0x1e, 0xac, 0x96, 0xfc, - 0x1b, 0x98, 0xb7, 0xa0, 0xca, 0xe2, 0x63, 0xaf, 0x9f, 0xba, 0x56, 0xc7, 0xe9, 0x56, 0x98, 0xa1, - 0xb0, 0x6f, 0xe2, 0x70, 0x32, 0x8a, 0x94, 0xc8, 0x46, 0x51, 0xc1, 0xa0, 0xb7, 0x61, 0x09, 0x9b, - 0x48, 0x45, 0x59, 0xe8, 0xaa, 0x23, 0xfd, 0xd1, 0x82, 0xc6, 0x16, 0x3f, 0x41, 0x20, 0x29, 0x79, - 0x0c, 0xf5, 0xac, 0x25, 0xf0, 0x52, 0xf3, 0xe1, 0xdb, 0x45, 0xe2, 0xf3, 0x6b, 0xeb, 0xd9, 0x9d, - 0xcd, 0x48, 0x26, 0x53, 0x96, 0xab, 0xdc, 0xf9, 0x0c, 0x5a, 0x33, 0x22, 0xe5, 0xef, 0x50, 0x4c, - 0xb3, 0xac, 0x1e, 0x8a, 0xa9, 0x8a, 0xf5, 0x08, 0xab, 0x64, 0xeb, 0x58, 0x91, 0xf8, 0xd4, 0xfe, - 0xd8, 0xa2, 0xbb, 0x40, 0x36, 0x12, 0xc1, 0xa5, 0x40, 0x27, 0x5b, 0x22, 0x4d, 0xf9, 0xbe, 0xb8, - 0x28, 0xe3, 0x4e, 0x39, 0xe3, 0x79, 0x76, 0xed, 0x52, 0x76, 0xe9, 0x7d, 0x20, 0x7d, 0x11, 0x0a, - 0x29, 0xcc, 0x90, 0xff, 0x83, 0x5d, 0x3a, 0xc8, 0x30, 0x5c, 0x7c, 0x97, 0xdc, 0x83, 0x8a, 0xda, - 0x18, 0xe8, 0xac, 0xf9, 0xf0, 0x7a, 0x91, 0xa7, 0x7c, 0x99, 0x30, 0xbc, 0x40, 0xc3, 0xcc, 0x28, - 0xa2, 0xbc, 0x64, 0x60, 0x33, 0xad, 0x74, 0xdf, 0xb8, 0x72, 0xd0, 0xd5, 0xad, 0xc2, 0x55, 0x79, - 0xdb, 0x18, 0x6f, 0x4f, 0xb2, 0x70, 0xaf, 0xea, 0x8d, 0xfa, 0xf0, 0x86, 0xb6, 0xf0, 0xc5, 0x11, - 0x0f, 0x42, 0xbe, 0x17, 0xfe, 0xab, 0x8a, 0xcc, 0x00, 0x77, 0xa1, 0x86, 0xba, 0x5e, 0xdf, 0xf4, - 0x76, 0x46, 0xd2, 0xef, 0xa0, 0x18, 0x93, 0x6d, 0x3e, 0x12, 0xc6, 0x1a, 0x9e, 0xf3, 0x78, 0xed, - 0x8b, 0xe3, 0xc5, 0xb1, 0x0f, 0xc4, 0xb1, 0xda, 0xd8, 0x8e, 0x72, 0x8c, 0x04, 0x7d, 0x04, 0xd5, - 0x81, 0x7f, 0x20, 0x46, 0x9c, 0xbc, 0x07, 0x35, 0x44, 0x28, 0x52, 0xd3, 0xd1, 0xd7, 0xe6, 0x2a, - 0xc5, 0x32, 0x39, 0xed, 0x9b, 0xc8, 0x16, 0x62, 0xba, 0x07, 0x55, 0xf4, 0x9e, 0xba, 0x95, 0x79, - 0x33, 0xc8, 0x67, 0x46, 0x4c, 0x37, 0xc1, 0x79, 0xc9, 0x3c, 0x35, 0xa9, 0x88, 0x20, 0xb3, 0x62, - 0x28, 0x65, 0xfb, 0xcb, 0x38, 0x95, 0x26, 0x4f, 0x78, 0x56, 0xbc, 0x17, 0x71, 0x22, 0x31, 0x47, - 0x2d, 0x86, 0x67, 0x9a, 0x42, 0x65, 0x3b, 0x1e, 0x0a, 0xb2, 0x02, 0xb6, 0xd7, 0x37, 0x36, 0x6c, - 0xaf, 0x4f, 0xde, 0x42, 0xf3, 0x26, 0x35, 0xad, 0x02, 0xc4, 0x4b, 0xe6, 0x31, 0x74, 0x7c, 0x17, - 0x5a, 0x5e, 0xba, 0x11, 0xc7, 0xc9, 0x30, 0x88, 0xb8, 0x8c, 0x13, 0xf3, 0x94, 0xcd, 0x32, 0x71, - 0x56, 0x24, 0x97, 0xfa, 0x91, 0x69, 0x30, 0x4d, 0xd0, 0x27, 0xd0, 0x56, 0x4e, 0x91, 0xc8, 0xea, - 0x7d, 0x0b, 0xaa, 0x8a, 0x97, 0x83, 0x30, 0x54, 0x61, 0xc1, 0x2e, 0x5b, 0xf8, 0x5a, 0x5b, 0xd8, - 0x3c, 0x12, 0x91, 0x2c, 0x75, 0x0c, 0xd2, 0x68, 0xa0, 0xc5, 0x34, 0x41, 0xa8, 0x0e, 0xd0, 0x44, - 0xb2, 0x52, 0x44, 0xa2, 0xb8, 0x0c, 0x65, 0xf4, 0x67, 0x0b, 0x20, 0x03, 0x34, 0x49, 0x73, 0x15, - 0xeb, 0x7c, 0x15, 0xd2, 0xcd, 0x2a, 0x6f, 0xa6, 0xa5, 0x5d, 0xdc, 0xd2, 0x7c, 0x96, 0x75, 0xc6, - 0x07, 0x45, 0x67, 0xe8, 0x92, 0xde, 0x9c, 0xeb, 0x0c, 0xed, 0xb5, 0xe8, 0x8f, 0x17, 0xd0, 0x2c, - 0xf1, 0x17, 0x76, 0xc9, 0xfb, 0x79, 0x97, 0xd8, 0xf3, 0x26, 0x91, 0x6f, 0x4c, 0x66, 0xbd, 0xf2, - 0x1c, 0x9a, 0x25, 0xf6, 0x42, 0x8b, 0x5d, 0xb8, 0x36, 0x3b, 0x87, 0xd9, 0x7e, 0x9f, 0x67, 0xd3, - 0x00, 0x5a, 0x1b, 0xe1, 0x24, 0x95, 0x22, 0x31, 0xe6, 0xd4, 0xa3, 0xa0, 0x19, 0x79, 0xf1, 0x0a, - 0xc6, 0xe2, 0xfa, 0x91, 0xbb, 0xb0, 0xa4, 0xd2, 0xa8, 0xc7, 0xe9, 0x6c, 0x8e, 0xb5, 0x90, 0xee, - 0x42, 0xbd, 0x37, 0xf0, 0x9e, 0x25, 0xf1, 0x64, 0xbc, 0x10, 0x74, 0xf6, 0xe1, 0x63, 0x97, 0x3e, - 0x7c, 0xda, 0xfa, 0xd1, 0x77, 0xf0, 0x1d, 0xc6, 0x17, 0xbe, 0xad, 0x5f, 0xf8, 0x8a, 0xe1, 0x70, - 0xb5, 0x7f, 0x57, 0xf5, 0xaa, 0x54, 0x53, 0x7c, 0x95, 0x85, 0x93, 0x3d, 0xba, 0x4e, 0xf1, 0xe8, - 0x2a, 0xa3, 0x7a, 0x9f, 0xfd, 0x9f, 0x46, 0x7f, 0xb5, 0x61, 0x95, 0x89, 0x34, 0x78, 0x25, 0xbc, - 0x28, 0x95, 0xc9, 0xc4, 0x57, 0x3b, 0x49, 0xe9, 0x7f, 0x15, 0xef, 0x99, 0x6c, 0x3b, 0x4c, 0x13, - 0x97, 0xe9, 0x74, 0xf2, 0x00, 0x9a, 0xf3, 0x33, 0x7b, 0xf6, 0x6a, 0xf9, 0x0a, 0x79, 0x00, 0xb5, - 0x41, 0x3c, 0x49, 0xfc, 0xbc, 0x7d, 0x4b, 0x7b, 0x52, 0x23, 0xd3, 0x62, 0x96, 0x5d, 0x23, 0x1f, - 0x96, 0x87, 0xc9, 0xad, 0xa1, 0x8b, 0x1b, 0xb3, 0x2e, 0x4c, 0x7f, 0x96, 0x87, 0xee, 0xf1, 0x5c, - 0x5b, 0xb9, 0x55, 0x54, 0x7c, 0xbd, 0x50, 0x9c, 0x11, 0xb3, 0xd9, 0xdb, 0xf4, 0x27, 0x0b, 0x96, - 0xcb, 0x70, 0x2e, 0x35, 0xc4, 0x79, 0x75, 0xec, 0x8b, 0x5f, 0xfd, 0xac, 0x3a, 0x95, 0x45, 0xdf, - 0x59, 0x4b, 0xe5, 0x2f, 0x81, 0x43, 0xb8, 0x7d, 0xa6, 0x64, 0x1b, 0xf1, 0x68, 0xac, 0x7a, 0xe3, - 0x3f, 0x94, 0x4e, 0xad, 0xb7, 0x24, 0x31, 0x45, 0x6b, 0x30, 0x4d, 0xd0, 0x4f, 0xe0, 0xe6, 0x40, - 0xc8, 0x52, 0xc1, 0xb2, 0xce, 0xeb, 0x80, 0xb3, 0x2d, 0x8e, 0xcf, 0x09, 0x5f, 0x89, 0xe8, 0xe7, - 0xe0, 0xbe, 0x1c, 0x0f, 0xb9, 0x14, 0x57, 0xd2, 0xee, 0x41, 0x7d, 0x27, 0x1e, 0xc7, 0x61, 0xbc, - 0x3f, 0xbd, 0x60, 0x03, 0xb8, 0x50, 0xd3, 0xbb, 0x5c, 0xaf, 0x94, 0x06, 0xcb, 0x48, 0x7a, 0x5d, - 0x35, 0xb7, 0xcf, 0x43, 0x7f, 0x12, 0x2a, 0x18, 0xea, 0xdb, 0x31, 0xed, 0xb5, 0x7f, 0x3b, 0x5d, - 0xb3, 0x7e, 0x3f, 0x5d, 0xb3, 0xfe, 0x38, 0x5d, 0xb3, 0x7e, 0xf9, 0x73, 0xed, 0xb5, 0xbd, 0x2a, - 0xfe, 0x4a, 0x3d, 0xfa, 0x3b, 0x00, 0x00, 0xff, 0xff, 0x50, 0x3d, 0x18, 0x31, 0x5b, 0x0d, 0x00, - 0x00, + // 1287 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xdb, 0x72, 0xdb, 0xc4, + 0x1b, 0xff, 0x4b, 0x72, 0x62, 0xfb, 0x73, 0x9c, 0x3a, 0xdb, 0x93, 0xda, 0x3f, 0x13, 0xcc, 0xd2, + 0xa1, 0xa6, 0x33, 0x84, 0x4e, 0x0b, 0x33, 0x9c, 0x3a, 0x53, 0x12, 0xa7, 0xc5, 0x94, 0x84, 0x76, + 0x9d, 0xe6, 0x8e, 0x8b, 0xad, 0xbc, 0xd3, 0x68, 0x22, 0x4b, 0x46, 0x5a, 0xe5, 0xd0, 0x0b, 0x6e, + 0x61, 0x86, 0x17, 0xe0, 0x91, 0xb8, 0xe4, 0x11, 0x98, 0xf2, 0x02, 0xdc, 0x72, 0xc7, 0xec, 0xb7, + 0xbb, 0x3a, 0xb8, 0x09, 0x29, 0x29, 0x77, 0xfa, 0xce, 0xbf, 0xfd, 0x4e, 0xbb, 0x82, 0xee, 0x2c, + 0x0d, 0x0f, 0xb8, 0x14, 0x6b, 0xb3, 0x34, 0x91, 0x09, 0x69, 0x85, 0xb1, 0x14, 0x69, 0xcc, 0x23, + 0xfa, 0x10, 0xda, 0xa3, 0x78, 0x22, 0x8e, 0xb6, 0x84, 0xe4, 0x84, 0x40, 0xe3, 0x91, 0x38, 0xce, + 0x7c, 0xaf, 0xef, 0x0c, 0x5a, 0x0c, 0xbf, 0xc9, 0x7b, 0xb0, 0xbc, 0x93, 0xf2, 0x60, 0x7f, 0xf3, + 0x28, 0xcc, 0xa4, 0x88, 0x03, 0xe1, 0x37, 0x50, 0x3a, 0xc7, 0xa5, 0x7f, 0xba, 0xb0, 0xf4, 0x20, + 0x14, 0xd1, 0xe4, 0xdb, 0x99, 0x0c, 0x93, 0x38, 0x53, 0xce, 0x76, 0x8e, 0x67, 0xc2, 0x6f, 0xf5, + 0x9d, 0x41, 0x9b, 0xe1, 0x37, 0x79, 0x0b, 0xda, 0x1b, 0x3c, 0xd8, 0x13, 0x28, 0xf0, 0x50, 0x50, + 0x32, 0x0a, 0xe9, 0x38, 0x7c, 0xa1, 0xa3, 0x74, 0x59, 0xc9, 0x20, 0x7d, 0xe8, 0xec, 0x84, 0x53, + 0xf1, 0x24, 0xe7, 0xb1, 0xcc, 0xa7, 0xfe, 0x02, 0x5a, 0x57, 0x59, 0x05, 0xfc, 0x4e, 0x1d, 0xfe, + 0x76, 0x32, 0x96, 0x3c, 0x9e, 0xf0, 0x74, 0xb2, 0x1b, 0x8a, 0x43, 0x7f, 0x49, 0xc3, 0xaf, 0x73, + 0x95, 0xed, 0x3a, 0xcf, 0x84, 0xdf, 0xed, 0x3b, 0x03, 0x8f, 0xe1, 0x37, 0xb9, 0x0e, 0xad, 0xf5, + 0x50, 0x0e, 0xc5, 0x4c, 0xee, 0xf9, 0xcb, 0x7d, 0x67, 0xd0, 0x60, 0x05, 0x4d, 0x2e, 0xc1, 0xc2, + 0x38, 0xe0, 0x91, 0xf0, 0x2f, 0xa0, 0x81, 0x26, 0x08, 0x85, 0xa5, 0x07, 0x49, 0x2a, 0xc2, 0xe7, + 0x31, 0x26, 0xd5, 0xef, 0x21, 0xc8, 0x1a, 0x8f, 0xbc, 0x0b, 0xde, 0x56, 0x18, 0xfb, 0x2b, 0x7d, + 0x67, 0xd0, 0xb9, 0xb3, 0xb2, 0x66, 0x2b, 0xb1, 0x36, 0x14, 0x41, 0x38, 0xe5, 0x11, 0x53, 0x52, + 0x54, 0xe2, 0x47, 0x3e, 0x39, 0x5d, 0x89, 0x1f, 0xd1, 0x8f, 0xa1, 0x69, 0x68, 0x05, 0x67, 0x97, + 0x47, 0xb9, 0xf0, 0x1d, 0x0d, 0x07, 0x89, 0x12, 0xa4, 0x5b, 0x01, 0x49, 0x29, 0x2c, 0x8f, 0xa6, + 0xb3, 0x24, 0x95, 0x4c, 0x64, 0xb3, 0x24, 0xce, 0x04, 0xe9, 0x81, 0xb7, 0x99, 0xa6, 0x68, 0xdb, + 0x66, 0xea, 0x93, 0xfe, 0x00, 0xbd, 0xf5, 0x28, 0x09, 0xf6, 0x87, 0x5c, 0x72, 0x26, 0xbe, 0xcf, + 0x45, 0x26, 0x95, 0x37, 0x7d, 0x2a, 0xad, 0xa7, 0x09, 0xc5, 0xc5, 0xb2, 0x63, 0x8c, 0x36, 0xd3, + 0x84, 0x4a, 0x27, 0x26, 0x5b, 0x57, 0x09, 0xbf, 0x11, 0xcd, 0x1e, 0x4f, 0x27, 0x58, 0xda, 0x06, + 0xd3, 0x84, 0xe2, 0x62, 0x24, 0x6c, 0x87, 0x06, 0xd3, 0x04, 0x1d, 0xc1, 0x4a, 0x25, 0xbe, 0x81, + 0x79, 0x05, 0x16, 0x59, 0x72, 0x38, 0x1a, 0x66, 0xbe, 0xd3, 0xf7, 0x06, 0x0d, 0x66, 0x28, 0xec, + 0x9b, 0x24, 0xca, 0xa7, 0xb1, 0x12, 0xb9, 0x28, 0x2a, 0x19, 0xf4, 0x1a, 0x2c, 0x60, 0x13, 0xa9, + 0x53, 0x96, 0xb6, 0xea, 0x93, 0xfe, 0xe8, 0x40, 0x7b, 0x8b, 0x1f, 0x21, 0x90, 0x8c, 0xdc, 0x83, + 0x96, 0x6d, 0x09, 0x54, 0xea, 0xdc, 0x79, 0xa7, 0x4c, 0x7c, 0xa1, 0xb6, 0x66, 0x75, 0x36, 0x63, + 0x99, 0x1e, 0xb3, 0xc2, 0xe4, 0xfa, 0xe7, 0xd0, 0xad, 0x89, 0x54, 0xbc, 0x7d, 0x71, 0x6c, 0xb3, + 0xba, 0x2f, 0x8e, 0xd5, 0x59, 0x0f, 0xb0, 0x4a, 0xae, 0x3e, 0x2b, 0x12, 0x9f, 0xb9, 0x9f, 0x38, + 0x74, 0x17, 0xc8, 0x46, 0x2a, 0xb8, 0x14, 0x18, 0x64, 0x4b, 0x64, 0x19, 0x7f, 0x2e, 0xce, 0xca, + 0xb8, 0x57, 0xcd, 0x78, 0x91, 0x5d, 0xb7, 0x92, 0x5d, 0x7a, 0x0b, 0xc8, 0x50, 0x44, 0x42, 0x0a, + 0x33, 0xe4, 0xff, 0xe0, 0x97, 0x8e, 0x2d, 0x86, 0xb3, 0x75, 0xc9, 0x4d, 0x68, 0xa8, 0x8d, 0x81, + 0xc1, 0x3a, 0x77, 0x2e, 0x96, 0x79, 0x2a, 0x96, 0x09, 0x43, 0x05, 0x1a, 0x59, 0xa7, 0x88, 0xf2, + 0x35, 0x0f, 0x56, 0x6b, 0xa5, 0x5b, 0x26, 0x94, 0x87, 0xa1, 0xae, 0x94, 0xa1, 0xaa, 0xdb, 0xc6, + 0x44, 0xbb, 0x6f, 0x8f, 0x7b, 0xde, 0x68, 0x34, 0x80, 0xff, 0x6b, 0x0f, 0x5f, 0x1e, 0xf0, 0x30, + 0xe2, 0xcf, 0xa2, 0x7f, 0x55, 0x91, 0x1a, 0x70, 0x1f, 0x9a, 0x68, 0x3b, 0x1a, 0x9a, 0xde, 0xb6, + 0x24, 0xfd, 0x0e, 0xca, 0x31, 0xd9, 0xe6, 0x53, 0x61, 0xbc, 0xe1, 0x77, 0x71, 0x5e, 0xf7, 0xec, + 0xf3, 0xe2, 0xd8, 0x87, 0xe2, 0x50, 0x6d, 0x6c, 0x4f, 0x05, 0x46, 0x82, 0xde, 0x85, 0xc5, 0x71, + 0xb0, 0x27, 0xa6, 0x9c, 0xbc, 0x0f, 0x4d, 0x44, 0x28, 0x32, 0xd3, 0xd1, 0x17, 0xe6, 0x2a, 0xc5, + 0xac, 0x9c, 0x66, 0xe6, 0x64, 0x27, 0x62, 0xfa, 0x00, 0x9a, 0x26, 0x30, 0x4e, 0xf4, 0x29, 0x15, + 0xb7, 0x3a, 0xe4, 0x26, 0x2c, 0x22, 0xd8, 0xcc, 0x6f, 0xcc, 0x47, 0x45, 0x3e, 0x33, 0x62, 0xba, + 0x09, 0xde, 0x53, 0x36, 0x52, 0x83, 0x8d, 0x80, 0x6d, 0x50, 0x43, 0x29, 0x28, 0x5f, 0x25, 0x99, + 0x34, 0x69, 0xc5, 0x6f, 0xc5, 0x7b, 0x9c, 0xa4, 0x12, 0x53, 0xda, 0x65, 0xf8, 0x4d, 0x33, 0x68, + 0x6c, 0x27, 0x13, 0x41, 0x96, 0xc1, 0x1d, 0x0d, 0x8d, 0x0f, 0x77, 0x34, 0x24, 0x6f, 0xa3, 0x7b, + 0x93, 0xc9, 0x6e, 0x09, 0xe2, 0x29, 0x1b, 0x31, 0x0c, 0x7c, 0x03, 0xba, 0xa3, 0x6c, 0x23, 0x49, + 0xd2, 0x49, 0x18, 0x73, 0x99, 0xa4, 0xe6, 0xe6, 0xab, 0x33, 0x71, 0xb4, 0x24, 0x97, 0xfa, 0x4e, + 0x6a, 0x33, 0x4d, 0xd0, 0xfb, 0xd0, 0x53, 0x41, 0x91, 0xb0, 0xed, 0x71, 0x05, 0x16, 0x15, 0xaf, + 0x00, 0x61, 0xa8, 0xd2, 0x83, 0x5b, 0xf5, 0xf0, 0x8d, 0xf6, 0xb0, 0x79, 0x20, 0x62, 0x59, 0x69, + 0x30, 0xa4, 0xd1, 0x41, 0x97, 0x69, 0x82, 0x50, 0x7d, 0x40, 0x73, 0x92, 0xe5, 0xf2, 0x24, 0x8a, + 0xcb, 0x50, 0x46, 0x7f, 0x76, 0x00, 0x2c, 0xa0, 0x3c, 0x2b, 0x4c, 0x9c, 0xd3, 0x4d, 0xc8, 0xc0, + 0x36, 0x8a, 0x19, 0xae, 0x5e, 0xa9, 0xa5, 0xf9, 0xcc, 0x36, 0xd2, 0x87, 0x65, 0x23, 0xe9, 0x92, + 0x5e, 0x9e, 0x6b, 0x00, 0x1d, 0xb5, 0x6c, 0xa7, 0xc7, 0xd0, 0xa9, 0xf0, 0x4f, 0x69, 0x2a, 0xdb, + 0x25, 0xee, 0xbc, 0x4b, 0xe4, 0x1b, 0x97, 0xb6, 0x57, 0x1e, 0x41, 0xa7, 0xc2, 0x3e, 0xd1, 0xe3, + 0x00, 0x2e, 0xd4, 0xc7, 0xd6, 0x5e, 0x07, 0xf3, 0x6c, 0x1a, 0x42, 0x77, 0x23, 0xca, 0x33, 0x29, + 0x52, 0xe3, 0x4e, 0xdd, 0x21, 0x9a, 0x51, 0x14, 0xaf, 0x64, 0x9c, 0x5c, 0x3f, 0x72, 0x03, 0x16, + 0x54, 0x1a, 0xf5, 0xf4, 0xbd, 0x9a, 0x63, 0x2d, 0xa4, 0xbb, 0xd0, 0x5a, 0x1f, 0x8f, 0x1e, 0xa6, + 0x49, 0x3e, 0x3b, 0x11, 0xb4, 0x7d, 0x27, 0xb9, 0x95, 0x77, 0x52, 0x4f, 0xbf, 0x11, 0x3c, 0xbc, + 0xb6, 0xf1, 0x41, 0xd0, 0xd3, 0x0f, 0x82, 0x86, 0xe1, 0x70, 0xb5, 0xae, 0x57, 0xf4, 0x66, 0x55, + 0x43, 0x7f, 0x9e, 0xfd, 0x64, 0xef, 0x68, 0xaf, 0xbc, 0xa3, 0x95, 0x53, 0xbd, 0xfe, 0xfe, 0x4b, + 0xa7, 0x7f, 0xb9, 0xb0, 0xc2, 0x44, 0x16, 0xbe, 0x10, 0xa3, 0x38, 0x93, 0x69, 0x1e, 0xa8, 0x2d, + 0xa1, 0xec, 0xbf, 0x4e, 0x9e, 0x99, 0x6c, 0x7b, 0x4c, 0x13, 0xaf, 0xd3, 0xe9, 0xe4, 0x36, 0x74, + 0xe6, 0x67, 0xf6, 0x55, 0xd5, 0xaa, 0x0a, 0xb9, 0x0d, 0xcd, 0x71, 0x92, 0xa7, 0x41, 0xd1, 0xbe, + 0x95, 0xb5, 0xaa, 0x91, 0x69, 0x31, 0xb3, 0x6a, 0xe4, 0x09, 0x90, 0x9d, 0x94, 0xc7, 0x59, 0xc4, + 0x15, 0x58, 0x6b, 0xdc, 0x9a, 0x7f, 0x16, 0x54, 0x74, 0x6a, 0x7e, 0x4e, 0x30, 0x26, 0x1f, 0x55, + 0xe7, 0xd3, 0x6f, 0x22, 0xea, 0x4b, 0x75, 0xd4, 0xa6, 0xe5, 0xab, 0x73, 0x7c, 0x6f, 0xae, 0x53, + 0xfd, 0x45, 0x34, 0xbc, 0x5a, 0x1a, 0xd6, 0xc4, 0xac, 0xae, 0x4d, 0x7f, 0x72, 0x60, 0xa9, 0x8a, + 0xec, 0xb5, 0xf6, 0x42, 0x51, 0x70, 0xf7, 0xec, 0x77, 0x87, 0x2d, 0x78, 0xe3, 0xa4, 0x97, 0xde, + 0x42, 0xf5, 0x2d, 0x92, 0xc3, 0xd5, 0x53, 0xd2, 0xf5, 0x06, 0xa0, 0xfa, 0xd0, 0x79, 0xcc, 0x53, + 0x19, 0x2a, 0x97, 0xe6, 0xa2, 0x5d, 0x60, 0x55, 0x16, 0xdd, 0x87, 0x6b, 0xaf, 0x34, 0xdf, 0x46, + 0x32, 0x9d, 0xa9, 0x2e, 0x7f, 0x83, 0x26, 0x54, 0x8b, 0x3a, 0x4d, 0x4d, 0xfb, 0xb5, 0x99, 0x26, + 0xe8, 0xa7, 0x70, 0x79, 0x2c, 0x64, 0xa5, 0xf5, 0xec, 0x0c, 0xf5, 0xc1, 0xdb, 0x16, 0x87, 0xa7, + 0x1c, 0x50, 0x89, 0xe8, 0x17, 0xe0, 0x3f, 0x9d, 0x4d, 0xb8, 0x14, 0xe7, 0xb2, 0x5e, 0x87, 0xd6, + 0x4e, 0x32, 0x4b, 0xa2, 0xe4, 0xf9, 0xf1, 0x19, 0xbb, 0xcc, 0x87, 0xa6, 0xbe, 0x95, 0xf4, 0x72, + 0x6c, 0x33, 0x4b, 0xd2, 0x8b, 0x6a, 0x4c, 0x03, 0x1e, 0x05, 0x79, 0xa4, 0x60, 0xa8, 0x47, 0x73, + 0xb6, 0xde, 0xfb, 0xf5, 0xe5, 0xaa, 0xf3, 0xdb, 0xcb, 0x55, 0xe7, 0xf7, 0x97, 0xab, 0xce, 0x2f, + 0x7f, 0xac, 0xfe, 0xef, 0xd9, 0x22, 0xfe, 0x43, 0xde, 0xfd, 0x3b, 0x00, 0x00, 0xff, 0xff, 0xa0, + 0x67, 0xe6, 0x48, 0x54, 0x0e, 0x00, 0x00, } func (m *IndexMeta) Marshal() (dAtA []byte, err error) { @@ -3109,6 +3192,18 @@ func (m *Index) MarshalToSizedBuffer(dAtA []byte) (int, error) { i -= len(m.XXX_unrecognized) copy(dAtA[i:], m.XXX_unrecognized) } + if m.Options != nil { + { + size, err := m.Options.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintPrivate(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x2a + } if len(m.Fields) > 0 { for iNdEx := len(m.Fields) - 1; iNdEx >= 0; iNdEx-- { { @@ -3465,20 +3560,20 @@ func (m *FieldStatus) MarshalToSizedBuffer(dAtA []byte) (int, error) { copy(dAtA[i:], m.XXX_unrecognized) } if len(m.AvailableShards) > 0 { - dAtA17 := make([]byte, len(m.AvailableShards)*10) - var j16 int + dAtA18 := make([]byte, len(m.AvailableShards)*10) + var j17 int for _, num := range m.AvailableShards { for num >= 1<<7 { - dAtA17[j16] = uint8(uint64(num)&0x7f | 0x80) + dAtA18[j17] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 - j16++ + j17++ } - dAtA17[j16] = uint8(num) - j16++ + dAtA18[j17] = uint8(num) + j17++ } - i -= j16 - copy(dAtA[i:], dAtA17[:j16]) - i = encodeVarintPrivate(dAtA, i, uint64(j16)) + i -= j17 + copy(dAtA[i:], dAtA18[:j17]) + i = encodeVarintPrivate(dAtA, i, uint64(j17)) i-- dAtA[i] = 0x12 } @@ -3718,6 +3813,20 @@ func (m *ResizeInstruction) MarshalToSizedBuffer(dAtA []byte) (int, error) { i -= len(m.XXX_unrecognized) copy(dAtA[i:], m.XXX_unrecognized) } + if len(m.TranslationSources) > 0 { + for iNdEx := len(m.TranslationSources) - 1; iNdEx >= 0; iNdEx-- { + { + size, err := m.TranslationSources[iNdEx].MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintPrivate(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0x42 + } + } if m.NodeStatus != nil { { size, err := m.NodeStatus.MarshalToSizedBuffer(dAtA[:i]) @@ -3853,6 +3962,57 @@ func (m *ResizeSource) MarshalToSizedBuffer(dAtA []byte) (int, error) { return len(dAtA) - i, nil } +func (m *TranslationResizeSource) 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 *TranslationResizeSource) MarshalTo(dAtA []byte) (int, error) { + size := m.Size() + return m.MarshalToSizedBuffer(dAtA[:size]) +} + +func (m *TranslationResizeSource) 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) + } + if m.PartitionID != 0 { + i = encodeVarintPrivate(dAtA, i, uint64(m.PartitionID)) + i-- + dAtA[i] = 0x18 + } + if len(m.Index) > 0 { + i -= len(m.Index) + copy(dAtA[i:], m.Index) + i = encodeVarintPrivate(dAtA, i, uint64(len(m.Index))) + i-- + dAtA[i] = 0x12 + } + if m.Node != nil { + { + size, err := m.Node.MarshalToSizedBuffer(dAtA[:i]) + if err != nil { + return 0, err + } + i -= size + i = encodeVarintPrivate(dAtA, i, uint64(size)) + } + i-- + dAtA[i] = 0xa + } + return len(dAtA) - i, nil +} + func (m *ResizeInstructionComplete) Marshal() (dAtA []byte, err error) { size := m.Size() dAtA = make([]byte, size) @@ -4450,6 +4610,10 @@ func (m *Index) Size() (n int) { n += 1 + l + sovPrivate(uint64(l)) } } + if m.Options != nil { + l = m.Options.Size() + n += 1 + l + sovPrivate(uint64(l)) + } if m.XXX_unrecognized != nil { n += len(m.XXX_unrecognized) } @@ -4747,6 +4911,12 @@ func (m *ResizeInstruction) Size() (n int) { l = m.NodeStatus.Size() n += 1 + l + sovPrivate(uint64(l)) } + if len(m.TranslationSources) > 0 { + for _, e := range m.TranslationSources { + l = e.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + } if m.XXX_unrecognized != nil { n += len(m.XXX_unrecognized) } @@ -4784,6 +4954,29 @@ func (m *ResizeSource) Size() (n int) { return n } +func (m *TranslationResizeSource) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if m.Node != nil { + l = m.Node.Size() + n += 1 + l + sovPrivate(uint64(l)) + } + l = len(m.Index) + if l > 0 { + n += 1 + l + sovPrivate(uint64(l)) + } + if m.PartitionID != 0 { + n += 1 + sovPrivate(uint64(m.PartitionID)) + } + if m.XXX_unrecognized != nil { + n += len(m.XXX_unrecognized) + } + return n +} + func (m *ResizeInstructionComplete) Size() (n int) { if m == nil { return 0 @@ -7303,6 +7496,42 @@ func (m *Index) Unmarshal(dAtA []byte) error { return err } iNdEx = postIndex + case 5: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Options", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Options == nil { + m.Options = &IndexMeta{} + } + if err := m.Options.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex default: iNdEx = preIndex skippy, err := skipPrivate(dAtA[iNdEx:]) @@ -9142,6 +9371,40 @@ func (m *ResizeInstruction) Unmarshal(dAtA []byte) error { return err } iNdEx = postIndex + case 8: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field TranslationSources", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.TranslationSources = append(m.TranslationSources, &TranslationResizeSource{}) + if err := m.TranslationSources[len(m.TranslationSources)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex default: iNdEx = preIndex skippy, err := skipPrivate(dAtA[iNdEx:]) @@ -9372,6 +9635,147 @@ func (m *ResizeSource) Unmarshal(dAtA []byte) error { } return nil } +func (m *TranslationResizeSource) 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: TranslationResizeSource: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: TranslationResizeSource: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Node", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Node == nil { + m.Node = &Node{} + } + if err := m.Node.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Index", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return ErrInvalidLengthPrivate + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Index = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field PartitionID", wireType) + } + m.PartitionID = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.PartitionID |= int32(b&0x7F) << shift + if b < 0x80 { + break + } + } + 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 *ResizeInstructionComplete) Unmarshal(dAtA []byte) error { l := len(dAtA) iNdEx := 0 diff --git a/internal/private.proto b/internal/private.proto index 23b6ba279..df4365c06 100644 --- a/internal/private.proto +++ b/internal/private.proto @@ -96,6 +96,7 @@ message Schema { message Index { string Name = 1; + IndexMeta Options = 5; repeated Field Fields = 4; } @@ -168,6 +169,7 @@ message ResizeInstruction { Node Node = 2; Node Coordinator = 3; repeated ResizeSource Sources = 4; + repeated TranslationResizeSource TranslationSources = 8; NodeStatus NodeStatus = 7; ClusterStatus ClusterStatus = 6; } @@ -180,6 +182,12 @@ message ResizeSource { uint64 Shard = 5; } +message TranslationResizeSource { + Node Node = 1; + string Index = 2; + int32 PartitionID = 3; +} + message ResizeInstructionComplete { int64 JobID = 1; Node Node = 2; diff --git a/mock/translator.go b/mock/translator.go index 9de9fb444..baf7de524 100644 --- a/mock/translator.go +++ b/mock/translator.go @@ -16,6 +16,7 @@ package mock import ( "context" + "io" "github.com/pilosa/pilosa/v2" ) @@ -80,6 +81,16 @@ func (s *TranslateStore) EntryReader(ctx context.Context, offset uint64) (pilosa return s.EntryReaderFunc(ctx, offset) } +// TODO: implement this +func (s *TranslateStore) WriteTo(w io.Writer) (int64, error) { + return 0, nil +} + +// TODO: implement this +func (s *TranslateStore) ReadFrom(r io.Reader) (int64, error) { + return 0, nil +} + var _ pilosa.TranslateEntryReader = (*TranslateEntryReader)(nil) type TranslateEntryReader struct { diff --git a/pilosa.go b/pilosa.go index 369b72998..86ecd303a 100644 --- a/pilosa.go +++ b/pilosa.go @@ -67,6 +67,7 @@ var ( ErrNodeIDNotExists = errors.New("node with provided ID does not exist") ErrNodeNotCoordinator = errors.New("node is not the coordinator") ErrResizeNotRunning = errors.New("no resize job currently running") + ErrResizeNoReplicas = errors.New("not enough data to perform resize (replica factor may need to be increased)") ErrNotImplemented = errors.New("not implemented") ErrFieldsArgumentRequired = errors.New("fields argument required") diff --git a/server/cluster_test.go b/server/cluster_test.go index b9fe66903..b05fd752e 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -491,6 +491,71 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { t.Fatalf("unexpected result: %s", res) } }) + t.Run("WithIndexKeys", func(t *testing.T) { + // Configure node0 + m0 := test.MustRunCluster(t, 1)[0] + defer m0.Close() + + seed := m0.GossipAddress() + + // Create a client for each node. + client0 := m0.Client() + + // Create indexes and fields on one node. + if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{Keys: true}); err != nil && err != pilosa.ErrIndexExists { + t.Fatal(err) + } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { + t.Fatal(err) + } + + // Write data on first node. + if _, err := m0.Query(t, "i", "", ` + Set('col1', f=1) + Set('col2', f=1) + `); err != nil { + t.Fatal(err) + } + + // exp is the expected result for the Row queries that follow. + exp := `{"results":[{"attrs":{},"columns":[],"keys":["col2","col1"]}]}` + "\n" + + // Verify the data exists on the single node. + if res, err := m0.Query(t, "i", "", `Row(f=1)`); err != nil { + t.Fatal(err) + } else if res != exp { + t.Fatalf("unexpected result: %s", res) + } + + // Configure node1 + m1 := test.NewCommandNode(false) + m1.Config.Gossip.Port = "0" + m1.Config.Gossip.Seeds = []string{seed} + errc := make(chan error, 1) + go func() { + _, err := m0.API.CreateIndex(context.Background(), "blah", pilosa.IndexOptions{}) + errc <- err + }() + err := m1.Start() + if err != nil { + t.Fatalf("starting second main: %v", err) + } + defer m1.Close() + + if !checkClusterState(m0, pilosa.ClusterStateNormal, 1000) { + t.Fatalf("unexpected node0 cluster state: %s", m0.API.State()) + } else if !checkClusterState(m1, pilosa.ClusterStateNormal, 1000) { + t.Fatalf("unexpected node1 cluster state: %s", m1.API.State()) + } + + // Verify the data exists on both nodes. + for i, node := range []*test.Command{m0, m1} { + if res, err := node.Query(t, "i", "", `Row(f=1)`); err != nil { + t.Fatal(err) + } else if res != exp { + t.Fatalf("node%d expected: %s, but got: %s", i, exp, res) + } + } + }) } // Ensure that redundant gossip seeds are used diff --git a/translate.go b/translate.go index b5f469f2f..6088774c2 100644 --- a/translate.go +++ b/translate.go @@ -77,6 +77,15 @@ type TranslateStore interface { // Returns a reader from the given ID offset. EntryReader(ctx context.Context, offset uint64) (TranslateEntryReader, error) + + // WriteTo ensures that the TranslateStore implements io.WriterTo. + // It should write the contents of the store to the writer. + WriteTo(io.Writer) (int64, error) + + // ReadFrom ensures that the TranslateStore implements io.ReaderFrom. + // It should read from the reader and replace the data store with + // the read payload. + ReadFrom(io.Reader) (int64, error) } // OpenTranslateStoreFunc represents a function for instantiating and opening a TranslateStore. @@ -405,6 +414,18 @@ func (s *InMemTranslateStore) EntryReader(ctx context.Context, offset uint64) (T return newInMemTranslateEntryReader(ctx, s, offset), nil } +// TODO: implement this +// WriteTo writes the contents of the store to the writer. +func (s *InMemTranslateStore) WriteTo(w io.Writer) (int64, error) { + return 0, nil +} + +// TODO: implement this +// ReadFrom ensures that the TranslateStore implements io.ReaderFrom. +func (s *InMemTranslateStore) ReadFrom(r io.Reader) (int64, error) { + return 0, nil +} + // MaxID returns the highest identifier in the store. func (s *InMemTranslateStore) MaxID() (uint64, error) { s.mu.RLock()