Prefer values from old tree when merging, COUCHDB-968

This commit represents a substantial refactor of the key tree merging
logic, some of which is not strictly necessary for the resolution of
COUCHDB-968.

Two etap test cases checking the ability to merge in a non-linear tree
are removed because the functionality is no longer supported.  CouchDB
only ever merged a linear revision history into an existing revision
tree.

git-svn-id: https://svn.apache.org/repos/asf/couchdb/branches/1.1.x@1043460 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 45 insertions, 60 deletions

diff --git a/src/couchdb/couch_key_tree.erl b/src/couchdb/couch_key_tree.erl
index 4fe09bf3..bc3775c4 100644
--- a/src/couchdb/couch_key_tree.erl
+++ b/src/couchdb/couch_key_tree.erl
@@ -43,50 +43,53 @@ merge(A, B) ->
 
 merge_one([], Insert, OutAcc, ConflictsAcc) ->
     {ok, [Insert | OutAcc], ConflictsAcc};
-merge_one([{Start, Tree}|Rest], {StartInsert, TreeInsert}, OutAcc, ConflictsAcc) ->
-    if Start =< StartInsert ->
-        StartA = Start,
-        StartB = StartInsert,
-        TreeA = Tree,
-        TreeB = TreeInsert;
-    true ->
-        StartB = Start,
-        StartA = StartInsert,
-        TreeB = Tree,
-        TreeA = TreeInsert
-    end,
-    case merge_at([TreeA], StartB - StartA, TreeB) of
-    {ok, [CombinedTrees], Conflicts} ->
-        merge_one(Rest, {StartA, CombinedTrees}, OutAcc, Conflicts or ConflictsAcc);
+merge_one([{Start, Tree}|Rest], {StartInsert, TreeInsert}, Acc, HasConflicts) ->
+    case merge_at([Tree], StartInsert - Start, [TreeInsert]) of
+    {ok, [Merged], Conflicts} ->
+        MergedStart = lists:min([Start, StartInsert]),
+        merge_one(Rest, {MergedStart, Merged}, Acc, Conflicts or HasConflicts);
     no ->
-        merge_one(Rest, {StartB, TreeB}, [{StartA, TreeA} | OutAcc], ConflictsAcc)
+        AccOut = [{Start, Tree} | Acc],
+        merge_one(Rest, {StartInsert, TreeInsert}, AccOut, HasConflicts)
     end.
 
+merge_at(_Ours, _Place, []) ->
+    no;
 merge_at([], _Place, _Insert) ->
     no;
-merge_at([{Key, Value, SubTree}|Sibs], 0, {InsertKey, InsertValue, InsertSubTree}) ->
-    if Key == InsertKey ->
-        {Merge, Conflicts} = merge_simple(SubTree, InsertSubTree),
-        {ok, [{Key, Value, Merge} | Sibs], Conflicts};
-    true ->
-        case merge_at(Sibs, 0, {InsertKey, InsertValue, InsertSubTree}) of
-        {ok, Merged, Conflicts} ->
-            {ok, [{Key, Value, SubTree} | Merged], Conflicts};
-        no ->
-            no
-        end
-    end;
-merge_at([{Key, Value, SubTree}|Sibs], Place, Insert) ->
-    case merge_at(SubTree, Place - 1,Insert) of
+merge_at([{Key, Value, SubTree}|Sibs], Place, InsertTree) when Place > 0 ->
+    % inserted starts later than committed, need to drill into committed subtree
+    case merge_at(SubTree, Place - 1, InsertTree) of
     {ok, Merged, Conflicts} ->
         {ok, [{Key, Value, Merged} | Sibs], Conflicts};
     no ->
-        case merge_at(Sibs, Place, Insert) of
+        case merge_at(Sibs, Place, InsertTree) of
         {ok, Merged, Conflicts} ->
             {ok, [{Key, Value, SubTree} | Merged], Conflicts};
         no ->
             no
         end
+    end;
+merge_at(OurTree, Place, [{Key, Value, SubTree}]) when Place < 0 ->
+    % inserted starts earlier than committed, need to drill into insert subtree
+    case merge_at(OurTree, Place + 1, SubTree) of
+    {ok, Merged, Conflicts} ->
+        {ok, [{Key, Value, Merged}], Conflicts};
+    no ->
+        no
+    end;
+merge_at([{Key, Value, SubTree}|Sibs], 0, [{Key, _Value, InsertSubTree}]) ->
+    {Merged, Conflicts} = merge_simple(SubTree, InsertSubTree),
+    {ok, [{Key, Value, Merged} | Sibs], Conflicts};
+merge_at([{OurKey, _, _} | _], 0, [{Key, _, _}]) when OurKey > Key ->
+    % siblings keys are ordered, no point in continuing
+    no;
+merge_at([Tree | Sibs], 0, InsertTree) ->
+    case merge_at(Sibs, 0, InsertTree) of
+    {ok, Merged, Conflicts} ->
+        {ok, [Tree | Merged], Conflicts};
+    no ->
+        no
     end.
 
 % key tree functions
@@ -94,22 +97,16 @@ merge_simple([], B) ->
     {B, false};
 merge_simple(A, []) ->
     {A, false};
-merge_simple([ATree | ANextTree], [BTree | BNextTree]) ->
-    {AKey, AValue, ASubTree} = ATree,
-    {BKey, _BValue, BSubTree} = BTree,
-    if
-    AKey == BKey ->
-        %same key
-        {MergedSubTree, Conflict1} = merge_simple(ASubTree, BSubTree),
-        {MergedNextTree, Conflict2} = merge_simple(ANextTree, BNextTree),
-        {[{AKey, AValue, MergedSubTree} | MergedNextTree], Conflict1 or Conflict2};
-    AKey < BKey ->
-        {MTree, _} = merge_simple(ANextTree, [BTree | BNextTree]),
-        {[ATree | MTree], true};
-    true ->
-        {MTree, _} = merge_simple([ATree | ANextTree], BNextTree),
-        {[BTree | MTree], true}
-    end.
+merge_simple([{Key, Value, SubA} | NextA], [{Key, _, SubB} | NextB]) ->
+    {MergedSubTree, Conflict1} = merge_simple(SubA, SubB),
+    {MergedNextTree, Conflict2} = merge_simple(NextA, NextB),
+    {[{Key, Value, MergedSubTree} | MergedNextTree], Conflict1 or Conflict2};
+merge_simple([{A, _, _} = Tree | Next], [{B, _, _} | _] = Insert) when A < B ->
+    {Merged, _} = merge_simple(Next, Insert),
+    {[Tree | Merged], true};
+merge_simple(Ours, [Tree | Next]) ->
+    {Merged, _} = merge_simple(Ours, Next),
+    {[Tree | Merged], true}.
 
 find_missing(_Tree, []) ->
     [];
diff --git a/test/etap/060-kt-merging.t b/test/etap/060-kt-merging.t
index d6b13d6d..73744e52 100755
--- a/test/etap/060-kt-merging.t
+++ b/test/etap/060-kt-merging.t
@@ -15,7 +15,7 @@
 
 main(_) ->
     test_util:init_code_path(),
-    etap:plan(16),
+    etap:plan(14),
     case (catch test()) of
         ok ->
             etap:end_tests();
@@ -89,24 +89,12 @@ test() ->
     ),
 
     etap:is(
-        {TwoChildSibs, no_conflicts},
-        couch_key_tree:merge(Stemmed1b, TwoChildSibs),
-        "Merging in the opposite direction."
-    ),
-
-    etap:is(
         {TwoChildSibs2, no_conflicts},
         couch_key_tree:merge(TwoChildSibs2, Stemmed1bb),
         "Merging a stem at a deeper level."
     ),
 
     etap:is(
-        {TwoChildSibs2, no_conflicts},
-        couch_key_tree:merge(Stemmed1bb, TwoChildSibs2),
-        "Merging a deeper level in opposite order."
-    ),
-
-    etap:is(
         {TwoChild, no_conflicts},
         couch_key_tree:merge(TwoChild, Stemmed1aa),
         "Merging a single tree with a deeper stem."