Add downgrade function to downgrade database with clustered purgeCOUCHDB-3226-downgrade-clustered-purge

  - provide downgrade functionality in case there is need to downgrade
   database which was modified by builds with clustered purge feature.
   After database is modified by builds with clustered purge feature,
   the disk version is bumped to 7. Also, there are two newly introduced
   trees: PurgeTreeState and PurgeSeqTreeState.

   Once downgrade function is called, the disk version is back to the
   latest version of current build. Also, purge_seq and purged_docs
   are re-introduced again.

COUCHDB-3226

Add test

7 files changed, 262 insertions, 4 deletions

diff --git a/src/couch/src/couch_bt_engine.erl b/src/couch/src/couch_bt_engine.erl
index ee0d6d864..2899e4556 100644
--- a/src/couch/src/couch_bt_engine.erl
+++ b/src/couch/src/couch_bt_engine.erl
@@ -85,7 +85,11 @@
     seq_tree_reduce/2,
 
     local_tree_split/1,
-    local_tree_join/2
+    local_tree_join/2,
+
+    purge_tree_reduce/2,
+    purge_seq_tree_split/1,
+    purge_seq_tree_join/2
 ]).
 
 
@@ -101,6 +105,9 @@
 -include("couch_bt_engine.hrl").
 
 
+-define(CLUSTERED_PURGE_DISK_VERSION, 7).
+
+
 exists(FilePath) ->
     case filelib:is_file(FilePath) of
         true ->
@@ -627,6 +634,21 @@ local_tree_join(Id, {Rev, BodyData}) when is_integer(Rev) ->
     }.
 
 
+purge_seq_tree_split({PurgeSeq, UUID, DocId, Revs}) ->
+    {PurgeSeq, {UUID, DocId, Revs}}.
+
+
+purge_seq_tree_join(PurgeSeq, {UUID, DocId, Revs}) ->
+    {PurgeSeq, UUID, DocId, Revs}.
+
+
+purge_tree_reduce(reduce, IdRevs) ->
+    % count the number of purge requests
+    length(IdRevs);
+purge_tree_reduce(rereduce, Reds) ->
+    lists:sum(Reds).
+
+
 set_update_seq(#st{header = Header} = St, UpdateSeq) ->
     {ok, St#st{
         header = couch_bt_engine_header:set(Header, [
@@ -681,8 +703,17 @@ init_state(FilePath, Fd, Header0, Options) ->
 
     Compression = couch_compress:get_compression_method(),
 
-    Header1 = couch_bt_engine_header:upgrade(Header0),
-    Header = set_default_security_object(Fd, Header1, Compression, Options),
+    DiskVersion = couch_bt_engine_header:disk_version(Header0),
+    Latest = couch_bt_engine_header:latest_disk_version(),
+    Header1 = case DiskVersion of
+        N when N =< Latest ->
+            Header0;
+        ?CLUSTERED_PURGE_DISK_VERSION ->
+            downgrade_purge_info(Fd, Header0)
+    end,
+
+    Header2 = couch_bt_engine_header:upgrade(Header1),
+    Header = set_default_security_object(Fd, Header2, Compression, Options),
 
     IdTreeState = couch_bt_engine_header:id_tree_state(Header),
     {ok, IdTree} = couch_btree:open(IdTreeState, Fd, [
@@ -727,7 +758,7 @@ init_state(FilePath, Fd, Header0, Options) ->
     % to be written to disk.
     case Header /= Header0 of
         true ->
-            {ok, NewSt} = commit_data(St),
+            {ok, NewSt} = commit_data(St#st{needs_commit = true}),
             NewSt;
         false ->
             St
@@ -763,6 +794,63 @@ set_default_security_object(Fd, Header, Compression, Options) ->
     end.
 
 
+% Rollback for clustered purge.
+downgrade_purge_info(Fd, Header) ->
+    {
+        db_header,
+        _DiskVer,
+        UpSeq,
+        _Unused,
+        IdTreeState,
+        SeqTreeState,
+        LocalTreeState,
+        PurgeTreeState,
+        PurgeSeqTreeState,
+        SecurityPtr,
+        RevsLimit,
+        Uuid,
+        Epochs,
+        CompactedSeq,
+        _PDocsLimit
+    } = Header,
+
+    {PSeq, PurgedDocsPtr} = case PurgeTreeState of
+        nil ->
+            {0, nil};
+        PurgeSeqInOldVer when is_integer(PurgeSeqInOldVer)->
+            {PurgeSeqInOldVer, nil};
+        _ when is_tuple(PurgeTreeState) ->
+            {ok, PSTree} = couch_btree:open(PurgeSeqTreeState, Fd, [
+                {split, fun ?MODULE:purge_seq_tree_split/1},
+                {join, fun ?MODULE:purge_seq_tree_join/2},
+                {reduce, fun ?MODULE:purge_tree_reduce/2}
+            ]),
+            Fun = fun({PurgeSeq, _, _, _}, _Reds, _Acc) ->
+                {stop, PurgeSeq}
+            end,
+            {ok, _, PurgeSeq} = couch_btree:fold(PSTree, Fun, 0, [{dir, rev}]),
+            Compression = couch_compress:get_compression_method(),
+            AppendOps = [{compression, Compression}],
+            {ok, Ptr, _} = couch_file:append_term(Fd, [], AppendOps),
+            if PurgeSeq > 0 -> {PurgeSeq + 2, Ptr}; true -> {0, Ptr} end
+    end,
+
+    NewHeader = couch_bt_engine_header:new(),
+    couch_bt_engine_header:set(NewHeader, [
+        {update_seq, UpSeq},
+        {id_tree_state, IdTreeState},
+        {seq_tree_state, SeqTreeState},
+        {local_tree_state, LocalTreeState},
+        {purge_seq, PSeq},
+        {purged_docs, PurgedDocsPtr},
+        {security_ptr, SecurityPtr},
+        {revs_limit, RevsLimit},
+        {uuid, Uuid},
+        {epochs, Epochs},
+        {compacted_seq, CompactedSeq}
+    ]).
+
+
 delete_compaction_files(FilePath) ->
     RootDir = config:get("couchdb", "database_dir", "."),
     DelOpts = [{context, compaction}],
diff --git a/src/couch/src/couch_bt_engine_header.erl b/src/couch/src/couch_bt_engine_header.erl
index 3d24f3189..2eaa6ff8b 100644
--- a/src/couch/src/couch_bt_engine_header.erl
+++ b/src/couch/src/couch_bt_engine_header.erl
@@ -26,6 +26,7 @@
 
 -export([
     disk_version/1,
+    latest_disk_version/0,
     update_seq/1,
     id_tree_state/1,
     seq_tree_state/1,
@@ -134,6 +135,10 @@ disk_version(Header) ->
     get_field(Header, disk_version).
 
 
+latest_disk_version() ->
+    ?LATEST_DISK_VERSION.
+
+
 update_seq(Header) ->
     get_field(Header, update_seq).
 
diff --git a/src/couch/test/couch_bt_engine_downgrade_tests.erl b/src/couch/test/couch_bt_engine_downgrade_tests.erl
new file mode 100644
index 000000000..b080de351
--- /dev/null
+++ b/src/couch/test/couch_bt_engine_downgrade_tests.erl
@@ -0,0 +1,165 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+%   http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_bt_engine_downgrade_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+
+setup() ->
+    Ctx = test_util:start_couch(),
+    DbDir = config:get("couchdb", "database_dir"),
+    DbFileNames = [
+        "db_with_v6_and_0_purge_req.couch",
+        "db_with_v7_and_0_purge_req.couch",
+        "db_with_v7_and_1_purge_req.couch",
+        "db_with_v7_and_2_purge_req.couch"
+    ],
+    NewPaths = lists:map(fun(DbFileName) ->
+        OldDbFilePath = filename:join([?FIXTURESDIR, DbFileName]),
+        NewDbFilePath = filename:join([DbDir, DbFileName]),
+        ok = filelib:ensure_dir(NewDbFilePath),
+        file:delete(NewDbFilePath),
+        {ok, _} = file:copy(OldDbFilePath, NewDbFilePath),
+        NewDbFilePath
+    end, DbFileNames),
+    {Ctx, NewPaths}.
+
+
+teardown({Ctx, Paths}) ->
+    test_util:stop_couch(Ctx),
+    lists:foreach(fun(Path) ->
+        file:delete(Path)
+    end, Paths).
+
+
+downgrade_test_() ->
+    {
+        "Couch Bt Engine downgrade tests",
+        {
+            setup,
+            fun setup/0,
+            fun teardown/1,
+            [
+                t_no_downgrade_without_purge_req(),
+                t_downgrade_without_purge_req(),
+                t_downgrade_with_1_purge_req(),
+                t_downgrade_with_2_purge_req()
+            ]
+        }
+    }.
+
+
+t_no_downgrade_without_purge_req() ->
+    ?_test(begin
+        % There are three documents in the fixture
+        % db with zero purge entries
+        DbName = <<"db_with_v6_and_0_purge_req">>,
+        ?assertEqual(6, get_disk_version_from_header(DbName)),
+        {ok, PurgeSeq} = couch_util:with_db(DbName, fun(Db) ->
+            couch_db:get_purge_seq(Db)
+        end),
+        ?assertEqual(0, PurgeSeq),
+
+        {ok, _} = save_doc(DbName, {[{<<"_id">>, <<"doc4">>}, {<<"v">>, 1}]}),
+        {ok, _} = save_doc(DbName, {[{<<"_id">>, <<"doc5">>}, {<<"v">>, 2}]}),
+
+        couch_util:with_db(DbName, fun(Db) ->
+            ?assertEqual({ok, 5}, couch_db:get_doc_count(Db)),
+            ?assertEqual({ok, 0}, couch_db:get_purge_seq(Db)),
+            ?assertEqual(6, couch_db_engine:get_disk_version(Db))
+        end)
+
+    end).
+
+t_downgrade_without_purge_req() ->
+    ?_test(begin
+        % There are three documents in the fixture
+        % db with zero purge entries
+        DbName = <<"db_with_v7_and_0_purge_req">>,
+        ?assertEqual(7, get_disk_version_from_header(DbName)),
+        {ok, PurgeSeq} = couch_util:with_db(DbName, fun(Db) ->
+            couch_db:get_purge_seq(Db)
+        end),
+        ?assertEqual(0, PurgeSeq),
+
+        {ok, _} = save_doc(DbName, {[{<<"_id">>, <<"doc4">>}, {<<"v">>, 1}]}),
+        {ok, _} = save_doc(DbName, {[{<<"_id">>, <<"doc5">>}, {<<"v">>, 2}]}),
+
+        couch_util:with_db(DbName, fun(Db) ->
+            ?assertEqual({ok, 5}, couch_db:get_doc_count(Db)),
+            ?assertEqual({ok, 0}, couch_db:get_purge_seq(Db)),
+            ?assertEqual(6, couch_db_engine:get_disk_version(Db))
+        end)
+    end).
+
+
+t_downgrade_with_1_purge_req() ->
+    ?_test(begin
+        % There are two documents in the fixture database
+        % with a single purge entry
+        DbName = <<"db_with_v7_and_1_purge_req">>,
+        ?assertEqual(7, get_disk_version_from_header(DbName)),
+        {ok, PurgeReq} = couch_util:with_db(DbName, fun(Db) ->
+            couch_db:get_purge_seq(Db)
+        end),
+        ?assertEqual(3, PurgeReq),
+
+        {ok, _} = save_doc(DbName, {[{<<"_id">>, <<"doc4">>}, {<<"v">>, 1}]}),
+        {ok, _} = save_doc(DbName, {[{<<"_id">>, <<"doc5">>}, {<<"v">>, 2}]}),
+
+        couch_util:with_db(DbName, fun(Db) ->
+            ?assertEqual({ok, 4}, couch_db:get_doc_count(Db)),
+            ?assertEqual({ok, 3}, couch_db:get_purge_seq(Db)),
+            ?assertEqual(6, couch_db_engine:get_disk_version(Db))
+        end)
+    end).
+
+
+t_downgrade_with_2_purge_req() ->
+    ?_test(begin
+        % There is one document in the fixture database
+        % with two docs that have been purged
+        DbName = <<"db_with_v7_and_2_purge_req">>,
+        ?assertEqual(7, get_disk_version_from_header(DbName)),
+        {ok, PurgeReq} = couch_util:with_db(DbName, fun(Db) ->
+            couch_db:get_purge_seq(Db)
+        end),
+        ?assertEqual(4, PurgeReq),
+
+        {ok, _} = save_doc(DbName, {[{<<"_id">>, <<"doc4">>}, {<<"v">>, 1}]}),
+        {ok, _} = save_doc(DbName, {[{<<"_id">>, <<"doc5">>}, {<<"v">>, 2}]}),
+
+        couch_util:with_db(DbName, fun(Db) ->
+            ?assertEqual({ok, 3}, couch_db:get_doc_count(Db)),
+            ?assertEqual({ok, 4}, couch_db:get_purge_seq(Db)),
+            ?assertEqual(6, couch_db_engine:get_disk_version(Db))
+         end)
+    end).
+
+
+get_disk_version_from_header(DbFileName) ->
+    DbDir = config:get("couchdb", "database_dir"),
+    DbFilePath = filename:join([DbDir, ?l2b(?b2l(DbFileName) ++ ".couch")]),
+    {ok, Fd} = couch_file:open(DbFilePath, []),
+    {ok, Header} = couch_file:read_header(Fd),
+    DiskVerison = couch_bt_engine_header:disk_version(Header),
+    couch_file:close(Fd),
+    DiskVerison.
+
+
+save_doc(DbName, Json) ->
+    Doc = couch_doc:from_json_obj(Json),
+    couch_util:with_db(DbName, fun(Db) ->
+        couch_db:update_doc(Db, Doc, [])
+    end).
diff --git a/src/couch/test/fixtures/db_with_v6_and_0_purge_req.couch b/src/couch/test/fixtures/db_with_v6_and_0_purge_req.couch
new file mode 100644
index 000000000..814feb8e1
--- /dev/null
+++ b/src/couch/test/fixtures/db_with_v6_and_0_purge_req.couch
diff --git a/src/couch/test/fixtures/db_with_v7_and_0_purge_req.couch b/src/couch/test/fixtures/db_with_v7_and_0_purge_req.couch
new file mode 100644
index 000000000..98a8a18b1
--- /dev/null
+++ b/src/couch/test/fixtures/db_with_v7_and_0_purge_req.couch
diff --git a/src/couch/test/fixtures/db_with_v7_and_1_purge_req.couch b/src/couch/test/fixtures/db_with_v7_and_1_purge_req.couch
new file mode 100644
index 000000000..0996a88b1
--- /dev/null
+++ b/src/couch/test/fixtures/db_with_v7_and_1_purge_req.couch
diff --git a/src/couch/test/fixtures/db_with_v7_and_2_purge_req.couch b/src/couch/test/fixtures/db_with_v7_and_2_purge_req.couch
new file mode 100644
index 000000000..2f17a0d65
--- /dev/null
+++ b/src/couch/test/fixtures/db_with_v7_and_2_purge_req.couch