Wait for indexing to complete in Alter

dgraph/cmd/alpha/http.go

@@ -21,8 +21,6 @@ import (
 	"compress/gzip"
 	"context"
 	"encoding/json"
-	"github.com/dgraph-io/dgraph/graphql/schema"
-	"github.com/dgraph-io/dgraph/graphql/web"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -35,14 +33,15 @@ import (
 	"github.com/dgraph-io/dgo/v2/protos/api"
 	"github.com/dgraph-io/dgraph/edgraph"
 	"github.com/dgraph-io/dgraph/gql"
+	"github.com/dgraph-io/dgraph/graphql/schema"
+	"github.com/dgraph-io/dgraph/graphql/web"
 	"github.com/dgraph-io/dgraph/query"
 	"github.com/dgraph-io/dgraph/worker"
 	"github.com/dgraph-io/dgraph/x"
 
 	"github.com/golang/glog"
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/pkg/errors"
-
 	"google.golang.org/grpc/metadata"
 )
 
@@ -551,6 +550,13 @@ func alterHandler(w http.ResponseWriter, r *http.Request) {
 		op.Schema = string(b)
 	}
 
+	runInBackground, err := parseBool(r, "run_in_background")
+	if err != nil {
+		x.SetStatus(w, x.ErrorInvalidRequest, err.Error())
+		return
+	}
+	op.RunInBackground = runInBackground
+
 	glog.Infof("Got alter request via HTTP from %s\n", r.RemoteAddr)
 	fwd := r.Header.Get("X-Forwarded-For")
 	if len(fwd) > 0 {

dgraph/cmd/alpha/run_test.go

@@ -109,7 +109,7 @@ func runJSONMutation(m string) error {
 func alterSchema(s string) error {
 	for {
 		_, _, err := runWithRetries("PUT", "", addr+"/alter", s)
-		if err != nil && strings.Contains(err.Error(), "is already being modified") {
+		if err != nil && strings.Contains(err.Error(), "errIndexingInProgress") {
 			time.Sleep(time.Second)
 			continue
 		} else if err != nil {

dgraph/cmd/live/run.go

@@ -43,7 +43,6 @@ import (
 	"github.com/dgraph-io/dgo/v2/protos/api"
 
 	"github.com/dgraph-io/dgraph/chunker"
-	"github.com/dgraph-io/dgraph/testutil"
 	"github.com/dgraph-io/dgraph/types"
 	"github.com/dgraph-io/dgraph/x"
 	"github.com/dgraph-io/dgraph/xidmap"
@@ -195,11 +194,7 @@ func processSchemaFile(ctx context.Context, file string, dgraphClient *dgo.Dgrap
 
 	op := &api.Operation{}
 	op.Schema = string(b)
-	if err := dgraphClient.Alter(ctx, op); err != nil {
-		return err
-	}
-	// TODO(Aman): avoid using functions from testutil.
-	return testutil.WaitForAlter(ctx, dgraphClient, op.Schema)
+	return dgraphClient.Alter(ctx, op)
 }
 
 func (l *loader) uid(val string) string {

edgraph/server.go

@@ -88,7 +88,7 @@ var (
 )
 
 var (
-	errIndexingInProgress = errors.New("schema is already being modified. Please retry")
+	errIndexingInProgress = errors.New("errIndexingInProgress. Please retry")
 )
 
 // Server implements protos.DgraphServer
@@ -245,8 +245,17 @@ func (s *Server) Alter(ctx context.Context, op *api.Operation) (*api.Payload, er
 	}
 
 	// If a background task is already running, we should reject all the new alter requests.
-	if schema.State().IndexingInProgress() {
-		return nil, errIndexingInProgress
+	const numTries = 3
+	for i := 0; i < numTries; i++ {
+		if !schema.State().IndexingInProgress() {
+			break
+		} else if i == numTries-1 {
+			return nil, errIndexingInProgress
+		}
+
+		// Let's wait a bit to see if some really simple indexing
+		// tasks can finish before we reject this request.
+		time.Sleep(time.Second)
 	}
 
 	for _, update := range result.Preds {
@@ -268,6 +277,16 @@ func (s *Server) Alter(ctx context.Context, op *api.Operation) (*api.Payload, er
 	m.Schema = result.Preds
 	m.Types = result.Types
 	_, err = query.ApplyMutations(ctx, m)
+
+	// wait for indexing to complete.
+	for !op.RunInBackground {
+		if !schema.State().IndexingInProgress() {
+			break
+		}
+
+		time.Sleep(time.Second * 2)
+	}
+
 	return empty, err
 }
 

go.mod

@@ -17,7 +17,7 @@ require (
 	github.com/blevesearch/snowballstem v0.0.0-20180110192139-26b06a2c243d // indirect
 	github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd
 	github.com/dgraph-io/badger/v2 v2.0.1-rc1.0.20200316175624-91c31ebe8c22
-	github.com/dgraph-io/dgo/v2 v2.2.0
+	github.com/dgraph-io/dgo/v2 v2.2.1-0.20200319183917-53c7d5bc32a7
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/dgryski/go-farm v0.0.0-20191112170834-c2139c5d712b
 	github.com/dgryski/go-groupvarint v0.0.0-20190318181831-5ce5df8ca4e1

go.sum

@@ -78,8 +78,8 @@ github.com/dgraph-io/badger v1.6.0 h1:DshxFxZWXUcO0xX476VJC07Xsr6ZCBVRHKZ93Oh7Ev
 github.com/dgraph-io/badger v1.6.0/go.mod h1:zwt7syl517jmP8s94KqSxTlM6IMsdhYy6psNgSztDR4=
 github.com/dgraph-io/badger/v2 v2.0.1-rc1.0.20200316175624-91c31ebe8c22 h1:X2g/STOldw/iNkdP9BHI/8zHe4N/Lkp8Z4jLyywtnCI=
 github.com/dgraph-io/badger/v2 v2.0.1-rc1.0.20200316175624-91c31ebe8c22/go.mod h1:3KY8+bsP8wI0OEnQJAKpd4wIJW/Mm32yw2j/9FUVnIM=
-github.com/dgraph-io/dgo/v2 v2.2.0 h1:qYbm6mEF3wuKiRpgNOldk6PmPbBJFwj6vL7I7dTSdyc=
-github.com/dgraph-io/dgo/v2 v2.2.0/go.mod h1:LJCkLxm5fUMcU+yb8gHFjHt7ChgNuz3YnQQ6MQkmscI=
+github.com/dgraph-io/dgo/v2 v2.2.1-0.20200319183917-53c7d5bc32a7 h1:9oFXHEReyRIB291rbdGwRk1PYegGO2XBtZ8muXPKqPA=
+github.com/dgraph-io/dgo/v2 v2.2.1-0.20200319183917-53c7d5bc32a7/go.mod h1:LJCkLxm5fUMcU+yb8gHFjHt7ChgNuz3YnQQ6MQkmscI=
 github.com/dgraph-io/ristretto v0.0.2-0.20200115201040-8f368f2f2ab3 h1:MQLRM35Pp0yAyBYksjbj1nZI/w6eyRY/mWoM1sFf4kU=
 github.com/dgraph-io/ristretto v0.0.2-0.20200115201040-8f368f2f2ab3/go.mod h1:KPxhHT9ZxKefz+PCeOGsrHpl1qZ7i70dGTu2u+Ahh6E=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=

systest/bgindex/count_test.go

@@ -86,7 +86,8 @@ func TestCountIndex(t *testing.T) {
 
 	fmt.Println("building indexes in background")
 	if err := dg.Alter(context.Background(), &api.Operation{
-		Schema: "value: [string] @count .",
+		Schema:          "value: [string] @count .",
+		RunInBackground: true,
 	}); err != nil {
 		t.Fatalf("error in adding indexes :: %v\n", err)
 	}

systest/bgindex/parallel_test.go

@@ -77,6 +77,7 @@ func TestParallelIndexing(t *testing.T) {
 			balance_str: string .
 			balance_float: float .
 		`,
+		RunInBackground: true,
 	}); err != nil {
 		t.Fatalf("error in setting up schema :: %v\n", err)
 	}
@@ -102,6 +103,7 @@ func TestParallelIndexing(t *testing.T) {
 			balance_int: int @index(int) .
 			balance_str: string @index(fulltext, term, exact) .
 		`,
+		RunInBackground: true,
 	}); err != nil {
 		t.Fatalf("error in adding indexes :: %v\n", err)
 	}
@@ -111,15 +113,17 @@ func TestParallelIndexing(t *testing.T) {
 			balance_int: int @index(int) .
 			balance_str: string @index(fulltext, term, exact) .
 		`,
-	}); err != nil && !strings.Contains(err.Error(), "is already being modified") {
+		RunInBackground: true,
+	}); err != nil && !strings.Contains(err.Error(), "errIndexingInProgress") {
 		t.Fatalf("error in adding indexes :: %v\n", err)
 	}
 
 	// Wait until previous indexing is complete.
 	for {
 		if err := dg.Alter(context.Background(), &api.Operation{
-			Schema: `balance_float: float @index(float) .`,
-		}); err != nil && !strings.Contains(err.Error(), "is already being modified") {
+			Schema:          `balance_float: float @index(float) .`,
+			RunInBackground: true,
+		}); err != nil && !strings.Contains(err.Error(), "errIndexingInProgress") {
 			t.Fatalf("error in adding indexes :: %v\n", err)
 		} else if err == nil {
 			break

systest/bgindex/reverse_test.go

@@ -76,7 +76,8 @@ func TestReverseIndex(t *testing.T) {
 
 	fmt.Println("building indexes in background")
 	if err := dg.Alter(context.Background(), &api.Operation{
-		Schema: "balance: [uid] @reverse .",
+		Schema:          "balance: [uid] @reverse .",
+		RunInBackground: true,
 	}); err != nil {
 		t.Fatalf("error in adding indexes :: %v\n", err)
 	}

systest/bgindex/string_test.go

@@ -82,7 +82,8 @@ func TestStringIndex(t *testing.T) {
 
 	fmt.Println("building indexes in background")
 	if err := dg.Alter(context.Background(), &api.Operation{
-		Schema: "balance: string @index(fulltext, term, exact) .",
+		Schema:          "balance: string @index(fulltext, term, exact) .",
+		RunInBackground: true,
 	}); err != nil {
 		t.Fatalf("error in adding indexes :: %v\n", err)
 	}

testutil/client.go

@@ -402,7 +402,7 @@ func VerifyCurlCmd(t *testing.T, args []string, failureConfig *CurlFailureConfig
 		require.NoError(t, json.Unmarshal(output, &co),
 			"unable to unmarshal the curl output")
 		if len(co.Errors) > 0 {
-			if strings.Contains(co.Errors[0].Message, "schema is already being modified") {
+			if strings.Contains(co.Errors[0].Message, "errIndexingInProgress") {
 				time.Sleep(time.Second)
 				continue
 			}