New txn operation

chunker/rdf/parse.go

@@ -78,7 +78,13 @@ L:
 				return rnq, x.Errorf("Expected '(', found: %s", item.Val)
 			}
 			it.Next()
-			if item = it.Item(); item.Typ != itemVarName {
+			item = it.Item()
+			switch item.Typ {
+			case itemSubjectVarName:
+				rnq.SubjectVar = item.Val
+			case itemObjectVarName:
+				rnq.ObjectVar = item.Val
+			default:
 				return rnq, x.Errorf("Expected variable name, found: %s", item.Val)
 			}
 
@@ -183,14 +189,14 @@ L:
 	if seenOval && rnq.ObjectValue == nil {
 		rnq.ObjectValue = &api.Value{Val: &api.Value_DefaultVal{DefaultVal: oval}}
 	}
-	if len(rnq.Subject) == 0 || len(rnq.Predicate) == 0 {
+	if (len(rnq.Subject) == 0 && len(rnq.SubjectVar) == 0) || len(rnq.Predicate) == 0 {
 		return rnq, x.Errorf("Empty required fields in NQuad. Input: [%s]", line)
 	}
-	if len(rnq.ObjectId) == 0 && rnq.ObjectValue == nil {
+	if len(rnq.ObjectId) == 0 && rnq.ObjectValue == nil && len(rnq.ObjectVar) == 0 {
 		return rnq, x.Errorf("No Object in NQuad. Input: [%s]", line)
 	}
-	if !sane(rnq.Subject) || !sane(rnq.Predicate) ||
-		!sane(rnq.ObjectId) || !sane(rnq.Label) {
+	if (!sane(rnq.Subject) && len(rnq.SubjectVar) == 0) || !sane(rnq.Predicate) ||
+		(!sane(rnq.ObjectId) && len(rnq.ObjectVar) == 0) || !sane(rnq.Label) {
 		return rnq, x.Errorf("NQuad failed sanity check:%+v", rnq)
 	}
 

chunker/rdf/state.go

@@ -26,23 +26,24 @@ import (
 
 // The constants represent different types of lexed Items possible for an rdf N-Quad.
 const (
-	itemText       lex.ItemType = 5 + iota // plain text
-	itemSubject                            // subject, 6
-	itemPredicate                          // predicate, 7
-	itemObject                             // object, 8
-	itemLabel                              // label, 9
-	itemLiteral                            // literal, 10
-	itemLanguage                           // language, 11
-	itemObjectType                         // object type, 12
-	itemValidEnd                           // end with dot, 13
-	itemComment                            // comment, 14
-	itemComma                              // comma, 15
-	itemEqual                              // equal, 16
-	itemLeftRound                          // '(', 17
-	itemRightRound                         // ')', 18
-	itemStar                               // *, 19
-	itemVarKeyword                         // var, 20
-	itemVarName                            // 21
+	itemText           lex.ItemType = 5 + iota // plain text
+	itemSubject                                // subject, 6
+	itemPredicate                              // predicate, 7
+	itemObject                                 // object, 8
+	itemLabel                                  // label, 9
+	itemLiteral                                // literal, 10
+	itemLanguage                               // language, 11
+	itemObjectType                             // object type, 12
+	itemValidEnd                               // end with dot, 13
+	itemComment                                // comment, 14
+	itemComma                                  // comma, 15
+	itemEqual                                  // equal, 16
+	itemLeftRound                              // '(', 17
+	itemRightRound                             // ')', 18
+	itemStar                                   // *, 19
+	itemVarKeyword                             // var, 20
+	itemSubjectVarName                         // 21
+	itemObjectVarName                          // 22
 )
 
 // These constants keep a track of the depth while parsing an rdf N-Quad.
@@ -459,7 +460,11 @@ func lexVariable(l *lex.Lexer) lex.StateFn {
 			break
 		}
 	}
-	l.Emit(itemVarName)
+	if l.Depth == atSubject {
+		l.Emit(itemSubjectVarName)
+	} else {
+		l.Emit(itemObjectVarName)
+	}
 
 	l.IgnoreRun(isSpace)
 

edgraph/server.go

@@ -384,6 +384,60 @@ func (s *Server) Mutate(ctx context.Context, mu *api.Mutation) (resp *api.Assign
 	return s.doMutate(ctx, mu)
 }
 
+// doCondQuery processes a conditional query within the same transaction of a mutation.
+// Returns a map with the vars created from the query, otherwise nil and an error.
+func doCondQuery(ctx context.Context, l *query.Latency, req *api.Request,
+) (map[string][]string, error) {
+	var queryVars map[string][]string
+
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+
+	// Nothing to do
+	if req.Query == "" {
+		return queryVars, nil
+	}
+	x.AssertTruef(req.StartTs != 0, "Transaction timestamp is zero")
+
+	parsedReq, err := gql.Parse(gql.Request{
+		Str:       req.Query,
+		Variables: make(map[string]string),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	if err = validateQuery(parsedReq.Query); err != nil {
+		return nil, err
+	}
+
+	qr := query.QueryRequest{
+		Latency:  l,
+		GqlQuery: &parsedReq,
+		ReadTs:   req.StartTs,
+	}
+	if err = qr.ProcessQuery(ctx); err != nil {
+		return nil, x.Wrapf(err, "while processing query: %q", req.Query)
+	}
+
+	queryVars = qr.GetUids()
+	if len(queryVars) == 0 {
+		glog.V(2).Infof("No variables defined in conditional query: %q", req.Query)
+		return nil, nil
+	}
+	glog.V(3).Infof("CondQuery: qr=%+v vars=%v", qr, queryVars)
+
+	// NOTE: This is a temporary check until the mutation logic is defined clearly.
+	for varName, uids := range queryVars {
+		if len(uids) > 1 {
+			return nil, x.Errorf("Too many Uids for %q, expected 1 got %d", varName, len(uids))
+		}
+	}
+
+	return queryVars, nil
+}
+
 func (s *Server) doMutate(ctx context.Context, mu *api.Mutation) (resp *api.Assigned, rerr error) {
 	if ctx.Err() != nil {
 		return nil, ctx.Err()
@@ -433,6 +487,17 @@ func (s *Server) doMutate(ctx context.Context, mu *api.Mutation) (resp *api.Assi
 
 	var l query.Latency
 	l.Start = time.Now()
+
+	// Parse query and process
+	queryVars, err := doCondQuery(ctx, &l, &api.Request{
+		StartTs: mu.StartTs,
+		Query:   mu.CondQuery,
+	})
+	// Quit early if we got an error or expected queryVars and got none.
+	if err != nil || (mu.CondQuery != "" && queryVars == nil) {
+		return resp, err
+	}
+
 	gmu, err := parseMutationObject(mu)
 	if err != nil {
 		return resp, err
@@ -449,7 +514,7 @@ func (s *Server) doMutate(ctx context.Context, mu *api.Mutation) (resp *api.Assi
 		}
 	}()
 
-	newUids, err := query.AssignUids(ctx, gmu.Set)
+	newUids, err := query.AssignUids(ctx, gmu.Set, queryVars)
 	if err != nil {
 		return resp, err
 	}

gql/parser.go

@@ -586,8 +586,14 @@ func checkDependency(vl []*Vars) error {
 	}
 
 	if len(defines) > len(needs) {
-		return x.Errorf("Some variables are defined but not used\nDefined:%v\nUsed:%v\n",
+		// NOTE: This error prevents conditional mutations to work, because we expect variables
+		//       created from a query that are used in a following mutation block. For now,
+		//       we notify the user in the output. I will revisit this later.
+		// return x.Errorf("Some variables are defined but not used\nDefined:%v\nUsed:%v\n",
+		// 	defines, needs)
+		glog.Warningf("Some variables are defined but not used\nDefined:%v\nUsed:%v\n",
 			defines, needs)
+		return nil
 	}
 
 	if len(defines) < len(needs) {

gql/parser_mutation.go

@@ -24,16 +24,35 @@ import (
 	"github.com/dgraph-io/dgraph/x"
 )
 
+// ParseMutation parses a block of text into a mutation.
+// TODO: add examples.
+// Returns an object with a mutation or a transaction query with mutation, otherwise returns
+// nil with an error.
 func ParseMutation(mutation string) (*api.Mutation, error) {
 	lexer := lex.NewLexer(mutation)
 	lexer.Run(lexInsideMutation)
 	it := lexer.NewIterator()
 	var mu api.Mutation
+	var inTxn bool
 
 	if !it.Next() {
 		return nil, errors.New("Invalid mutation")
 	}
 	item := it.Item()
+	// Inside txn{ ... } block.
+	// Here we switch into txn mode and try to fetch any query inside a txn block.
+	// If no query text is found, this txn is a no-op.
+	if item.Typ == itemMutationTxn {
+		var err error
+		// Get the query text: txn{ query { ... }}
+		mu.CondQuery, err = parseMutationCondQuery(it)
+		if err != nil {
+			return nil, err
+		}
+		inTxn = true
+		item = it.Item()
+		// fallthrough to regular mutation parsing.
+	}
 	if item.Typ != itemLeftCurl {
 		return nil, x.Errorf("Expected { at the start of block. Got: [%s]", item.Val)
 	}
@@ -45,7 +64,7 @@ func ParseMutation(mutation string) (*api.Mutation, error) {
 		}
 		if item.Typ == itemRightCurl {
 			// mutations must be enclosed in a single block.
-			if it.Next() && it.Item().Typ != lex.ItemEOF {
+			if !inTxn && it.Next() && it.Item().Typ != lex.ItemEOF {
 				return nil, x.Errorf("Unexpected %s after the end of the block.", it.Item().Val)
 			}
 			return &mu, nil
@@ -59,6 +78,45 @@ func ParseMutation(mutation string) (*api.Mutation, error) {
 	return nil, x.Errorf("Invalid mutation.")
 }
 
+// parseMutationCondQuery gets the text inside a txn query block. It is possible that there's
+// no query to be found, in that case it's the caller's responsbility to fail.
+// Returns the query text if any is found, otherwise an empty string with error.
+func parseMutationCondQuery(it *lex.ItemIterator) (string, error) {
+	var query string
+	var parse bool
+	for it.Next() {
+		item := it.Item()
+		switch item.Typ {
+		case itemLeftCurl:
+			continue
+		case itemMutationOpContent:
+			if !parse {
+				return "", x.Errorf("Invalid query block.")
+			}
+			query = item.Val
+		case itemMutationTxnOp:
+			if item.Val == "query" {
+				if parse {
+					return "", x.Errorf("Too many query blocks in txn")
+				}
+				parse = true
+				continue
+			}
+			// TODO: mutation conditionals
+			if item.Val != "mutation" {
+				return "", x.Errorf("Invalid txn operator %q.", item.Val)
+			}
+			if !it.Next() {
+				return "", errors.New("Invalid mutation block")
+			}
+			return query, nil
+		default:
+			return "", x.Errorf("Unexpected %q inside of txn block.", item.Val)
+		}
+	}
+	return query, nil
+}
+
 // parseMutationOp parses and stores set or delete operation string in Mutation.
 func parseMutationOp(it *lex.ItemIterator, op string, mu *api.Mutation) error {
 	parse := false
@@ -73,7 +131,7 @@ func parseMutationOp(it *lex.ItemIterator, op string, mu *api.Mutation) error {
 			}
 			parse = true
 		}
-		if item.Typ == itemMutationContent {
+		if item.Typ == itemMutationOpContent {
 			if !parse {
 				return x.Errorf("Mutation syntax invalid.")
 			}