Add support for inclusive_end to fold_reduce, BugzID 11069

1 files changed, 26 insertions, 40 deletions

diff --git a/apps/couch/src/couch_btree.erl b/apps/couch/src/couch_btree.erl
index 4ed3fe54..34c756f9 100644
--- a/apps/couch/src/couch_btree.erl
+++ b/apps/couch/src/couch_btree.erl
@@ -75,18 +75,18 @@ final_reduce(Reduce, {KVs, Reductions}) ->
 
 fold_reduce(#btree{root=Root}=Bt, Fun, Acc, Options) ->
     Dir = couch_util:get_value(dir, Options, fwd),
-    StartKey = couch_util:get_value(start_key, Options),
-    EndKey = couch_util:get_value(end_key, Options),
+    InRangeFun = make_key_in_end_range_function(Bt,Dir,Options),
     KeyGroupFun = couch_util:get_value(key_group_fun, Options, fun(_,_) -> true end),
-    {StartKey2, EndKey2} =
     case Dir of
-        rev -> {EndKey, StartKey};
-        fwd -> {StartKey, EndKey}
+    fwd ->
+        StartKey = couch_util:get_value(start_key, Options);
+    rev ->
+        StartKey = couch_util:get_value(end_key, Options)
     end,
     try
         {ok, Acc2, GroupedRedsAcc2, GroupedKVsAcc2, GroupedKey2} =
-            reduce_stream_node(Bt, Dir, Root, StartKey2, EndKey2, undefined, [], [],
-            KeyGroupFun, Fun, Acc),
+            reduce_stream_node(Bt, Dir, Root, StartKey, InRangeFun, undefined,
+                [], [], KeyGroupFun, Fun, Acc),
         if GroupedKey2 == undefined ->
             {ok, Acc2};
         true ->
@@ -461,21 +461,21 @@ modify_kvnode(Bt, NodeTuple, LowerBound, [{ActionType, ActionKey, ActionValue} |
     end.
 
 
-reduce_stream_node(_Bt, _Dir, nil, _KeyStart, _KeyEnd, GroupedKey, GroupedKVsAcc,
+reduce_stream_node(_Bt, _Dir, nil, _KeyStart, _InRange, GroupedKey, GroupedKVsAcc,
         GroupedRedsAcc, _KeyGroupFun, _Fun, Acc) ->
     {ok, Acc, GroupedRedsAcc, GroupedKVsAcc, GroupedKey};
-reduce_stream_node(Bt, Dir, {P, _R}, KeyStart, KeyEnd, GroupedKey, GroupedKVsAcc,
+reduce_stream_node(Bt, Dir, {P, _R}, KeyStart, InRange, GroupedKey, GroupedKVsAcc,
         GroupedRedsAcc, KeyGroupFun, Fun, Acc) ->
     case get_node(Bt, P) of
     {kp_node, NodeList} ->
-        reduce_stream_kp_node(Bt, Dir, NodeList, KeyStart, KeyEnd, GroupedKey,
+        reduce_stream_kp_node(Bt, Dir, NodeList, KeyStart, InRange, GroupedKey,
                 GroupedKVsAcc, GroupedRedsAcc, KeyGroupFun, Fun, Acc);
     {kv_node, KVs} ->
-        reduce_stream_kv_node(Bt, Dir, KVs, KeyStart, KeyEnd, GroupedKey,
+        reduce_stream_kv_node(Bt, Dir, KVs, KeyStart, InRange, GroupedKey,
                 GroupedKVsAcc, GroupedRedsAcc, KeyGroupFun, Fun, Acc)
     end.
 
-reduce_stream_kv_node(Bt, Dir, KVs, KeyStart, KeyEnd,
+reduce_stream_kv_node(Bt, Dir, KVs, KeyStart, InRange,
                         GroupedKey, GroupedKVsAcc, GroupedRedsAcc,
                         KeyGroupFun, Fun, Acc) ->
 
@@ -486,16 +486,7 @@ reduce_stream_kv_node(Bt, Dir, KVs, KeyStart, KeyEnd,
     _ ->
         lists:dropwhile(fun({Key,_}) -> less(Bt, Key, KeyStart) end, KVs)
     end,
-    KVs2 =
-    case KeyEnd of
-    undefined ->
-        GTEKeyStartKVs;
-    _ ->
-        lists:takewhile(
-            fun({Key,_}) ->
-                not less(Bt, KeyEnd, Key)
-            end, GTEKeyStartKVs)
-    end,
+    KVs2 = lists:takewhile(fun({Key,_}) -> InRange(Key) end, GTEKeyStartKVs),
     reduce_stream_kv_node2(Bt, adjust_dir(Dir, KVs2), GroupedKey, GroupedKVsAcc, GroupedRedsAcc,
                         KeyGroupFun, Fun, Acc).
 
@@ -527,7 +518,7 @@ reduce_stream_kv_node2(Bt, [{Key, Value}| RestKVs], GroupedKey, GroupedKVsAcc,
         end
     end.
 
-reduce_stream_kp_node(Bt, Dir, NodeList, KeyStart, KeyEnd,
+reduce_stream_kp_node(Bt, Dir, NodeList, KeyStart, InRange,
                         GroupedKey, GroupedKVsAcc, GroupedRedsAcc,
                         KeyGroupFun, Fun, Acc) ->
     Nodes =
@@ -540,29 +531,24 @@ reduce_stream_kp_node(Bt, Dir, NodeList, KeyStart, KeyEnd,
                 less(Bt, Key, KeyStart)
             end, NodeList)
     end,
-    NodesInRange =
-    case KeyEnd of
-    undefined ->
-        Nodes;
-    _ ->
-        {InRange, MaybeInRange} = lists:splitwith(
-            fun({Key,_}) ->
-                less(Bt, Key, KeyEnd)
-            end, Nodes),
-        InRange ++ case MaybeInRange of [] -> []; [FirstMaybe|_] -> [FirstMaybe] end
+    case lists:splitwith(fun({Key,_}) -> InRange(Key) end, Nodes) of
+    {InTheRange, []} ->
+        NodesInRange = InTheRange;
+    {InTheRange, [FirstMaybe|_]} ->
+        NodesInRange = InTheRange ++ [FirstMaybe]
     end,
-    reduce_stream_kp_node2(Bt, Dir, adjust_dir(Dir, NodesInRange), KeyStart, KeyEnd,
+    reduce_stream_kp_node2(Bt, Dir, adjust_dir(Dir, NodesInRange), KeyStart, InRange,
         GroupedKey, GroupedKVsAcc, GroupedRedsAcc, KeyGroupFun, Fun, Acc).
 
 
-reduce_stream_kp_node2(Bt, Dir, [{_Key, NodeInfo} | RestNodeList], KeyStart, KeyEnd,
+reduce_stream_kp_node2(Bt, Dir, [{_Key, NodeInfo} | RestNodeList], KeyStart, InRange,
                         undefined, [], [], KeyGroupFun, Fun, Acc) ->
     {ok, Acc2, GroupedRedsAcc2, GroupedKVsAcc2, GroupedKey2} =
-            reduce_stream_node(Bt, Dir, NodeInfo, KeyStart, KeyEnd, undefined,
+            reduce_stream_node(Bt, Dir, NodeInfo, KeyStart, InRange, undefined,
                 [], [], KeyGroupFun, Fun, Acc),
-    reduce_stream_kp_node2(Bt, Dir, RestNodeList, KeyStart, KeyEnd, GroupedKey2,
+    reduce_stream_kp_node2(Bt, Dir, RestNodeList, KeyStart, InRange, GroupedKey2,
             GroupedKVsAcc2, GroupedRedsAcc2, KeyGroupFun, Fun, Acc2);
-reduce_stream_kp_node2(Bt, Dir, NodeList, KeyStart, KeyEnd,
+reduce_stream_kp_node2(Bt, Dir, NodeList, KeyStart, InRange,
         GroupedKey, GroupedKVsAcc, GroupedRedsAcc, KeyGroupFun, Fun, Acc) ->
     {Grouped0, Ungrouped0} = lists:splitwith(fun({Key,_}) ->
         KeyGroupFun(GroupedKey, Key) end, NodeList),
@@ -578,9 +564,9 @@ reduce_stream_kp_node2(Bt, Dir, NodeList, KeyStart, KeyEnd,
     case UngroupedNodes of
     [{_Key, NodeInfo}|RestNodes] ->
         {ok, Acc2, GroupedRedsAcc2, GroupedKVsAcc2, GroupedKey2} =
-            reduce_stream_node(Bt, Dir, NodeInfo, KeyStart, KeyEnd, GroupedKey,
+            reduce_stream_node(Bt, Dir, NodeInfo, KeyStart, InRange, GroupedKey,
                 GroupedKVsAcc, GroupedReds ++ GroupedRedsAcc, KeyGroupFun, Fun, Acc),
-        reduce_stream_kp_node2(Bt, Dir, RestNodes, KeyStart, KeyEnd, GroupedKey2,
+        reduce_stream_kp_node2(Bt, Dir, RestNodes, KeyStart, InRange, GroupedKey2,
                 GroupedKVsAcc2, GroupedRedsAcc2, KeyGroupFun, Fun, Acc2);
     [] ->
         {ok, Acc, GroupedReds ++ GroupedRedsAcc, GroupedKVsAcc, GroupedKey}