anchor.diff:          E ->
               E
       end.
  +
  +count_parts(Parts) ->
  +    Gets = [extract_gets(Part) || Part <- Parts],
  +    Gets1 = lists:flatten(Gets),
  +    Total = length(Gets1),
  +    Unique = length(lists:usort(Gets1)),
  +    {Total, Unique}.
  +
  +extract_gets({named, _, P}) ->
  +    extract_gets(P);
  +extract_gets({combine, _Fun, Parts}) ->
  +    [extract_gets(P) || P <- Parts];
  +extract_gets({calc, _, {get, {B, M}}}) ->
  +    {get, {B, M}}.
   
   expand_parts(Parts, Buckets) ->
       Parts1 = [expand_part(P, Buckets) || P <- Parts],
  
positive.code:             pdebug('prepare', "Buckets fetched.", []),
              Parts1 = expand_parts(Parts, Buckets1),
              pdebug('prepare', "Parts expanded.", []),
              case name_parts(Parts1, [], Aliases, Buckets1) of
                  {ok, Parts2} ->
                      pdebug('prepare', "Naing applied.", []),
  
                      {ok, {Parts2, Start, Count}};
                  E ->
                      pdebug('prepare', "Naing failed.", []),
                      E
  
negative.code: run(Query, Timeout) ->
      put(start, erlang:system_time()),
      case prepare(Query) of
          {ok, {Total, Unique, Parts, Start, Count}} ->
              pdebug('query', "preperation done.", []),
              WaitRef = make_ref(),
              Funnel = {dqe_funnel, [[{dqe_collect, [Part]} || Part <- Parts]]},
              Sender = {dflow_send, [self(), WaitRef, Funnel]},
              %% We only optimize the flow when there are at least 10% duplicate
              %% gets, or in other words if less then 90% of the requests are
              %% unique
              FlowOpts = case Unique / Total of
                             UniquePercentage when UniquePercentage > 0.9 ->
                                 [terminate_when_done];
                             _ ->
                                 [optimize, terminate_when_done]
                         end,
              {ok, _Ref, Flow} = dflow:build(Sender, FlowOpts),
              pdebug('query', "flow generated.", []),
              dflow:start(Flow, {Start, Count}),
              case  dflow_send:recv(WaitRef, Timeout) of
  