68276 new purge api

src/couch_mrview.erl

@@ -413,8 +413,13 @@ all_docs_fold(Db, #mrargs{keys=undefined}=Args, Callback, UAcc) ->
         update_seq=UpdateSeq,
         args=Args
     },
-    [Opts] = couch_mrview_util:all_docs_key_opts(Args),
-    {ok, Offset, FinalAcc} = couch_db:enum_docs(Db, fun map_fold/3, Acc, Opts),
+    [Opts1] = couch_mrview_util:all_docs_key_opts(Args),
+    % TODO: This is a terrible hack for now. We'll probably have
+    % to rewrite _all_docs to not be part of mrview and not expect
+    % a btree. For now non-btree's will just have to pass 0 or
+    % some fake reductions to get an offset.
+    Opts2 = [include_reductions | Opts1],
+    {ok, Offset, FinalAcc} = couch_db:fold_docs(Db, fun map_fold/3, Acc, Opts2),
     finish_fold(FinalAcc, [{total, Total}, {offset, Offset}]);
 all_docs_fold(Db, #mrargs{direction=Dir, keys=Keys0}=Args, Callback, UAcc) ->
     {ok, Info} = couch_db:get_db_info(Db),

src/couch_mrview_cleanup.erl

@@ -41,7 +41,23 @@ run(Db) ->
 
     lists:foreach(fun(FN) ->
         couch_log:debug("Deleting stale view file: ~s", [FN]),
-        couch_file:delete(RootDir, FN, [sync])
+        couch_file:delete(RootDir, FN, [sync]),
+        Sig = couch_mrview_util:get_signature_from_filename(FN),
+        if length(Sig) < 16 -> ok; true ->
+            case re:run(Sig,"^[a-fA-F0-9]+$",[{capture, none}]) of
+                match ->
+                    DocId = couch_mrview_util:get_local_purge_doc_id(Sig),
+                    case couch_db:open_doc(Db, DocId, []) of
+                        {ok, LocalPurgeDoc} ->
+                            couch_db:update_doc(Db,
+                                LocalPurgeDoc#doc{deleted=true}, [?ADMIN_CTX]);
+                        {not_found, _} ->
+                            ok
+                    end;
+                _ ->
+                    ok
+            end
+        end
     end, ToDelete),
 
     ok.

src/couch_mrview_compactor.erl

@@ -52,8 +52,7 @@ compact(State) ->
         {ok, Fd} = couch_mrview_util:open_file(CompactFName),
         ESt = couch_mrview_util:reset_index(Db, Fd, State),
 
-        {ok, DbReduce} = couch_btree:full_reduce(Db#db.id_tree),
-        Count = element(1, DbReduce),
+        {ok, Count} = couch_db:get_doc_count(Db),
 
         {ESt, Count}
     end),

src/couch_mrview_index.erl

@@ -18,6 +18,7 @@
 -export([start_update/3, purge/4, process_doc/3, finish_update/1, commit/1]).
 -export([compact/3, swap_compacted/2]).
 -export([index_file_exists/1]).
+-export([update_local_purge_doc/2, verify_index_exists/1]).
 
 -include_lib("couch/include/couch_db.hrl").
 -include_lib("couch_mrview/include/couch_mrview.hrl").
@@ -134,14 +135,17 @@ open(Db, State) ->
                 {ok, {OldSig, Header}} ->
                     % Matching view signatures.
                     NewSt = couch_mrview_util:init_state(Db, Fd, State, Header),
+                    maybe_create_local_purge_doc(Db, NewSt),
                     {ok, NewSt};
                 % end of upgrade code for <= 1.2.x
                 {ok, {Sig, Header}} ->
                     % Matching view signatures.
                     NewSt = couch_mrview_util:init_state(Db, Fd, State, Header),
+                    maybe_create_local_purge_doc(Db, NewSt),
                     {ok, NewSt};
                 _ ->
                     NewSt = couch_mrview_util:reset_index(Db, Fd, State),
+                    maybe_create_local_purge_doc(Db, NewSt),
                     {ok, NewSt}
             end;
         {error, Reason} = Error ->
@@ -204,3 +208,60 @@ index_file_exists(State) ->
     } = State,
     IndexFName = couch_mrview_util:index_file(DbName, Sig),
     filelib:is_file(IndexFName).
+
+
+update_local_purge_doc(Db, State) ->
+    Sig = couch_index_util:hexsig(get(signature, State)),
+    Doc = couch_doc:from_json_obj({[
+        {<<"_id">>, couch_mrview_util:get_local_purge_doc_id(Sig)},
+        {<<"purge_seq">>, get(purge_seq, State)},
+        {<<"timestamp_utc">>, list_to_binary(couch_util:utc_string())},
+        {<<"verify_module">>, <<"couch_mrview_index">>},
+        {<<"verify_function">>, <<"verify_index_exists">>},
+        {<<"verify_options">>, {[
+            {<<"dbname">>, get(db_name, State)},
+            {<<"ddoc_id">>, get(idx_name, State)},
+            {<<"signature">>, Sig}
+        ]}},
+        {<<"type">>, <<"mrview">>}
+    ]}),
+    couch_db:update_doc(Db, Doc, []).
+
+
+verify_index_exists(Options) ->
+    ShardDbName = couch_mrview_util:get_value_from_options(<<"dbname">>, Options),
+    DDocId = couch_mrview_util:get_value_from_options(<<"ddoc_id">>, Options),
+    SigInLocal = couch_mrview_util:get_value_from_options(<<"signature">>, Options),
+    case couch_db:open_int(ShardDbName, []) of
+        {ok, Db} ->
+            try
+                DbName = mem3:dbname(Db#db.name),
+                case ddoc_cache:open(DbName, DDocId) of
+                    {ok, DDoc} ->
+                        {ok, IdxState} = couch_mrview_util:ddoc_to_mrst(ShardDbName, DDoc),
+                        couch_index_util:hexsig(IdxState#mrst.sig) == SigInLocal;
+                    _Else ->
+                        false
+                end
+            catch E:T ->
+                Stack = erlang:get_stacktrace(),
+                couch_log:error("Error occurs when verifying existence of ~s/~s :: ~p ~p",
+                    [ShardDbName, DDocId, {E, T}, Stack]),
+                false
+            after
+                catch couch_db:close(Db)
+            end;
+        _ ->
+            false
+    end.
+
+
+maybe_create_local_purge_doc(Db, State) ->
+    Sig = couch_index_util:hexsig(get(signature, State)),
+    LocalPurgeDocId = couch_mrview_util:get_local_purge_doc_id(Sig),
+    case couch_db:open_doc(Db, LocalPurgeDocId, []) of
+        {not_found, _Reason} ->
+            update_local_purge_doc(Db, State);
+        {ok, _LocalPurgeDoc} ->
+            ok
+    end.

src/couch_mrview_util.erl

@@ -12,6 +12,8 @@
 
 -module(couch_mrview_util).
 
+-export([get_local_purge_doc_id/1, get_value_from_options/2]).
+-export([get_signature_from_filename/1]).
 -export([get_view/4]).
 -export([ddoc_to_mrst/2, init_state/4, reset_index/3]).
 -export([make_header/1]).
@@ -39,6 +41,26 @@
 -include_lib("couch_mrview/include/couch_mrview.hrl").
 
 
+get_local_purge_doc_id(Sig) ->
+    list_to_binary(?LOCAL_DOC_PREFIX ++ "purge-mrview-" ++ Sig).
+
+
+get_value_from_options(Key, Options) ->
+    case couch_util:get_value(Key, Options) of
+        undefined ->
+            Reason = binary_to_list(Key) ++ " must exist in Options.",
+            throw({bad_request, Reason});
+        Value -> Value
+    end.
+
+
+get_signature_from_filename(FileName) ->
+    FilePathList = filename:split(FileName),
+    [PureFN] = lists:nthtail(length(FilePathList) - 1, FilePathList),
+    PureFNExt = filename:extension(PureFN),
+    filename:basename(PureFN, PureFNExt).
+
+
 get_view(Db, DDoc, ViewName, Args0) ->
     ArgCheck = fun(InitState) ->
         Args1 = set_view_type(Args0, ViewName, InitState#mrst.views),
@@ -195,9 +217,10 @@ view_sig_term(BaseSig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed, Args) ->
 
 
 init_state(Db, Fd, #mrst{views=Views}=State, nil) ->
+    PurgeSeq = couch_db:get_purge_seq(Db),
     Header = #mrheader{
         seq=0,
-        purge_seq=couch_db:get_purge_seq(Db),
+        purge_seq=PurgeSeq,
         id_btree_state=nil,
         log_btree_state=nil,
         view_states=[make_view_state(#mrview{}) || _ <- Views]
@@ -232,7 +255,9 @@ init_state(Db, Fd, State, Header) ->
         view_states=ViewStates
     } = Header,
 
-    IdBtOpts = [{compression, couch_db:compression(Db)}],
+    IdBtOpts = [
+        {compression, couch_compress:get_compression_method()}
+    ],
     {ok, IdBtree} = couch_btree:open(IdBtreeState, Fd, IdBtOpts),
     {ok, LogBtree} = case SeqIndexed orelse KeySeqIndexed of
         true -> couch_btree:open(LogBtreeState, Fd, IdBtOpts);
@@ -252,10 +277,10 @@ init_state(Db, Fd, State, Header) ->
         views=Views2
     }.
 
-open_view(Db, Fd, Lang, ViewState, View) ->
+open_view(_Db, Fd, Lang, ViewState, View) ->
     ReduceFun = make_reduce_fun(Lang, View#mrview.reduce_funs),
     LessFun = maybe_define_less_fun(View),
-    Compression = couch_db:compression(Db),
+    Compression = couch_compress:get_compression_method(),
     BTState = get_key_btree_state(ViewState),
     ViewBtOpts = [
         {less, LessFun},
@@ -264,7 +289,7 @@ open_view(Db, Fd, Lang, ViewState, View) ->
     ],
     {ok, Btree} = couch_btree:open(BTState, Fd, ViewBtOpts),
 
-    BySeqReduceFun = fun couch_db_updater:btree_by_seq_reduce/2,
+    BySeqReduceFun = fun couch_bt_engine:seq_tree_reduce/2,
     {ok, SeqBtree} = if View#mrview.seq_indexed ->
         SeqBTState = get_seq_btree_state(ViewState),
         ViewSeqBtOpts = [{reduce, BySeqReduceFun},
@@ -316,7 +341,7 @@ get_row_count(#mrview{btree=Bt}) ->
 
 
 all_docs_reduce_to_count(Reductions) ->
-    Reduce = fun couch_db_updater:btree_by_id_reduce/2,
+    Reduce = fun couch_bt_engine:id_tree_reduce/2,
     {Count, _, _} = couch_btree:final_reduce(Reduce, Reductions),
     Count.
 

test/couch_mrview_purge_docs_fabric_tests.erl

@@ -0,0 +1,93 @@
+% 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_mrview_purge_docs_fabric_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+
+-define(TIMEOUT, 1000).
+
+
+setup() ->
+    DbName = ?tempdb(),
+    ok = fabric:create_db(DbName, [?ADMIN_CTX]),
+    DbName.
+
+
+teardown(DbName) ->
+    ok = fabric:delete_db(DbName, [?ADMIN_CTX]).
+
+
+view_purge_fabric_test_() ->
+    {
+        "Map views",
+        {
+            setup,
+            fun() -> test_util:start_couch([fabric, mem3]) end,
+            fun test_util:stop_couch/1,
+            {
+                foreach,
+                fun setup/0, fun teardown/1,
+                [
+                    fun test_purge_verify_index/1
+                ]
+            }
+        }
+    }.
+
+
+test_purge_verify_index(DbName) ->
+    ?_test(begin
+        Docs1 = couch_mrview_test_util:make_docs(5),
+        {ok, _} = fabric:update_docs(DbName, Docs1, [?ADMIN_CTX]),
+        {ok, _} = fabric:update_doc(DbName, couch_mrview_test_util:ddoc(map), [?ADMIN_CTX]),
+
+        purge_docs(DbName, [<<"1">>]),
+
+        Result2 = fabric:query_view(DbName, <<"bar">>, <<"baz">>, #mrargs{}),
+        Expect2 = {ok, [
+            {meta, [{total, 4}, {offset, 0}]},
+            {row, [{id, <<"2">>}, {key, 2}, {value, 2}]},
+            {row, [{id, <<"3">>}, {key, 3}, {value, 3}]},
+            {row, [{id, <<"4">>}, {key, 4}, {value, 4}]},
+            {row, [{id, <<"5">>}, {key, 5}, {value, 5}]}
+        ]},
+        ?assertEqual(Expect2, Result2),
+
+        {ok, DDoc} = fabric:open_doc(DbName, <<"_design/bar">>, []),
+        {ok, IdxState} = couch_mrview_util:ddoc_to_mrst(DbName, DDoc),
+        Sig = IdxState#mrst.sig,
+        HexSig = list_to_binary(couch_index_util:hexsig(Sig)),
+        DocId = couch_mrview_util:get_local_purge_doc_id(HexSig),
+        {ok, LocPurgeDoc} = fabric:open_doc(DbName, DocId, []),
+        {Props} = couch_doc:to_json_obj(LocPurgeDoc,[]),
+        {Options} = couch_util:get_value(<<"verify_options">>, Props),
+        ?assertEqual(true, couch_mrview_index:verify_index_exists(Options)),
+
+        ok
+    end).
+
+get_rev(#full_doc_info{} = FDI) ->
+    #doc_info{
+        revs = [#rev_info{} = PrevRev | _]
+    } = couch_doc:to_doc_info(FDI),
+    PrevRev#rev_info.rev.
+
+
+purge_docs(DbName, DocIds) ->
+    lists:foreach(fun(DocId) ->
+        FDI = fabric:get_full_doc_info(DbName, DocId, []),
+        Rev = get_rev(FDI),
+        {ok, {_, [{ok, _}]}} = fabric:purge_docs(DbName, [{DocId, [Rev]}], [])
+    end, DocIds).