Adds @noconflict directive to prevent conflict detection.

dgraph/cmd/live/batch.go

@@ -237,6 +237,15 @@ func fingerprintEdge(t *pb.DirectedEdge, pred *predicate) uint64 {
 }
 
 func (l *loader) conflictKeysForNQuad(nq *api.NQuad) ([]uint64, error) {
+	pred, found := l.schema.preds[nq.Predicate]
+
+	// We dont' need to generate conflict keys for predicate with noconflict directive.
+	if found && pred.NoConflict {
+		return nil, nil
+	}
+
+	keys := make([]uint64, 0)
+
 	// Calculates the conflict keys, inspired by the logic in
 	// addMutationInteration in posting/list.go.
 	sid, err := strconv.ParseUint(nq.Subject, 0, 64)
@@ -256,9 +265,8 @@ func (l *loader) conflictKeysForNQuad(nq *api.NQuad) ([]uint64, error) {
 		x.Check(err)
 	}
 
-	keys := make([]uint64, 0, 1)
-	pred, ok := l.schema.preds[nq.Predicate]
-	if !ok {
+	// If the predicate is not found in schema then we don't have to generate any more keys.
+	if !found {
 		return keys, nil
 	}
 

dgraph/cmd/live/run.go

@@ -69,16 +69,17 @@ type options struct {
 }
 
 type predicate struct {
-	Predicate string   `json:"predicate,omitempty"`
-	Type      string   `json:"type,omitempty"`
-	Tokenizer []string `json:"tokenizer,omitempty"`
-	Count     bool     `json:"count,omitempty"`
-	List      bool     `json:"list,omitempty"`
-	Lang      bool     `json:"lang,omitempty"`
-	Index     bool     `json:"index,omitempty"`
-	Upsert    bool     `json:"upsert,omitempty"`
-	Reverse   bool     `json:"reverse,omitempty"`
-	ValueType types.TypeID
+	Predicate  string   `json:"predicate,omitempty"`
+	Type       string   `json:"type,omitempty"`
+	Tokenizer  []string `json:"tokenizer,omitempty"`
+	Count      bool     `json:"count,omitempty"`
+	List       bool     `json:"list,omitempty"`
+	Lang       bool     `json:"lang,omitempty"`
+	Index      bool     `json:"index,omitempty"`
+	Upsert     bool     `json:"upsert,omitempty"`
+	Reverse    bool     `json:"reverse,omitempty"`
+	NoConflict bool     `json:"no_conflict,omitempty"`
+	ValueType  types.TypeID
 }
 
 type schema struct {

posting/list.go

@@ -402,6 +402,8 @@ func (l *List) addMutationInternal(ctx context.Context, txn *Txn, t *pb.Directed
 		return err
 	}
 	switch {
+	case schema.State().HasNoConflict(t.Attr):
+		break
 	case schema.State().HasUpsert(t.Attr):
 		// Consider checking to see if a email id is unique. A user adds:
 		// <uid> <email> "[email protected]", and there's a string equal tokenizer

protos/pb.proto

@@ -378,6 +378,7 @@ message SchemaNode {
 	bool list = 7;
 	bool upsert = 8;
 	bool lang = 9;
+	bool no_conflict = 10;
 }
 
 message SchemaResult {
@@ -408,6 +409,8 @@ message SchemaUpdate {
 	// custom name. This field stores said name.
 	string object_type_name = 12;
 
+	bool no_conflict = 13;
+
 	// Deleted field:
 	reserved 7;
 	reserved "explicit";

query/common_test.go

@@ -285,6 +285,7 @@ newage                         : int .
 boss                           : uid .
 newfriend                      : [uid] .
 owner                          : [uid] .
+noconflict_pred                : string @noconflict .
 `
 
 func populateCluster() {

query/query4_test.go

@@ -18,9 +18,17 @@ package query
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"testing"
 
+	"github.com/dgraph-io/dgo/v2"
+	"github.com/dgraph-io/dgo/v2/protos/api"
+	"github.com/dgraph-io/dgraph/testutil"
+	"github.com/dgraph-io/dgraph/x"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
 )
 
 func TestBigMathValue(t *testing.T) {
@@ -269,6 +277,167 @@ func TestDeleteAndReaddReverse(t *testing.T) {
 	setSchema(testSchema)
 }
 
+func TestSchemaUpdateNoConflict(t *testing.T) {
+	// Verify schema is as expected for the predicate with noconflict directive.
+	q1 := `schema(pred: [noconflict_pred]) { }`
+	js := processQueryNoErr(t, q1)
+	require.JSONEq(t, `{
+		"data": {
+			"schema": [{
+				"predicate": "noconflict_pred",
+				"type": "string",
+				"no_conflict": true
+			}]
+		}
+	}`, js)
+
+	// Verify schema is as expected for the predicate without noconflict directive.
+	q1 = `schema(pred: [name]) { }`
+	js = processQueryNoErr(t, q1)
+	require.JSONEq(t, `{
+		"data": {
+			"schema": [{
+				"predicate": "name",
+				"type": "string",
+				"index": true,
+				"tokenizer": ["term", "exact", "trigram"],
+				"count": true,
+				"lang": true
+			}]
+		}
+	}`, js)
+}
+
+func TestNoConflictQuery(t *testing.T) {
+	conn, err := grpc.Dial(testutil.SockAddr, grpc.WithInsecure())
+	require.NoError(t, err)
+
+	dg := dgo.NewDgraphClient(api.NewDgraphClient(conn))
+	defer dg.Alter(context.Background(), &api.Operation{DropAll: true})
+
+	schema := `
+		type node {
+		name_noconflict: string
+		child: uid
+		}
+
+		name_noconflict: string @noconflict .
+		child: uid .
+	`
+	err = dg.Alter(context.Background(), &api.Operation{Schema: schema})
+	require.NoError(t, err)
+
+	type node struct {
+		ID    string `json:"uid"`
+		Name  string `json:"name_noconflict"`
+		Child *node  `json:"child"`
+	}
+
+	child := node{ID: "_:blank-0", Name: "child"}
+	js, err := json.Marshal(child)
+	require.NoError(t, err)
+
+	res, err := dg.NewTxn().Mutate(context.Background(),
+		&api.Mutation{SetJson: js, CommitNow: true})
+	require.NoError(t, err)
+
+	in := []node{}
+	for i := 0; i < 5; i++ {
+		in = append(in, node{ID: "_:blank-0", Name: fmt.Sprintf("%d", i+1),
+			Child: &node{ID: res.GetUids()["blank-0"]}})
+	}
+
+	errChan := make(chan error)
+	for i := range in {
+		go func(n node) {
+			js, err := json.Marshal(n)
+			require.NoError(t, err)
+
+			_, err = dg.NewTxn().Mutate(context.Background(),
+				&api.Mutation{SetJson: js, CommitNow: true})
+			errChan <- err
+		}(in[i])
+	}
+
+	errs := []error{}
+	for i := 0; i < len(in); i++ {
+		errs = append(errs, <-errChan)
+	}
+
+	for _, e := range errs {
+		assert.NoError(t, e)
+	}
+}
+
+func TestNoConflictQuery2(t *testing.T) {
+	conn, err := grpc.Dial(testutil.SockAddr, grpc.WithInsecure())
+	require.NoError(t, err)
+
+	dg := dgo.NewDgraphClient(api.NewDgraphClient(conn))
+	defer dg.Alter(context.Background(), &api.Operation{DropAll: true})
+
+	schema := `
+		type node {
+		name_noconflict: string
+		address_conflict: string
+		child: uid
+		}
+
+		name_noconflict: string @noconflict .
+		address_conflict: string .
+		child: uid .
+	`
+	err = dg.Alter(context.Background(), &api.Operation{Schema: schema})
+	require.NoError(t, err)
+
+	type node struct {
+		ID      string `json:"uid"`
+		Name    string `json:"name_noconflict"`
+		Child   *node  `json:"child"`
+		Address string `json:"address_conflict"`
+	}
+
+	child := node{ID: "_:blank-0", Name: "child", Address: "dgraph labs"}
+	js, err := json.Marshal(child)
+	require.NoError(t, err)
+
+	res, err := dg.NewTxn().Mutate(context.Background(),
+		&api.Mutation{SetJson: js, CommitNow: true})
+	require.NoError(t, err)
+
+	in := []node{}
+	for i := 0; i < 5; i++ {
+		in = append(in, node{ID: "_:blank-0", Name: fmt.Sprintf("%d", i+1),
+			Child: &node{ID: res.GetUids()["blank-0"]}})
+	}
+
+	errChan := make(chan error)
+	for i := range in {
+		go func(n node) {
+			js, err := json.Marshal(n)
+			require.NoError(t, err)
+
+			_, err = dg.NewTxn().Mutate(context.Background(),
+				&api.Mutation{SetJson: js, CommitNow: true})
+			errChan <- err
+		}(in[i])
+	}
+
+	errs := []error{}
+	for i := 0; i < len(in); i++ {
+		errs = append(errs, <-errChan)
+	}
+
+	hasError := false
+	for _, e := range errs {
+		if e != nil {
+			hasError = true
+			require.Contains(t, e.Error(), "Transaction has been aborted. Please retry")
+		}
+	}
+	x.AssertTrue(hasError)
+}
+
 func TestDropPredicate(t *testing.T) {
 	// Add new predicate with several indices.
 	s1 := testSchema + "\n numerology: string @index(term) .\n"

schema/parse.go

@@ -70,6 +70,8 @@ func parseDirective(it *lex.ItemIterator, schema *pb.SchemaUpdate, t types.TypeI
 		schema.Count = true
 	case "upsert":
 		schema.Upsert = true
+	case "noconflict":
+		schema.NoConflict = true
 	case "lang":
 		if t != types.StringID || schema.List {
 			return next.Errorf("@lang directive can only be specified for string type."+

schema/parse_test.go

@@ -265,6 +265,13 @@ func TestParse8_Error(t *testing.T) {
 	require.Nil(t, result)
 }
 
+func TestParse9_Error(t *testing.T) {
+	reset()
+	result, err := Parse("age:uid @noconflict .")
+	require.NotNil(t, result)
+	require.NoError(t, err)
+}
+
 func TestParseScalarList(t *testing.T) {
 	reset()
 	result, err := Parse(`

schema/schema.go

@@ -301,6 +301,12 @@ func (s *state) HasLang(pred string) bool {
 	return false
 }
 
+func (s *state) HasNoConflict(pred string) bool {
+	s.RLock()
+	defer s.RUnlock()
+	return s.predicate[pred].GetNoConflict()
+}
+
 // Init resets the schema state, setting the underlying DB to the given pointer.
 func Init(ps *badger.DB) {
 	pstore = ps

worker/schema.go

@@ -56,7 +56,7 @@ func getSchema(ctx context.Context, s *pb.SchemaRequest) (*pb.SchemaResult, erro
 		fields = s.Fields
 	} else {
 		fields = []string{"type", "index", "tokenizer", "reverse", "count", "list", "upsert",
-			"lang"}
+			"lang", "noconflict"}
 	}
 
 	for _, attr := range predicates {
@@ -105,6 +105,8 @@ func populateSchema(attr string, fields []string) *pb.SchemaNode {
 			schemaNode.Upsert = schema.State().HasUpsert(attr)
 		case "lang":
 			schemaNode.Lang = schema.State().HasLang(attr)
+		case "noconflict":
+			schemaNode.NoConflict = schema.State().HasNoConflict(attr)
 		default:
 			//pass
 		}