fix(backup/restore): fixes backup and restore with DROP operations (GRAPHQL-735)

dgraph/cmd/bulk/systest/test-bulk-schema.sh

@@ -201,6 +201,7 @@ EOF
   diff <(LC_ALL=C sort all_dbs.out | uniq -c) - <<EOF
       1 dgraph.acl.rule
       1 dgraph.cors
+      1 dgraph.drop.op
       1 dgraph.graphql.p_query
       1 dgraph.graphql.p_sha256hash
       1 dgraph.graphql.schema

edgraph/server.go

@@ -289,6 +289,21 @@ func parseSchemaFromAlterOperation(op *api.Operation) (*schema.ParsedSchema, err
 	return result, nil
 }
 
+func insertDropRecord(ctx context.Context, dropOp string) error {
+	_, err := (&Server{}).doQuery(context.WithValue(ctx, IsGraphql, true),
+		&api.Request{
+			Mutations: []*api.Mutation{{
+				Set: []*api.NQuad{{
+					Subject:     "_:r",
+					Predicate:   "dgraph.drop.op",
+					ObjectValue: &api.Value{Val: &api.Value_StrVal{StrVal: dropOp}},
+				}},
+			}},
+			CommitNow: true,
+		}, NoAuthorize)
+	return err
+}
+
 // Alter handles requests to change the schema or remove parts or all of the data.
 func (s *Server) Alter(ctx context.Context, op *api.Operation) (*api.Payload, error) {
 	ctx, span := otrace.StartSpan(ctx, "Server.Alter")
@@ -321,6 +336,12 @@ func (s *Server) Alter(ctx context.Context, op *api.Operation) (*api.Payload, er
 			return empty, err
 		}
 
+		// insert a helper record for backup & restore, indicating that drop_all was done
+		err = insertDropRecord(ctx, "DROP_ALL;")
+		if err != nil {
+			return empty, err
+		}
+
 		// insert empty GraphQL schema, so all alphas get notified to
 		// reset their in-memory GraphQL schema
 		_, err = UpdateGQLSchema(ctx, "", "")
@@ -347,6 +368,12 @@ func (s *Server) Alter(ctx context.Context, op *api.Operation) (*api.Payload, er
 			return empty, err
 		}
 
+		// insert a helper record for backup & restore, indicating that drop_data was done
+		err = insertDropRecord(ctx, "DROP_DATA;")
+		if err != nil {
+			return empty, err
+		}
+
 		// just reinsert the GraphQL schema, no need to alter dgraph schema as this was drop_data
 		_, err = UpdateGQLSchema(ctx, graphQLSchema, "")
 		// recreate the admin account after a drop data operation
@@ -386,6 +413,12 @@ func (s *Server) Alter(ctx context.Context, op *api.Operation) (*api.Payload, er
 		edges := []*pb.DirectedEdge{edge}
 		m.Edges = edges
 		_, err = query.ApplyMutations(ctx, m)
+		if err != nil {
+			return empty, err
+		}
+
+		// insert a helper record for backup & restore, indicating that drop_attr was done
+		err = insertDropRecord(ctx, "DROP_ATTR;"+attr)
 		return empty, err
 	}
 

ee/acl/acl_test.go

@@ -2146,6 +2146,10 @@ func TestSchemaQueryWithACL(t *testing.T) {
       	],
       	"upsert": true
 	},
+	{
+		"predicate":"dgraph.drop.op",
+		"type":"string"
+	},
 	{
 		"predicate":"dgraph.graphql.p_query",
 		"type":"string"

graphql/e2e/common/admin.go

@@ -53,6 +53,10 @@ const (
             ],
             "upsert": true
         },
+        {
+            "predicate": "dgraph.drop.op",
+            "type": "string"
+        },
         {
             "predicate":"dgraph.graphql.p_query",
             "type":"string"
@@ -149,6 +153,10 @@ const (
             ],
             "upsert": true
         },
+        {
+            "predicate": "dgraph.drop.op",
+            "type": "string"
+        },
         {
             "predicate":"dgraph.graphql.p_query",
             "type":"string"
@@ -268,6 +276,10 @@ const (
             ],
             "upsert": true
         },
+        {
+            "predicate": "dgraph.drop.op",
+            "type": "string"
+        },
         {
             "predicate":"dgraph.graphql.p_query",
             "type":"string"
@@ -398,6 +410,10 @@ const (
             ],
             "upsert": true
         },
+        {
+            "predicate": "dgraph.drop.op",
+            "type": "string"
+        },
         {
             "predicate":"dgraph.graphql.p_query",
             "type":"string"

graphql/e2e/directives/schema_response.json

@@ -256,6 +256,10 @@
             "list": true,
             "upsert": true
         },
+        {
+            "predicate": "dgraph.drop.op",
+            "type": "string"
+        },
         {
             "predicate":"dgraph.graphql.p_query",
             "type":"string"

graphql/e2e/normal/schema_response.json

@@ -420,6 +420,10 @@
             "list": true,
             "upsert": true
         },
+        {
+            "predicate": "dgraph.drop.op",
+            "type": "string"
+        },
         {
             "predicate": "dgraph.graphql.schema",
             "type": "string"

graphql/e2e/schema/schema_test.go

@@ -219,6 +219,10 @@ func TestConcurrentSchemaUpdates(t *testing.T) {
       		],
       		"upsert": true
 		},
+		{
+			"predicate":"dgraph.drop.op",
+			"type":"string"
+		},
 		{
 			"predicate":"dgraph.graphql.p_query",
 			"type":"string"

protos/pb.proto

@@ -537,7 +537,7 @@ service Worker {
 	rpc StreamSnapshot (stream Snapshot)    returns (stream KVS) {}
 	rpc Sort (SortMessage)                  returns (SortResult) {}
 	rpc Schema (SchemaRequest)              returns (SchemaResult) {}
-	rpc Backup (BackupRequest)              returns (Status) {}
+	rpc Backup (BackupRequest)              returns (BackupResponse) {}
 	rpc Restore (RestoreRequest)            returns (Status) {}
 	rpc Export (ExportRequest)              returns (ExportResponse) {}
 	rpc ReceivePredicate(stream KVS)        returns (api.Payload) {}
@@ -599,6 +599,20 @@ message BackupRequest {
 	repeated string predicates = 10;
 }
 
+message BackupResponse {
+	repeated DropOperation drop_operations = 1;
+}
+
+message DropOperation {
+	enum DropOp {
+		ALL = 0;
+		DATA = 1;
+		ATTR = 2;
+	}
+	DropOp drop_op    = 1;
+	string drop_value = 2;
+}
+
 message ExportRequest {
 	uint32  group_id = 1;  // Group id to back up.
 	uint64  read_ts  = 2;