Fix reduce view row collation with unicode equivalent keys

Previously, view reduce collation with keys relied on the keys in the
rows returned from the view shards to exactly match (=:=) the keys
specified in the args. However, in the case when there are multiple
rows which compare equal with the unicode collator, that may not
always be the case.

In that case when the rows are fetched from the row dict by key, they
should be matched using the same collation algorithm as the one used
on the view shards.

src/fabric/src/fabric_view.erl

@@ -242,9 +242,8 @@ get_next_row(#collector{reducer = RedSrc} = St) when RedSrc =/= undefined ->
         collation = Collation
     } = St,
     {Key, RestKeys} = find_next_key(Keys, Dir, Collation, RowDict),
-    case dict:find(Key, RowDict) of
-    {ok, Records} ->
-        NewRowDict = dict:erase(Key, RowDict),
+    case reduce_row_dict_take(Key, RowDict, Collation) of
+    {Records, NewRowDict} ->
         Counters = lists:foldl(fun(#view_row{worker={Worker,From}}, CntrsAcc) ->
             case From of
                 {Pid, _} when is_pid(Pid) ->
@@ -269,6 +268,22 @@ get_next_row(State) ->
     Counters1 = fabric_dict:update_counter(Worker, -1, Counters0),
     {Row, State#collector{rows = Rest, counters=Counters1}}.
 
+reduce_row_dict_take(Key, Dict, <<"raw">>) ->
+    dict:take(Key, Dict);
+reduce_row_dict_take(Key, Dict, _Collation) ->
+    IsEq = fun(K, _) -> couch_ejson_compare:less(K, Key) =:= 0 end,
+    KVs = dict:to_list(dict:filter(IsEq, Dict)),
+    case KVs of
+        [] ->
+            error;
+        [_ | _] ->
+            {Keys, Vals} = lists:unzip(KVs),
+            NewDict = lists:foldl(fun(K, Acc) ->
+                dict:erase(K, Acc)
+            end, Dict, Keys),
+            {lists:flatten(Vals), NewDict}
+    end.
+
 %% TODO: rectify nil <-> undefined discrepancies
 find_next_key(nil, Dir, Collation, RowDict) ->
     find_next_key(undefined, Dir, Collation, RowDict);