diff --git a/cluster.go b/cluster.go index eecd8385d..40193415a 100644 --- a/cluster.go +++ b/cluster.go @@ -135,63 +135,6 @@ func (c *cluster) unprotectedPrimaryNode() *disco.Node { return snap.PrimaryFieldTranslationNode() } -func (c *cluster) applySchemaWithNewShards(schema *Schema) error { - if schema == nil || len(schema.Indexes) == 0 { - return nil - } - - if err := c.holder.applySchema(schema); err != nil { - return errors.Wrap(err, "applying schema") - } - - // Get and set the shards for each field. - for _, idx := range c.holder.indexes { - for _, fld := range idx.fields { - err := fld.loadAvailableShards() - if err != nil { - return errors.Wrapf(err, "getting shards for field: %s/%s", idx.name, fld.name) - } - } - } - - return nil -} - -// unprotectedStatus returns the the cluster's status including what nodes it contains, its ID, and current state. -func (c *cluster) unprotectedStatus() (*ClusterStatus, error) { - state, err := c.noder.ClusterState(context.Background()) - if err != nil { - return nil, err - } - - indexes, err := c.holder.Schema() - if err != nil { - return nil, errors.Wrap(err, "getting schema") - } - - return &ClusterStatus{ - State: string(state), - Nodes: c.Nodes(), - Schema: &Schema{Indexes: indexes}, - }, nil -} - -func (c *cluster) remoteSchema() (*Schema, error) { - for _, n := range c.noder.Nodes() { - if c.disCo.ID() == n.ID { - continue - } - - ii, err := c.InternalClient.SchemaNode(context.Background(), &n.URI, true) - if err != nil { - return nil, errors.Wrapf(err, "getting schema from %s (%v)", n.ID, n.URI) - } - - return &Schema{ii}, nil - } - return nil, nil -} - // nodeIDs returns the list of IDs in the cluster. func (c *cluster) nodeIDs() []string { return disco.Nodes(c.Nodes()).IDs() diff --git a/executor.go b/executor.go index 4f9035834..a5026b565 100644 --- a/executor.go +++ b/executor.go @@ -6321,7 +6321,6 @@ type job struct { ctx context.Context memoryAvailable *int64 // shared, atomic value resultChan chan mapResponse - idleHands bool } // doOneJob had one job. *disappointed sigh* @@ -6341,24 +6340,6 @@ func (e *executor) doOneJob() { j.resultChan <- mapResponse{result: result, err: err} } -func (e *executor) worker(work chan job) { - for j := range work { - e.Holder.Stats.Count("job_total", 1, 0) - if j.idleHands { - return - } - // Skip out early if the context is done, but still send - // an ack so mapperLocal can be sure we aren't about to - // work on something it sent us. - if err := j.ctx.Err(); err != nil { - j.resultChan <- mapResponse{result: nil, err: err} - continue - } - result, err := j.mapFn(j.ctx, j.shard, &mapOptions{memoryAvailable: j.memoryAvailable}) - j.resultChan <- mapResponse{result: result, err: err} - } -} - var errShutdown = errors.New("executor has shut down") // mapperLocal performs map & reduce entirely on the local node. diff --git a/executor_internal_test.go b/executor_internal_test.go index 6add1b1f9..1a2cc4a45 100644 --- a/executor_internal_test.go +++ b/executor_internal_test.go @@ -592,6 +592,9 @@ func TestExecutor_DeleteRows(t *testing.T) { } changed, err = DeleteRows(ctx, row, idx, 0) + if err != nil { + t.Fatalf("deleting rows: %v", err) + } if changed { t.Fatalf("expected delete to not clear bit but it did") } diff --git a/executor_test.go b/executor_test.go index 699e542cd..9ee6f8473 100644 --- a/executor_test.go +++ b/executor_test.go @@ -7567,8 +7567,7 @@ func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) { } type testCase struct { - query string - // qrVerifier func(t *testing.T, resp pilosa.QueryResponse) + query string csvVerifier string } @@ -7602,9 +7601,6 @@ func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) { t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) { // resp := c.Query(t, "users2", tst.query) tr := c.QueryGRPC(t, "users2", tst.query) - // if tst.qrVerifier != nil { - // tst.qrVerifier(t, resp) - // } csvString, err := tableResponseToCSVString(tr) if err != nil { t.Fatal(err) @@ -7673,7 +7669,6 @@ func variousQueriesOnTimeFields(t *testing.T, c *test.Cluster) { type testCase struct { query string - qrVerifier func(t *testing.T, resp pilosa.QueryResponse) csvVerifier string } @@ -7724,11 +7719,7 @@ func variousQueriesOnTimeFields(t *testing.T, c *test.Cluster) { for i, tst := range tests { t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) { - resp := c.Query(t, "t_index", tst.query) tr := c.QueryGRPC(t, "t_index", tst.query) - if tst.qrVerifier != nil { - tst.qrVerifier(t, resp) - } csvString, err := tableResponseToCSVString(tr) if err != nil { t.Fatal(err) @@ -7795,7 +7786,6 @@ func variousQueriesOnIntFields(t *testing.T, c *test.Cluster) { type testCase struct { query string - qrVerifier func(t *testing.T, resp pilosa.QueryResponse) csvVerifier string } @@ -7813,11 +7803,7 @@ userG,-1,10,10,10 for i, tst := range tests { t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) { - resp := c.Query(t, index, tst.query) tr := c.QueryGRPC(t, index, tst.query) - if tst.qrVerifier != nil { - tst.qrVerifier(t, resp) - } csvString, err := tableResponseToCSVString(tr) if err != nil { t.Fatal(err) @@ -7888,7 +7874,6 @@ func variousQueriesOnTimestampFields(t *testing.T, c *test.Cluster) { type testCase struct { query string - qrVerifier func(t *testing.T, resp pilosa.QueryResponse) csvVerifier string } @@ -8136,7 +8121,6 @@ func variousQueriesOnLargeEpoch(t *testing.T, c *test.Cluster) { type testCase struct { query string - qrVerifier func(t *testing.T, resp pilosa.QueryResponse) csvVerifier string } diff --git a/field_internal_test.go b/field_internal_test.go index 1193d9ef1..ff4511225 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -811,7 +811,7 @@ func TestBSIGroup_TxReopenDB(t *testing.T) { } // loop // the test: can we re-open a BSI fragment under Tx store - f, err := reopenTestField(t, f) + _, err := reopenTestField(t, f) if err != nil { t.Fatalf("reopening test field: %v", err) } diff --git a/http_handler_test.go b/http_handler_test.go index 6a055cacf..541f8a22c 100644 --- a/http_handler_test.go +++ b/http_handler_test.go @@ -9,7 +9,6 @@ import ( "io" "net" "net/http" - gohttp "net/http" "os" "path" "reflect" @@ -178,13 +177,13 @@ func TestUpdateFieldTTL(t *testing.T) { t.Run(test.name, func(t *testing.T) { c.CreateField(t, indexName, pilosa.IndexOptions{}, test.name, pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMD"), test.ttl)) nodeURL := fmt.Sprintf("%s/index/%s/field/%s", c.Nodes[0].URL(), c, test.field) - req, err := gohttp.NewRequest("PATCH", nodeURL, strings.NewReader(test.ttlOption)) + req, err := http.NewRequest("PATCH", nodeURL, strings.NewReader(test.ttlOption)) if err != nil { t.Fatal(err) } req.Header.Set("Content-Type", "application/json") - resp, err := gohttp.DefaultClient.Do(req) + resp, err := http.DefaultClient.Do(req) if err != nil { t.Fatalf("doing option request: %v", err) @@ -295,13 +294,13 @@ func TestUpdateFieldNoStandardView(t *testing.T) { t.Run(test.name, func(t *testing.T) { c.CreateField(t, indexName, pilosa.IndexOptions{}, test.name, pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMD"), "0")) nodeURL := fmt.Sprintf("%s/index/%s/field/%s", c.Nodes[0].URL(), c, test.field) - req, err := gohttp.NewRequest("PATCH", nodeURL, strings.NewReader(test.fieldOption)) + req, err := http.NewRequest("PATCH", nodeURL, strings.NewReader(test.fieldOption)) if err != nil { t.Fatal(err) } req.Header.Set("Content-Type", "application/json") - resp, err := gohttp.DefaultClient.Do(req) + resp, err := http.DefaultClient.Do(req) if err != nil { t.Fatalf("doing option request: %v", err) @@ -425,12 +424,12 @@ func TestIngestSchemaHandler(t *testing.T) { m := c.GetPrimary() schemaURL := fmt.Sprintf("%s/internal/schema", m.URL()) resp := test.Do(t, "POST", schemaURL, string(schema)) - if resp.StatusCode != gohttp.StatusOK { + if resp.StatusCode != http.StatusOK { t.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } // now, try again, expecting a failure: resp = test.Do(t, "POST", schemaURL, string(schema)) - if resp.StatusCode != gohttp.StatusConflict { + if resp.StatusCode != http.StatusConflict { t.Errorf("invalid status: expected 409, got %d, body=%s", resp.StatusCode, resp.Body) } } @@ -451,7 +450,7 @@ func TestPostFieldWithTTL(t *testing.T) { m := c.GetPrimary() schemaURL := fmt.Sprintf("%s/internal/schema", m.URL()) resp := test.Do(t, "POST", schemaURL, string(schema)) - if resp.StatusCode != gohttp.StatusOK { + if resp.StatusCode != http.StatusOK { t.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } @@ -567,7 +566,7 @@ func TestGetViewAndDelete(t *testing.T) { m := c.GetPrimary() schemaURL := fmt.Sprintf("%s/internal/schema", m.URL()) resp := test.Do(t, "POST", schemaURL, string(schema)) - if resp.StatusCode != gohttp.StatusOK { + if resp.StatusCode != http.StatusOK { t.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } @@ -577,7 +576,7 @@ func TestGetViewAndDelete(t *testing.T) { Set(1,test_view=1,2001-02-03T04:05) ` respQuery := test.Do(t, "POST", postQueryUrl, string(queryOption)) - if respQuery.StatusCode != gohttp.StatusOK { + if respQuery.StatusCode != http.StatusOK { t.Errorf("posting query, status: %d, body=%s", respQuery.StatusCode, respQuery.Body) } @@ -593,7 +592,7 @@ func TestGetViewAndDelete(t *testing.T) { // Call view to get data viewUrl := fmt.Sprintf("%s/index/%s/field/test_view/view", m.URL(), c) respView := test.Do(t, "GET", viewUrl, "") - if respView.StatusCode != gohttp.StatusOK { + if respView.StatusCode != http.StatusOK { t.Errorf("view handler, status: %d, body=%s", respView.StatusCode, respView.Body) } @@ -623,7 +622,7 @@ func TestGetViewAndDelete(t *testing.T) { // call delete on view standard_2001020304 deleteViewUrl := fmt.Sprintf("%s/index/%s/field/test_view/view/standard_2001020304", m.URL(), c) respDelete := test.Do(t, "DELETE", deleteViewUrl, "") - if respDelete.StatusCode != gohttp.StatusOK { + if respDelete.StatusCode != http.StatusOK { t.Errorf("delete handler, status: %d, body=%s", respDelete.StatusCode, respDelete.Body) } @@ -633,7 +632,7 @@ func TestGetViewAndDelete(t *testing.T) { // call view again viewUrl = fmt.Sprintf("%s/index/%s/field/test_view/view", m.URL(), c) respView = test.Do(t, "GET", viewUrl, "") - if respView.StatusCode != gohttp.StatusOK { + if respView.StatusCode != http.StatusOK { t.Errorf("view handler after delete, status: %d, body=%s", respView.StatusCode, respView.Body) } @@ -684,7 +683,7 @@ func TestTranslationHandlers(t *testing.T) { m := c.GetPrimary() schemaURL := fmt.Sprintf("%s/internal/schema", m.URL()) resp := test.Do(t, "POST", schemaURL, string(schema)) - if resp.StatusCode != gohttp.StatusOK { + if resp.StatusCode != http.StatusOK { t.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } baseURLs := []string{ @@ -700,11 +699,11 @@ func TestTranslationHandlers(t *testing.T) { if expectFailure { resp := test.Do(t, "POST", findURL, names) - if resp.StatusCode != gohttp.StatusInternalServerError { + if resp.StatusCode != http.StatusInternalServerError { t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } resp = test.Do(t, "POST", createURL, names) - if resp.StatusCode != gohttp.StatusInternalServerError { + if resp.StatusCode != http.StatusInternalServerError { t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } continue @@ -712,7 +711,7 @@ func TestTranslationHandlers(t *testing.T) { // try to find them when they don't exist resp := test.Do(t, "POST", findURL, names) - if resp.StatusCode != gohttp.StatusOK { + if resp.StatusCode != http.StatusOK { t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } err := json.Unmarshal([]byte(resp.Body), &results) @@ -725,13 +724,13 @@ func TestTranslationHandlers(t *testing.T) { // try to create them, but malformed, so we expect an error resp = test.Do(t, "POST", createURL, names[:6]) - if resp.StatusCode != gohttp.StatusBadRequest { + if resp.StatusCode != http.StatusBadRequest { t.Fatalf("invalid status: expected 400, got %d, body=%s", resp.StatusCode, resp.Body) } // try to create them resp = test.Do(t, "POST", createURL, names) - if resp.StatusCode != gohttp.StatusOK { + if resp.StatusCode != http.StatusOK { t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } err = json.Unmarshal([]byte(resp.Body), &results) @@ -744,7 +743,7 @@ func TestTranslationHandlers(t *testing.T) { // try to find them now that they exist resp = test.Do(t, "POST", findURL, names) - if resp.StatusCode != gohttp.StatusOK { + if resp.StatusCode != http.StatusOK { t.Fatalf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) } err = json.Unmarshal([]byte(resp.Body), &results) @@ -904,14 +903,14 @@ admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` for _, ipTest := range IPTests { for _, test := range tests { t.Run(ipTest.TestName+"-"+test.testName, func(t *testing.T) { - var req *gohttp.Request + var req *http.Request if test.body != "" { - req, err = gohttp.NewRequest(test.method, test.url, strings.NewReader(test.body)) + req, err = http.NewRequest(test.method, test.url, strings.NewReader(test.body)) if err != nil { t.Fatal(err) } } else { - req, err = gohttp.NewRequest(test.method, test.url, nil) + req, err = http.NewRequest(test.method, test.url, nil) if err != nil { t.Fatal(err) } @@ -919,7 +918,7 @@ admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"` req.Header.Set("Content-Type", "application/json") req.Header.Set("X-Forwarded-For", ipTest.ClientIP) - resp, err := gohttp.DefaultClient.Do(req) + resp, err := http.DefaultClient.Do(req) if err != nil { t.Fatalf("failed to send request: %v", err) }