From f1a460aca7f85d0d86e5a6367ec4e2060e77686b Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Thu, 16 Aug 2018 11:27:08 -0500 Subject: [PATCH] adds view parameter to sync logic for syncing time fields --- api.go | 6 ++--- client.go | 8 +++---- fragment.go | 4 ++-- holder.go | 2 +- holder_test.go | 54 +++++++++++++++++++++++++++++++++++++++++++++ http/client.go | 6 +++-- http/client_test.go | 4 ++-- http/handler.go | 4 ++-- test/holder.go | 9 ++++++-- 9 files changed, 79 insertions(+), 18 deletions(-) diff --git a/api.go b/api.go index 14ab78969..db741b652 100644 --- a/api.go +++ b/api.go @@ -391,7 +391,7 @@ func (api *API) FragmentBlockData(_ context.Context, body io.Reader) ([]byte, er } // Retrieve fragment from holder. - f := api.holder.fragment(req.Index, req.Field, viewStandard, req.Shard) + f := api.holder.fragment(req.Index, req.Field, req.View, req.Shard) if f == nil { return nil, ErrFragmentNotFound } @@ -409,13 +409,13 @@ func (api *API) FragmentBlockData(_ context.Context, body io.Reader) ([]byte, er } // FragmentBlocks returns the checksums and block ids for all blocks in the specified fragment. -func (api *API) FragmentBlocks(_ context.Context, indexName string, fieldName string, shard uint64) ([]FragmentBlock, error) { +func (api *API) FragmentBlocks(_ context.Context, indexName, fieldName, viewName string, shard uint64) ([]FragmentBlock, error) { if err := api.validate(apiFragmentBlocks); err != nil { return nil, errors.Wrap(err, "validating api method") } // Retrieve fragment from holder. - f := api.holder.fragment(indexName, fieldName, viewStandard, shard) + f := api.holder.fragment(indexName, fieldName, viewName, shard) if f == nil { return nil, ErrFragmentNotFound } diff --git a/client.go b/client.go index 9c5016646..a0b89f604 100644 --- a/client.go +++ b/client.go @@ -45,8 +45,8 @@ type InternalClient interface { ImportValueK(ctx context.Context, index, field string, vals []FieldValue) error ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error CreateField(ctx context.Context, index, field string) error - FragmentBlocks(ctx context.Context, uri *URI, index, field string, shard uint64) ([]FragmentBlock, error) - BlockData(ctx context.Context, uri *URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) + FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error) + BlockData(ctx context.Context, uri *URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) SendMessage(ctx context.Context, uri *URI, msg []byte) error @@ -122,10 +122,10 @@ func (n nopInternalClient) ExportCSV(ctx context.Context, index, field string, s return nil } func (n nopInternalClient) CreateField(ctx context.Context, index, field string) error { return nil } -func (n nopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field string, shard uint64) ([]FragmentBlock, error) { +func (n nopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error) { return nil, nil } -func (n nopInternalClient) BlockData(ctx context.Context, uri *URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) { +func (n nopInternalClient) BlockData(ctx context.Context, uri *URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) { return nil, nil, nil } func (n nopInternalClient) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) { diff --git a/fragment.go b/fragment.go index 2f57cccce..e4fbcbbda 100644 --- a/fragment.go +++ b/fragment.go @@ -1837,7 +1837,7 @@ func (s *fragmentSyncer) syncFragment() error { } // Retrieve remote blocks. - blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), &node.URI, s.Fragment.index, s.Fragment.field, s.Fragment.shard) + blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), &node.URI, s.Fragment.index, s.Fragment.field, s.Fragment.view, s.Fragment.shard) if err != nil && err != ErrFragmentNotFound { return errors.Wrap(err, "getting blocks") } @@ -1916,7 +1916,7 @@ func (s *fragmentSyncer) syncBlock(id int) error { uris = append(uris, uri) // Only sync the standard block. - rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(context.Background(), &node.URI, f.index, f.field, f.shard, id) + rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(context.Background(), &node.URI, f.index, f.field, f.view, f.shard, id) if err != nil { return errors.Wrap(err, "getting block") } diff --git a/holder.go b/holder.go index d8d3fea81..943384dc3 100644 --- a/holder.go +++ b/holder.go @@ -657,7 +657,7 @@ func (s *holderSyncer) SyncHolder() error { // Sync fragment if own it. if err := s.syncFragment(di.Name, fi.Name, vi.Name, shard); err != nil { - return fmt.Errorf("fragment sync error: index=%s, field=%s, shard=%d, err=%s", di.Name, fi.Name, shard, err) + return fmt.Errorf("fragment sync error: index=%s, field=%s, view=%s, shard=%d, err=%s", di.Name, fi.Name, vi.Name, shard, err) } } } diff --git a/holder_test.go b/holder_test.go index d4886ff30..3920e4cc7 100644 --- a/holder_test.go +++ b/holder_test.go @@ -22,6 +22,7 @@ import ( "reflect" "strings" "testing" + "time" "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/test" @@ -362,3 +363,56 @@ func TestHolderSyncer_SyncHolder(t *testing.T) { } } } + +// Ensure holder can sync time quantum views with a remote holder. +func TestHolderSyncer_TimeQuantum(t *testing.T) { + c := test.MustNewCluster(t, 2) + c[0].Config.Cluster.ReplicaN = 2 + c[0].Config.AntiEntropy.Interval = 0 + c[1].Config.Cluster.ReplicaN = 2 + c[1].Config.AntiEntropy.Interval = 0 + err := c.Start() + if err != nil { + t.Fatalf("starting cluster: %v", err) + } + defer c.Close() + + quantum := "D" + + _, err = c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}) + if err != nil { + t.Fatalf("creating index i: %v", err) + } + _, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeTime(pilosa.TimeQuantum(quantum))) + if err != nil { + t.Fatalf("creating field f: %v", err) + } + + hldr0 := &test.Holder{Holder: c[0].Server.Holder()} + hldr1 := &test.Holder{Holder: c[1].Server.Holder()} + + // Set data on the local holder. + t1 := time.Date(2018, 8, 1, 12, 30, 0, 0, time.UTC) + t2 := time.Date(2018, 8, 2, 12, 30, 0, 0, time.UTC) + hldr0.SetBitTime("i", "f", 0, 1, &t1) + hldr0.SetBitTime("i", "f", 0, 2, &t2) + + err = c[0].Server.SyncData() + if err != nil { + t.Fatalf("syncing node 0: %v", err) + } + err = c[1].Server.SyncData() + if err != nil { + t.Fatalf("syncing node 1: %v", err) + } + + // Verify data is the same on both nodes. + for i, hldr := range []*test.Holder{hldr0, hldr1} { + if a := hldr.RowTime("i", "f", 0, t1, quantum).Columns(); !reflect.DeepEqual(a, []uint64{1}) { + t.Errorf("unexpected columns(%d/0): %+v", i, a) + } + if a := hldr.RowTime("i", "f", 0, t2, quantum).Columns(); !reflect.DeepEqual(a, []uint64{2}) { + t.Errorf("unexpected columns(%d/0): %+v", i, a) + } + } +} diff --git a/http/client.go b/http/client.go index 0bfdbdddd..d996b9e57 100644 --- a/http/client.go +++ b/http/client.go @@ -688,7 +688,7 @@ func (c *InternalClient) CreateField(ctx context.Context, index, field string) e // FragmentBlocks returns a list of block checksums for a fragment on a host. // Only returns blocks which contain data. -func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64) ([]pilosa.FragmentBlock, error) { +func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64) ([]pilosa.FragmentBlock, error) { if uri == nil { uri = c.defaultURI } @@ -696,6 +696,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in u.RawQuery = url.Values{ "index": {index}, "field": {field}, + "view": {view}, "shard": {strconv.FormatUint(shard, 10)}, }.Encode() @@ -733,13 +734,14 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in } // BlockData returns row/column id pairs for a block. -func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) { +func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) { if uri == nil { panic("need to pass a URI to BlockData") } buf, err := c.serializer.Marshal(&pilosa.BlockDataRequest{ Index: index, Field: field, + View: view, Shard: shard, Block: uint64(block), }) diff --git a/http/client_test.go b/http/client_test.go index 702860c23..87c433f05 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -502,7 +502,7 @@ func TestClient_FragmentBlocks(t *testing.T) { // Set a bit on a different shard. hldr.SetBit("i", "f", 0, 1) c := MustNewClient(cmd.URL(), http.GetHTTPClient(nil)) - blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", 0) + blocks, err := c.FragmentBlocks(context.Background(), nil, "i", "f", "standard", 0) if err != nil { t.Fatal(err) } else if len(blocks) != 2 { @@ -514,7 +514,7 @@ func TestClient_FragmentBlocks(t *testing.T) { } // Verify data matches local blocks. - if a, err := cmd.API.FragmentBlocks(context.Background(), "i", "f", 0); err != nil { + if a, err := cmd.API.FragmentBlocks(context.Background(), "i", "f", "standard", 0); err != nil { t.Fatal(err) } else if !reflect.DeepEqual(a, blocks) { t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks)) diff --git a/http/handler.go b/http/handler.go index 6462f793a..0624cfb99 100644 --- a/http/handler.go +++ b/http/handler.go @@ -176,7 +176,7 @@ func (h *Handler) populateValidators() { h.validators["GetExport"] = queryValidationSpecRequired("index", "field", "shard") h.validators["GetFragmentData"] = queryValidationSpecRequired("index", "field", "shard") h.validators["PostFragmentData"] = queryValidationSpecRequired("index", "field", "shard") - h.validators["GetFragmentBlocks"] = queryValidationSpecRequired("index", "field", "shard") + h.validators["GetFragmentBlocks"] = queryValidationSpecRequired("index", "field", "view", "shard") } func (h *Handler) queryArgValidator(next http.Handler) http.Handler { @@ -1113,7 +1113,7 @@ func (h *Handler) handleGetFragmentBlocks(w http.ResponseWriter, r *http.Request return } - blocks, err := h.api.FragmentBlocks(r.Context(), q.Get("index"), q.Get("field"), shard) + blocks, err := h.api.FragmentBlocks(r.Context(), q.Get("index"), q.Get("field"), q.Get("view"), shard) if err != nil { if errors.Cause(err) == pilosa.ErrFragmentNotFound { http.Error(w, err.Error(), http.StatusNotFound) diff --git a/test/holder.go b/test/holder.go index 1d9790e2f..80778fe2a 100644 --- a/test/holder.go +++ b/test/holder.go @@ -113,14 +113,19 @@ func (h *Holder) RowTime(index, field string, rowID uint64, t time.Time, quantum return row } -// SetBit clears a bit on the given field. +// SetBit sets a bit on the given field. func (h *Holder) SetBit(index, field string, rowID, columnID uint64) { + h.SetBitTime(index, field, rowID, columnID, nil) +} + +// SetBitTime sets a bit with timestamp on the given field. +func (h *Holder) SetBitTime(index, field string, rowID, columnID uint64, t *time.Time) { idx := h.MustCreateIndexIfNotExists(index, pilosa.IndexOptions{}) f, err := idx.CreateFieldIfNotExists(field, pilosa.OptFieldTypeDefault()) if err != nil { panic(err) } - _, err = f.SetBit(rowID, columnID, nil) + _, err = f.SetBit(rowID, columnID, t) if err != nil { panic(err) }