func positionCmd(mysqld *mysqlctl.Mysqld, subFlags *flag.FlagSet, args []string) error {
	subFlags.Parse(args)
	if len(args) < 3 {
		return fmt.Errorf("Not enough arguments for position operation.")
	}

	pos1, err := myproto.DecodeReplicationPosition(args[1])
	if err != nil {
		return err
	}

	switch args[0] {
	case "equal":
		pos2, err := myproto.DecodeReplicationPosition(args[2])
		if err != nil {
			return err
		}
		fmt.Println(pos1.Equal(pos2))
	case "at_least":
		pos2, err := myproto.DecodeReplicationPosition(args[2])
		if err != nil {
			return err
		}
		fmt.Println(pos1.AtLeast(pos2))
	case "append":
		gtid, err := myproto.DecodeGTID(args[2])
		if err != nil {
			return err
		}
		fmt.Println(myproto.AppendGTID(pos1, gtid))
	}

	return nil
}

// StreamTables is part of the proto.UpdateStream interface
func (updateStream *UpdateStream) StreamTables(position string, tables []string, charset *mproto.Charset, sendReply func(reply *proto.BinlogTransaction) error) (err error) {
	pos, err := myproto.DecodeReplicationPosition(position)
	if err != nil {
		return err
	}

	updateStream.actionLock.Lock()
	if !updateStream.isEnabled() {
		updateStream.actionLock.Unlock()
		log.Errorf("Unable to serve client request: Update stream service is not enabled")
		return fmt.Errorf("update stream service is not enabled")
	}
	updateStream.stateWaitGroup.Add(1)
	updateStream.actionLock.Unlock()
	defer updateStream.stateWaitGroup.Done()

	streamCount.Add("Tables", 1)
	defer streamCount.Add("Tables", -1)
	log.Infof("ServeUpdateStream starting @ %#v", pos)

	// Calls cascade like this: BinlogStreamer->TablesFilterFunc->func(*proto.BinlogTransaction)->sendReply
	f := TablesFilterFunc(tables, func(reply *proto.BinlogTransaction) error {
		keyrangeStatements.Add(int64(len(reply.Statements)))
		keyrangeTransactions.Add(1)
		return sendReply(reply)
	})
	bls := NewBinlogStreamer(updateStream.dbname, updateStream.mysqld, charset, pos, f)

	svm := &sync2.ServiceManager{}
	svm.Go(bls.Stream)
	updateStream.streams.Add(svm)
	defer updateStream.streams.Delete(svm)
	return svm.Join()
}

// ServeUpdateStream is part of the proto.UpdateStream interface
func (updateStream *UpdateStream) ServeUpdateStream(position string, sendReply func(reply *proto.StreamEvent) error) (err error) {
	pos, err := myproto.DecodeReplicationPosition(position)
	if err != nil {
		return err
	}

	updateStream.actionLock.Lock()
	if !updateStream.isEnabled() {
		updateStream.actionLock.Unlock()
		log.Errorf("Unable to serve client request: update stream service is not enabled")
		return fmt.Errorf("update stream service is not enabled")
	}
	updateStream.stateWaitGroup.Add(1)
	updateStream.actionLock.Unlock()
	defer updateStream.stateWaitGroup.Done()

	streamCount.Add("Updates", 1)
	defer streamCount.Add("Updates", -1)
	log.Infof("ServeUpdateStream starting @ %#v", pos)

	evs := NewEventStreamer(updateStream.dbname, updateStream.mysqld, pos, func(reply *proto.StreamEvent) error {
		if reply.Category == "ERR" {
			updateStreamErrors.Add("UpdateStream", 1)
		} else {
			updateStreamEvents.Add(reply.Category, 1)
		}
		return sendReply(reply)
	})

	svm := &sync2.ServiceManager{}
	svm.Go(evs.Stream)
	updateStream.streams.Add(svm)
	defer updateStream.streams.Delete(svm)
	return svm.Join()
}

// WaitBlpPosition will wait for the filtered replication to reach at least
// the provided position.
func WaitBlpPosition(mysqld MysqlDaemon, bp *blproto.BlpPosition, waitTimeout time.Duration) error {
	timeOut := time.Now().Add(waitTimeout)
	for {
		if time.Now().After(timeOut) {
			break
		}

		cmd := binlogplayer.QueryBlpCheckpoint(bp.Uid)
		qr, err := mysqld.FetchSuperQuery(cmd)
		if err != nil {
			return err
		}
		if len(qr.Rows) != 1 {
			return fmt.Errorf("QueryBlpCheckpoint(%v) returned unexpected row count: %v", bp.Uid, len(qr.Rows))
		}
		var pos proto.ReplicationPosition
		if !qr.Rows[0][0].IsNull() {
			pos, err = proto.DecodeReplicationPosition(qr.Rows[0][0].String())
			if err != nil {
				return err
			}
		}
		if pos.AtLeast(bp.Position) {
			return nil
		}

		log.Infof("Sleeping 1 second waiting for binlog replication(%v) to catch up: %v != %v", bp.Uid, pos, bp.Position)
		time.Sleep(1 * time.Second)
	}

	return fmt.Errorf("WaitBlpPosition(%v) timed out", bp.Uid)
}

// ProtoToBlpPosition converts a proto to a BlpPosition
func ProtoToBlpPosition(b *pbt.BlpPosition) *BlpPosition {
	pos, err := myproto.DecodeReplicationPosition(b.Position)
	if err != nil {
		panic(fmt.Errorf("cannot decode position: %v", err))
	}
	return &BlpPosition{
		Uid:      b.Uid,
		Position: pos,
	}
}

func (s *server) InitSlave(ctx context.Context, request *pb.InitSlaveRequest) (*pb.InitSlaveResponse, error) {
	ctx = callinfo.GRPCCallInfo(ctx)
	response := &pb.InitSlaveResponse{}
	return response, s.agent.RPCWrapLockAction(ctx, actionnode.TabletActionInitSlave, request, response, true, func() error {
		position, err := myproto.DecodeReplicationPosition(request.ReplicationPosition)
		if err != nil {
			return err
		}
		return s.agent.InitSlave(ctx, topo.ProtoToTabletAlias(request.Parent), position, request.TimeCreatedNs)
	})
}

func (s *server) PopulateReparentJournal(ctx context.Context, request *pb.PopulateReparentJournalRequest) (*pb.PopulateReparentJournalResponse, error) {
	ctx = callinfo.GRPCCallInfo(ctx)
	response := &pb.PopulateReparentJournalResponse{}
	return response, s.agent.RPCWrap(ctx, actionnode.TabletActionPopulateReparentJournal, request, response, func() error {
		position, err := myproto.DecodeReplicationPosition(request.ReplicationPosition)
		if err != nil {
			return err
		}
		return s.agent.PopulateReparentJournal(ctx, request.TimeCreatedNs, request.ActionName, topo.ProtoToTabletAlias(request.MasterAlias), position)
	})
}

func (s *server) PromoteSlaveWhenCaughtUp(ctx context.Context, request *pb.PromoteSlaveWhenCaughtUpRequest) (*pb.PromoteSlaveWhenCaughtUpResponse, error) {
	ctx = callinfo.GRPCCallInfo(ctx)
	response := &pb.PromoteSlaveWhenCaughtUpResponse{}
	return response, s.agent.RPCWrapLockAction(ctx, actionnode.TabletActionPromoteSlaveWhenCaughtUp, request, response, true, func() error {
		position, err := myproto.DecodeReplicationPosition(request.Position)
		if err != nil {
			return err
		}
		position, err = s.agent.PromoteSlaveWhenCaughtUp(ctx, position)
		if err == nil {
			response.Position = myproto.EncodeReplicationPosition(position)
		}
		return err
	})
}

func (s *server) StopSlaveMinimum(ctx context.Context, request *pb.StopSlaveMinimumRequest) (*pb.StopSlaveMinimumResponse, error) {
	ctx = callinfo.GRPCCallInfo(ctx)
	response := &pb.StopSlaveMinimumResponse{}
	return response, s.agent.RPCWrapLock(ctx, actionnode.TabletActionStopSlaveMinimum, request, response, true, func() error {
		position, err := myproto.DecodeReplicationPosition(request.Position)
		if err != nil {
			return err
		}
		position, err = s.agent.StopSlaveMinimum(ctx, position, time.Duration(request.WaitTimeout))
		if err == nil {
			response.Position = myproto.EncodeReplicationPosition(position)
		}
		return err
	})
}

// PromoteSlave is part of the tmclient.TabletManagerClient interface
func (client *Client) PromoteSlave(ctx context.Context, tablet *topo.TabletInfo) (myproto.ReplicationPosition, error) {
	cc, c, err := client.dial(ctx, tablet)
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	defer cc.Close()
	response, err := c.PromoteSlave(ctx, &pb.PromoteSlaveRequest{})
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	position, err := myproto.DecodeReplicationPosition(response.Position)
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	return position, err
}

// ReadStartPosition will return the current start position and the flags for
// the provided binlog player.
func ReadStartPosition(dbClient VtClient, uid uint32) (*proto.BlpPosition, string, error) {
	selectRecovery := QueryBlpCheckpoint(uid)
	qr, err := dbClient.ExecuteFetch(selectRecovery, 1, true)
	if err != nil {
		return nil, "", fmt.Errorf("error %v in selecting from recovery table %v", err, selectRecovery)
	}
	if qr.RowsAffected != 1 {
		return nil, "", fmt.Errorf("checkpoint information not available in db for %v", uid)
	}
	pos, err := myproto.DecodeReplicationPosition(qr.Rows[0][0].String())
	if err != nil {
		return nil, "", err
	}
	return &proto.BlpPosition{
		Uid:      uid,
		Position: pos,
	}, string(qr.Rows[0][1].Raw()), nil
}

// RunBlpUntil is part of the tmclient.TabletManagerClient interface
func (client *Client) RunBlpUntil(ctx context.Context, tablet *topo.TabletInfo, positions *blproto.BlpPositionList, waitTime time.Duration) (myproto.ReplicationPosition, error) {
	cc, c, err := client.dial(ctx, tablet)
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	defer cc.Close()
	response, err := c.RunBlpUntil(ctx, &pb.RunBlpUntilRequest{
		BlpPositions: blproto.BlpPositionListToProto(positions),
		WaitTimeout:  int64(waitTime),
	})
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	position, err := myproto.DecodeReplicationPosition(response.Position)
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	return position, err
}

// StopSlaveMinimum is part of the tmclient.TabletManagerClient interface
func (client *Client) StopSlaveMinimum(ctx context.Context, tablet *topo.TabletInfo, minPos myproto.ReplicationPosition, waitTime time.Duration) (myproto.ReplicationPosition, error) {
	cc, c, err := client.dial(ctx, tablet)
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	defer cc.Close()
	response, err := c.StopSlaveMinimum(ctx, &pb.StopSlaveMinimumRequest{
		Position:    myproto.EncodeReplicationPosition(minPos),
		WaitTimeout: int64(waitTime),
	})
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	position, err := myproto.DecodeReplicationPosition(response.Position)
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	return position, err
}