Function runMutation

worker/mutation.go:59–120

runMutation goes through all the edges and applies them.

(ctx context.Context, edge *pb.DirectedEdge, txn *posting.Txn)

// runMutation goes through all the edges and applies them.
func runMutation(ctx context.Context, edge *pb.DirectedEdge, txn *posting.Txn) error {
	ctx = schema.GetWriteContext(ctx)

	// We shouldn't check whether this Alpha serves this predicate or not. Membership information
	// isn't consistent across the entire cluster. We should just apply whatever is given to us.
	su, ok := schema.State().Get(ctx, edge.Attr)
	if edge.Op != pb.DirectedEdge_DEL {
		if !ok {
			return errors.Errorf("runMutation: Unable to find schema for %s", edge.Attr)
		}
	}

	if isDeletePredicateEdge(edge) {
		return errors.New("We should never reach here")
	}

	// Once mutation comes via raft we do best effort conversion
	// Type check is done before proposing mutation, in case schema is not
	// present, some invalid entries might be written initially
	if err := ValidateAndConvert(edge, &su); err != nil {
		return err
	}

	key := x.DataKey(edge.Attr, edge.Entity)
	// The following is a performance optimization which allows us to not read a posting list from
	// disk. We calculate this based on how AddMutationWithIndex works. The general idea is that if
	// we're not using the read posting list, we don't need to retrieve it. We need the posting list
	// if we're doing count index or delete operation. For scalar predicates, we just get the last item merged.
	// In other cases, we can just create a posting list facade in memory and use it to store the delta in Badger.
	// Later, the rollup operation would consolidate all these deltas into a posting list.
	isList := su.GetList()
	var getFn func(key []byte) (*posting.List, error)
	switch {
	case len(edge.Lang) == 0 && !isList:
		// Scalar Predicates, without lang
		getFn = txn.GetScalarList
	case len(edge.Lang) > 0 || su.GetCount():
		// Language or Count Index
		getFn = txn.Get
	case edge.Op == pb.DirectedEdge_DEL:
		// Covers various delete cases to keep things simple.
		getFn = txn.Get
	default:
		// Only count index needs to be read. For other indexes on list, we don't need to read any data.
		// For indexes on scalar prediactes, only the last element needs to be left.
		// Delete cases covered above.
		getFn = txn.GetFromDelta
	}

	t := time.Now()
	plist, err := getFn(key)
	if dur := time.Since(t); dur > time.Millisecond {
		span := trace.SpanFromContext(ctx)
		span.AddEvent("Slow GetLru", trace.WithAttributes(
			attribute.Bool("slow-get", true),
			attribute.String("duration", dur.String())))
	}
	if err != nil {