Function ProcessTaskOverNetwork

worker/task.go:123–168

ProcessTaskOverNetwork is used to process the query and get the result from the instance which stores posting list corresponding to the predicate in the query.

(ctx context.Context, q *pb.Query)

// the instance which stores posting list corresponding to the predicate in the
// query.
func ProcessTaskOverNetwork(ctx context.Context, q *pb.Query) (*pb.Result, error) {
	attr := q.Attr
	gid, err := groups().BelongsToReadOnly(attr, q.ReadTs)
	switch {
	case err != nil:
		return nil, err
	case gid == 0:
		return nil, errNonExistentTablet
	}

	span := trace.SpanFromContext(ctx)
	span.AddEvent("ProcessTaskOverNetwork", trace.WithAttributes(
		attribute.String("attr", attr),
		attribute.String("gid", fmt.Sprintf("%d", gid)),
		attribute.String("readTs", fmt.Sprintf("%d", q.ReadTs)),
		attribute.String("node_id", fmt.Sprintf("%d", groups().Node.Id))))

	if groups().ServesGroup(gid) {
		// No need for a network call, as this should be run from within this instance.
		return processTask(ctx, q, gid)
	}

	// Add span for cross-alpha network call
	ctx, networkSpan := otel.Tracer("").Start(ctx, "ProcessTaskOverNetwork.RemoteCall")
	networkSpan.SetAttributes(
		attribute.String("target_group", fmt.Sprintf("%d", gid)),
		attribute.String("predicate", x.SafeUTF8(attr)),
		attribute.Bool("is_remote", true),
	)

	result, err := processWithBackupRequest(ctx, gid,
		func(ctx context.Context, c pb.WorkerClient) (interface{}, error) {
			return c.ServeTask(ctx, q)
		})
	networkSpan.End()
	if err != nil {
		return nil, err
	}

	reply := result.(*pb.Result)
	span.AddEvent("Reply from server", trace.WithAttributes(
		attribute.Int("len", len(reply.UidMatrix)),
		attribute.Int64("gid", int64(gid)),
		attribute.String("attr", attr)))
	return reply, nil
}

// convertValue converts the data to the schema.State() type of predicate.
func convertValue(attr, data string) (types.Val, error) {