From 2724ecfd5f51c83b4884250cc64fe4beba021718 Mon Sep 17 00:00:00 2001 From: Travis Date: Sat, 21 Mar 2020 17:31:00 -0500 Subject: [PATCH 1/4] WIP: include translate partitions in cluster resize instructions This commit adds `TranslationSources` to the cluster `ResizeInstruction`. These are the sources of translation partitions which the receiving node needs in order to support partition distribution in the new, resized cluster. This also fixes a bug where index options were not being encode in the proto Index object. That meant that the schema transferred via protobuf was not correct. The reason why things normally worked is because index creation typically happens on the CreateIndex message, which does include the options. TODO: - [ ] implement the TranslateStore interface for `InMemTranslateStore` and `mock.TranslateStore` - [ ] surely need some more tests around the `ReadFrom` and `WriteTo` --- api.go | 30 +- apimethod_string.go | 33 ++- boltdb/translate.go | 38 +++ boltdb/translate_test.go | 67 +++++ client.go | 4 + cluster.go | 194 ++++++++++-- encoding/proto/proto.go | 52 +++- holder.go | 5 +- http/client.go | 37 +++ http/handler.go | 24 +- internal/private.pb.go | 618 ++++++++++++++++++++++++++++++++------- internal/private.proto | 8 + mock/translator.go | 11 + pilosa.go | 1 + server/cluster_test.go | 65 ++++ translate.go | 21 ++ 16 files changed, 1047 insertions(+), 161 deletions(-) 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..48462d5f9 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -17,6 +17,8 @@ package boltdb import ( "bytes" "context" + "io" + "io/ioutil" "os" "path/filepath" "sync" @@ -343,6 +345,42 @@ 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) (int64, error) { + // Close store. + if err := s.Close(); err != nil { + return 0, errors.Wrap(err, "closing store") + } + + buf := bytes.NewBuffer(nil) + n, err := buf.ReadFrom(r) + if err != nil { + return n, errors.Wrap(err, "reading from reader") + } + + // Overwrite the store file. + if err := ioutil.WriteFile(s.Path, buf.Bytes(), 0666); err != nil { + return n, errors.Wrap(err, "writing file") + } + + // 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..ef289b4b1 100644 --- a/cluster.go +++ b/cluster.go @@ -864,6 +864,63 @@ 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 only need the first node in the + // list because that's the primary. 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. + if tNodes[0].ID != fNodes[i].ID { + m[tNodes[0].ID] = append(m[tNodes[0].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 +1328,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 +1487,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 +1509,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 +1523,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 +1535,32 @@ 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 + + // 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) + } + + // Get the translate store for this index/partition. + idx := c.holder.Index(src.Index) + if idx == nil { + return ErrIndexNotFound + } + + store := idx.TranslateStore(src.PartitionID) + if _, err = store.ReadFrom(rd); err != nil { + return errors.Wrap(err, "reading from reader") + } + } + return nil }(); err != nil { complete.Error = err.Error() @@ -2291,12 +2417,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 +2436,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/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/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() From 4c311aa1a789a14cf4cc17b75e83283695cc3ba5 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 25 Mar 2020 15:37:42 -0500 Subject: [PATCH 2/4] write to temp partition file. use io.Copy --- boltdb/translate.go | 31 +++++++++++++++++++++++-------- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/boltdb/translate.go b/boltdb/translate.go index 48462d5f9..cb8088674 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -18,7 +18,6 @@ import ( "bytes" "context" "io" - "io/ioutil" "os" "path/filepath" "sync" @@ -35,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) @@ -356,21 +360,32 @@ func (s *TranslateStore) WriteTo(w io.Writer) (int64, error) { } // ReadFrom reads the content and overwrites the existing store. -func (s *TranslateStore) ReadFrom(r io.Reader) (int64, error) { +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") } - buf := bytes.NewBuffer(nil) - n, err := buf.ReadFrom(r) + // Create a temporary file to snapshot to. + snapshotPath := s.Path + snapshotExt + file, err := os.Create(snapshotPath) if err != nil { - return n, errors.Wrap(err, "reading from reader") + return n, errors.Wrap(err, "creating snapshot file") } - // Overwrite the store file. - if err := ioutil.WriteFile(s.Path, buf.Bytes(), 0666); err != nil { - return n, errors.Wrap(err, "writing 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. From 98c5603965710bfa935eee32f35063b0a88b6a0e Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 25 Mar 2020 22:39:35 -0500 Subject: [PATCH 3/4] close reader. include all replias in translation partition rebalance --- cluster.go | 46 +++++++++++++++++++++++++++------------------- 1 file changed, 27 insertions(+), 19 deletions(-) diff --git a/cluster.go b/cluster.go index ef289b4b1..54f8795ab 100644 --- a/cluster.go +++ b/cluster.go @@ -886,11 +886,12 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize fNodes := c.partitionNodes(pid) tNodes := to.partitionNodes(pid) - // For `to` cluster, we only need the first node in the - // list because that's the primary. 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 + // 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 { @@ -902,12 +903,14 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize // 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. - if tNodes[0].ID != fNodes[i].ID { - m[tNodes[0].ID] = append(m[tNodes[0].ID], - &translationResizeNode{ - node: fNodes[i], - partitionID: pid, - }) + 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 @@ -1540,6 +1543,11 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { 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) @@ -1549,15 +1557,15 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { return fmt.Errorf("partition %d doesn't exist on host: %s", src.PartitionID, src.Node.URI) } - // Get the translate store for this index/partition. - idx := c.holder.Index(src.Index) - if idx == nil { - return ErrIndexNotFound - } - - store := idx.TranslateStore(src.PartitionID) - if _, err = store.ReadFrom(rd); err != nil { + // 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") } } From ac76f6227d34ec3ef46891e451749f4d770c472d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Thu, 26 Mar 2020 00:19:00 +0100 Subject: [PATCH 4/4] Make internal.IndexMeta.TrackExistence true --- go.mod | 2 +- go.sum | 1 - index.go | 13 +++++++++---- 3 files changed, 10 insertions(+), 6 deletions(-) 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/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 }