You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by da...@apache.org on 2018/03/16 19:59:54 UTC
[couchdb] branch COUCHDB-3326-clustered-purge-davisp-refactor
updated (bc6e6e0 -> 98be828)
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a change to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git.
discard bc6e6e0 Temporarily disable should_compare_compression_methods/1
discard b1efd23 Fix tests to work with clustered purge
discard 83fbc6b Add metrics for clustered purge
discard c4222da Implement clustered purge HTTP endpoint
discard b505b10 Update fabric_doc_open_revs to handle purges
discard de91d70 Implement clustered purge API
discard 92b35a3 Add internal replication of purge requests
discard 177b4ff Use EPI to create local purge doc for indexers
discard ae6e83a Update view engine to use new purge API
discard 340d1f2 Implement new purge API
new 6ba2ae7 ss - typos in couch_db_engine.erl
new 1b9ed8f [WIP] - Declare new purge storage engine APIs
new 63d39c2 WIP - couch_db_updater
new 8dae2f0 WIP - couch_db.erl
new a719552 WIP - couch_db_engine.erl
new c8e80d9 WIP - couch_httpd_db.erl
new aafbd35 WIP - couch_db_updater.erl:
new 334af01 WIP - couch_db.erl
new 38d0bb0 WIP - couch_db_engine.erl
new dd43b72 WIP - couch_bt_engine implementation
new 425a60d WIP - add test suite
new dc370c0 Update view engine to use new purge API
new f826388 Use EPI to create local purge doc for indexers
new 0f44181 Add internal replication of purge requests
new c3cc8d4 Implement clustered purge API
new da8f86c Update fabric_doc_open_revs to handle purges
new f60758a Implement clustered purge HTTP endpoint
new 986d39e Add metrics for clustered purge
new 85d3022 Fix tests to work with clustered purge
new 98be828 Temporarily disable should_compare_compression_methods/1
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (bc6e6e0)
\
N -- N -- N refs/heads/COUCHDB-3326-clustered-purge-davisp-refactor (98be828)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 20 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
src/couch/src/couch_bt_engine.erl | 335 ++++++++-------------
src/couch/src/couch_bt_engine.hrl | 2 +-
src/couch/src/couch_bt_engine_compactor.erl | 199 ++++++------
src/couch/src/couch_bt_engine_header.erl | 34 +--
src/couch/src/couch_db.erl | 173 +++++++----
src/couch/src/couch_db_engine.erl | 100 +++---
...ch_bt_engine_header.erl => couch_db_header.erl} | 68 +----
src/couch/src/couch_db_updater.erl | 218 ++++----------
src/couch/src/couch_httpd_db.erl | 21 +-
src/couch/src/couch_key_tree.erl | 2 +-
src/couch/src/couch_util.erl | 11 -
11 files changed, 508 insertions(+), 655 deletions(-)
copy src/couch/src/{couch_bt_engine_header.erl => couch_db_header.erl} (87%)
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 11/20: WIP - add test suite
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 425a60ddbe98f9880d3e62b591a792848d144830
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Wed Mar 14 14:09:16 2018 -0500
WIP - add test suite
---
src/couch/src/test_engine_compaction.erl | 116 +++++++-
src/couch/src/test_engine_fold_purged_docs.erl | 133 +++++++++
src/couch/src/test_engine_get_set_props.erl | 3 +-
src/couch/src/test_engine_purge_docs.erl | 35 ++-
src/couch/src/test_engine_util.erl | 162 +++++++----
src/couch/test/couch_db_purge_docs_tests.erl | 360 +++++++++++++++++++++++++
6 files changed, 744 insertions(+), 65 deletions(-)
diff --git a/src/couch/src/test_engine_compaction.erl b/src/couch/src/test_engine_compaction.erl
index 09a1e4e..51970ee 100644
--- a/src/couch/src/test_engine_compaction.erl
+++ b/src/couch/src/test_engine_compaction.erl
@@ -84,10 +84,8 @@ cet_compact_with_everything() ->
BarRev = test_engine_util:prev_rev(BarFDI),
Actions3 = [
- {batch, [
- {purge, {<<"foo">>, FooRev#rev_info.rev}},
- {purge, {<<"bar">>, BarRev#rev_info.rev}}
- ]}
+ {purge, {<<"foo">>, FooRev#rev_info.rev}},
+ {purge, {<<"bar">>, BarRev#rev_info.rev}}
],
{ok, St6} = test_engine_util:apply_actions(Engine, St5, Actions3),
@@ -97,7 +95,8 @@ cet_compact_with_everything() ->
{<<"foo">>, [FooRev#rev_info.rev]}
],
- ?assertEqual(PurgedIdRevs, lists:sort(Engine:get_last_purged(St6))),
+ {ok, PIdRevs6} = Engine:fold_purged_docs(St6, 0, fun fold_fun/2, [], []),
+ ?assertEqual(PurgedIdRevs, PIdRevs6),
{ok, St7} = try
[Att0, Att1, Att2, Att3, Att4] = test_engine_util:prep_atts(Engine, St6, [
@@ -131,6 +130,9 @@ cet_compact_with_everything() ->
end),
{ok, St10, undefined} = Engine:finish_compaction(St9, DbName, [], Term),
+ {ok, PIdRevs11} = Engine:fold_purged_docs(St10, 0, fun fold_fun/2, [], []),
+ ?assertEqual(PurgedIdRevs, PIdRevs11),
+
Db2 = test_engine_util:db_as_term(Engine, St10),
Diff = test_engine_util:term_diff(Db1, Db2),
?assertEqual(nodiff, Diff).
@@ -175,6 +177,106 @@ cet_recompact_updates() ->
?assertEqual(nodiff, Diff).
+cet_recompact_purge() ->
+ {ok, Engine, Path, St1} = test_engine_util:init_engine(dbpath),
+
+ Actions1 = [
+ {create, {<<"foo">>, []}},
+ {create, {<<"bar">>, []}},
+ {conflict, {<<"bar">>, [{<<"vsn">>, 2}]}},
+ {create, {<<"baz">>, []}}
+ ],
+
+ {ok, St2} = test_engine_util:apply_actions(Engine, St1, Actions1),
+ {ok, St3, DbName, _, Term} = test_engine_util:compact(Engine, St2, Path),
+
+ [BarFDI, BazFDI] = Engine:open_docs(St3, [<<"bar">>, <<"baz">>]),
+ BarRev = test_engine_util:prev_rev(BarFDI),
+ BazRev = test_engine_util:prev_rev(BazFDI),
+ Actions2 = [
+ {purge, {<<"bar">>, BarRev#rev_info.rev}},
+ {purge, {<<"baz">>, BazRev#rev_info.rev}}
+ ],
+ {ok, St4} = test_engine_util:apply_actions(Engine, St3, Actions2),
+ Db1 = test_engine_util:db_as_term(Engine, St4),
+
+ {ok, St5, NewPid} = Engine:finish_compaction(St4, DbName, [], Term),
+
+ ?assertEqual(true, is_pid(NewPid)),
+ Ref = erlang:monitor(process, NewPid),
+
+ NewTerm = receive
+ {'$gen_cast', {compact_done, Engine, Term0}} ->
+ Term0;
+ {'DOWN', Ref, _, _, Reason} ->
+ erlang:error({compactor_died, Reason});
+ {'$gen_call', {NewPid, Ref2}, get_disposable_purge_seq} ->
+ NewPid!{Ref2, {ok, 0}},
+ receive
+ {'$gen_cast', {compact_done, Engine, Term0}} ->
+ Term0;
+ {'DOWN', Ref, _, _, Reason} ->
+ erlang:error({compactor_died, Reason})
+ after 10000 ->
+ erlang:error(compactor_timed_out)
+ end
+ after 10000 ->
+ erlang:error(compactor_timed_out)
+ end,
+
+ {ok, St6, undefined} = Engine:finish_compaction(St5, DbName, [], NewTerm),
+ Db2 = test_engine_util:db_as_term(Engine, St6),
+ Diff = test_engine_util:term_diff(Db1, Db2),
+ ?assertEqual(nodiff, Diff).
+
+
+% temporary ignoring this test as it times out
+ignore_cet_compact_purged_docs_limit() ->
+ {ok, Engine, Path, St1} = test_engine_util:init_engine(dbpath),
+ % create NumDocs docs
+ NumDocs = 1200,
+ {RActions, RIds} = lists:foldl(fun(Id, {CActions, CIds}) ->
+ Id1 = docid(Id),
+ Action = {create, {Id1, [{<<"int">>, Id}]}},
+ {[Action| CActions], [Id1| CIds]}
+ end, {[], []}, lists:seq(1, NumDocs)),
+ Ids = lists:reverse(RIds),
+ {ok, St2} = test_engine_util:apply_actions(Engine, St1,
+ lists:reverse(RActions)),
+
+ % purge NumDocs docs
+ FDIs = Engine:open_docs(St2, Ids),
+ RevActions2 = lists:foldl(fun(FDI, CActions) ->
+ Id = FDI#full_doc_info.id,
+ PrevRev = test_engine_util:prev_rev(FDI),
+ Rev = PrevRev#rev_info.rev,
+ [{purge, {Id, Rev}}| CActions]
+ end, [], FDIs),
+ {ok, St3} = test_engine_util:apply_actions(Engine, St2,
+ lists:reverse(RevActions2)),
+
+ % check that before compaction all NumDocs of purge_requests
+ % are in purge_tree,
+ % even if NumDocs=1200 is greater than purged_docs_limit=1000
+ {ok, PurgedIdRevs} = Engine:fold_purged_docs(St3, 0, fun fold_fun/2, [], []),
+ ?assertEqual(1, Engine:get_oldest_purge_seq(St3)),
+ ?assertEqual(NumDocs, length(PurgedIdRevs)),
+
+ % compact db
+ {ok, St4, DbName, _, Term} = test_engine_util:compact(Engine, St3, Path),
+ {ok, St5, undefined} = Engine:finish_compaction(St4, DbName, [], Term),
+
+ % check that after compaction only purged_docs_limit purge_requests
+ % are in purge_tree
+ PurgedDocsLimit = Engine:get_purged_docs_limit(St5),
+ OldestPSeq = Engine:get_oldest_purge_seq(St5),
+ {ok, PurgedIdRevs2} = Engine:fold_purged_docs(
+ St5, OldestPSeq - 1, fun fold_fun/2, [], []),
+ ExpectedOldestPSeq = NumDocs - PurgedDocsLimit + 1,
+ ?assertEqual(ExpectedOldestPSeq, OldestPSeq),
+ ?assertEqual(PurgedDocsLimit, length(PurgedIdRevs2)).
+
+
docid(I) ->
Str = io_lib:format("~4..0b", [I]),
iolist_to_binary(Str).
@@ -183,3 +285,7 @@ docid(I) ->
local_docid(I) ->
Str = io_lib:format("_local/~4..0b", [I]),
iolist_to_binary(Str).
+
+
+fold_fun({_PSeq, _UUID, Id, Revs}, Acc) ->
+ [{Id, Revs} | Acc].
diff --git a/src/couch/src/test_engine_fold_purged_docs.erl b/src/couch/src/test_engine_fold_purged_docs.erl
new file mode 100644
index 0000000..1dc0885
--- /dev/null
+++ b/src/couch/src/test_engine_fold_purged_docs.erl
@@ -0,0 +1,133 @@
+% 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(test_engine_fold_purged_docs).
+-compile(export_all).
+
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+
+-define(NUM_DOCS, 100).
+
+
+cet_empty_purged_docs() ->
+ {ok, Engine, St} = test_engine_util:init_engine(),
+ ?assertEqual({ok, []}, Engine:fold_purged_docs(St, 0, fun fold_fun/2, [], [])).
+
+
+cet_all_purged_docs() ->
+ {ok, Engine, St1} = test_engine_util:init_engine(),
+
+ {RActions, RIds} = lists:foldl(fun(Id, {CActions, CIds}) ->
+ Id1 = docid(Id),
+ Action = {create, {Id1, [{<<"int">>, Id}]}},
+ {[Action| CActions], [Id1| CIds]}
+ end, {[], []}, lists:seq(1, ?NUM_DOCS)),
+ Actions = lists:reverse(RActions),
+ Ids = lists:reverse(RIds),
+ {ok, St2} = test_engine_util:apply_actions(Engine, St1, Actions),
+
+ FDIs = Engine:open_docs(St2, Ids),
+ {RevActions2, RevIdRevs} = lists:foldl(fun(FDI, {CActions, CIdRevs}) ->
+ Id = FDI#full_doc_info.id,
+ PrevRev = test_engine_util:prev_rev(FDI),
+ Rev = PrevRev#rev_info.rev,
+ Action = {purge, {Id, Rev}},
+ {[Action| CActions], [{Id, [Rev]}| CIdRevs]}
+ end, {[], []}, FDIs),
+ {Actions2, IdsRevs} = {lists:reverse(RevActions2), lists:reverse(RevIdRevs)},
+
+ {ok, St3} = test_engine_util:apply_actions(Engine, St2, Actions2),
+ {ok, PurgedIdRevs} = Engine:fold_purged_docs(St3, 0, fun fold_fun/2, [], []),
+ ?assertEqual(IdsRevs, lists:reverse(PurgedIdRevs)).
+
+
+cet_start_seq() ->
+ {ok, Engine, St1} = test_engine_util:init_engine(),
+ Actions1 = [
+ {create, {docid(1), [{<<"int">>, 1}]}},
+ {create, {docid(2), [{<<"int">>, 2}]}},
+ {create, {docid(3), [{<<"int">>, 3}]}},
+ {create, {docid(4), [{<<"int">>, 4}]}},
+ {create, {docid(5), [{<<"int">>, 5}]}}
+ ],
+ Ids = [docid(1), docid(2), docid(3), docid(4), docid(5)],
+ {ok, St2} = test_engine_util:apply_actions(Engine, St1, Actions1),
+
+ FDIs = Engine:open_docs(St2, Ids),
+ {RActions2, RIdRevs} = lists:foldl(fun(FDI, {CActions, CIdRevs}) ->
+ Id = FDI#full_doc_info.id,
+ PrevRev = test_engine_util:prev_rev(FDI),
+ Rev = PrevRev#rev_info.rev,
+ Action = {purge, {Id, Rev}},
+ {[Action| CActions], [{Id, [Rev]}| CIdRevs]}
+ end, {[], []}, FDIs),
+ {ok, St3} = test_engine_util:apply_actions(Engine, St2, lists:reverse(RActions2)),
+
+ StartSeq = 3,
+ StartSeqIdRevs = lists:nthtail(StartSeq, lists:reverse(RIdRevs)),
+ {ok, PurgedIdRevs} = Engine:fold_purged_docs(St3, StartSeq, fun fold_fun/2, [], []),
+ ?assertEqual(StartSeqIdRevs, lists:reverse(PurgedIdRevs)).
+
+
+cet_id_rev_repeated() ->
+ {ok, Engine, St1} = test_engine_util:init_engine(),
+
+ Actions1 = [
+ {create, {<<"foo">>, [{<<"vsn">>, 1}]}},
+ {conflict, {<<"foo">>, [{<<"vsn">>, 2}]}}
+ ],
+ {ok, St2} = test_engine_util:apply_actions(Engine, St1, Actions1),
+
+ [FDI1] = Engine:open_docs(St2, [<<"foo">>]),
+ PrevRev1 = test_engine_util:prev_rev(FDI1),
+ Rev1 = PrevRev1#rev_info.rev,
+ Actions2 = [
+ {purge, {<<"foo">>, Rev1}}
+ ],
+ {ok, St3} = test_engine_util:apply_actions(Engine, St2, Actions2),
+ PurgedIdRevs0 = [{<<"foo">>, [Rev1]}],
+ {ok, PurgedIdRevs1} = Engine:fold_purged_docs(St3, 0, fun fold_fun/2, [], []),
+ ?assertEqual(PurgedIdRevs0, PurgedIdRevs1),
+ ?assertEqual(1, Engine:get_purge_seq(St3)),
+
+ % purge the same Id,Rev when the doc still exists
+ {ok, St4} = test_engine_util:apply_actions(Engine, St3, Actions2),
+ {ok, PurgedIdRevs2} = Engine:fold_purged_docs(St4, 0, fun fold_fun/2, [], []),
+ ?assertEqual(PurgedIdRevs0, PurgedIdRevs2),
+ ?assertEqual(1, Engine:get_purge_seq(St4)),
+
+ [FDI2] = Engine:open_docs(St4, [<<"foo">>]),
+ PrevRev2 = test_engine_util:prev_rev(FDI2),
+ Rev2 = PrevRev2#rev_info.rev,
+ Actions3 = [
+ {purge, {<<"foo">>, Rev2}}
+ ],
+ {ok, St5} = test_engine_util:apply_actions(Engine, St4, Actions3),
+ PurgedIdRevs00 = [{<<"foo">>, [Rev1]}, {<<"foo">>, [Rev2]}],
+
+ % purge the same Id,Rev when the doc was completely purged
+ {ok, St6} = test_engine_util:apply_actions(Engine, St5, Actions3),
+ {ok, PurgedIdRevs3} = Engine:fold_purged_docs(St6, 0, fun fold_fun/2, [], []),
+ ?assertEqual(PurgedIdRevs00, lists:reverse(PurgedIdRevs3)),
+ ?assertEqual(2, Engine:get_purge_seq(St6)).
+
+
+fold_fun({_PSeq, _UUID, Id, Revs}, Acc) ->
+ [{Id, Revs} | Acc].
+
+
+docid(I) ->
+ Str = io_lib:format("~4..0b", [I]),
+ iolist_to_binary(Str).
diff --git a/src/couch/src/test_engine_get_set_props.erl b/src/couch/src/test_engine_get_set_props.erl
index 6d2a447..ac6aca8 100644
--- a/src/couch/src/test_engine_get_set_props.erl
+++ b/src/couch/src/test_engine_get_set_props.erl
@@ -34,7 +34,8 @@ cet_default_props() ->
?assertEqual(true, is_integer(Engine:get_disk_version(St))),
?assertEqual(0, Engine:get_update_seq(St)),
?assertEqual(0, Engine:get_purge_seq(St)),
- ?assertEqual([], Engine:get_last_purged(St)),
+ ?assertEqual(true, is_integer(Engine:get_purged_docs_limit(St))),
+ ?assertEqual(true, Engine:get_purged_docs_limit(St) > 0),
?assertEqual(dso, Engine:get_security(St)),
?assertEqual(1000, Engine:get_revs_limit(St)),
?assertMatch(<<_:32/binary>>, Engine:get_uuid(St)),
diff --git a/src/couch/src/test_engine_purge_docs.erl b/src/couch/src/test_engine_purge_docs.erl
index e5bf249..a1dbae7 100644
--- a/src/couch/src/test_engine_purge_docs.erl
+++ b/src/couch/src/test_engine_purge_docs.erl
@@ -25,12 +25,13 @@ cet_purge_simple() ->
{create, {<<"foo">>, [{<<"vsn">>, 1}]}}
],
{ok, St2} = test_engine_util:apply_actions(Engine, St1, Actions1),
+ {ok, PIdRevs2} = Engine:fold_purged_docs(St2, 0, fun fold_fun/2, [], []),
?assertEqual(1, Engine:get_doc_count(St2)),
?assertEqual(0, Engine:get_del_doc_count(St2)),
?assertEqual(1, Engine:get_update_seq(St2)),
?assertEqual(0, Engine:get_purge_seq(St2)),
- ?assertEqual([], Engine:get_last_purged(St2)),
+ ?assertEqual([], PIdRevs2),
[FDI] = Engine:open_docs(St2, [<<"foo">>]),
PrevRev = test_engine_util:prev_rev(FDI),
@@ -40,12 +41,13 @@ cet_purge_simple() ->
{purge, {<<"foo">>, Rev}}
],
{ok, St3} = test_engine_util:apply_actions(Engine, St2, Actions2),
+ {ok, PIdRevs3} = Engine:fold_purged_docs(St3, 0, fun fold_fun/2, [], []),
?assertEqual(0, Engine:get_doc_count(St3)),
?assertEqual(0, Engine:get_del_doc_count(St3)),
?assertEqual(2, Engine:get_update_seq(St3)),
?assertEqual(1, Engine:get_purge_seq(St3)),
- ?assertEqual([{<<"foo">>, [Rev]}], Engine:get_last_purged(St3)).
+ ?assertEqual([{<<"foo">>, [Rev]}], PIdRevs3).
cet_purge_conflicts() ->
@@ -56,12 +58,13 @@ cet_purge_conflicts() ->
{conflict, {<<"foo">>, [{<<"vsn">>, 2}]}}
],
{ok, St2} = test_engine_util:apply_actions(Engine, St1, Actions1),
+ {ok, PIdRevs2} = Engine:fold_purged_docs(St2, 0, fun fold_fun/2, [], []),
?assertEqual(1, Engine:get_doc_count(St2)),
?assertEqual(0, Engine:get_del_doc_count(St2)),
?assertEqual(2, Engine:get_update_seq(St2)),
?assertEqual(0, Engine:get_purge_seq(St2)),
- ?assertEqual([], Engine:get_last_purged(St2)),
+ ?assertEqual([], PIdRevs2),
[FDI1] = Engine:open_docs(St2, [<<"foo">>]),
PrevRev1 = test_engine_util:prev_rev(FDI1),
@@ -71,12 +74,13 @@ cet_purge_conflicts() ->
{purge, {<<"foo">>, Rev1}}
],
{ok, St3} = test_engine_util:apply_actions(Engine, St2, Actions2),
+ {ok, PIdRevs3} = Engine:fold_purged_docs(St3, 0, fun fold_fun/2, [], []),
?assertEqual(1, Engine:get_doc_count(St3)),
?assertEqual(0, Engine:get_del_doc_count(St3)),
- ?assertEqual(4, Engine:get_update_seq(St3)),
+ ?assertEqual(3, Engine:get_update_seq(St3)),
?assertEqual(1, Engine:get_purge_seq(St3)),
- ?assertEqual([{<<"foo">>, [Rev1]}], Engine:get_last_purged(St3)),
+ ?assertEqual([{<<"foo">>, [Rev1]}], PIdRevs3),
[FDI2] = Engine:open_docs(St3, [<<"foo">>]),
PrevRev2 = test_engine_util:prev_rev(FDI2),
@@ -86,12 +90,13 @@ cet_purge_conflicts() ->
{purge, {<<"foo">>, Rev2}}
],
{ok, St4} = test_engine_util:apply_actions(Engine, St3, Actions3),
+ {ok, PIdRevs4} = Engine:fold_purged_docs(St4, 0, fun fold_fun/2, [], []),
?assertEqual(0, Engine:get_doc_count(St4)),
?assertEqual(0, Engine:get_del_doc_count(St4)),
- ?assertEqual(5, Engine:get_update_seq(St4)),
+ ?assertEqual(4, Engine:get_update_seq(St4)),
?assertEqual(2, Engine:get_purge_seq(St4)),
- ?assertEqual([{<<"foo">>, [Rev2]}], Engine:get_last_purged(St4)).
+ ?assertEqual([{<<"foo">>, [Rev2]}, {<<"foo">>, [Rev1]}], PIdRevs4).
cet_add_delete_purge() ->
@@ -103,12 +108,13 @@ cet_add_delete_purge() ->
],
{ok, St2} = test_engine_util:apply_actions(Engine, St1, Actions1),
+ {ok, PIdRevs2} = Engine:fold_purged_docs(St2, 0, fun fold_fun/2, [], []),
?assertEqual(0, Engine:get_doc_count(St2)),
?assertEqual(1, Engine:get_del_doc_count(St2)),
?assertEqual(2, Engine:get_update_seq(St2)),
?assertEqual(0, Engine:get_purge_seq(St2)),
- ?assertEqual([], Engine:get_last_purged(St2)),
+ ?assertEqual([], PIdRevs2),
[FDI] = Engine:open_docs(St2, [<<"foo">>]),
PrevRev = test_engine_util:prev_rev(FDI),
@@ -118,12 +124,13 @@ cet_add_delete_purge() ->
{purge, {<<"foo">>, Rev}}
],
{ok, St3} = test_engine_util:apply_actions(Engine, St2, Actions2),
+ {ok, PIdRevs3} = Engine:fold_purged_docs(St3, 0, fun fold_fun/2, [], []),
?assertEqual(0, Engine:get_doc_count(St3)),
?assertEqual(0, Engine:get_del_doc_count(St3)),
?assertEqual(3, Engine:get_update_seq(St3)),
?assertEqual(1, Engine:get_purge_seq(St3)),
- ?assertEqual([{<<"foo">>, [Rev]}], Engine:get_last_purged(St3)).
+ ?assertEqual([{<<"foo">>, [Rev]}], PIdRevs3).
cet_add_two_purge_one() ->
@@ -135,12 +142,13 @@ cet_add_two_purge_one() ->
],
{ok, St2} = test_engine_util:apply_actions(Engine, St1, Actions1),
+ {ok, PIdRevs2} = Engine:fold_purged_docs(St2, 0, fun fold_fun/2, [], []),
?assertEqual(2, Engine:get_doc_count(St2)),
?assertEqual(0, Engine:get_del_doc_count(St2)),
?assertEqual(2, Engine:get_update_seq(St2)),
?assertEqual(0, Engine:get_purge_seq(St2)),
- ?assertEqual([], Engine:get_last_purged(St2)),
+ ?assertEqual([], PIdRevs2),
[FDI] = Engine:open_docs(St2, [<<"foo">>]),
PrevRev = test_engine_util:prev_rev(FDI),
@@ -150,9 +158,14 @@ cet_add_two_purge_one() ->
{purge, {<<"foo">>, Rev}}
],
{ok, St3} = test_engine_util:apply_actions(Engine, St2, Actions2),
+ {ok, PIdRevs3} = Engine:fold_purged_docs(St3, 0, fun fold_fun/2, [], []),
?assertEqual(1, Engine:get_doc_count(St3)),
?assertEqual(0, Engine:get_del_doc_count(St3)),
?assertEqual(3, Engine:get_update_seq(St3)),
?assertEqual(1, Engine:get_purge_seq(St3)),
- ?assertEqual([{<<"foo">>, [Rev]}], Engine:get_last_purged(St3)).
+ ?assertEqual([{<<"foo">>, [Rev]}], PIdRevs3).
+
+
+fold_fun({_Pseq, _UUID, Id, Revs}, Acc) ->
+ [{Id, Revs} | Acc].
\ No newline at end of file
diff --git a/src/couch/src/test_engine_util.erl b/src/couch/src/test_engine_util.erl
index 8999753..bd49969 100644
--- a/src/couch/src/test_engine_util.erl
+++ b/src/couch/src/test_engine_util.erl
@@ -24,6 +24,7 @@
test_engine_attachments,
test_engine_fold_docs,
test_engine_fold_changes,
+ test_engine_fold_purged_docs,
test_engine_purge_docs,
test_engine_compaction,
test_engine_ref_counting
@@ -132,28 +133,35 @@ apply_action(Engine, St, Action) ->
apply_batch(Engine, St, [Action]).
+apply_batch(Engine, St, [{purge, {Id, Revs}}]) ->
+ UpdateSeq = Engine:get_update_seq(St) + 1,
+ case gen_write(Engine, St, {purge, {Id, Revs}}, UpdateSeq) of
+ {_, _, purged_before}->
+ St;
+ {Pair, _, {Id, PRevs}} ->
+ UUID = couch_uuids:new(),
+ {ok, NewSt} = Engine:purge_doc_revs(
+ St, [Pair], [{UUID, Id, PRevs}]),
+ NewSt
+ end;
+
apply_batch(Engine, St, Actions) ->
UpdateSeq = Engine:get_update_seq(St) + 1,
- AccIn = {UpdateSeq, [], [], []},
+ AccIn = {UpdateSeq, [], []},
AccOut = lists:foldl(fun(Action, Acc) ->
- {SeqAcc, DocAcc, LDocAcc, PurgeAcc} = Acc,
+ {SeqAcc, DocAcc, LDocAcc} = Acc,
case Action of
{_, {<<"_local/", _/binary>>, _}} ->
LDoc = gen_local_write(Engine, St, Action),
- {SeqAcc, DocAcc, [LDoc | LDocAcc], PurgeAcc};
+ {SeqAcc, DocAcc, [LDoc | LDocAcc]};
_ ->
- case gen_write(Engine, St, Action, SeqAcc) of
- {_OldFDI, _NewFDI} = Pair ->
- {SeqAcc + 1, [Pair | DocAcc], LDocAcc, PurgeAcc};
- {Pair, NewSeqAcc, NewPurgeInfo} ->
- NewPurgeAcc = [NewPurgeInfo | PurgeAcc],
- {NewSeqAcc, [Pair | DocAcc], LDocAcc, NewPurgeAcc}
- end
+ {OldFDI, NewFDI} = gen_write(Engine, St, Action, SeqAcc),
+ {SeqAcc + 1, [{OldFDI, NewFDI} | DocAcc], LDocAcc}
end
end, AccIn, Actions),
- {_, Docs0, LDocs, PurgeIdRevs} = AccOut,
+ {_, Docs0, LDocs} = AccOut,
Docs = lists:reverse(Docs0),
- {ok, NewSt} = Engine:write_doc_infos(St, Docs, LDocs, PurgeIdRevs),
+ {ok, NewSt} = Engine:write_doc_infos(St, Docs, LDocs),
NewSt.
@@ -221,39 +229,71 @@ gen_write(Engine, St, {create, {DocId, Body, Atts0}}, UpdateSeq) ->
}};
gen_write(Engine, St, {purge, {DocId, PrevRevs0, _}}, UpdateSeq) ->
- [#full_doc_info{} = PrevFDI] = Engine:open_docs(St, [DocId]),
- PrevRevs = if is_list(PrevRevs0) -> PrevRevs0; true -> [PrevRevs0] end,
-
- #full_doc_info{
- rev_tree = PrevTree
- } = PrevFDI,
-
- {NewTree, RemRevs} = couch_key_tree:remove_leafs(PrevTree, PrevRevs),
- RemovedAll = lists:sort(RemRevs) == lists:sort(PrevRevs),
- if RemovedAll -> ok; true ->
- % If we didn't purge all the requested revisions
- % then its a bug in the test.
- erlang:error({invalid_purge_test_revs, PrevRevs})
- end,
+ case Engine:open_docs(St, [DocId]) of
+ [not_found] ->
+ % Check if this doc has been purged before
+ FoldFun = fun({_PSeq, _UUID, Id, _Revs}, _Acc) ->
+ case Id of
+ DocId -> true;
+ _ -> false
+ end
+ end,
+ {ok, IsPurgedBefore} = Engine:fold_purged_docs(
+ St, 0, FoldFun, false, []),
+ case IsPurgedBefore of
+ true -> {{}, UpdateSeq, purged_before};
+ false -> erlang:error({invalid_purge_test_id, DocId})
+ end;
+ [#full_doc_info{} = PrevFDI] ->
+ PrevRevs = if is_list(PrevRevs0) -> PrevRevs0; true -> [PrevRevs0] end,
+
+ #full_doc_info{
+ rev_tree = PrevTree
+ } = PrevFDI,
+
+ {NewTree, RemRevs0} = couch_key_tree:remove_leafs(PrevTree, PrevRevs),
+ {RemRevs, NotRemRevs} = lists:partition(fun(R) ->
+ lists:member(R, RemRevs0) end, PrevRevs),
+
+ if NotRemRevs == [] -> ok; true ->
+ % Check if these Revs have been purged before
+ FoldFun = fun({_Pseq, _UUID, Id, Revs}, Acc) ->
+ case Id of
+ DocId -> Acc ++ Revs;
+ _ -> Acc
+ end
+ end,
+ {ok, PurgedRevs} = Engine:fold_purged_docs(St, 0, FoldFun, [], []),
+ case lists:subtract(PrevRevs, PurgedRevs) of [] -> ok; _ ->
+ % If we didn't purge all the requested revisions
+ % and they haven't been purged before
+ % then its a bug in the test.
+ erlang:error({invalid_purge_test_revs, PrevRevs})
+ end
+ end,
+
+ case {RemRevs, NewTree} of
+ {[], _} ->
+ {{PrevFDI, PrevFDI}, UpdateSeq, purged_before};
+ {_, []} ->
+ % We've completely purged the document
+ {{PrevFDI, not_found}, UpdateSeq, {DocId, RemRevs}};
+ _ ->
+ % We have to relabel the update_seq of all
+ % leaves. See couch_db_updater for details.
+ {NewNewTree, NewUpdateSeq} = couch_key_tree:mapfold(fun
+ (_RevId, Leaf, leaf, InnerSeqAcc) ->
+ {Leaf#leaf{seq = InnerSeqAcc}, InnerSeqAcc + 1};
+ (_RevId, Value, _Type, InnerSeqAcc) ->
+ {Value, InnerSeqAcc}
+ end, UpdateSeq, NewTree),
+ NewFDI = PrevFDI#full_doc_info{
+ update_seq = NewUpdateSeq - 1,
+ rev_tree = NewNewTree
+ },
+ {{PrevFDI, NewFDI}, NewUpdateSeq, {DocId, RemRevs}}
- case NewTree of
- [] ->
- % We've completely purged the document
- {{PrevFDI, not_found}, UpdateSeq, {DocId, RemRevs}};
- _ ->
- % We have to relabel the update_seq of all
- % leaves. See couch_db_updater for details.
- {NewNewTree, NewUpdateSeq} = couch_key_tree:mapfold(fun
- (_RevId, Leaf, leaf, InnerSeqAcc) ->
- {Leaf#leaf{seq = InnerSeqAcc}, InnerSeqAcc + 1};
- (_RevId, Value, _Type, InnerSeqAcc) ->
- {Value, InnerSeqAcc}
- end, UpdateSeq, NewTree),
- NewFDI = PrevFDI#full_doc_info{
- update_seq = NewUpdateSeq - 1,
- rev_tree = NewNewTree
- },
- {{PrevFDI, NewFDI}, NewUpdateSeq, {DocId, RemRevs}}
+ end
end;
gen_write(Engine, St, {Action, {DocId, Body, Atts0}}, UpdateSeq) ->
@@ -408,7 +448,8 @@ db_as_term(Engine, St) ->
{props, db_props_as_term(Engine, St)},
{docs, db_docs_as_term(Engine, St)},
{local_docs, db_local_docs_as_term(Engine, St)},
- {changes, db_changes_as_term(Engine, St)}
+ {changes, db_changes_as_term(Engine, St)},
+ {purged_docs, db_purged_docs_as_term(Engine, St)}
].
@@ -419,7 +460,7 @@ db_props_as_term(Engine, St) ->
get_disk_version,
get_update_seq,
get_purge_seq,
- get_last_purged,
+ get_purged_docs_limit,
get_security,
get_revs_limit,
get_uuid,
@@ -452,6 +493,15 @@ db_changes_as_term(Engine, St) ->
end, Changes)).
+db_purged_docs_as_term(Engine, St) ->
+ StartPSeq = Engine:get_oldest_purge_seq(St) - 1,
+ FoldFun = fun({PSeq, UUID, Id, Revs}, Acc) ->
+ [{PSeq, UUID, Id, Revs} | Acc]
+ end,
+ {ok, PDocs} = Engine:fold_purged_docs(St, StartPSeq, FoldFun, [], []),
+ lists:reverse(PDocs).
+
+
fdi_to_term(Engine, St, FDI) ->
#full_doc_info{
id = DocId,
@@ -577,8 +627,24 @@ compact(Engine, St1, DbPath) ->
Term = receive
{'$gen_cast', {compact_done, Engine, Term0}} ->
Term0;
- {'DOWN', Ref, _, _, Reason} ->
- erlang:error({compactor_died, Reason})
+ erlang:error({compactor_died, Reason});
+ {'$gen_call', {Pid, Ref2}, get_disposable_purge_seq} ->
+ % assuming no client exists (no internal replications or indexes)
+ PSeq = Engine:get_purge_seq(St2),
+ OldestPSeq = Engine:get_oldest_purge_seq(St2),
+ PDocsLimit = Engine:get_purged_docs_limit(St2),
+ ExpectedDispPSeq = PSeq - PDocsLimit,
+ DisposablePSeq = if ExpectedDispPSeq > 0 -> ExpectedDispPSeq;
+ true -> OldestPSeq - 1 end,
+ Pid!{Ref2, {ok, DisposablePSeq}},
+ receive
+ {'$gen_cast', {compact_done, Engine, Term0}} ->
+ Term0;
+ {'DOWN', Ref, _, _, Reason} ->
+ erlang:error({compactor_died, Reason})
+ after 10000 ->
+ erlang:error(compactor_timed_out)
+ end
after ?COMPACTOR_TIMEOUT ->
erlang:error(compactor_timed_out)
end,
diff --git a/src/couch/test/couch_db_purge_docs_tests.erl b/src/couch/test/couch_db_purge_docs_tests.erl
new file mode 100644
index 0000000..1608957
--- /dev/null
+++ b/src/couch/test/couch_db_purge_docs_tests.erl
@@ -0,0 +1,360 @@
+% 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_db_purge_docs_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+
+setup() ->
+ DbName = ?tempdb(),
+ {ok, _Db} = create_db(DbName),
+ DbName.
+
+teardown(DbName) ->
+ delete_db(DbName),
+ ok.
+
+couch_db_purge_docs_test_() ->
+ {
+ "Couch_db purge_docs",
+ [
+ {
+ setup,
+ fun test_util:start_couch/0, fun test_util:stop_couch/1,
+ [couch_db_purge_docs()]
+ },
+ purge_with_replication()
+ ]
+
+ }.
+
+
+couch_db_purge_docs() ->
+ {
+ foreach,
+ fun setup/0, fun teardown/1,
+ [
+ fun purge_simple/1,
+ fun add_delete_purge/1,
+ fun add_two_purge_one/1,
+ fun purge_id_not_exist/1,
+ fun purge_non_leaf_rev/1,
+ fun purge_conflicts/1,
+ fun purge_deep_tree/1
+ ]
+ }.
+
+
+purge_simple(DbName) ->
+ ?_test(
+ begin
+ {ok, Db} = couch_db:open_int(DbName, []),
+ Doc1 = {[{<<"_id">>, <<"foo1">>}, {<<"vsn">>, 1.1}]},
+ Doc2 = {[{<<"_id">>, <<"foo2">>}, {<<"vsn">>, 1.2}]},
+ {ok, Rev} = save_doc(Db, Doc1),
+ {ok, Rev2} = save_doc(Db, Doc2),
+ couch_db:ensure_full_commit(Db),
+
+ {ok, Db2} = couch_db:reopen(Db),
+ ?assertEqual(2, couch_db_engine:get_doc_count(Db2)),
+ ?assertEqual(0, couch_db_engine:get_del_doc_count(Db2)),
+ ?assertEqual(2, couch_db_engine:get_update_seq(Db2)),
+ ?assertEqual(0, couch_db_engine:get_purge_seq(Db2)),
+
+ UUID = couch_uuids:new(), UUID2 = couch_uuids:new(),
+ {ok, [{ok, PRevs}, {ok, PRevs2}]} = couch_db:purge_docs(
+ Db2, [{UUID, <<"foo1">>, [Rev]}, {UUID2, <<"foo2">>, [Rev2]}]
+ ),
+
+ ?assertEqual([Rev], PRevs),
+ ?assertEqual([Rev2], PRevs2),
+
+ {ok, Db3} = couch_db:reopen(Db2),
+ {ok, PIdsRevs} = couch_db:fold_purged_docs(
+ Db3, 0, fun fold_fun/2, [], []),
+ ?assertEqual(0, couch_db_engine:get_doc_count(Db3)),
+ ?assertEqual(0, couch_db_engine:get_del_doc_count(Db3)),
+ ?assertEqual(3, couch_db_engine:get_update_seq(Db3)),
+ ?assertEqual(2, couch_db_engine:get_purge_seq(Db3)),
+ ?assertEqual([{<<"foo2">>, [Rev2]}, {<<"foo1">>, [Rev]}], PIdsRevs)
+ end).
+
+
+add_delete_purge(DbName) ->
+ ?_test(
+ begin
+ {ok, Db0} = couch_db:open_int(DbName, []),
+ Doc0 = {[{<<"_id">>,<<"foo">>}, {<<"vsn">>, 1}]},
+ {ok, Rev} = save_doc(Db0, Doc0),
+ couch_db:ensure_full_commit(Db0),
+ {ok, Db1} = couch_db:reopen(Db0),
+
+ Doc1 = {[
+ {<<"_id">>, <<"foo">>}, {<<"vsn">>, 2},
+ {<<"_rev">>, couch_doc:rev_to_str(Rev)},
+ {<<"_deleted">>, true}]
+ },
+ {ok, Rev2} = save_doc(Db1, Doc1),
+ couch_db:ensure_full_commit(Db1),
+
+ {ok, Db2} = couch_db:reopen(Db1),
+ {ok, PIdsRevs1} = couch_db:fold_purged_docs(
+ Db2, 0, fun fold_fun/2, [], []),
+ ?assertEqual(0, couch_db_engine:get_doc_count(Db2)),
+ ?assertEqual(1, couch_db_engine:get_del_doc_count(Db2)),
+ ?assertEqual(2, couch_db_engine:get_update_seq(Db2)),
+ ?assertEqual(0, couch_db_engine:get_purge_seq(Db2)),
+ ?assertEqual([], PIdsRevs1),
+
+ UUID = couch_uuids:new(),
+ {ok, [{ok, PRevs}]} = couch_db:purge_docs(
+ Db2, [{UUID, <<"foo">>, [Rev2]}]),
+ ?assertEqual([Rev2], PRevs),
+
+ {ok, Db3} = couch_db:reopen(Db2),
+ {ok, PIdsRevs2} = couch_db:fold_purged_docs(
+ Db3, 0, fun fold_fun/2, [], []),
+ ?assertEqual(0, couch_db_engine:get_doc_count(Db3)),
+ ?assertEqual(0, couch_db_engine:get_del_doc_count(Db3)),
+ ?assertEqual(3, couch_db_engine:get_update_seq(Db3)),
+ ?assertEqual(1, couch_db_engine:get_purge_seq(Db3)),
+ ?assertEqual([{<<"foo">>, [Rev2]}], PIdsRevs2)
+ end).
+
+
+add_two_purge_one(DbName) ->
+ ?_test(
+ begin
+ {ok, Db} = couch_db:open_int(DbName, []),
+ Doc1 = {[{<<"_id">>, <<"foo1">>}, {<<"vsn">>, 1}]},
+ Doc2 = {[{<<"_id">>, <<"foo2">>}, {<<"vsn">>, 2}]},
+ {ok, Rev} = save_doc(Db, Doc1),
+ {ok, _Rev2} = save_doc(Db, Doc2),
+ couch_db:ensure_full_commit(Db),
+
+ {ok, Db2} = couch_db:reopen(Db),
+ ?assertEqual(2, couch_db_engine:get_doc_count(Db2)),
+ ?assertEqual(0, couch_db_engine:get_del_doc_count(Db2)),
+ ?assertEqual(2, couch_db_engine:get_update_seq(Db2)),
+ ?assertEqual(0, couch_db_engine:get_purge_seq(Db2)),
+
+ UUID = couch_uuids:new(),
+ {ok, [{ok, PRevs}]} = couch_db:purge_docs(Db2,
+ [{UUID, <<"foo1">>, [Rev]}]),
+ ?assertEqual([Rev], PRevs),
+
+ {ok, Db3} = couch_db:reopen(Db2),
+ {ok, PIdsRevs} = couch_db:fold_purged_docs(
+ Db3, 0, fun fold_fun/2, [], []),
+ ?assertEqual(1, couch_db_engine:get_doc_count(Db3)),
+ ?assertEqual(0, couch_db_engine:get_del_doc_count(Db3)),
+ ?assertEqual(3, couch_db_engine:get_update_seq(Db3)),
+ ?assertEqual(1, couch_db_engine:get_purge_seq(Db3)),
+ ?assertEqual([{<<"foo1">>, [Rev]}], PIdsRevs)
+ end).
+
+
+purge_id_not_exist(DbName) ->
+ ?_test(
+ begin
+ {ok, Db} = couch_db:open_int(DbName, []),
+ UUID = couch_uuids:new(),
+ {ok, [{ok, PRevs}]} = couch_db:purge_docs(Db,
+ [{UUID, <<"foo">>, [{0, <<0>>}]}]),
+ ?assertEqual([], PRevs),
+
+ {ok, Db2} = couch_db:reopen(Db),
+ {ok, PIdsRevs} = couch_db:fold_purged_docs(
+ Db2, 0, fun fold_fun/2, [], []),
+ ?assertEqual(0, couch_db_engine:get_doc_count(Db2)),
+ ?assertEqual(0, couch_db_engine:get_del_doc_count(Db2)),
+ ?assertEqual(0, couch_db_engine:get_update_seq(Db2)),
+ ?assertEqual(0, couch_db_engine:get_purge_seq(Db2)),
+ ?assertEqual([], PIdsRevs)
+ end).
+
+
+purge_non_leaf_rev(DbName) ->
+ ?_test(
+ begin
+ {ok, Db} = couch_db:open_int(DbName, []),
+ Doc0 = {[{<<"_id">>, <<"foo">>}, {<<"vsn">>, 1}]},
+ {ok, Rev} = save_doc(Db, Doc0),
+ couch_db:ensure_full_commit(Db),
+ {ok, Db2} = couch_db:reopen(Db),
+
+ Doc1 = {[
+ {<<"_id">>, <<"foo">>}, {<<"vsn">>, 2},
+ {<<"_rev">>, couch_doc:rev_to_str(Rev)}
+ ]},
+ {ok, _Rev2} = save_doc(Db2, Doc1),
+ couch_db:ensure_full_commit(Db2),
+ {ok, Db3} = couch_db:reopen(Db2),
+
+ UUID = couch_uuids:new(),
+ {ok, [{ok, PRevs}]} = couch_db:purge_docs(Db3,
+ [{UUID, <<"foo">>, [Rev]}]),
+ ?assertEqual([], PRevs),
+
+ {ok, Db4} = couch_db:reopen(Db3),
+ {ok, PIdsRevs} = couch_db:fold_purged_docs(Db4, 0, fun fold_fun/2, [], []),
+ ?assertEqual(1, couch_db_engine:get_doc_count(Db4)),
+ ?assertEqual(2, couch_db_engine:get_update_seq(Db4)),
+ ?assertEqual(0, couch_db_engine:get_purge_seq(Db4)),
+ ?assertEqual([], PIdsRevs)
+ end).
+
+
+purge_conflicts(DbName) ->
+ ?_test(
+ begin
+ {ok, Db} = couch_db:open_int(DbName, []),
+ Doc = {[{<<"_id">>, <<"foo">>}, {<<"vsn">>, <<"v1.1">>}]},
+ {ok, Rev} = save_doc(Db, Doc),
+ couch_db:ensure_full_commit(Db),
+ {ok, Db2} = couch_db:reopen(Db),
+
+ % create a conflict
+ DocConflict = #doc{
+ id = <<"foo">>,
+ revs = {1, [crypto:hash(md5, <<"v1.2">>)]},
+ body = {[ {<<"vsn">>, <<"v1.2">>}]}
+ },
+ {ok, _} = couch_db:update_doc(Db2, DocConflict, [], replicated_changes),
+ couch_db:ensure_full_commit(Db2),
+ {ok, Db3} = couch_db:reopen(Db2),
+
+ UUID = couch_uuids:new(),
+ {ok, [{ok, PRevs}]} = couch_db:purge_docs(Db3,
+ [{UUID, <<"foo">>, [Rev]}]),
+ ?assertEqual([Rev], PRevs),
+
+ {ok, Db4} = couch_db:reopen(Db3),
+ {ok, PIdsRevs} = couch_db:fold_purged_docs(
+ Db4, 0, fun fold_fun/2, [], []),
+ % still has one doc
+ ?assertEqual(1, couch_db_engine:get_doc_count(Db4)),
+ ?assertEqual(0, couch_db_engine:get_del_doc_count(Db4)),
+ ?assertEqual(3, couch_db_engine:get_update_seq(Db4)),
+ ?assertEqual(1, couch_db_engine:get_purge_seq(Db4)),
+ ?assertEqual([{<<"foo">>, [Rev]}], PIdsRevs)
+ end).
+
+
+purge_deep_tree(DbName) ->
+ ?_test(
+ begin
+ NRevs = 100,
+ {ok, Db0} = couch_db:open_int(DbName, []),
+ Doc0 = {[{<<"_id">>, <<"bar">>}, {<<"vsn">>, 0}]},
+ {ok, InitRev} = save_doc(Db0, Doc0),
+ ok = couch_db:close(Db0),
+ LastRev = lists:foldl(fun(V, PrevRev) ->
+ {ok, Db} = couch_db:open_int(DbName, []),
+ {ok, Rev} = save_doc(Db,
+ {[{<<"_id">>, <<"bar">>},
+ {<<"vsn">>, V},
+ {<<"_rev">>, couch_doc:rev_to_str(PrevRev)}]}
+ ),
+ ok = couch_db:close(Db),
+ Rev
+ end, InitRev, lists:seq(2, NRevs)),
+ {ok, Db1} = couch_db:open_int(DbName, []),
+
+ % purge doc
+ UUID = couch_uuids:new(),
+ {ok, [{ok, PRevs}]} = couch_db:purge_docs(Db1,
+ [{UUID, <<"bar">>, [LastRev]}]),
+ ?assertEqual([LastRev], PRevs),
+
+ {ok, Db2} = couch_db:reopen(Db1),
+ % no docs left
+ ?assertEqual(0, couch_db_engine:get_doc_count(Db2)),
+ ?assertEqual(0, couch_db_engine:get_del_doc_count(Db2)),
+ ?assertEqual(1, couch_db_engine:get_purge_seq(Db2)),
+ ?assertEqual(NRevs + 1 , couch_db_engine:get_update_seq(Db2))
+ end).
+
+
+purge_with_replication() ->
+ ?_test(
+ begin
+ Ctx = test_util:start_couch([couch_replicator]),
+ Source = ?tempdb(),
+ {ok, SourceDb} = create_db(Source),
+ Target = ?tempdb(),
+ {ok, _Db} = create_db(Target),
+
+ % create Doc and do replication to Target
+ {ok, Rev} = save_doc(SourceDb,
+ {[{<<"_id">>, <<"foo">>}, {<<"vsn">>, 1}]}),
+ couch_db:ensure_full_commit(SourceDb),
+ {ok, SourceDb2} = couch_db:reopen(SourceDb),
+ RepObject = {[
+ {<<"source">>, Source},
+ {<<"target">>, Target}
+ ]},
+ {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
+ {ok, TargetDb} = couch_db:open_int(Target, []),
+ {ok, Doc} = couch_db:get_doc_info(TargetDb, <<"foo">>),
+
+ % purge Doc on Source and do replication to Target
+ % assert purges don't get replicated to Target
+ UUID = couch_uuids:new(),
+ {ok, _} = couch_db:purge_docs(SourceDb2, [{UUID, <<"foo">>, [Rev]}]),
+ {ok, SourceDb3} = couch_db:reopen(SourceDb2),
+ {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
+ {ok, TargetDb2} = couch_db:open_int(Target, []),
+ {ok, Doc2} = couch_db:get_doc_info(TargetDb2, <<"foo">>),
+ [Rev2] = Doc2#doc_info.revs,
+ ?assertEqual(Rev, Rev2#rev_info.rev),
+ ?assertEqual(Doc, Doc2),
+ ?assertEqual(0, couch_db_engine:get_doc_count(SourceDb3)),
+ ?assertEqual(1, couch_db_engine:get_purge_seq(SourceDb3)),
+ ?assertEqual(1, couch_db_engine:get_doc_count(TargetDb2)),
+ ?assertEqual(0, couch_db_engine:get_purge_seq(TargetDb2)),
+
+ % replicate from Target to Source
+ % assert that Doc reappears on Source
+ RepObject2 = {[
+ {<<"source">>, Target},
+ {<<"target">>, Source}
+ ]},
+ {ok, _} = couch_replicator:replicate(RepObject2, ?ADMIN_USER),
+ {ok, SourceDb4} = couch_db:reopen(SourceDb3),
+ {ok, Doc3} = couch_db:get_doc_info(SourceDb4, <<"foo">>),
+ [Rev3] = Doc3#doc_info.revs,
+ ?assertEqual(Rev, Rev3#rev_info.rev),
+ ?assertEqual(1, couch_db_engine:get_doc_count(SourceDb4)),
+ ?assertEqual(1, couch_db_engine:get_purge_seq(SourceDb4)),
+
+ delete_db(Source),
+ delete_db(Target),
+ ok = application:stop(couch_replicator),
+ ok = test_util:stop_couch(Ctx)
+ end).
+
+
+create_db(DbName) ->
+ couch_db:create(DbName, [?ADMIN_CTX, overwrite]).
+
+delete_db(DbName) ->
+ couch_server:delete(DbName, [?ADMIN_CTX]).
+
+save_doc(Db, Json) ->
+ Doc = couch_doc:from_json_obj(Json),
+ couch_db:update_doc(Db, Doc, []).
+
+fold_fun({_PSeq, _UUID, Id, Revs}, Acc) ->
+ [{Id, Revs} | Acc].
\ No newline at end of file
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 14/20: Add internal replication of purge requests
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 0f44181de0cb729883c11326da284cf687e3a70e
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Mon May 1 14:36:39 2017 -0400
Add internal replication of purge requests
* Add initial pull replication of purge requests
- Each internal replication job starts by pulling purge requests from
target and applying them on source. If a source and target were
disconnected during a purge request, it's possible that the target
has received a purge request not yet present on the source. Given
that internal replication is push oriented it would be possible for the
source and target to reconnect and have the source push a revision that
has since been purged. To avoid this we should pull purge requests
from the target to ensure we're up to date before beginning internal
replication.
- Add _local/purge-mem3-$hash docs in mem3_rep. mem3 writes a
_local/purge-mem3-$hash document once purge requests have been
replicated. This document will exist on the target and the
purge_seq value will be the target's purge_seq that has been processed
during *pull* replication.
* Add push replication of purge requests
- Push new purge requests from source to target, and apply them on
target
- Update checkpoint docs to store the purge_seq on source
COUCHDB-3326
---
src/mem3/src/mem3_rep.erl | 142 +++++++++++++++++++++++++++++++++++++++++-----
src/mem3/src/mem3_rpc.erl | 128 +++++++++++++++++++++++++++++++++++++++--
2 files changed, 252 insertions(+), 18 deletions(-)
diff --git a/src/mem3/src/mem3_rep.erl b/src/mem3/src/mem3_rep.erl
index 942f8a8..44aee49 100644
--- a/src/mem3/src/mem3_rep.erl
+++ b/src/mem3/src/mem3_rep.erl
@@ -17,7 +17,9 @@
go/2,
go/3,
make_local_id/2,
- find_source_seq/4
+ make_local_purge_id/2,
+ find_source_seq/4,
+ mem3_sync_purge/1
]).
-export([
@@ -39,7 +41,8 @@
target,
filter,
db,
- history = {[]}
+ history = {[]},
+ purge_seq = 0
}).
@@ -119,6 +122,12 @@ make_local_id(SourceThing, TargetThing, Filter) ->
<<"_local/shard-sync-", S/binary, "-", T/binary, F/binary>>.
+make_local_purge_id(SourceUUID, TargetUUID) ->
+ V = ?l2b("v" ++ config:get("purge", "version", "1") ++ "-"),
+ <<"_local/purge-", V/binary, "mem3-",
+ SourceUUID/binary, "-", TargetUUID/binary>>.
+
+
%% @doc Find and return the largest update_seq in SourceDb
%% that the client has seen from TargetNode.
%%
@@ -172,18 +181,58 @@ find_source_seq_int(#doc{body={Props}}, SrcNode0, TgtNode0, TgtUUID, TgtSeq) ->
repl(Db, Acc0) ->
erlang:put(io_priority, {internal_repl, couch_db:name(Db)}),
- #acc{seq=Seq} = Acc1 = calculate_start_seq(Acc0#acc{source = Db}),
- case Seq >= couch_db:get_update_seq(Db) of
- true ->
- {ok, 0};
- false ->
- Fun = fun ?MODULE:changes_enumerator/2,
- {ok, Acc2} = couch_db:fold_changes(Db, Seq, Fun, Acc1),
- {ok, #acc{seq = LastSeq}} = replicate_batch(Acc2),
- {ok, couch_db:count_changes_since(Db, LastSeq)}
+ #acc{source = Db2} = Acc1 = pull_purges_from_target(Db, Acc0),
+ #acc{seq=Seq} = Acc2 = calculate_start_seq(Acc1),
+ try
+ % this throws an exception: {invalid_start_purge_seq, PurgeSeq0}
+ Acc3 = replicate_purged_docs(Acc2),
+ Fun = fun ?MODULE:changes_enumerator/2,
+ {ok, Acc4} = couch_db:fold_changes(Db2, Seq, Fun, Acc3),
+ {ok, #acc{seq = LastSeq}} = replicate_batch(Acc4),
+ {ok, couch_db:count_changes_since(Db2, LastSeq)}
+ catch
+ throw:{invalid_start_purge_seq, PurgeSeq} ->
+ couch_log:error(
+ "Oldest_purge_seq on source is greated than "
+ "the last source's purge_seq: ~p known to target!"
+ "Can't synchronize purges between: ~p and ~p!",
+ [PurgeSeq, Acc2#acc.source, Acc2#acc.target]
+ )
end.
+pull_purges_from_target(Db, #acc{target=#shard{node=TNode, name=DbName}}=Acc) ->
+ SourceUUID = couch_db:get_uuid(Db),
+ {TUUIDsIdsRevs, TargetPDocID, TargetPSeq} =
+ mem3_rpc:load_purges(TNode, DbName, SourceUUID),
+ Acc2 = case TUUIDsIdsRevs of
+ [] -> Acc#acc{source = Db};
+ _ ->
+ % check which Target UUIDs have not been applied to Source
+ UUIDs = [UUID || {UUID, _Id, _Revs} <- TUUIDsIdsRevs],
+ PurgedDocs = couch_db:open_purged_docs(Db, UUIDs),
+ Results = lists:zip(TUUIDsIdsRevs, PurgedDocs),
+ Unapplied = lists:filtermap(fun
+ ({UUIDIdRevs, not_found}) -> {true, UUIDIdRevs};
+ (_) -> false
+ end, Results),
+ Acc1 = case Unapplied of
+ [] -> Acc#acc{source = Db};
+ _ ->
+ % purge Db on Source and reopen it
+ couch_db:purge_docs(Db, Unapplied),
+ couch_db:close(Db),
+ {ok, Db2} = couch_db:open(DbName, [?ADMIN_CTX]),
+ Acc#acc{source = Db2}
+ end,
+ % update on Target target_purge_seq known to Source
+ mem3_rpc:save_purge_checkpoint(TNode, DbName, TargetPDocID,
+ TargetPSeq, node()),
+ Acc1
+ end,
+ Acc2.
+
+
calculate_start_seq(Acc) ->
#acc{
source = Db,
@@ -215,7 +264,33 @@ calculate_start_seq(Acc) ->
Seq = TargetSeq,
History = couch_util:get_value(<<"history">>, TProps, {[]})
end,
- Acc1#acc{seq = Seq, history = History};
+ SourcePurgeSeq0 = couch_util:get_value(<<"purge_seq">>, SProps),
+ TargetPurgeSeq0 = couch_util:get_value(<<"purge_seq">>, TProps),
+ % before purge upgrade, purge_seq was not saved in checkpoint file,
+ % thus get purge_seq directly from dbs
+ SourcePurgeSeq = case is_integer(SourcePurgeSeq0) of
+ true ->
+ SourcePurgeSeq0;
+ false ->
+ {ok, SPS} = couch_db:get_purge_seq(Db),
+ SPS
+ end,
+ TargetPurgeSeq = case is_integer(TargetPurgeSeq0) of
+ true ->
+ TargetPurgeSeq0;
+ false ->
+ mem3_rpc:get_purge_seq(Node, Name, [
+ {io_priority, {internal_repl, Name}},
+ ?ADMIN_CTX
+ ])
+ end,
+ case SourcePurgeSeq =< TargetPurgeSeq of
+ true ->
+ PurgeSeq = SourcePurgeSeq;
+ false ->
+ PurgeSeq = TargetPurgeSeq
+ end,
+ Acc1#acc{seq = Seq, history = History, purge_seq = PurgeSeq};
{not_found, _} ->
compare_epochs(Acc1)
end.
@@ -251,6 +326,27 @@ changes_enumerator(#full_doc_info{}=FDI, #acc{revcount=C, infos=Infos}=Acc0) ->
{Go, Acc1}.
+replicate_purged_docs(Acc0) ->
+ #acc{
+ source = Db,
+ target = #shard{node=Node, name=Name},
+ purge_seq = PurgeSeq0
+ } = Acc0,
+ PFoldFun = fun({_PSeq, UUID, Id, Revs}, Acc) ->
+ [{UUID, Id, Revs} | Acc]
+ end,
+
+ {ok, UUIDsIdsRevs} = couch_db:fold_purged_docs(Db, PurgeSeq0, PFoldFun, [], []),
+ case UUIDsIdsRevs of
+ [] ->
+ Acc0;
+ _ ->
+ ok = purge_on_target(Node, Name, UUIDsIdsRevs),
+ {ok, PurgeSeq} = couch_db:get_purge_seq(Db),
+ Acc0#acc{purge_seq = PurgeSeq}
+ end.
+
+
replicate_batch(#acc{target = #shard{node=Node, name=Name}} = Acc) ->
case find_missing_revs(Acc) of
[] ->
@@ -324,8 +420,19 @@ save_on_target(Node, Name, Docs) ->
ok.
+purge_on_target(Node, Name, UUIdsIdsRevs) ->
+ mem3_rpc:purge_docs(Node, Name, UUIdsIdsRevs,[
+ replicated_changes,
+ full_commit,
+ ?ADMIN_CTX,
+ {io_priority, {internal_repl, Name}}
+ ]),
+ ok.
+
+
update_locals(Acc) ->
- #acc{seq=Seq, source=Db, target=Target, localid=Id, history=History} = Acc,
+ #acc{seq=Seq, source=Db, target=Target, localid=Id,
+ history=History, purge_seq = PurgeSeq} = Acc,
#shard{name=Name, node=Node} = Target,
NewEntry = [
{<<"source_node">>, atom_to_binary(node(), utf8)},
@@ -333,7 +440,8 @@ update_locals(Acc) ->
{<<"source_seq">>, Seq},
{<<"timestamp">>, list_to_binary(iso8601_timestamp())}
],
- NewBody = mem3_rpc:save_checkpoint(Node, Name, Id, Seq, NewEntry, History),
+ NewBody = mem3_rpc:save_checkpoint(Node, Name, Id, Seq, PurgeSeq,
+ NewEntry, History),
{ok, _} = couch_db:update_doc(Db, #doc{id = Id, body = NewBody}, []).
@@ -369,6 +477,12 @@ find_repl_doc(SrcDb, TgtUUIDPrefix) ->
end.
+% used during compaction to check if _local/purge doc is current
+mem3_sync_purge(Opts)->
+ Node = couch_util:get_value(<<"node">>, Opts),
+ lists:member(Node, mem3:nodes()).
+
+
is_prefix(Prefix, Subject) ->
binary:longest_common_prefix([Prefix, Subject]) == size(Prefix).
diff --git a/src/mem3/src/mem3_rpc.erl b/src/mem3/src/mem3_rpc.erl
index c2bd58f..b9c1b39 100644
--- a/src/mem3/src/mem3_rpc.erl
+++ b/src/mem3/src/mem3_rpc.erl
@@ -19,15 +19,22 @@
find_common_seq/4,
get_missing_revs/4,
update_docs/4,
+ get_purge_seq/3,
+ purge_docs/4,
load_checkpoint/4,
- save_checkpoint/6
+ save_checkpoint/7,
+ load_purges/3,
+ save_purge_checkpoint/5
]).
% Private RPC callbacks
-export([
find_common_seq_rpc/3,
load_checkpoint_rpc/3,
- save_checkpoint_rpc/5
+ save_checkpoint_rpc/5,
+ save_checkpoint_rpc/6,
+ load_purges_rpc/2,
+ save_purge_checkpoint_rpc/4
]).
@@ -43,16 +50,34 @@ update_docs(Node, DbName, Docs, Options) ->
rexi_call(Node, {fabric_rpc, update_docs, [DbName, Docs, Options]}).
+get_purge_seq(Node, DbName, Options) ->
+ rexi_call(Node, {fabric_rpc, get_purge_seq, [DbName, Options]}).
+
+
+purge_docs(Node, DbName, PUUIdsIdsRevs, Options) ->
+ rexi_call(Node, {fabric_rpc, purge_docs, [DbName, PUUIdsIdsRevs, Options]}).
+
+
load_checkpoint(Node, DbName, SourceNode, SourceUUID) ->
Args = [DbName, SourceNode, SourceUUID],
rexi_call(Node, {mem3_rpc, load_checkpoint_rpc, Args}).
-save_checkpoint(Node, DbName, DocId, Seq, Entry, History) ->
- Args = [DbName, DocId, Seq, Entry, History],
+save_checkpoint(Node, DbName, DocId, Seq, PurgeSeq, Entry, History) ->
+ Args = [DbName, DocId, Seq, PurgeSeq, Entry, History],
rexi_call(Node, {mem3_rpc, save_checkpoint_rpc, Args}).
+load_purges(Node, DbName, SourceUUID) ->
+ Args = [DbName, SourceUUID],
+ rexi_call(Node, {mem3_rpc, load_purges_rpc, Args}).
+
+
+save_purge_checkpoint(Node, DbName, DocId, PurgeSeq, SourceNode) ->
+ Args = [DbName, DocId, PurgeSeq, SourceNode],
+ rexi_call(Node, {mem3_rpc, save_purge_checkpoint_rpc, Args}).
+
+
find_common_seq(Node, DbName, SourceUUID, SourceEpochs) ->
Args = [DbName, SourceUUID, SourceEpochs],
rexi_call(Node, {mem3_rpc, find_common_seq_rpc, Args}).
@@ -81,6 +106,7 @@ load_checkpoint_rpc(DbName, SourceNode, SourceUUID) ->
end.
+% Remove after all nodes in the cluster are upgrades to clustered purge
save_checkpoint_rpc(DbName, Id, SourceSeq, NewEntry0, History0) ->
erlang:put(io_priority, {internal_repl, DbName}),
case get_or_create_db(DbName, [?ADMIN_CTX]) of
@@ -111,6 +137,40 @@ save_checkpoint_rpc(DbName, Id, SourceSeq, NewEntry0, History0) ->
rexi:reply(Error)
end.
+
+save_checkpoint_rpc(DbName, Id, SourceSeq, SourcePurgeSeq,
+ NewEntry0, History0) ->
+ erlang:put(io_priority, {internal_repl, DbName}),
+ case get_or_create_db(DbName, [?ADMIN_CTX]) of
+ {ok, Db} ->
+ NewEntry = {[
+ {<<"target_node">>, atom_to_binary(node(), utf8)},
+ {<<"target_uuid">>, couch_db:get_uuid(Db)},
+ {<<"target_seq">>, couch_db:get_update_seq(Db)}
+ ] ++ NewEntry0},
+ Body = {[
+ {<<"seq">>, SourceSeq},
+ {<<"purge_seq">>, SourcePurgeSeq},
+ {<<"target_uuid">>, couch_db:get_uuid(Db)},
+ {<<"history">>, add_checkpoint(NewEntry, History0)}
+ ]},
+ Doc = #doc{id = Id, body = Body},
+ rexi:reply(try couch_db:update_doc(Db, Doc, []) of
+ {ok, _} ->
+ {ok, Body};
+ Else ->
+ {error, Else}
+ catch
+ Exception ->
+ Exception;
+ error:Reason ->
+ {error, Reason}
+ end);
+ Error ->
+ rexi:reply(Error)
+ end.
+
+
find_common_seq_rpc(DbName, SourceUUID, SourceEpochs) ->
erlang:put(io_priority, {internal_repl, DbName}),
case get_or_create_db(DbName, [?ADMIN_CTX]) of
@@ -128,6 +188,66 @@ find_common_seq_rpc(DbName, SourceUUID, SourceEpochs) ->
end.
+load_purges_rpc(DbName, SourceUUID) ->
+ erlang:put(io_priority, {internal_repl, DbName}),
+ case get_or_create_db(DbName, [?ADMIN_CTX]) of
+ {ok, Db} ->
+ TargetUUID = couch_db:get_uuid(Db),
+ DocId = mem3_rep:make_local_purge_id(SourceUUID, TargetUUID),
+ LastPSeq = case couch_db:open_doc(Db, DocId, []) of
+ {ok, #doc{body={Props}} } ->
+ couch_util:get_value(<<"purge_seq">>, Props);
+ {not_found, _} ->
+ % synchronize only last purge
+ {ok, OldestPSeq} = couch_db:get_oldest_purge_seq(Db),
+ erlang:max(OldestPSeq-1, 0)
+ end,
+ {ok, CurPSeq} = couch_db:get_purge_seq(Db),
+ UUIDsIdsRevs = if (LastPSeq == CurPSeq) -> []; true ->
+ FoldFun = fun({_PSeq, UUID, Id, Revs}, Acc) ->
+ [{UUID, Id, Revs} | Acc]
+ end,
+ {ok, UUIDsIdsRevs0} = couch_db:fold_purged_docs(
+ Db, LastPSeq, FoldFun, [], []
+ ),
+ UUIDsIdsRevs0
+ end,
+ rexi:reply({ok, {UUIDsIdsRevs, DocId, CurPSeq}});
+ Error ->
+ rexi:reply(Error)
+ end.
+
+
+save_purge_checkpoint_rpc(DbName, Id, PurgeSeq, Node) ->
+ erlang:put(io_priority, {internal_repl, DbName}),
+ case get_or_create_db(DbName, [?ADMIN_CTX]) of
+ {ok, Db} ->
+ Timestamp = couch_util:utc_string(),
+ Body = {[
+ {<<"purge_seq">>, PurgeSeq},
+ {<<"timestamp_utc">>, Timestamp},
+ {<<"verify_module">>, <<"mem3_rep">>},
+ {<<"verify_function">>, <<"mem3_sync_purge">>},
+ {<<"verify_options">>, {[{<<"node">>, Node}]}},
+ {<<"type">>, <<"internal_replication">>}
+ ]},
+ Doc = #doc{id = Id, body = Body},
+ rexi:reply(try couch_db:update_doc(Db, Doc, []) of
+ {ok, _} ->
+ {ok, Body};
+ Else ->
+ {error, Else}
+ catch
+ Exception ->
+ Exception;
+ error:Reason ->
+ {error, Reason}
+ end);
+ Error ->
+ rexi:reply(Error)
+ end.
+
+
%% @doc Return the sequence where two files with the same UUID diverged.
compare_epochs(SourceEpochs, TargetEpochs) ->
compare_rev_epochs(
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 06/20: WIP - couch_httpd_db.erl
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit c8e80d9b86ff7cd59812beb9f9446cf49da5084a
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Wed Mar 14 13:25:07 2018 -0500
WIP - couch_httpd_db.erl
---
src/couch/src/couch_httpd_db.erl | 22 +++++++++++++---------
1 file changed, 13 insertions(+), 9 deletions(-)
diff --git a/src/couch/src/couch_httpd_db.erl b/src/couch/src/couch_httpd_db.erl
index 99b1192..ed8a47b 100644
--- a/src/couch/src/couch_httpd_db.erl
+++ b/src/couch/src/couch_httpd_db.erl
@@ -377,16 +377,20 @@ db_req(#httpd{path_parts=[_,<<"_bulk_docs">>]}=Req, _Db) ->
db_req(#httpd{method='POST',path_parts=[_,<<"_purge">>]}=Req, Db) ->
couch_httpd:validate_ctype(Req, "application/json"),
- {IdsRevs} = couch_httpd:json_body_obj(Req),
- IdsRevs2 = [{Id, couch_doc:parse_revs(Revs)} || {Id, Revs} <- IdsRevs],
+ {IdRevs} = couch_httpd:json_body_obj(Req),
+ PurgeReqs = lists:map(fun({Id, JsonRevs} ->
+ {couch_uuids:new(), Id, couch_doc:parse_revs(Revs)}
+ end, IdRevs),
- case couch_db:purge_docs(Db, IdsRevs2) of
- {ok, PurgeSeq, PurgedIdsRevs} ->
- PurgedIdsRevs2 = [{Id, couch_doc:revs_to_strs(Revs)} || {Id, Revs} <- PurgedIdsRevs],
- send_json(Req, 200, {[{<<"purge_seq">>, PurgeSeq}, {<<"purged">>, {PurgedIdsRevs2}}]});
- Error ->
- throw(Error)
- end;
+ {ok, Replies} = couch_db:purge_docs(Db, PurgeReqs),
+
+ Results = lists:zipwith(fun({{Id, _}, Reply}) ->
+ {Id, couch_doc:revs_to_strs(Reply)}
+ end, IdRevs, PurgeReqs),
+
+ {ok, Db2} = couch_db:reopen(Db),
+ {ok, PurgeSeq} = couch_db:get_purge_seq(Db2),
+ send_json(Req, 200, {[{purge_seq, PurgeSeq}, {purged, {Results}}]});
db_req(#httpd{path_parts=[_,<<"_purge">>]}=Req, _Db) ->
send_method_not_allowed(Req, "POST");
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 05/20: WIP - couch_db_engine.erl
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit a7195520d2f76b8e4dcb5ee16096117e53d20da7
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Wed Mar 14 13:19:05 2018 -0500
WIP - couch_db_engine.erl
---
src/couch/src/couch_db_engine.erl | 20 ++++++++++++--------
1 file changed, 12 insertions(+), 8 deletions(-)
diff --git a/src/couch/src/couch_db_engine.erl b/src/couch/src/couch_db_engine.erl
index 0e7a1cf..50682b2 100644
--- a/src/couch/src/couch_db_engine.erl
+++ b/src/couch/src/couch_db_engine.erl
@@ -398,13 +398,14 @@
% and as such is guaranteed single threaded for the given
% DbHandle.
%
-% The DocPair argument is a 2-tuple of #full_doc_info{} records. The
-% first element of th epair is the #full_doc_info{} that exists
+% Each doc_pair() is a 2-tuple of #full_doc_info{} records. The
+% first element of the pair is the #full_doc_info{} that exists
% on disk. The second element is the new version that should be
-% written to disk. There are two basic cases that should be considered:
+% written to disk. There are three basic cases that should be considered:
%
% 1. {#full_doc_info{}, #full_doc_info{}} - A document was partially purged
% 2. {#full_doc_info{}, not_found} - A document was completely purged
+% 3. {not_found, not_found} - A no-op purge
%
% In case 1, non-tail-append engines may have to remove revisions
% specifically rather than rely on compaction to remove them. Also
@@ -415,11 +416,14 @@
% means it needs to be removed from the database including the
% update sequence.
%
-% The PurgeInfo contains the purge_seq, uuid, docid and revisions that
-% were requested to be purged. This should be persisted in such a way
-% that we can efficiently load purge_info() by its UUID as well as
-% iterate over purge_info() entries in order of their PurgeSeq.
--callback purge_doc(DbHandle::db_handle(), doc_pair(), purge_info()) ->
+% In case 3 we just need to store the purge_info() to know that it
+% was processed even though it produced no changes to the database.
+%
+% The purge_info() tuples contain the purge_seq, uuid, docid and
+% revisions that were requested to be purged. This should be persisted
+% in such a way that we can efficiently load purge_info() by its UUID
+% as well as iterate over purge_info() entries in order of their PurgeSeq.
+-callback purge_docs(DbHandle::db_handle(), [doc_pair()], [purge_info()]) ->
{ok, NewDbHandle::db_handle()}.
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 18/20: Add metrics for clustered purge
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 986d39e07f8a30fbb21d62e44f3003ca9247435a
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Tue Jun 27 14:59:03 2017 -0400
Add metrics for clustered purge
COUCHDB-3326
---
src/chttpd/src/chttpd_db.erl | 1 +
src/couch/priv/stats_descriptions.cfg | 12 ++++++++++++
src/couch/src/couch_db.erl | 1 +
src/couch/src/couch_httpd_db.erl | 1 +
src/couch_index/src/couch_index_updater.erl | 17 ++++++++++++++---
src/couch_mrview/src/couch_mrview_index.erl | 11 ++++++++---
src/couch_mrview/src/couch_mrview_updater.erl | 14 +++++++++-----
7 files changed, 46 insertions(+), 11 deletions(-)
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index f469b98..fbfae66 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -496,6 +496,7 @@ db_req(#httpd{path_parts=[_, <<"_bulk_get">>]}=Req, _Db) ->
db_req(#httpd{method='POST',path_parts=[_,<<"_purge">>]}=Req, Db) ->
+ couch_stats:increment_counter([couchdb, httpd, purge_requests]),
chttpd:validate_ctype(Req, "application/json"),
W = chttpd:qs_value(Req, "w", integer_to_list(mem3:quorum(Db))),
Options = [{user_ctx, Req#httpd.user_ctx}, {w, W}],
diff --git a/src/couch/priv/stats_descriptions.cfg b/src/couch/priv/stats_descriptions.cfg
index f091978..bceb0ce 100644
--- a/src/couch/priv/stats_descriptions.cfg
+++ b/src/couch/priv/stats_descriptions.cfg
@@ -34,6 +34,10 @@
{type, counter},
{desc, <<"number of times a document was read from a database">>}
]}.
+{[couchdb, database_purges], [
+ {type, counter},
+ {desc, <<"number of times a database was purged">>}
+]}.
{[couchdb, db_open_time], [
{type, histogram},
{desc, <<"milliseconds required to open a database">>}
@@ -46,6 +50,10 @@
{type, counter},
{desc, <<"number of document write operations">>}
]}.
+{[couchdb, document_purges], [
+ {type, counter},
+ {desc, <<"number of document purge operations">>}
+]}.
{[couchdb, local_document_writes], [
{type, counter},
{desc, <<"number of _local document write operations">>}
@@ -74,6 +82,10 @@
{type, counter},
{desc, <<"number of clients for continuous _changes">>}
]}.
+{[couchdb, httpd, purge_requests], [
+ {type, counter},
+ {desc, <<"number of purge requests">>}
+]}.
{[couchdb, httpd_request_methods, 'COPY'], [
{type, counter},
{desc, <<"number of HTTP COPY requests">>}
diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index 3ef6ab0..2a2095c 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -382,6 +382,7 @@ get_full_doc_infos(Db, Ids) ->
Rev :: {non_neg_integer(), binary()},
Reply :: {ok, []} | {ok, [Rev]}.
purge_docs(#db{main_pid = Pid}, UUIdsIdsRevs) ->
+ increment_stat(Db, [couchdb, database_purges]),
gen_server:call(Pid, {purge_docs, UUIdsIdsRevs});
-spec get_purge_infos(#db{}, [UUId]) -> [PurgeInfo] when
diff --git a/src/couch/src/couch_httpd_db.erl b/src/couch/src/couch_httpd_db.erl
index ed8a47b..b407c02 100644
--- a/src/couch/src/couch_httpd_db.erl
+++ b/src/couch/src/couch_httpd_db.erl
@@ -376,6 +376,7 @@ db_req(#httpd{path_parts=[_,<<"_bulk_docs">>]}=Req, _Db) ->
send_method_not_allowed(Req, "POST");
db_req(#httpd{method='POST',path_parts=[_,<<"_purge">>]}=Req, Db) ->
+ couch_stats:increment_counter([couchdb, httpd, purge_requests]),
couch_httpd:validate_ctype(Req, "application/json"),
{IdRevs} = couch_httpd:json_body_obj(Req),
PurgeReqs = lists:map(fun({Id, JsonRevs} ->
diff --git a/src/couch_index/src/couch_index_updater.erl b/src/couch_index/src/couch_index_updater.erl
index b416d17..6dc587d 100644
--- a/src/couch_index/src/couch_index_updater.erl
+++ b/src/couch_index/src/couch_index_updater.erl
@@ -141,10 +141,11 @@ update(Idx, Mod, IdxState) ->
DbUpdateSeq = couch_db:get_update_seq(Db),
DbCommittedSeq = couch_db:get_committed_update_seq(Db),
+ NumUpdateChanges = couch_db:count_changes_since(Db, CurrSeq),
+ NumPurgeChanges = count_pending_purged_docs_since(Db, Mod, IdxState),
+ TotalChanges = NumUpdateChanges + NumPurgeChanges,
{ok, PurgedIdxState} = purge_index(Db, Mod, IdxState),
- NumChanges = couch_db:count_changes_since(Db, CurrSeq),
-
GetSeq = fun
(#full_doc_info{update_seq=Seq}) -> Seq;
(#doc_info{high_seq=Seq}) -> Seq
@@ -182,8 +183,13 @@ update(Idx, Mod, IdxState) ->
{ok, {NewSt, true}}
end
end,
+ {ok, InitIdxState} = Mod:start_update(
+ Idx,
+ PurgedIdxState,
+ TotalChanges,
+ NumPurgeChanges
+ ),
- {ok, InitIdxState} = Mod:start_update(Idx, PurgedIdxState, NumChanges),
Acc0 = {InitIdxState, true},
{ok, Acc} = couch_db:fold_changes(Db, CurrSeq, Proc, Acc0, []),
{ProcIdxSt, SendLast} = Acc,
@@ -223,3 +229,8 @@ purge_index(Db, Mod, IdxState) ->
Mod:update_local_purge_doc(Db, NewStateAcc),
{ok, NewStateAcc}
end.
+
+count_pending_purged_docs_since(Db, Mod, IdxState) ->
+ {ok, DbPurgeSeq} = couch_db:get_purge_seq(Db),
+ IdxPurgeSeq = Mod:get(purge_seq, IdxState),
+ DbPurgeSeq - IdxPurgeSeq.
diff --git a/src/couch_mrview/src/couch_mrview_index.erl b/src/couch_mrview/src/couch_mrview_index.erl
index 6c63eaf..3f04f8f 100644
--- a/src/couch_mrview/src/couch_mrview_index.erl
+++ b/src/couch_mrview/src/couch_mrview_index.erl
@@ -15,7 +15,7 @@
-export([get/2]).
-export([init/2, open/2, close/1, reset/1, delete/1, shutdown/1]).
--export([start_update/3, purge/4, process_doc/3, finish_update/1, commit/1]).
+-export([start_update/4, purge/4, process_doc/3, finish_update/1, commit/1]).
-export([compact/3, swap_compacted/2, remove_compacted/1]).
-export([index_file_exists/1]).
-export([update_local_purge_doc/2, verify_index_exists/1]).
@@ -172,8 +172,13 @@ reset(State) ->
end).
-start_update(PartialDest, State, NumChanges) ->
- couch_mrview_updater:start_update(PartialDest, State, NumChanges).
+start_update(PartialDest, State, NumChanges, NumChangesDone) ->
+ couch_mrview_updater:start_update(
+ PartialDest,
+ State,
+ NumChanges,
+ NumChangesDone
+ ).
purge(Db, PurgeSeq, PurgedIdRevs, State) ->
diff --git a/src/couch_mrview/src/couch_mrview_updater.erl b/src/couch_mrview/src/couch_mrview_updater.erl
index 214f487..3383b49 100644
--- a/src/couch_mrview/src/couch_mrview_updater.erl
+++ b/src/couch_mrview/src/couch_mrview_updater.erl
@@ -12,15 +12,14 @@
-module(couch_mrview_updater).
--export([start_update/3, purge/4, process_doc/3, finish_update/1]).
+-export([start_update/4, purge/4, process_doc/3, finish_update/1]).
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
-define(REM_VAL, removed).
-
-start_update(Partial, State, NumChanges) ->
+start_update(Partial, State, NumChanges, NumChangesDone) ->
MaxSize = config:get_integer("view_updater", "queue_memory_cap", 100000),
MaxItems = config:get_integer("view_updater", "queue_item_cap", 500),
QueueOpts = [{max_size, MaxSize}, {max_items, MaxItems}],
@@ -36,14 +35,19 @@ start_update(Partial, State, NumChanges) ->
},
Self = self(),
+
MapFun = fun() ->
+ Progress = case NumChanges of
+ 0 -> 0;
+ _ -> (NumChangesDone * 100) div NumChanges
+ end,
couch_task_status:add_task([
{indexer_pid, ?l2b(pid_to_list(Partial))},
{type, indexer},
{database, State#mrst.db_name},
{design_document, State#mrst.idx_name},
- {progress, 0},
- {changes_done, 0},
+ {progress, Progress},
+ {changes_done, NumChangesDone},
{total_changes, NumChanges}
]),
couch_task_status:set_update_frequency(500),
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 10/20: WIP - couch_bt_engine implementation
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit dd43b72330627e579fc03459d16e773c1e1d5a11
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Fri Mar 16 14:57:36 2018 -0500
WIP - couch_bt_engine implementation
---
src/couch/src/couch_bt_engine.erl | 236 +++++++++++++++++++++++-----
src/couch/src/couch_bt_engine.hrl | 4 +-
src/couch/src/couch_bt_engine_compactor.erl | 107 +++++++++++--
src/couch/src/couch_bt_engine_header.erl | 35 +++--
4 files changed, 316 insertions(+), 66 deletions(-)
diff --git a/src/couch/src/couch_bt_engine.erl b/src/couch/src/couch_bt_engine.erl
index 9406897..485d21d 100644
--- a/src/couch/src/couch_bt_engine.erl
+++ b/src/couch/src/couch_bt_engine.erl
@@ -35,8 +35,9 @@
get_disk_version/1,
get_doc_count/1,
get_epochs/1,
- get_last_purged/1,
get_purge_seq/1,
+ get_oldest_purge_seq/1,
+ get_purge_infos_limit/1,
get_revs_limit/1,
get_security/1,
get_size_info/1,
@@ -44,15 +45,19 @@
get_uuid/1,
set_revs_limit/2,
+ set_purge_infos_limit/2,
set_security/2,
+ set_purged_docs_limit/2,
open_docs/2,
open_local_docs/2,
read_doc_body/2,
+ load_purge_infos/2,
serialize_doc/2,
write_doc_body/2,
- write_doc_infos/4,
+ write_doc_infos/3,
+ purge_docs/3,
commit_data/1,
@@ -63,6 +68,7 @@
fold_docs/4,
fold_local_docs/4,
fold_changes/5,
+ fold_purge_infos/5,
count_changes_since/2,
start_compaction/4,
@@ -85,7 +91,13 @@
seq_tree_reduce/2,
local_tree_split/1,
- local_tree_join/2
+ local_tree_join/2,
+
+ purge_tree_split/1,
+ purge_tree_join/2,
+ purge_tree_reduce/2,
+ purge_seq_tree_split/1,
+ purge_seq_tree_join/2
]).
@@ -216,18 +228,24 @@ get_epochs(#st{header = Header}) ->
couch_bt_engine_header:get(Header, epochs).
-get_last_purged(#st{header = Header} = St) ->
- case couch_bt_engine_header:get(Header, purged_docs) of
- nil ->
- [];
- Pointer ->
- {ok, PurgeInfo} = couch_file:pread_term(St#st.fd, Pointer),
- PurgeInfo
- end.
+get_purge_seq(#st{purge_seq_tree = PurgeSeqTree}) ->
+ Fun = fun({PurgeSeq, _, _, _}, _Reds, _Acc) ->
+ {stop, PurgeSeq}
+ end,
+ {ok, _, PurgeSeq} = couch_btree:fold(PurgeSeqTree, Fun, 0, [{dir, rev}]),
+ PurgeSeq.
+
+
+get_oldest_purge_seq(#st{purge_seq_tree = PurgeSeqTree}) ->
+ Fun = fun({PurgeSeq, _, _, _}, _Reds, _Acc) ->
+ {stop, PurgeSeq}
+ end,
+ {ok, _, PurgeSeq} = couch_btree:fold(PurgeSeqTree, Fun, 0, []),
+ PurgeSeq.
-get_purge_seq(#st{header = Header}) ->
- couch_bt_engine_header:get(Header, purge_seq).
+get_purge_infos_limit(#st{header = Header}) ->
+ couch_bt_engine_header:get(Header, purge_infos_limit).
get_revs_limit(#st{header = Header}) ->
@@ -283,6 +301,16 @@ set_revs_limit(#st{header = Header} = St, RevsLimit) ->
{ok, increment_update_seq(NewSt)}.
+set_purge_infos_limit(#st{header = Header} = St, PurgeInfosLimit) ->
+ NewSt = St#st{
+ header = couch_bt_engine_header:set(Header, [
+ {purged_docs_limit, PurgeInfosLimit}
+ ]),
+ needs_commit = true
+ },
+ {ok, increment_update_seq(NewSt)}.
+
+
set_security(#st{header = Header} = St, NewSecurity) ->
Options = [{compression, St#st.compression}],
{ok, Ptr, _} = couch_file:append_term(St#st.fd, NewSecurity, Options),
@@ -319,6 +347,14 @@ read_doc_body(#st{} = St, #doc{} = Doc) ->
}.
+load_purge_infos(St, UUIDs) ->
+ Results = couch_btree:lookup(St#st.upurge_tree, UUIDs),
+ lists:map(fun
+ ({ok, Info}) -> Info;
+ (not_found) -> not_found
+ end, Results).
+
+
serialize_doc(#st{} = St, #doc{} = Doc) ->
Compress = fun(Term) ->
case couch_compress:is_compressed(Term, St#st.compression) of
@@ -350,7 +386,7 @@ write_doc_body(St, #doc{} = Doc) ->
{ok, Doc#doc{body = Ptr}, Written}.
-write_doc_infos(#st{} = St, Pairs, LocalDocs, PurgedIdRevs) ->
+write_doc_infos(#st{} = St, Pairs, LocalDocs) ->
#st{
id_tree = IdTree,
seq_tree = SeqTree,
@@ -390,23 +426,9 @@ write_doc_infos(#st{} = St, Pairs, LocalDocs, PurgedIdRevs) ->
erlang:max(Seq, Acc)
end, get_update_seq(St), Add),
- NewHeader = case PurgedIdRevs of
- [] ->
- couch_bt_engine_header:set(St#st.header, [
- {update_seq, NewUpdateSeq}
- ]);
- _ ->
- {ok, Ptr, _} = couch_file:append_term(St#st.fd, PurgedIdRevs),
- OldPurgeSeq = couch_bt_engine_header:get(St#st.header, purge_seq),
- % We bump NewUpdateSeq because we have to ensure that
- % indexers see that they need to process the new purge
- % information.
- couch_bt_engine_header:set(St#st.header, [
- {update_seq, NewUpdateSeq + 1},
- {purge_seq, OldPurgeSeq + 1},
- {purged_docs, Ptr}
- ])
- end,
+ NewHeader = couch_bt_engine_header:set(St#st.header, [
+ {update_seq, NewUpdateSeq}
+ ]),
{ok, St#st{
header = NewHeader,
@@ -417,6 +439,46 @@ write_doc_infos(#st{} = St, Pairs, LocalDocs, PurgedIdRevs) ->
}}.
+purge_docs(#st{} = St, Pairs, PurgeInfos) ->
+ #st{
+ id_tree = IdTree,
+ seq_tree = SeqTree,
+ purge_tree = PurgeTree,
+ upurge_tree = UPurgeTree
+ } = St,
+
+ RemDocIds = [Old#full_doc_info.doc_id || {Old, not_found} <- Pairs],
+ RemSeqs = [Old#full_doc_info.update_seq || {Old, _} <- Pairs],
+ DocsToAdd = [New || {_, New} <- Pairs, New /= not_found],
+ CurrSeq = couch_bt_engine_header:get(St#st.header, update_seq),
+ Seqs = [FDI#full_doc_info.update_seq || FDI <- DocsToAdd],
+ NewSeq = lists:max([CurrSeq | Seqs]),
+
+ % We bump NewUpdateSeq because we have to ensure that
+ % indexers see that they need to process the new purge
+ % information.
+ UpdateSeq = case NewSeq == CurrSeq of
+ true -> InitUpdateSeq + 1;
+ false -> NewUpdateSeq
+ end,
+ Header = couch_bt_engine_header:set(St#st.header, [
+ {update_seq, UpdateSeq}
+ ]),
+
+ {ok, IdTree2} = couch_btree:add_remove(IdTree, DocsToAdd, RemDocIds),
+ {ok, SeqTree2} = couch_btree:add_remove(SeqTree, DocsToAdd, RemSeqs),
+ {ok, PurgeTree2} = couch_btree:add(PurgeTree, PurgeInfos),
+ {ok, PurgeSeqTree2} = couch_btree:add(PurgeSeqTree, PurgeInfos),
+ {ok, St#st{
+ header = Header2,
+ id_tree = IdTree2,
+ seq_tree = SeqTree2,
+ purge_tree = PurgeTree2,
+ purge_seq_tree = PurgeSeqTree2,
+ needs_commit = true
+ }}.
+
+
commit_data(St) ->
#st{
fd = Fd,
@@ -479,6 +541,21 @@ fold_changes(St, SinceSeq, UserFun, UserAcc, Options) ->
{ok, FinalUserAcc}.
+fold_purge_infos(St, StartSeq0, UserFun, UserAcc, Options) ->
+ PurgeSeqTree = St#st.purge_seq_tree,
+ StartSeq = StartSeq0 + 1,
+ MinSeq = load_oldest_purge_seq(PurgeSeqTree),
+ if MinSeq =< StartSeq -> ok; true ->
+ throw({invalid_start_purge_seq, StartSeq0})
+ end,
+ Wrapper = fun(Info, _Reds, UAcc) ->
+ UserFun(Info, UAcc)
+ end,
+ Opts = [{start_key, StartSeq}] ++ Options,
+ {ok, _, OutAcc} = couch_btree:fold(PurgeSeqTree, Wrapper, UserAcc, Opts),
+ {ok, OutAcc};
+
+
count_changes_since(St, SinceSeq) ->
BTree = St#st.seq_tree,
FoldFun = fun(_SeqStart, PartialReds, 0) ->
@@ -618,6 +695,13 @@ local_tree_split(#doc{revs = {0, [Rev]}} = Doc) when is_integer(Rev) ->
{Id, {Rev, BodyData}}.
+local_tree_join(Id, {Rev, BodyData}) when is_binary(Rev) ->
+ #doc{
+ id = Id,
+ revs = {0, [Rev]},
+ body = BodyData
+ };
+
local_tree_join(Id, {Rev, BodyData}) when is_integer(Rev) ->
#doc{
id = Id,
@@ -626,6 +710,29 @@ local_tree_join(Id, {Rev, BodyData}) when is_integer(Rev) ->
}.
+purge_tree_split({PurgeSeq, UUID, DocId, Revs}) ->
+ {UUID, {PurgeSeq, DocId, Revs}}.
+
+
+purge_tree_join({UUID, {PurgeSeq, DocId, Revs}}) ->
+ {PurgeSeq, UUID, DocId, Revs}.
+
+
+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,7 +788,8 @@ 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),
+ Header2 = set_default_security_object(Fd, Header1, Compression, Options),
+ Header = upgrade_purge_info(Fd, Header2),
IdTreeState = couch_bt_engine_header:id_tree_state(Header),
{ok, IdTree} = couch_btree:open(IdTreeState, Fd, [
@@ -706,6 +814,20 @@ init_state(FilePath, Fd, Header0, Options) ->
{compression, Compression}
]),
+ PurgeTreeState = couch_bt_engine_header:purge_tree_state(Header),
+ {ok, PurgeTree} = couch_btree:open(PurgeTreeState, Fd, [
+ {split, fun ?MODULE:purge_tree_split/1},
+ {join, fun ?MODULE:purge_tree_join/2},
+ {reduce, fun ?MODULE:purge_tree_reduce/2}
+ ]),
+
+ PurgeSeqTreeState = couch_bt_engine_header:upurge_tree_state(Header),
+ {ok, PurgeSeqTree} = 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}
+ ]),
+
ok = couch_file:set_db_pid(Fd, self()),
St = #st{
@@ -718,7 +840,9 @@ init_state(FilePath, Fd, Header0, Options) ->
id_tree = IdTree,
seq_tree = SeqTree,
local_tree = LocalTree,
- compression = Compression
+ compression = Compression,
+ purge_tree = PurgeTree,
+ purge_seq_tree = PurgeSeqTree
},
% If this is a new database we've just created a
@@ -737,7 +861,9 @@ update_header(St, Header) ->
couch_bt_engine_header:set(Header, [
{seq_tree_state, couch_btree:get_state(St#st.seq_tree)},
{id_tree_state, couch_btree:get_state(St#st.id_tree)},
- {local_tree_state, couch_btree:get_state(St#st.local_tree)}
+ {local_tree_state, couch_btree:get_state(St#st.local_tree)},
+ {purge_tree_state, couch_btree:get_state(St#st.purge_tree)},
+ {purge_seq_tree_state, couch_btree:get_state(St#st.purge_seq_tree)}
]).
@@ -762,6 +888,45 @@ set_default_security_object(Fd, Header, Compression, Options) ->
end.
+% This function is here, and not in couch_bt_engine_header
+% because it requires modifying file contents
+upgrade_purge_info(Fd, Header) ->
+ case couch_bt_engine_header:get(Header, purge_tree_state) of
+ nil ->
+ Header;
+ Ptr when is_tuple(Ptr) ->
+ Header;
+ PurgeSeq when is_integer(PurgeSeq)->
+ {ok, PurgedIdsRevs} = couch_file:pread_term(Fd, Ptr),
+
+ {Infos, NewSeq} = lists:foldl(fun({Id, Revs}, {InfoAcc, PSeq}) ->
+ Info = {PSeq, couch_uuids:random(), Id, Revs},
+ {[Info | InfoAcc], PSeq + 1}
+ end, {[], PurgeSeq}, PurgedIdsRevs),
+
+ {ok, PurgeTree} = couch_btree:open(nil, Fd, [
+ {split, fun ?MODULE:purge_tree_split/1},
+ {join, fun ?MODULE:purge_tree_join/2},
+ {reduce, fun ?MODULE:purge_tree_reduce/2}
+ ]),
+ {ok, PurgeTree2} = couch_btree:add(PurgeTree, Infos),
+ {ok, PurgeTreeSt} = couch_btree:get_state(PurgeTree2),
+
+ {ok, PurgeSeqTree} = couch_btree:open(nil, Fd, [
+ {split, fun ?MODULE:purge_seq_tree_split/1},
+ {join, fun ?MODULE:purge_seq_tree_join/2},
+ {reduce, fun ?MODULE:purge_tree_reduce/2}
+ ]),
+ {ok, PurgeSeqTree2} = couch_btree:add(PurgeSeqTree, Infos),
+ {ok, PurgeSeqTreeSt} = couch_btree:get_state(PurgeSeqTree2),
+
+ couch_bt_engine_header:set(Header, [
+ {purge_tree_state, PTreeState},
+ {purge_seq_tree_state, UPTreeState}
+ ])
+ end.
+
+
delete_compaction_files(FilePath) ->
RootDir = config:get("couchdb", "database_dir", "."),
DelOpts = [{context, delete}],
@@ -932,7 +1097,8 @@ finish_compaction_int(#st{} = OldSt, #st{} = NewSt1) ->
{ok, NewSt2} = commit_data(NewSt1#st{
header = couch_bt_engine_header:set(Header, [
{compacted_seq, get_update_seq(OldSt)},
- {revs_limit, get_revs_limit(OldSt)}
+ {revs_limit, get_revs_limit(OldSt)},
+ {purge_infos_limit, get_purged_docs_limit(OldSt)}
]),
local_tree = NewLocal2
}),
diff --git a/src/couch/src/couch_bt_engine.hrl b/src/couch/src/couch_bt_engine.hrl
index 7f52d8f..1f5bcc9 100644
--- a/src/couch/src/couch_bt_engine.hrl
+++ b/src/couch/src/couch_bt_engine.hrl
@@ -20,5 +20,7 @@
id_tree,
seq_tree,
local_tree,
- compression
+ compression,
+ purge_tree,
+ purge_seq_tree
}).
diff --git a/src/couch/src/couch_bt_engine_compactor.erl b/src/couch/src/couch_bt_engine_compactor.erl
index 2f7a351..7b11b44 100644
--- a/src/couch/src/couch_bt_engine_compactor.erl
+++ b/src/couch/src/couch_bt_engine_compactor.erl
@@ -54,7 +54,7 @@ start(#st{} = St, DbName, Options, Parent) ->
% and hope everything works out for the best.
unlink(DFd),
- NewSt1 = copy_purge_info(St, NewSt),
+ NewSt1 = copy_purge_info(DbName, St, NewSt, Retry),
NewSt2 = copy_compact(DbName, St, NewSt1, Retry),
NewSt3 = sort_meta_data(NewSt2),
NewSt4 = commit_compaction_data(NewSt3),
@@ -97,22 +97,97 @@ open_compaction_files(SrcHdr, DbFilePath, Options) ->
end.
-copy_purge_info(OldSt, NewSt) ->
- OldHdr = OldSt#st.header,
- NewHdr = NewSt#st.header,
- OldPurgeSeq = couch_bt_engine_header:purge_seq(OldHdr),
- case OldPurgeSeq > 0 of
+copy_purge_info(DbName, OldSt, NewSt, Retry) ->
+ MinPurgeSeq = couch_util:with_db(DbName, fun(Db) ->
+ couch_db:get_minimum_purge_seq(Db)
+ end),
+ OldIdTree = OldSt#st.id_tree,
+ OldPSTree = OldSt#st.purge_seq_tree,
+ StartSeq = couch_bt_engine:get_purge_seq(NewSt) + 1,
+ BufferSize = config:get_integer(
+ "database_compaction", "doc_buffer_size", 524288),
+ CheckpointAfter = config:get(
+ "database_compaction", "checkpoint_after", BufferSize * 10),
+
+ EnumFun = fun(Info, _Reds, {StAcc, InfosAcc0, InfosSize, CopiedSize}) ->
+ NewInfosSize = InfosSize + ?term_size(Info),
+ if NewInfosSize >= BufferSize ->
+ StAcc1 = copy_purge_infos(
+ OldSt, StAcc0, [Info | InfosAcc], MinPurgeSeq, Retry),
+ NewCopiedSize = CopiedSize + NewInfosSize,
+ if NewCopiedSize >= CheckpointAfter ->
+ StAcc2 = commit_compaction_data(StAcc1),
+ {ok, {StAcc2, [], 0, 0}};
+ true ->
+ {ok, {StAcc2, [], 0, NewCopiedSize}}
+ end;
true ->
- Purged = couch_bt_engine:get_last_purged(OldSt),
- Opts = [{compression, NewSt#st.compression}],
- {ok, Ptr, _} = couch_file:append_term(NewSt#st.fd, Purged, Opts),
- NewNewHdr = couch_bt_engine_header:set(NewHdr, [
- {purge_seq, OldPurgeSeq},
- {purged_docs, Ptr}
- ]),
- NewSt#st{header = NewNewHdr};
- false ->
- NewSt
+ NewInfosAcc = [Info | InfosAcc]
+ {ok, {StAcc, NewInfosAcc, NewInfosSize, CopiedSize}}
+ end
+ end,
+
+ InitAcc = {NewSt, [], 0, 0},
+ Opts = [{start_key, StartSeq}],
+ {ok, _, FinalAcc} = couch_btree:fold(OldPSTree, EnumFun, InitAcc, Opts),
+ {NewStAcc, Infos, _, _} = FinalAcc,
+ copy_purge_infos(OldSt, NewStAcc, Infos, MinPurgeSeq, Retry).
+
+
+copy_purge_docs(OldSt, NewSt, Infos, MinPurgeSeq, Retry) ->
+ #st{
+ id_tree = OldIdTree
+ } = OldSt,
+ #st{
+ id_tree = NewIdTree0,
+ seq_tree = NewSeqTree0,
+ purge_tree = NewPurgeTree0,
+ purge_seq_tree = NewPurgeSeqTree0
+ } = NewSt,
+
+ % Copy over the purge infos
+ InfosToAdd = lists:dropwhile(fun({PSeq, _, _, _}) ->
+ PSeq < MinPurgeSeq
+ end, Infos),
+ {ok, NewPurgeTree1} = couch_btree:add(NewPurgeTree0, InfosToAdd),
+ {ok, NewPurgeSeqTree1} = couch_btree:add(NewPurgeSeqTree0, InfosToAdd),
+ NewSt1 = NewSt#st{
+ purge_tree = NewPurgeTree1,
+ purge_seq_tree = NewPurgeSeqTree1
+ },
+
+ % If we're peforming a retry compaction we have to check if
+ % any of the referenced docs have been completely purged
+ % from the database. Any doc that has been completely purged
+ % must then be removed from our partially compacted database.
+ if Retry == nil -> NewSt1; true ->
+ AllDocIds = [DocId || {_PurgeSeq, _UUID, DocId, _Revs} <- Infos],
+ UniqDocIds = lists:usort(AllDocIds),
+ {ok, OldIdResults} = couch_btree:lookup(OldIdTree, UniqDocIds),
+ OldZipped = lists:zip(UniqDocIds, Results),
+
+ % The list of non-existant docs in the database being compacted
+ MaybeRemDocIds = [DocId || {DocId, not_found} <- Zipped],
+
+ % Removing anything that exists in the partially compacted database
+ {ok, NewIdResults} = couch_btree:lookup(NewIdTree, MaybeRemDocIds),
+ ToRemove = [Doc || Doc <- NewIdResults, Doc /= not_found],
+
+ {RemIds, RemSeqs} = lists:unzip(lists:map(fun(FDI) ->
+ #full_doc_info{
+ id = Id,
+ update_seq = Seq
+ } = FDI,
+ {Id, Seq}
+ end, ToRemove)),
+
+ {ok, NewIdTree1} = couch_btree:add_remove(NewIdTree0, [], RemIds),
+ {ok, NewSeqTree1} = couch_btree:add_remove(NewSeqTree0, [], RemSeqs),
+
+ NewSt1#st{
+ id_tree = NewIdTree1,
+ seq_tree = NewSeqTree1
+ }
end.
diff --git a/src/couch/src/couch_bt_engine_header.erl b/src/couch/src/couch_bt_engine_header.erl
index 3d24f31..55246ac 100644
--- a/src/couch/src/couch_bt_engine_header.erl
+++ b/src/couch/src/couch_bt_engine_header.erl
@@ -31,8 +31,9 @@
seq_tree_state/1,
latest/1,
local_tree_state/1,
- purge_seq/1,
- purged_docs/1,
+ purge_tree_state/1,
+ purge_seq_tree_state/1,
+ purged_docs_limit/1,
security_ptr/1,
revs_limit/1,
uuid/1,
@@ -51,7 +52,7 @@
% if the disk revision is incremented, then new upgrade logic will need to be
% added to couch_db_updater:init_db.
--define(LATEST_DISK_VERSION, 6).
+-define(LATEST_DISK_VERSION, 7).
-record(db_header, {
disk_version = ?LATEST_DISK_VERSION,
@@ -60,13 +61,14 @@
id_tree_state = nil,
seq_tree_state = nil,
local_tree_state = nil,
- purge_seq = 0,
- purged_docs = nil,
+ purge_tree_state = nil,
+ purge_seq_tree_state = nil, %purge tree: purge_seq -> uuid
security_ptr = nil,
revs_limit = 1000,
uuid,
epochs,
- compacted_seq
+ compacted_seq,
+ purge_infos_limit = 1000
}).
@@ -150,12 +152,12 @@ local_tree_state(Header) ->
get_field(Header, local_tree_state).
-purge_seq(Header) ->
- get_field(Header, purge_seq).
+purge_tree_state(Header) ->
+ get_field(Header, purge_tree_state).
-purged_docs(Header) ->
- get_field(Header, purged_docs).
+purge_seq_tree_state(Header) ->
+ get_field(Header, purge_seq_tree_state).
security_ptr(Header) ->
@@ -178,6 +180,10 @@ compacted_seq(Header) ->
get_field(Header, compacted_seq).
+purge_infos_limit(Header) ->
+ get_field(Header, purge_infos_limit).
+
+
get_field(Header, Field) ->
get_field(Header, Field, undefined).
@@ -229,6 +235,7 @@ upgrade_disk_version(#db_header{}=Header) ->
3 -> throw({database_disk_version_error, ?OLD_DISK_VERSION_ERROR});
4 -> Header#db_header{security_ptr = nil}; % [0.10 - 0.11)
5 -> Header; % pre 1.2
+ 6 -> Header; % pre clustered purge
?LATEST_DISK_VERSION -> Header;
_ ->
Reason = "Incorrect disk header version",
@@ -322,8 +329,8 @@ mk_header(Vsn) ->
foo, % id_tree_state
bar, % seq_tree_state
bam, % local_tree_state
- 1, % purge_seq
- baz, % purged_docs
+ flam, % was purge_seq - now purge_tree_state
+ baz, % was purged_docs - now purge_seq_tree_state
bang, % security_ptr
999 % revs_limit
}.
@@ -342,8 +349,8 @@ upgrade_v3_test() ->
?assertEqual(foo, id_tree_state(NewHeader)),
?assertEqual(bar, seq_tree_state(NewHeader)),
?assertEqual(bam, local_tree_state(NewHeader)),
- ?assertEqual(1, purge_seq(NewHeader)),
- ?assertEqual(baz, purged_docs(NewHeader)),
+ ?assertEqual(flam, purge_tree_state(NewHeader)),
+ ?assertEqual(baz, purge_seq_tree_state(NewHeader)),
?assertEqual(bang, security_ptr(NewHeader)),
?assertEqual(999, revs_limit(NewHeader)),
?assertEqual(undefined, uuid(NewHeader)),
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 01/20: ss - typos in couch_db_engine.erl
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 6ba2ae71ea036ef7d23ebc21b6190609da2bbdad
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Wed Mar 14 12:19:16 2018 -0500
ss - typos in couch_db_engine.erl
---
src/couch/src/couch_db_engine.erl | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/src/couch/src/couch_db_engine.erl b/src/couch/src/couch_db_engine.erl
index 1a8df38..4974201 100644
--- a/src/couch/src/couch_db_engine.erl
+++ b/src/couch/src/couch_db_engine.erl
@@ -436,9 +436,9 @@
%
% 1. start_key - Start iteration at the provided key or
% or just after if the key doesn't exist
-% 2. end_key - Stop iteration prior to visiting the provided
+% 2. end_key - Stop iteration just after the provided key
+% 3. end_key_gt - Stop iteration prior to visiting the provided
% key
-% 3. end_key_gt - Stop iteration just after the provided key
% 4. dir - The atom fwd or rev. This is to be able to iterate
% over documents in reverse order. The logic for comparing
% start_key, end_key, and end_key_gt are then reversed (ie,
@@ -492,12 +492,12 @@
% This function is called to fold over the documents (not local
% documents) in order of their most recent update. Each document
% in the database should have exactly one entry in this sequence.
-% If a document is updated during a call to this funciton it should
+% If a document is updated during a call to this function it should
% not be included twice as that will probably lead to Very Bad Things.
%
% This should behave similarly to fold_docs/4 in that the supplied
% user function should be invoked with a #full_doc_info{} record
-% as the first arugment and the current user accumulator as the
+% as the first argument and the current user accumulator as the
% second argument. The same semantics for the return value from the
% user function should be handled as in fold_docs/4.
%
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 04/20: WIP - couch_db.erl
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 8dae2f081d20126bf1a208e83aa3a51729dc3ec3
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Wed Mar 14 13:18:35 2018 -0500
WIP - couch_db.erl
---
src/couch/src/couch_db.erl | 51 +++++++++++++++++++++++++++++++++++++++++-----
1 file changed, 46 insertions(+), 5 deletions(-)
diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index 93ea07e..e27f632 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -43,7 +43,6 @@
get_epochs/1,
get_filepath/1,
get_instance_start_time/1,
- get_last_purged/1,
get_pid/1,
get_revs_limit/1,
get_security/1,
@@ -51,12 +50,14 @@
get_user_ctx/1,
get_uuid/1,
get_purge_seq/1,
+ get_purge_infos_limit/1,
is_db/1,
is_system_db/1,
is_clustered/1,
set_revs_limit/2,
+ set_purge_infos_limit/2,
set_security/2,
set_user_ctx/2,
@@ -75,6 +76,7 @@
get_full_doc_infos/2,
get_missing_revs/2,
get_design_docs/1,
+ load_purge_infos/2,
update_doc/3,
update_doc/4,
@@ -84,6 +86,7 @@
delete_doc/3,
purge_docs/2,
+ purge_docs/3,
with_stream/3,
open_write_stream/2,
@@ -97,6 +100,8 @@
fold_changes/4,
fold_changes/5,
count_changes_since/2,
+ fold_purge_infos/4,
+ fold_purge_infos/5,
calculate_start_seq/3,
owner_of/2,
@@ -366,8 +371,31 @@ get_full_doc_info(Db, Id) ->
get_full_doc_infos(Db, Ids) ->
couch_db_engine:open_docs(Db, Ids).
-purge_docs(#db{main_pid=Pid}, IdsRevs) ->
- gen_server:call(Pid, {purge_docs, IdsRevs}).
+
+-spec purge_docs(#db{}, [{UUId, Id, [Rev]}]) ->
+ {ok, [Reply]} when
+ UUId :: binary(),
+ Id :: binary(),
+ Rev :: {non_neg_integer(), binary()},
+ Reply :: {ok, []} | {ok, [Rev]}.
+purge_docs(#db{main_pid = Pid}, UUIdsIdsRevs) ->
+ gen_server:call(Pid, {purge_docs, UUIdsIdsRevs});
+
+-spec load_purge_infos(#db{}, [UUId]) -> [PurgeInfo] when
+ UUId :: binary(),
+ PurgeInfo :: {PurgeSeq, UUId, Id, [Rev]} | not_found,
+ Id :: binary(),
+ Rev :: {non_neg_integer(), binary()}.
+load_purge_infos(Db, UUIDs) ->
+ couch_db_engine:load_purge_infos(Db, UUIDs).
+
+
+set_purge_infos_limit(#db{main_pid=Pid}=Db, Limit) when Limit > 0 ->
+ check_is_admin(Db),
+ gen_server:call(Pid, {set_purge_infos_limit, Limit}, infinity);
+set_purge_infos_limit(_Db, _Limit) ->
+ throw(invalid_purge_infos_limit).
+
get_after_doc_read_fun(#db{after_doc_read = Fun}) ->
Fun.
@@ -389,8 +417,13 @@ get_user_ctx(?OLD_DB_REC = Db) ->
get_purge_seq(#db{}=Db) ->
{ok, couch_db_engine:get_purge_seq(Db)}.
-get_last_purged(#db{}=Db) ->
- {ok, couch_db_engine:get_last_purged(Db)}.
+get_oldest_purge_seq(#db{}=Db) ->
+ {ok, StartSeq} = get_purge_seq(Db),
+ FoldFun = fun({_UUId, PurgeSeq, _, _}, _) -> {stop, PurgeSeq} end,
+ fold_purge_infos(Db, StartSeq, FoldFun, StartSeq).
+
+get_purge_infos_limit(#db{}=Db) ->
+ couch_db_engine:get_purge_infos_limit(Db).
get_pid(#db{main_pid = Pid}) ->
Pid.
@@ -1403,6 +1436,14 @@ fold_changes(Db, StartSeq, UserFun, UserAcc, Opts) ->
couch_db_engine:fold_changes(Db, StartSeq, UserFun, UserAcc, Opts).
+fold_purge_infos(Db, StartPurgeSeq, Fun, Acc) ->
+ fold_purge_infos(Db, StartPurgeSeq, Fun, Acc, []).
+
+
+fold_purge_infos(Db, StartPurgeSeq, UFun, UAcc, Opts) ->
+ couch_db_engine:fold_purge_infos(Db, StartPurgeSeq, UFun, UAcc, Opts).
+
+
count_changes_since(Db, SinceSeq) ->
couch_db_engine:count_changes_since(Db, SinceSeq).
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 02/20: [WIP] - Declare new purge storage engine APIs
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 1b9ed8fe58fefdf00164ab971805632b0f9ac431
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Wed Mar 14 12:19:37 2018 -0500
[WIP] - Declare new purge storage engine APIs
---
src/couch/src/couch_db_engine.erl | 148 +++++++++++++++++++++++++++++---------
1 file changed, 116 insertions(+), 32 deletions(-)
diff --git a/src/couch/src/couch_db_engine.erl b/src/couch/src/couch_db_engine.erl
index 4974201..0e7a1cf 100644
--- a/src/couch/src/couch_db_engine.erl
+++ b/src/couch/src/couch_db_engine.erl
@@ -22,6 +22,8 @@
-type rev() :: {non_neg_integer(), binary()}.
-type revs() :: [rev()].
-type json() :: any().
+-type uuid() :: binary().
+-type purge_seq() :: non_neg_integer().
-type doc_pair() :: {
#full_doc_info{} | not_found,
@@ -39,7 +41,7 @@
sync
].
--type purge_info() :: [{docid(), revs()}].
+-type purge_info() :: {purge_seq(), uuid(), docid(), revs()}.
-type epochs() :: [{Node::atom(), UpdateSeq::non_neg_integer()}].
-type size_info() :: [{Name::atom(), Size::non_neg_integer()}].
@@ -62,6 +64,10 @@
{dir, fwd | rev}
].
+-type purge_fold_options() :: [
+ % Need to enumerate these
+].
+
-type db_handle() :: any().
-type doc_fold_fun() :: fun((#full_doc_info{}, UserAcc::any()) ->
@@ -76,6 +82,10 @@
{ok, NewUserAcc::any()} |
{stop, NewUserAcc::any()}).
+-type purge_fold_fun() :: fun((purge_info(), UserAcc::any()) ->
+ {ok, NewUserAcc::any()} |
+ {stop, NewUserAcc::any()}).
+
% This is called by couch_server to determine which
% engine should be used for the given database. DbPath
@@ -206,13 +216,14 @@
-callback get_epochs(DbHandle::db_handle()) -> Epochs::epochs().
-% Get the last purge request performed.
--callback get_last_purged(DbHandle::db_handle()) -> LastPurged::purge_info().
-
-
% Get the current purge sequence. This should be incremented
% for every purge operation.
--callback get_purge_seq(DbHandle::db_handle()) -> PurgeSeq::non_neg_integer().
+-callback get_purge_seq(DbHandle::db_handle()) -> purge_seq().
+
+
+% Get the purged infos limit. This should just return the last
+% value that was passed to set_purged_docs_limit/2.
+-callback get_purge_infos_limit(DbHandle::db_handle()) -> pos_integer().
% Get the revision limit. This should just return the last
@@ -261,6 +272,11 @@
-callback set_revs_limit(DbHandle::db_handle(), RevsLimit::pos_integer()) ->
{ok, NewDbHandle::db_handle()}.
+
+-callback set_purge_infos_limit(DbHandle::db_handle(), Limit::pos_integer()) ->
+ {ok, NewDbHandle::db_handle()}.
+
+
-callback set_security(DbHandle::db_handle(), SecProps::any()) ->
{ok, NewDbHandle::db_handle()}.
@@ -301,6 +317,15 @@
doc().
+% This function will be called from many contexts concurrently.
+% If the storage engine has a purge_info() record for any of the
+% provided UUIDs, those purge_info() records should be returned. The
+% resulting list should have the same length as the input list of
+% UUIDs.
+-callback load_purge_infos(DbHandle::db_handle(), [uuid()]) ->
+ [purge_info() | not_found].
+
+
% This function is called concurrently by any client process
% that is writing a document. It should accept a #doc{}
% record and return a #doc{} record with a mutated body it
@@ -341,31 +366,20 @@
% #full_doc_info{} records. The first element of the pair is
% the #full_doc_info{} that exists on disk. The second element
% is the new version that should be written to disk. There are
-% three basic cases that should be followed:
+% two basic cases that should be followed:
%
% 1. {not_found, #full_doc_info{}} - A new document was created
% 2. {#full_doc_info{}, #full_doc_info{}} - A document was updated
-% 3. {#full_doc_info{}, not_found} - A document was purged completely
%
-% Number one and two are fairly straight forward as long as proper
-% accounting for moving entries in the udpate sequence are accounted
-% for. However, case 3 you'll notice is "purged completely" which
-% means it needs to be removed from the database including the
-% update sequence. Also, for engines that are not using append
-% only storage like the legacy engine, case 2 can be the result of
-% a purge so special care will be needed to see which revisions
-% should be removed.
+% The cases are fairly straight forward as long as proper
+% accounting for moving entries in the update sequence are accounted
+% for.
%
% The LocalDocs variable is applied separately. Its important to
% note for new storage engine authors that these documents are
% separate because they should *not* be included as part of the
% changes index for the database.
%
-% The PurgedDocIdRevs is the list of Ids and Revisions that were
-% purged during this update. While its not guaranteed by the API,
-% currently there will never be purge changes comingled with
-% standard updates.
-%
% Traditionally an invocation of write_doc_infos should be all
% or nothing in so much that if an error occurs (or the VM dies)
% then the database doesn't retain any of the changes. However
@@ -376,8 +390,36 @@
-callback write_doc_infos(
DbHandle::db_handle(),
Pairs::doc_pairs(),
- LocalDocs::[#doc{}],
- PurgedDocIdRevs::[{docid(), revs()}]) ->
+ LocalDocs::[#doc{}]) ->
+ {ok, NewDbHandle::db_handle()}.
+
+
+% This function is called from the context of couch_db_updater
+% and as such is guaranteed single threaded for the given
+% DbHandle.
+%
+% The DocPair argument is a 2-tuple of #full_doc_info{} records. The
+% first element of th epair is the #full_doc_info{} that exists
+% on disk. The second element is the new version that should be
+% written to disk. There are two basic cases that should be considered:
+%
+% 1. {#full_doc_info{}, #full_doc_info{}} - A document was partially purged
+% 2. {#full_doc_info{}, not_found} - A document was completely purged
+%
+% In case 1, non-tail-append engines may have to remove revisions
+% specifically rather than rely on compaction to remove them. Also
+% note that the new #full_doc_info{} will have a different update_seq
+% that will need to be reflected in the changes feed.
+%
+% In case 2 you'll notice is "purged completely" which
+% means it needs to be removed from the database including the
+% update sequence.
+%
+% The PurgeInfo contains the purge_seq, uuid, docid and revisions that
+% were requested to be purged. This should be persisted in such a way
+% that we can efficiently load purge_info() by its UUID as well as
+% iterate over purge_info() entries in order of their PurgeSeq.
+-callback purge_doc(DbHandle::db_handle(), doc_pair(), purge_info()) ->
{ok, NewDbHandle::db_handle()}.
@@ -518,6 +560,21 @@
% This function may be called by many processes concurrently.
%
+% This function is called to fold over purged requests in order of
+% their oldest purge (increasing purge_seq order)
+%
+% The StartPurgeSeq parameter indicates where the fold should start *after*.
+-callback fold_purge_infos(
+ DbHandle::db_handle(),
+ StartPurgeSeq::purge_seq(),
+ UserFold::purge_fold_fun(),
+ UserAcc::any(),
+ purge_fold_options()) ->
+ {ok, LastUserAcc::any()}.
+
+
+% This function may be called by many processes concurrently.
+%
% This function is called to count the number of documents changed
% since the given UpdateSeq (ie, not including the possible change
% at exactly UpdateSeq). It is currently only used internally to
@@ -597,8 +654,8 @@
get_disk_version/1,
get_doc_count/1,
get_epochs/1,
- get_last_purged/1,
get_purge_seq/1,
+ get_purge_infos_limit/1,
get_revs_limit/1,
get_security/1,
get_size_info/1,
@@ -607,14 +664,17 @@
set_revs_limit/2,
set_security/2,
+ set_purge_docs_limit/2,
open_docs/2,
open_local_docs/2,
read_doc_body/2,
+ load_purge_infos/2,
serialize_doc/2,
write_doc_body/2,
- write_doc_infos/4,
+ write_doc_infos/3,
+ purge_doc_revs/3,
commit_data/1,
open_write_stream/2,
@@ -624,6 +684,7 @@
fold_docs/4,
fold_local_docs/4,
fold_changes/5,
+ fold_purge_infos/5,
count_changes_since/2,
start_compaction/1,
@@ -737,14 +798,14 @@ get_epochs(#db{} = Db) ->
Engine:get_epochs(EngineState).
-get_last_purged(#db{} = Db) ->
+get_purge_seq(#db{} = Db) ->
#db{engine = {Engine, EngineState}} = Db,
- Engine:get_last_purged(EngineState).
+ Engine:get_purge_seq(EngineState).
-get_purge_seq(#db{} = Db) ->
+get_purge_infos_limit(#db{} = Db) ->
#db{engine = {Engine, EngineState}} = Db,
- Engine:get_purge_seq(EngineState).
+ Engine:get_purge_infos_limit(EngineState).
get_revs_limit(#db{} = Db) ->
@@ -777,6 +838,12 @@ set_revs_limit(#db{} = Db, RevsLimit) ->
{ok, Db#db{engine = {Engine, NewSt}}}.
+set_purge_infos_limit(#db{} = Db, PurgedDocsLimit) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ {ok, NewSt} = Engine:set_purge_infos_limit(EngineState, PurgedDocsLimit),
+ {ok, Db#db{engine = {Engine, NewSt}}}.
+
+
set_security(#db{} = Db, SecProps) ->
#db{engine = {Engine, EngineState}} = Db,
{ok, NewSt} = Engine:set_security(EngineState, SecProps),
@@ -798,6 +865,11 @@ read_doc_body(#db{} = Db, RawDoc) ->
Engine:read_doc_body(EngineState, RawDoc).
+load_purge_infos(#db{} = Db, UUIDs) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ Engine:open_purged_docs(EngineState, UUIDs).
+
+
serialize_doc(#db{} = Db, #doc{} = Doc) ->
#db{engine = {Engine, EngineState}} = Db,
Engine:serialize_doc(EngineState, Doc).
@@ -808,10 +880,16 @@ write_doc_body(#db{} = Db, #doc{} = Doc) ->
Engine:write_doc_body(EngineState, Doc).
-write_doc_infos(#db{} = Db, DocUpdates, LocalDocs, PurgedDocIdRevs) ->
+write_doc_infos(#db{} = Db, DocUpdates, LocalDocs) ->
#db{engine = {Engine, EngineState}} = Db,
- {ok, NewSt} = Engine:write_doc_infos(
- EngineState, DocUpdates, LocalDocs, PurgedDocIdRevs),
+ {ok, NewSt} = Engine:write_doc_infos(EngineState, DocUpdates, LocalDocs),
+ {ok, Db#db{engine = {Engine, NewSt}}}.
+
+
+purge_doc_revs(#db{} = Db, DocUpdates, Purges) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ {ok, NewSt} = Engine:purge_doc_revs(
+ EngineState, DocUpdates, Purges),
{ok, Db#db{engine = {Engine, NewSt}}}.
@@ -851,6 +929,12 @@ fold_changes(#db{} = Db, StartSeq, UserFun, UserAcc, Options) ->
Engine:fold_changes(EngineState, StartSeq, UserFun, UserAcc, Options).
+fold_purge_infos(#db{} = Db, StartPurgeSeq, UserFun, UserAcc, Options) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ Engine:fold_purge_infos(
+ EngineState, StartPurgeSeq, UserFun, UserAcc, Options).
+
+
count_changes_since(#db{} = Db, StartSeq) ->
#db{engine = {Engine, EngineState}} = Db,
Engine:count_changes_since(EngineState, StartSeq).
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 09/20: WIP - couch_db_engine.erl
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 38d0bb0f46917de9e8cdd6c6c49737fa1f0d9dc4
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Fri Mar 16 14:56:16 2018 -0500
WIP - couch_db_engine.erl
---
src/couch/src/couch_db_engine.erl | 14 ++++++++++++--
1 file changed, 12 insertions(+), 2 deletions(-)
diff --git a/src/couch/src/couch_db_engine.erl b/src/couch/src/couch_db_engine.erl
index 50682b2..b19514b 100644
--- a/src/couch/src/couch_db_engine.erl
+++ b/src/couch/src/couch_db_engine.erl
@@ -216,11 +216,15 @@
-callback get_epochs(DbHandle::db_handle()) -> Epochs::epochs().
-% Get the current purge sequence. This should be incremented
-% for every purge operation.
+% Get the current purge sequence known to the engine. This
+% value should be updated during calls to purge_docs.
-callback get_purge_seq(DbHandle::db_handle()) -> purge_seq().
+% Get the oldest purge sequence known to the engine
+-callback get_oldest_purge_seq(DbHandle::db_handle()) -> purge_seq().
+
+
% Get the purged infos limit. This should just return the last
% value that was passed to set_purged_docs_limit/2.
-callback get_purge_infos_limit(DbHandle::db_handle()) -> pos_integer().
@@ -659,6 +663,7 @@
get_doc_count/1,
get_epochs/1,
get_purge_seq/1,
+ get_oldest_purge_seq/1,
get_purge_infos_limit/1,
get_revs_limit/1,
get_security/1,
@@ -807,6 +812,11 @@ get_purge_seq(#db{} = Db) ->
Engine:get_purge_seq(EngineState).
+get_oldest_purge_seq(#db{} = Db) ->
+ #db{engine = {Engine, EngineState}} = Db,
+ Engine:get_oldest_purge_seq(EngineState).
+
+
get_purge_infos_limit(#db{} = Db) ->
#db{engine = {Engine, EngineState}} = Db,
Engine:get_purge_infos_limit(EngineState).
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 13/20: Use EPI to create local purge doc for indexers
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit f8263882962745a2043d26c9b92ff6d4b8e3a93e
Author: jiangphcn <ji...@cn.ibm.com>
AuthorDate: Fri Mar 9 15:29:49 2018 +0800
Use EPI to create local purge doc for indexers
- Before performing compaction, local purge document needs to be
created so that compaction can detect whether old purge requests
can be compacted. Otherwise, if some old purge requests are
compacted and some of indexers didn't process purge requests,
it will cause index reset.
- use EPI approach to create local purge documents (if not) before
compacting purge requests. To do so, all local purge documents
against valid design documents can be created and used by
compactor to decide whether to compact old purge requests.
- In ASF stream, the extensible plugin interface is implemented
for secondary index (mrview index).
COUCHDB-3226
---
src/couch/src/couch_bt_engine_compactor.erl | 18 ++++
src/couch/src/couch_db_plugin.erl | 8 +-
src/couch_index/src/couch_index_epi.erl | 5 +-
...dex_epi.erl => couch_index_plugin_couch_db.erl} | 39 ++-------
src/couch_mrview/src/couch_mrview_index.erl | 95 +++++++++++++++++-----
.../test/couch_mrview_purge_docs_fabric_tests.erl | 76 ++++++++++++++++-
6 files changed, 186 insertions(+), 55 deletions(-)
diff --git a/src/couch/src/couch_bt_engine_compactor.erl b/src/couch/src/couch_bt_engine_compactor.erl
index 7b11b44..f6e79b4 100644
--- a/src/couch/src/couch_bt_engine_compactor.erl
+++ b/src/couch/src/couch_bt_engine_compactor.erl
@@ -44,6 +44,12 @@ start(#st{} = St, DbName, Options, Parent) ->
} = St,
couch_log:debug("Compaction process spawned for db \"~s\"", [DbName]),
+ {ok, DDocs} = design_docs(DbName),
+ lists:map(fun(DDoc) ->
+ JsonDDoc = couch_doc:from_json_obj(DDoc),
+ couch_db_plugin:maybe_init_index_purge_state(DbName, JsonDDoc)
+ end, DDocs),
+
{ok, NewSt, DName, DFd, MFd, Retry} =
open_compaction_files(Header, FilePath, Options),
erlang:monitor(process, MFd),
@@ -569,3 +575,15 @@ update_compact_task(NumChanges) ->
end,
couch_task_status:update([{changes_done, Changes2}, {progress, Progress}]).
+
+design_docs(DbName) ->
+ try
+ case fabric:design_docs(mem3:dbname(DbName)) of
+ {error, {maintenance_mode, _, _Node}} ->
+ {ok, []};
+ Else ->
+ Else
+ end
+ catch error:database_does_not_exist ->
+ {ok, []}
+ end.
diff --git a/src/couch/src/couch_db_plugin.erl b/src/couch/src/couch_db_plugin.erl
index 740b812..0107403 100644
--- a/src/couch/src/couch_db_plugin.erl
+++ b/src/couch/src/couch_db_plugin.erl
@@ -18,7 +18,8 @@
after_doc_read/2,
validate_docid/1,
check_is_admin/1,
- on_delete/2
+ on_delete/2,
+ maybe_init_index_purge_state/2
]).
-define(SERVICE_ID, couch_db).
@@ -60,6 +61,11 @@ on_delete(DbName, Options) ->
Handle = couch_epi:get_handle(?SERVICE_ID),
couch_epi:apply(Handle, ?SERVICE_ID, on_delete, [DbName, Options], []).
+maybe_init_index_purge_state(DbName, DDoc) ->
+ Handle = couch_epi:get_handle(?SERVICE_ID),
+ couch_epi:apply(Handle, ?SERVICE_ID, maybe_init_index_purge_state,
+ [DbName, DDoc], []).
+
%% ------------------------------------------------------------------
%% Internal Function Definitions
%% ------------------------------------------------------------------
diff --git a/src/couch_index/src/couch_index_epi.erl b/src/couch_index/src/couch_index_epi.erl
index 946a590..1c4eb95 100644
--- a/src/couch_index/src/couch_index_epi.erl
+++ b/src/couch_index/src/couch_index_epi.erl
@@ -28,8 +28,9 @@ app() ->
couch_index.
providers() ->
- [].
-
+ [
+ {couch_db, couch_index_plugin_couch_db}
+ ].
services() ->
[
diff --git a/src/couch_index/src/couch_index_epi.erl b/src/couch_index/src/couch_index_plugin_couch_db.erl
similarity index 52%
copy from src/couch_index/src/couch_index_epi.erl
copy to src/couch_index/src/couch_index_plugin_couch_db.erl
index 946a590..3e3b711 100644
--- a/src/couch_index/src/couch_index_epi.erl
+++ b/src/couch_index/src/couch_index_plugin_couch_db.erl
@@ -2,7 +2,7 @@
% 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
+% 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
@@ -10,40 +10,15 @@
% License for the specific language governing permissions and limitations under
% the License.
--module(couch_index_epi).
-
--behaviour(couch_epi_plugin).
+-module(couch_index_plugin_couch_db).
-export([
- app/0,
- providers/0,
- services/0,
- data_subscriptions/0,
- data_providers/0,
- processes/0,
- notify/3
+ maybe_init_index_purge_state/2
]).
-app() ->
- couch_index.
-
-providers() ->
- [].
-
-
-services() ->
- [
- {couch_index, couch_index_plugin}
- ].
-
-data_subscriptions() ->
- [].
-
-data_providers() ->
- [].
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
-processes() ->
- [].
-notify(_Key, _Old, _New) ->
- ok.
+maybe_init_index_purge_state(DbName, DDoc) ->
+ couch_mrview_index:maybe_create_local_purge_doc(DbName, DDoc).
diff --git a/src/couch_mrview/src/couch_mrview_index.erl b/src/couch_mrview/src/couch_mrview_index.erl
index 6ab0812..6c63eaf 100644
--- a/src/couch_mrview/src/couch_mrview_index.erl
+++ b/src/couch_mrview/src/couch_mrview_index.erl
@@ -19,6 +19,7 @@
-export([compact/3, swap_compacted/2, remove_compacted/1]).
-export([index_file_exists/1]).
-export([update_local_purge_doc/2, verify_index_exists/1]).
+-export([maybe_create_local_purge_doc/2]).
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
@@ -213,24 +214,6 @@ index_file_exists(State) ->
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">>, 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">>,
@@ -274,7 +257,7 @@ verify_index_exists(Options) ->
end.
-maybe_create_local_purge_doc(Db, State) ->
+maybe_create_local_purge_doc(Db, #mrst{}=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
@@ -282,4 +265,78 @@ maybe_create_local_purge_doc(Db, State) ->
update_local_purge_doc(Db, State);
{ok, _LocalPurgeDoc} ->
ok
+ end;
+maybe_create_local_purge_doc(DbName, #doc{}=DDoc) ->
+ {ok, Db} = case couch_db:open_int(DbName, []) of
+ {ok, _} = Resp -> Resp;
+ Else -> exit(Else)
+ end,
+ try
+ {ok, DefaultPurgeSeq} = couch_db:get_purge_seq(Db),
+ case get_index_type(DDoc, <<"views">>) of true ->
+ try couch_mrview_util:ddoc_to_mrst(DbName, DDoc) of
+ {ok, MRSt} -> update_local_purge_doc(Db, MRSt, DefaultPurgeSeq)
+ catch _:_ ->
+ ok
+ end;
+ false ->
+ ok
+ end
+ catch E:T ->
+ Stack = erlang:get_stacktrace(),
+ couch_log:error(
+ "Error occurs when creating local purge document ~p ~p",
+ [{E, T}, Stack]
+ )
+ after
+ catch couch_db:close(Db)
+ end.
+
+
+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">>, 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, []).
+
+
+update_local_purge_doc(Db, State, PSeq) ->
+ Sig = couch_index_util:hexsig(State#mrst.sig),
+ DocId = couch_mrview_util:get_local_purge_doc_id(Sig),
+ case couch_db:open_doc(Db, DocId, []) of
+ {not_found, _Reason} ->
+ Doc = couch_doc:from_json_obj({[
+ {<<"_id">>, DocId},
+ {<<"purge_seq">>, PSeq},
+ {<<"timestamp_utc">>, couch_util:utc_string()},
+ {<<"verify_module">>, <<"couch_mrview_index">>},
+ {<<"verify_function">>, <<"verify_index_exists">>},
+ {<<"verify_options">>, {[
+ {<<"dbname">>, State#mrst.db_name},
+ {<<"ddoc_id">>, State#mrst.idx_name},
+ {<<"signature">>, Sig}
+ ]}},
+ {<<"type">>, <<"mrview">>}
+ ]}),
+ couch_db:update_doc(Db, Doc, []);
+ {ok, _LocalPurgeDoc} ->
+ ok
+ end.
+
+
+get_index_type(#doc{body={Props}}, IndexType) ->
+ case couch_util:get_value(IndexType, Props) of
+ undefined -> false;
+ _ -> true
end.
diff --git a/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl b/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl
index 0fa2a7c..b36cad5 100644
--- a/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl
+++ b/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl
@@ -14,6 +14,7 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
+-include_lib("mem3/include/mem3.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
-define(TIMEOUT, 1000).
@@ -22,10 +23,15 @@
setup() ->
DbName = ?tempdb(),
ok = fabric:create_db(DbName, [?ADMIN_CTX]),
+ meck:new(couch_mrview_index, [passthrough]),
+ meck:expect(couch_mrview_index, maybe_create_local_purge_doc, fun(A, B) ->
+ meck:passthrough([A, B])
+ end),
DbName.
teardown(DbName) ->
+ meck:unload(couch_mrview_index),
ok = fabric:delete_db(DbName, [?ADMIN_CTX]).
@@ -40,7 +46,8 @@ view_purge_fabric_test_() ->
foreach,
fun setup/0, fun teardown/1,
[
- fun test_purge_verify_index/1
+ fun test_purge_verify_index/1,
+ fun test_purge_hook_before_compaction/1
]
}
}
@@ -82,6 +89,40 @@ test_purge_verify_index(DbName) ->
ok
end).
+test_purge_hook_before_compaction(DbName) ->
+ ?_test(begin
+ Docs1 = couch_mrview_test_util:make_docs(normal, 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">>]),
+
+ Result1 = fabric:query_view(DbName, <<"bar">>, <<"baz">>, #mrargs{}),
+ Expect1 = {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(Expect1, Result1),
+
+ [FirstShDbName | _RestShDbNames] = local_shards(DbName),
+ {ok, Db} = couch_db:open_int(FirstShDbName, []),
+ {ok, _CompactPid} = couch_db:start_compact(Db),
+ wait_compaction(FirstShDbName, "database", ?LINE),
+ ok = couch_db:close(Db),
+
+ ?assertEqual(ok, meck:wait(1, couch_mrview_index,
+ maybe_create_local_purge_doc, '_', 5000)
+ ),
+ ok
+ end).
+
get_rev(#full_doc_info{} = FDI) ->
#doc_info{
revs = [#rev_info{} = PrevRev | _]
@@ -95,3 +136,36 @@ purge_docs(DbName, DocIds) ->
Rev = get_rev(FDI),
{ok, [{ok, _}]} = fabric:purge_docs(DbName, [{DocId, [Rev]}], [])
end, DocIds).
+
+wait_compaction(DbName, Kind, Line) ->
+ WaitFun = fun() ->
+ case is_compaction_running(DbName) of
+ true -> wait;
+ false -> ok
+ end
+ end,
+ case test_util:wait(WaitFun, 10000) of
+ timeout ->
+ erlang:error({assertion_failed,
+ [{module, ?MODULE},
+ {line, Line},
+ {reason, "Timeout waiting for "
+ ++ Kind
+ ++ " database compaction"}]});
+ _ ->
+ ok
+ end.
+
+is_compaction_running(DbName) ->
+ {ok, Db} = couch_db:open_int(DbName, []),
+ {ok, DbInfo} = couch_db:get_db_info(Db),
+ couch_db:close(Db),
+ couch_util:get_value(compact_running, DbInfo).
+
+local_shards(DbName) ->
+ try
+ [ShardName || #shard{name = ShardName} <- mem3:local_shards(DbName)]
+ catch
+ error:database_does_not_exist ->
+ []
+ end.
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 12/20: Update view engine to use new purge API
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit dc370c03193da9d96a4fef9b20c3c1c41dc7bcd5
Author: jiangphcn <ji...@cn.ibm.com>
AuthorDate: Tue May 2 11:32:48 2017 +0800
Update view engine to use new purge API
COUCHDB-3326
---
src/couch_index/src/couch_index_updater.erl | 24 +-
src/couch_mrview/src/couch_mrview_cleanup.erl | 18 +-
src/couch_mrview/src/couch_mrview_index.erl | 76 ++++++
src/couch_mrview/src/couch_mrview_test_util.erl | 5 +
src/couch_mrview/src/couch_mrview_util.erl | 23 ++
.../test/couch_mrview_purge_docs_fabric_tests.erl | 97 ++++++++
.../test/couch_mrview_purge_docs_tests.erl | 274 +++++++++++++++++++++
7 files changed, 507 insertions(+), 10 deletions(-)
diff --git a/src/couch_index/src/couch_index_updater.erl b/src/couch_index/src/couch_index_updater.erl
index 5ab9ea8..b416d17 100644
--- a/src/couch_index/src/couch_index_updater.erl
+++ b/src/couch_index/src/couch_index_updater.erl
@@ -141,10 +141,7 @@ update(Idx, Mod, IdxState) ->
DbUpdateSeq = couch_db:get_update_seq(Db),
DbCommittedSeq = couch_db:get_committed_update_seq(Db),
- PurgedIdxState = case purge_index(Db, Mod, IdxState) of
- {ok, IdxState0} -> IdxState0;
- reset -> exit({reset, self()})
- end,
+ {ok, PurgedIdxState} = purge_index(Db, Mod, IdxState),
NumChanges = couch_db:count_changes_since(Db, CurrSeq),
@@ -209,11 +206,20 @@ purge_index(Db, Mod, IdxState) ->
{ok, DbPurgeSeq} = couch_db:get_purge_seq(Db),
IdxPurgeSeq = Mod:get(purge_seq, IdxState),
if
- DbPurgeSeq == IdxPurgeSeq ->
+ IdxPurgeSeq == DbPurgeSeq ->
{ok, IdxState};
- DbPurgeSeq == IdxPurgeSeq + 1 ->
- {ok, PurgedIdRevs} = couch_db:get_last_purged(Db),
- Mod:purge(Db, DbPurgeSeq, PurgedIdRevs, IdxState);
true ->
- reset
+ FoldFun = fun({PurgeSeq, _UUId, Id, Revs}, Acc) ->
+ {ok, StateAcc} = Mod:purge(Db, PurgeSeq, [{Id, Revs}], Acc),
+ StateAcc
+ end,
+ {ok, NewStateAcc} = couch_db:fold_purged_docs(
+ Db,
+ IdxPurgeSeq,
+ FoldFun,
+ IdxState,
+ []
+ ),
+ Mod:update_local_purge_doc(Db, NewStateAcc),
+ {ok, NewStateAcc}
end.
diff --git a/src/couch_mrview/src/couch_mrview_cleanup.erl b/src/couch_mrview/src/couch_mrview_cleanup.erl
index 380376d..93c9387 100644
--- a/src/couch_mrview/src/couch_mrview_cleanup.erl
+++ b/src/couch_mrview/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.
diff --git a/src/couch_mrview/src/couch_mrview_index.erl b/src/couch_mrview/src/couch_mrview_index.erl
index aa1ee27..6ab0812 100644
--- a/src/couch_mrview/src/couch_mrview_index.erl
+++ b/src/couch_mrview/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, remove_compacted/1]).
-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").
@@ -121,14 +122,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 ->
@@ -207,3 +211,75 @@ 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">>, 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(couch_db:name(Db)),
+ case ddoc_cache:open(DbName, DDocId) of
+ {ok, DDoc} ->
+ {ok, IdxState} = couch_mrview_util:ddoc_to_mrst(
+ ShardDbName,
+ DDoc
+ ),
+ IdxSig = IdxState#mrst.sig,
+ couch_index_util:hexsig(IdxSig) == 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.
diff --git a/src/couch_mrview/src/couch_mrview_test_util.erl b/src/couch_mrview/src/couch_mrview_test_util.erl
index b07b076..35ab6c6 100644
--- a/src/couch_mrview/src/couch_mrview_test_util.erl
+++ b/src/couch_mrview/src/couch_mrview_test_util.erl
@@ -49,6 +49,11 @@ make_docs(local, Count) ->
make_docs(_, Count) ->
[doc(I) || I <- lists:seq(1, Count)].
+
+make_docs(_, Since, Count) ->
+ [doc(I) || I <- lists:seq(Since, Count)].
+
+
ddoc({changes, Opts}) ->
ViewOpts = case Opts of
seq_indexed ->
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index 0c6e5fc..710cb28 100644
--- a/src/couch_mrview/src/couch_mrview_util.erl
+++ b/src/couch_mrview/src/couch_mrview_util.erl
@@ -13,6 +13,8 @@
-module(couch_mrview_util).
-export([get_view/4, get_view_index_pid/4]).
+-export([get_local_purge_doc_id/1, get_value_from_options/2]).
+-export([get_signature_from_filename/1]).
-export([ddoc_to_mrst/2, init_state/4, reset_index/3]).
-export([make_header/1]).
-export([index_file/2, compaction_file/2, open_file/1]).
@@ -41,6 +43,27 @@
-include_lib("couch_mrview/include/couch_mrview.hrl").
+get_local_purge_doc_id(Sig) ->
+ Version = "v" ++ config:get("purge", "version", "1") ++ "-",
+ ?l2b(?LOCAL_DOC_PREFIX ++ "purge-" ++ Version ++ "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) ->
case get_view_index_state(Db, DDoc, ViewName, Args0) of
{ok, State, Args2} ->
diff --git a/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl b/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl
new file mode 100644
index 0000000..0fa2a7c
--- /dev/null
+++ b/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl
@@ -0,0 +1,97 @@
+% 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(normal, 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).
diff --git a/src/couch_mrview/test/couch_mrview_purge_docs_tests.erl b/src/couch_mrview/test/couch_mrview_purge_docs_tests.erl
new file mode 100644
index 0000000..5ab4f21
--- /dev/null
+++ b/src/couch_mrview/test/couch_mrview_purge_docs_tests.erl
@@ -0,0 +1,274 @@
+% 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_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() ->
+ {ok, Db} = couch_mrview_test_util:init_db(?tempdb(), map, 5),
+ Db.
+
+teardown(Db) ->
+ couch_db:close(Db),
+ couch_server:delete(couch_db:name(Db), [?ADMIN_CTX]),
+ ok.
+
+view_purge_test_() ->
+ {
+ "Map views",
+ {
+ setup,
+ fun test_util:start_couch/0, fun test_util:stop_couch/1,
+ {
+ foreach,
+ fun setup/0, fun teardown/1,
+ [
+ fun test_purge_single/1,
+ fun test_purge_multiple/1,
+ fun test_purge_with_compact1/1,
+ fun test_purge_with_compact2/1
+ ]
+ }
+ }
+ }.
+
+
+test_purge_single(Db) ->
+ ?_test(begin
+ Result = run_query(Db, []),
+ Expect = {ok, [
+ {meta, [{total, 5}, {offset, 0}]},
+ {row, [{id, <<"1">>}, {key, 1}, {value, 1}]},
+ {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(Expect, Result),
+
+ FDI = couch_db:get_full_doc_info(Db, <<"1">>),
+ Rev = get_rev(FDI),
+ {ok, [{ok, _PRevs}]} = couch_db:purge_docs(
+ Db,
+ [{<<"UUID1">>, <<"1">>, [Rev]}]
+ ),
+ {ok, Db2} = couch_db:reopen(Db),
+
+ Result2 = run_query(Db2, []),
+ 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
+ end).
+
+
+test_purge_multiple(Db) ->
+ ?_test(begin
+ Result = run_query(Db, []),
+ Expect = {ok, [
+ {meta, [{total, 5}, {offset, 0}]},
+ {row, [{id, <<"1">>}, {key, 1}, {value, 1}]},
+ {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(Expect, Result),
+
+ FDI1 = couch_db:get_full_doc_info(Db, <<"1">>), Rev1 = get_rev(FDI1),
+ FDI2 = couch_db:get_full_doc_info(Db, <<"2">>), Rev2 = get_rev(FDI2),
+ FDI5 = couch_db:get_full_doc_info(Db, <<"5">>), Rev5 = get_rev(FDI5),
+
+ IdsRevs = [
+ {<<"UUID1">>, <<"1">>, [Rev1]},
+ {<<"UUID2">>, <<"2">>, [Rev2]},
+ {<<"UUID5">>, <<"5">>, [Rev5]}
+ ],
+ {ok, _} = couch_db:purge_docs(Db, IdsRevs),
+ {ok, Db2} = couch_db:reopen(Db),
+
+ Result2 = run_query(Db2, []),
+ Expect2 = {ok, [
+ {meta, [{total, 2}, {offset, 0}]},
+ {row, [{id, <<"3">>}, {key, 3}, {value, 3}]},
+ {row, [{id, <<"4">>}, {key, 4}, {value, 4}]}
+ ]},
+ ?assertEqual(Expect2, Result2),
+
+ ok
+ end).
+
+
+test_purge_with_compact1(Db) ->
+ ?_test(begin
+ DbName = couch_db:name(Db),
+ Docs = couch_mrview_test_util:make_docs(normal, 6, 200),
+ {ok, Db1} = couch_mrview_test_util:save_docs(Db, Docs),
+ _Result = run_query(Db1, []),
+ DiskSizeBefore = db_disk_size(DbName),
+
+ PurgedDocsNum = 150,
+ IdsRevs = lists:foldl(fun(Id, CIdRevs) ->
+ Id1 = docid(Id),
+ FDI1 = couch_db:get_full_doc_info(Db1, Id1),
+ Rev1 = get_rev(FDI1),
+ UUID1 = uuid(Id),
+ [{UUID1, Id1, [Rev1]} | CIdRevs]
+ end, [], lists:seq(1, PurgedDocsNum)),
+ {ok, _} = couch_db:purge_docs(Db1, IdsRevs),
+
+ {ok, Db2} = couch_db:reopen(Db1),
+ _Result1 = run_query(Db2, []),
+ {ok, PurgedIdRevs} = couch_db:fold_purged_docs(
+ Db2,
+ 0,
+ fun fold_fun/2,
+ [],
+ []
+ ),
+ ?assertEqual(PurgedDocsNum, length(PurgedIdRevs)),
+ config:set("couchdb", "file_compression", "snappy", false),
+
+ {ok, Db3} = couch_db:open_int(DbName, []),
+ {ok, _CompactPid} = couch_db:start_compact(Db3),
+ wait_compaction(DbName, "database", ?LINE),
+ ok = couch_db:close(Db3),
+ DiskSizeAfter = db_disk_size(DbName),
+ ?assert(DiskSizeBefore > DiskSizeAfter),
+
+ ok
+ end).
+
+test_purge_with_compact2(Db) ->
+ ?_test(begin
+ DbName = couch_db:name(Db),
+ % add more documents to database for purge
+ Docs = couch_mrview_test_util:make_docs(normal, 6, 200),
+ {ok, Db1} = couch_mrview_test_util:save_docs(Db, Docs),
+
+ % change PurgedDocsLimit to 10 from 1000 to
+ % avoid timeout of eunit test
+ PurgedDocsLimit = 10,
+ couch_db:set_purged_docs_limit(Db1, PurgedDocsLimit),
+ _Result = run_query(Db1, []),
+
+ % purge 150 documents
+ PurgedDocsNum = 150,
+ IdsRevs = lists:foldl(fun(Id, CIdRevs) ->
+ Id1 = docid(Id),
+ FDI1 = couch_db:get_full_doc_info(Db1, Id1),
+ Rev1 = get_rev(FDI1),
+ UUID1 = uuid(Id),
+ [{UUID1, Id1, [Rev1]} | CIdRevs]
+ end, [], lists:seq(1, PurgedDocsNum)),
+ {ok, _} = couch_db:purge_docs(Db1, IdsRevs),
+
+ % run query again to reflect purge requests
+ % to mrview
+ {ok, Db2} = couch_db:reopen(Db1),
+ _Result1 = run_query(Db2, []),
+ {ok, PurgedIdRevs} = couch_db:fold_purged_docs(
+ Db2,
+ 0,
+ fun fold_fun/2,
+ [],
+ []
+ ),
+ ?assertEqual(PurgedDocsNum, length(PurgedIdRevs)),
+
+ % run compaction to trigger pruning of purge tree
+ {ok, Db3} = couch_db:open_int(DbName, []),
+ {ok, _CompactPid} = couch_db:start_compact(Db3),
+ wait_compaction(DbName, "database", ?LINE),
+ ok = couch_db:close(Db3),
+
+ % check the remaining purge requests in purge tree
+ {ok, Db4} = couch_db:reopen(Db3),
+ {ok, OldestPSeq} = couch_db:get_oldest_purge_seq(Db4),
+ {ok, PurgedIdRevs2} = couch_db:fold_purged_docs(
+ Db4,
+ OldestPSeq - 1,
+ fun fold_fun/2,
+ [],
+ []
+ ),
+ ?assertEqual(PurgedDocsLimit, length(PurgedIdRevs2)),
+
+ ok
+ end).
+
+
+run_query(Db, Opts) ->
+ couch_mrview:query_view(Db, <<"_design/bar">>, <<"baz">>, Opts).
+
+
+get_rev(#full_doc_info{} = FDI) ->
+ #doc_info{
+ revs = [#rev_info{} = PrevRev | _]
+ } = couch_doc:to_doc_info(FDI),
+ PrevRev#rev_info.rev.
+
+db_disk_size(DbName) ->
+ {ok, Db} = couch_db:open_int(DbName, []),
+ {ok, Info} = couch_db:get_db_info(Db),
+ ok = couch_db:close(Db),
+ active_size(Info).
+
+active_size(Info) ->
+ couch_util:get_nested_json_value({Info}, [sizes, active]).
+
+wait_compaction(DbName, Kind, Line) ->
+ WaitFun = fun() ->
+ case is_compaction_running(DbName) of
+ true -> wait;
+ false -> ok
+ end
+ end,
+ case test_util:wait(WaitFun, 10000) of
+ timeout ->
+ erlang:error({assertion_failed,
+ [{module, ?MODULE},
+ {line, Line},
+ {reason, "Timeout waiting for "
+ ++ Kind
+ ++ " database compaction"}]});
+ _ ->
+ ok
+ end.
+
+is_compaction_running(DbName) ->
+ {ok, Db} = couch_db:open_int(DbName, []),
+ {ok, DbInfo} = couch_db:get_db_info(Db),
+ couch_db:close(Db),
+ couch_util:get_value(compact_running, DbInfo).
+
+fold_fun({_PSeq, _UUID, Id, Revs}, Acc) ->
+ [{Id, Revs} | Acc].
+
+docid(I) ->
+ list_to_binary(integer_to_list(I)).
+
+uuid(I) ->
+ Str = io_lib:format("UUID~4..0b", [I]),
+ iolist_to_binary(Str).
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 07/20: WIP - couch_db_updater.erl:
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit aafbd35f6939aafb4d4d82d848005db68424a4d0
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Fri Mar 16 14:49:41 2018 -0500
WIP - couch_db_updater.erl:
---
src/couch/src/couch_db_updater.erl | 92 ++------------------------------------
1 file changed, 3 insertions(+), 89 deletions(-)
diff --git a/src/couch/src/couch_db_updater.erl b/src/couch/src/couch_db_updater.erl
index ce0d45f..b2bcf1e 100644
--- a/src/couch/src/couch_db_updater.erl
+++ b/src/couch/src/couch_db_updater.erl
@@ -688,99 +688,13 @@ purge_docs([Req | RestReqs], [FDI | RestInfos], USeq, PSeq, Acc) ->
end,
{Pairs, PInfos, Replies} = Acc,
NewAcc = {
- [Pair | Pairs],
- [{PSeq, UUID, DocId, Revs} | PInfos],
- [{ok, RemovedRevs} | Replies]
+ [Pair | Pairs],
+ [{PSeq, UUID, DocId, Revs} | PInfos],
+ [{ok, RemovedRevs} | Replies]
},
purge_docs(RestReqs, RestInfos, NewUSeq, PSeq + 1, NewAcc).
-% find purge seq such that all purge requests that happen before or
-% during it can be removed from purge trees
-get_disposable_purge_seq(#db{name=DbName} = Db) ->
- PSeq = couch_db_engine:get_purge_seq(Db),
- OldestPSeq = couch_db_engine:get_oldest_purge_seq(Db),
- PDocsLimit = couch_db_engine:get_purged_docs_limit(Db),
- ExpectedDispPSeq = PSeq - PDocsLimit,
- % client's purge_seq can be up to "allowed_purge_seq_lag"
- % behind ExpectedDispPSeq
- AllowedPSeqLag = config:get_integer("purge", "allowed_purge_seq_lag", 100),
- ClientAllowedMinPSeq = ExpectedDispPSeq - AllowedPSeqLag,
- DisposablePSeq = if OldestPSeq > ClientAllowedMinPSeq ->
- % DisposablePSeq is the last pseq we can remove;
- % it should be one less than OldestPSeq when #purges is within limit
- OldestPSeq - 1;
- true ->
- % Find the smallest checkpointed purge_seq among clients
- V = "v" ++ config:get("purge", "version", "1") ++ "-",
- Opts = [
- {start_key, list_to_binary(?LOCAL_DOC_PREFIX ++ "purge-" ++ V)},
- {end_key_gt, list_to_binary(?LOCAL_DOC_PREFIX ++ "purge1")}
- ],
- FoldFun = fun(#doc{id=DocID, body={Props}}, MinPSeq) ->
- ClientPSeq = couch_util:get_value(<<"purge_seq">>, Props),
- MinPSeq2 = if ClientPSeq >= ClientAllowedMinPSeq ->
- erlang:min(MinPSeq, ClientPSeq);
- true ->
- case check_client_exists(DbName, DocID, Props) of
- true -> erlang:min(MinPSeq, ClientPSeq);
- false -> MinPSeq % ignore nonexisting clients
- end
- end,
- {ok, MinPSeq2}
- end,
- {ok, ClientPSeq} = couch_db_engine:fold_local_docs(
- Db, FoldFun, PSeq, Opts),
- erlang:min(ClientPSeq, ExpectedDispPSeq)
- end,
- DisposablePSeq.
-
-
-check_client_exists(DbName, DocID, Props) ->
- % will warn about clients that have not
- % checkpointed more than "allowed_purge_time_lag"
- AllowedPTimeLag = config:get_integer("purge",
- "allowed_purge_time_lag", 86400), % secs in 1 day
- M0 = couch_util:get_value(<<"verify_module">>, Props),
- F0 = couch_util:get_value(<<"verify_function">>, Props),
- M = binary_to_atom(M0, latin1),
- F = binary_to_atom(F0, latin1),
- {A} = couch_util:get_value(<<"verify_options">>, Props),
- ClientExists = try erlang:apply(M, F, [A]) of
- true ->
- % warn if we haven't heard of this client more than AllowedPTimeLag
- ClientTime = ?b2l(couch_util:get_value(<<"timestamp_utc">>, Props)),
- {ok, [Y, Mon, D, H, Min, S], [] }=
- io_lib:fread("~4d-~2d-~2dT~2d:~2d:~2dZ", ClientTime),
- SecsClient = calendar:datetime_to_gregorian_seconds(
- {{Y, Mon, D}, {H, Min, S}}),
- SecsNow = calendar:datetime_to_gregorian_seconds(
- calendar:now_to_universal_time(os:timestamp())),
- if SecsClient + AllowedPTimeLag > SecsNow -> ok; true ->
- couch_log:warning(
- "Client: ~p hasn't processed purge requests for more than"
- " ~p secs. Check this client, as it prevents compaction of "
- "purge trees on db:~p.", [A, AllowedPTimeLag, DbName]
- )
- end,
- true;
- false ->
- couch_log:warning(
- "Client ~p doesn't exist, "
- "but its checkpoint purge doc: ~p is still available. "
- "Remove this doc from: ~p", [A, DocID, DbName]
- ),
- false
- catch
- error:Error ->
- couch_log:error(
- "error in evaluating if client: ~p exists: ~p", [A, Error]
- ),
- false
- end,
- ClientExists.
-
-
commit_data(Db) ->
commit_data(Db, false).
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 19/20: Fix tests to work with clustered purge
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 85d30224c065d92977d795bebdf95d8dd2548f64
Author: jiangphcn <ji...@cn.ibm.com>
AuthorDate: Thu Mar 1 17:46:17 2018 +0800
Fix tests to work with clustered purge
COUCHDB-3326
---
test/javascript/tests/erlang_views.js | 5 +++--
test/javascript/tests/purge.js | 27 ++++++++-------------------
2 files changed, 11 insertions(+), 21 deletions(-)
diff --git a/test/javascript/tests/erlang_views.js b/test/javascript/tests/erlang_views.js
index ec78e65..9b15e10 100644
--- a/test/javascript/tests/erlang_views.js
+++ b/test/javascript/tests/erlang_views.js
@@ -56,7 +56,7 @@ couchTests.erlang_views = function(debug) {
' {Info} = couch_util:get_value(<<"info">>, Req, {[]}), ' +
' Purged = couch_util:get_value(<<"purge_seq">>, Info, -1), ' +
' Verb = couch_util:get_value(<<"method">>, Req, <<"not_get">>), ' +
- ' R = list_to_binary(io_lib:format("~b - ~s", [Purged, Verb])), ' +
+ ' R = list_to_binary(io_lib:format("~s - ~s", [Purged, Verb])), ' +
' {[{<<"code">>, 200}, {<<"headers">>, {[]}}, {<<"body">>, R}]} ' +
'end.'
},
@@ -85,7 +85,8 @@ couchTests.erlang_views = function(debug) {
var url = "/" + db_name + "/_design/erlview/_show/simple/1";
var xhr = CouchDB.request("GET", url);
T(xhr.status == 200, "standard get should be 200");
- T(xhr.responseText == "0 - GET");
+ T(/0-/.test(xhr.responseText));
+ T(/- GET/.test(xhr.responseText));
var url = "/" + db_name + "/_design/erlview/_list/simple_list/simple_view";
var xhr = CouchDB.request("GET", url);
diff --git a/test/javascript/tests/purge.js b/test/javascript/tests/purge.js
index 38eca8d..6bfba02 100644
--- a/test/javascript/tests/purge.js
+++ b/test/javascript/tests/purge.js
@@ -11,7 +11,6 @@
// the License.
couchTests.purge = function(debug) {
- return console.log('TODO: this feature is not yet implemented');
var db_name = get_random_db_name();
var db = new CouchDB(db_name, {"X-Couch-Full-Commit":"false"});
db.createDb();
@@ -53,21 +52,13 @@ couchTests.purge = function(debug) {
var xhr = CouchDB.request("POST", "/" + db_name + "/_purge", {
body: JSON.stringify({"1":[doc1._rev], "2":[doc2._rev]})
});
- console.log(xhr.status);
- console.log(xhr.responseText);
- T(xhr.status == 200);
+ T(xhr.status == 201);
var result = JSON.parse(xhr.responseText);
var newInfo = db.info();
-
- // purging increments the update sequence
- T(info.update_seq+1 == newInfo.update_seq);
- // and it increments the purge_seq
- T(info.purge_seq+1 == newInfo.purge_seq);
- T(result.purge_seq == newInfo.purge_seq);
- T(result.purged["1"][0] == doc1._rev);
- T(result.purged["2"][0] == doc2._rev);
+ T(result.purged["1"].purged[0] == doc1._rev);
+ T(result.purged["2"].purged[0] == doc2._rev);
T(db.open("1") == null);
T(db.open("2") == null);
@@ -85,7 +76,6 @@ couchTests.purge = function(debug) {
// compaction isn't instantaneous, loop until done
while (db.info().compact_running) {};
var compactInfo = db.info();
- T(compactInfo.purge_seq == newInfo.purge_seq);
// purge documents twice in a row without loading views
// (causes full view rebuilds)
@@ -97,15 +87,14 @@ couchTests.purge = function(debug) {
body: JSON.stringify({"3":[doc3._rev]})
});
- T(xhr.status == 200);
+ T(xhr.status == 201);
xhr = CouchDB.request("POST", "/" + db_name + "/_purge", {
body: JSON.stringify({"4":[doc4._rev]})
});
- T(xhr.status == 200);
+ T(xhr.status == 201);
result = JSON.parse(xhr.responseText);
- T(result.purge_seq == db.info().purge_seq);
var rows = db.view("test/all_docs_twice").rows;
for (var i = 4; i < numDocs; i++) {
@@ -129,7 +118,7 @@ couchTests.purge = function(debug) {
var xhr = CouchDB.request("POST", "/" + dbB.name + "/_purge", {
body: JSON.stringify({"test":[docA._rev]})
});
- TEquals(200, xhr.status, "single rev purge after replication succeeds");
+ TEquals(201, xhr.status, "single rev purge after replication succeeds");
var xhr = CouchDB.request("GET", "/" + dbB.name + "/test?rev=" + docA._rev);
TEquals(404, xhr.status, "single rev purge removes revision");
@@ -137,14 +126,14 @@ couchTests.purge = function(debug) {
var xhr = CouchDB.request("POST", "/" + dbB.name + "/_purge", {
body: JSON.stringify({"test":[docB._rev]})
});
- TEquals(200, xhr.status, "single rev purge after replication succeeds");
+ TEquals(201, xhr.status, "single rev purge after replication succeeds");
var xhr = CouchDB.request("GET", "/" + dbB.name + "/test?rev=" + docB._rev);
TEquals(404, xhr.status, "single rev purge removes revision");
var xhr = CouchDB.request("POST", "/" + dbB.name + "/_purge", {
body: JSON.stringify({"test":[docA._rev, docB._rev]})
});
- TEquals(200, xhr.status, "all rev purge after replication succeeds");
+ TEquals(201, xhr.status, "all rev purge after replication succeeds");
// cleanup
db.deleteDb();
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 17/20: Implement clustered purge HTTP endpoint
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit f60758aefd5092e82da8a9a37f74fc39f817c1f1
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Mon May 1 16:44:25 2017 -0400
Implement clustered purge HTTP endpoint
* Implement clustered purge endpoint
* Add endpoint for setting purged_docs_limit
* Add endpoint for getting purged_docs_limit
COUCHDB-3326
---
src/chttpd/src/chttpd_db.erl | 42 ++++++++---
src/chttpd/test/chttpd_purge_tests.erl | 130 +++++++++++++++++++++++++++++++++
2 files changed, 161 insertions(+), 11 deletions(-)
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index 2c3ec63..f469b98 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -497,23 +497,22 @@ db_req(#httpd{path_parts=[_, <<"_bulk_get">>]}=Req, _Db) ->
db_req(#httpd{method='POST',path_parts=[_,<<"_purge">>]}=Req, Db) ->
chttpd:validate_ctype(Req, "application/json"),
+ W = chttpd:qs_value(Req, "w", integer_to_list(mem3:quorum(Db))),
+ Options = [{user_ctx, Req#httpd.user_ctx}, {w, W}],
{IdsRevs} = chttpd:json_body_obj(Req),
IdsRevs2 = [{Id, couch_doc:parse_revs(Revs)} || {Id, Revs} <- IdsRevs],
- case fabric:purge_docs(Db, IdsRevs2) of
- {ok, PurgeSeq, PurgedIdsRevs} ->
- PurgedIdsRevs2 = [{Id, couch_doc:revs_to_strs(Revs)} || {Id, Revs}
- <- PurgedIdsRevs],
- send_json(Req, 200, {[
- {<<"purge_seq">>, PurgeSeq},
- {<<"purged">>, {PurgedIdsRevs2}}
- ]});
- Error ->
- throw(Error)
- end;
+ {Status, Results} = fabric:purge_docs(Db, IdsRevs2, Options),
+ Code = case Status of
+ ok -> 201;
+ accepted -> 202
+ end,
+ Purged = lists:zipwith(fun purge_result_to_json/2, IdsRevs2, Results),
+ send_json(Req, Code, {[{<<"purged">>, {Purged}}]});
db_req(#httpd{path_parts=[_,<<"_purge">>]}=Req, _Db) ->
send_method_not_allowed(Req, "POST");
+
db_req(#httpd{method='GET',path_parts=[_,OP]}=Req, Db) when ?IS_ALL_DOCS(OP) ->
case chttpd:qs_json_value(Req, "keys", nil) of
Keys when is_list(Keys) ->
@@ -608,6 +607,20 @@ db_req(#httpd{method='GET',path_parts=[_,<<"_revs_limit">>]}=Req, Db) ->
db_req(#httpd{path_parts=[_,<<"_revs_limit">>]}=Req, _Db) ->
send_method_not_allowed(Req, "PUT,GET");
+db_req(#httpd{method='PUT',path_parts=[_,<<"_purged_docs_limit">>]}=Req, Db) ->
+ Limit = chttpd:json_body(Req),
+ Options = [{user_ctx, Req#httpd.user_ctx}],
+ case chttpd:json_body(Req) of
+ Limit when is_integer(Limit), Limit > 0 ->
+ ok = fabric:set_purged_docs_limit(Db, Limit, Options),
+ send_json(Req, {[{<<"ok">>, true}]});
+ _->
+ throw({bad_request, "`purged_docs_limit` must be positive integer"})
+ end;
+
+db_req(#httpd{method='GET',path_parts=[_,<<"_purged_docs_limit">>]}=Req, Db) ->
+ send_json(Req, fabric:get_purged_docs_limit(Db));
+
% Special case to enable using an unencoded slash in the URL of design docs,
% as slashes in document IDs must otherwise be URL encoded.
db_req(#httpd{method='GET', mochi_req=MochiReq, path_parts=[_DbName, <<"_design/", _/binary>> | _]}=Req, _Db) ->
@@ -955,6 +968,13 @@ update_doc_result_to_json(DocId, Error) ->
{_Code, ErrorStr, Reason} = chttpd:error_info(Error),
{[{id, DocId}, {error, ErrorStr}, {reason, Reason}]}.
+purge_result_to_json({DocId, _Revs}, {ok, PRevs}) ->
+ {DocId, {[{purged, couch_doc:revs_to_strs(PRevs)}, {ok, true}]}};
+purge_result_to_json({DocId, _Revs}, {accepted, PRevs}) ->
+ {DocId, {[{purged, couch_doc:revs_to_strs(PRevs)}, {accepted, true}]}};
+purge_result_to_json({DocId, _Revs}, Error) ->
+ {_Code, ErrorStr, Reason} = chttpd:error_info(Error),
+ {DocId, {[{error, ErrorStr}, {reason, Reason}]}}.
send_updated_doc(Req, Db, DocId, Json) ->
send_updated_doc(Req, Db, DocId, Json, []).
diff --git a/src/chttpd/test/chttpd_purge_tests.erl b/src/chttpd/test/chttpd_purge_tests.erl
new file mode 100644
index 0000000..7900090
--- /dev/null
+++ b/src/chttpd/test/chttpd_purge_tests.erl
@@ -0,0 +1,130 @@
+% 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(chttpd_purge_tests).
+
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+
+-define(USER, "chttpd_db_test_admin").
+-define(PASS, "pass").
+-define(AUTH, {basic_auth, {?USER, ?PASS}}).
+-define(CONTENT_JSON, {"Content-Type", "application/json"}).
+
+
+setup() ->
+ ok = config:set("admins", ?USER, ?PASS, _Persist=false),
+ TmpDb = ?tempdb(),
+ Addr = config:get("chttpd", "bind_address", "127.0.0.1"),
+ Port = mochiweb_socket_server:get(chttpd, port),
+ Url = lists:concat(["http://", Addr, ":", Port, "/", ?b2l(TmpDb)]),
+ create_db(Url),
+ Url.
+
+
+teardown(Url) ->
+ delete_db(Url),
+ ok = config:delete("admins", ?USER, _Persist=false).
+
+
+create_db(Url) ->
+ {ok, Status, _, _} = test_request:put(Url, [?CONTENT_JSON, ?AUTH], "{}"),
+ ?assert(Status =:= 201 orelse Status =:= 202).
+
+
+create_doc(Url, Id) ->
+ test_request:put(Url ++ "/" ++ Id,
+ [?CONTENT_JSON, ?AUTH], "{\"mr\": \"rockoartischocko\"}").
+
+
+delete_db(Url) ->
+ {ok, 200, _, _} = test_request:delete(Url, [?AUTH]).
+
+
+purge_test_() ->
+ {
+ "chttpd db tests",
+ {
+ setup,
+ fun chttpd_test_util:start_couch/0,
+ fun chttpd_test_util:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ fun test_empty_purge_request/1,
+ fun test_ok_purge_request/1,
+ fun should_error_set_purged_docs_limit_to0/1
+ ]
+ }
+ }
+ }.
+
+
+test_empty_purge_request(Url) ->
+ ?_test(begin
+ IdsRevs = "{}",
+ {ok, Status, _, ResultBody} = test_request:post(Url ++ "/_purge/",
+ [?CONTENT_JSON, ?AUTH], IdsRevs),
+ ResultJson = ?JSON_DECODE(ResultBody),
+ ?assert(Status =:= 201 orelse Status =:= 202),
+ ?assertEqual({[{<<"purged">>,{[]}}]}, ResultJson)
+ end).
+
+
+test_ok_purge_request(Url) ->
+ ?_test(begin
+ {ok, _, _, Body} = create_doc(Url, "doc1"),
+ {Json} = ?JSON_DECODE(Body),
+ Rev1 = couch_util:get_value(<<"rev">>, Json, undefined),
+ {ok, _, _, Body2} = create_doc(Url, "doc2"),
+ {Json2} = ?JSON_DECODE(Body2),
+ Rev2 = couch_util:get_value(<<"rev">>, Json2, undefined),
+ {ok, _, _, Body3} = create_doc(Url, "doc3"),
+ {Json3} = ?JSON_DECODE(Body3),
+ Rev3 = couch_util:get_value(<<"rev">>, Json3, undefined),
+ IdsRevs = "{\"doc1\": [\"" ++ ?b2l(Rev1) ++ "\"], \"doc2\": [\"" ++
+ ?b2l(Rev2) ++ "\"], \"doc3\": [\"" ++ ?b2l(Rev3) ++ "\"] }",
+
+ {ok, Status, _, ResultBody} = test_request:post(Url ++ "/_purge/",
+ [?CONTENT_JSON, ?AUTH], IdsRevs),
+ ResultJson = ?JSON_DECODE(ResultBody),
+ ?assert(Status =:= 201 orelse Status =:= 202),
+ ?assertEqual(
+ {[{<<"purged">>, {[
+ {<<"doc1">>, {[
+ {<<"purged">>,[Rev1]},
+ {<<"ok">>,true}
+ ]}},
+ {<<"doc2">>, {[
+ {<<"purged">>,[Rev2]},
+ {<<"ok">>,true}
+ ]}},
+ {<<"doc3">>, {[
+ {<<"purged">>,[Rev3]},
+ {<<"ok">>,true}
+ ]}}
+ ]}}]},
+ ResultJson
+ )
+ end).
+
+
+should_error_set_purged_docs_limit_to0(Url) ->
+ ?_test(begin
+ {ok, Status, _, _} = test_request:put(Url ++ "/_purged_docs_limit/",
+ [?CONTENT_JSON, ?AUTH], "0"),
+ ?assert(Status =:= 400)
+ end).
\ No newline at end of file
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 15/20: Implement clustered purge API
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit c3cc8d4d8c40d8b77c405585d796ac9535d87dc8
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Mon May 1 15:43:43 2017 -0400
Implement clustered purge API
* Implement clustered endpoint to purge docs on all nodes
- implement fabric:purge_docs(DbName, IdsRevs, Options)
- generate a unique ID: "UUID" for every purge request before
sending them to workers on specific node.
- fabric_rpc:purge_docs has an option of "replicated_changes" or
or "interactive edit" that are passed to couch_db:purge_docs/3.
This done so that "replicated_changes" updates will not reapply
purges with UUIDs that already exist on a node.
* Fix read-repair
- fix read-repair so not to recreate docs that have been purged before
on a certain node from nodes that are out of sync.
- In the open docs calls we track which nodes sent which revisions. If we
detect the need for read repair, we send this list of (node, rev) pairs
as an option to the fabric:update_docs call. When fabric:update_docs
receives a list of (node, rev) pairs it will use this information to know
whether it should apply the update or ignore it. It checks the
_local/purge-mem3.. docs to see if the purge_seq is up to date.
If not it should ignore the update request. As an optimization,
if the purge_seq is less than a configurable limit
out of sync, the updater sequentially scans the purge_seq tree
looking for purge requests for the given revision and if not found can
continue with the write.
* Implement clustered endpoint to set purged_docs_limit of Db on all nodes
- implement fabric:set_purged_docs_limit(DbName, Limit, Options)
* Implement clustered endpoint to get purged_docs_limit
- implement fabric:get_purged_docs_limit(DbName)
COUCHDB-3326
---
src/fabric/src/fabric.erl | 35 ++-
src/fabric/src/fabric_db_info.erl | 29 +--
src/fabric/src/fabric_db_meta.erl | 26 ++-
src/fabric/src/fabric_doc_open.erl | 42 ++--
src/fabric/src/fabric_doc_purge.erl | 414 ++++++++++++++++++++++++++++++++++++
src/fabric/src/fabric_rpc.erl | 102 ++++++++-
6 files changed, 612 insertions(+), 36 deletions(-)
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 4a07271..7221654 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -21,12 +21,13 @@
delete_db/2, get_db_info/1, get_doc_count/1, set_revs_limit/3,
set_security/2, set_security/3, get_revs_limit/1, get_security/1,
get_security/2, get_all_security/1, get_all_security/2,
+ get_purged_docs_limit/1, set_purged_docs_limit/3,
compact/1, compact/2]).
% Documents
-export([open_doc/3, open_revs/4, get_doc_info/3, get_full_doc_info/3,
get_missing_revs/2, get_missing_revs/3, update_doc/3, update_docs/3,
- purge_docs/2, att_receiver/2]).
+ purge_docs/3, att_receiver/2]).
% Views
-export([all_docs/4, all_docs/5, changes/4, query_view/3, query_view/4,
@@ -137,6 +138,18 @@ set_security(DbName, SecObj) ->
set_security(DbName, SecObj, Options) ->
fabric_db_meta:set_security(dbname(DbName), SecObj, opts(Options)).
+%% @doc sets the upper bound for the number of stored purge requests
+-spec set_purged_docs_limit(dbname(), pos_integer(), [option()]) -> ok.
+set_purged_docs_limit(DbName, Limit, Options)
+ when is_integer(Limit), Limit > 0 ->
+ fabric_db_meta:set_purged_docs_limit(dbname(DbName), Limit, opts(Options)).
+
+%% @doc retrieves the upper bound for the number of stored purge requests
+-spec get_purged_docs_limit(dbname()) -> pos_integer() | no_return().
+get_purged_docs_limit(DbName) ->
+ {ok, Db} = fabric_util:get_db(dbname(DbName), [?ADMIN_CTX]),
+ try couch_db:get_purged_docs_limit(Db) after catch couch_db:close(Db) end.
+
get_security(DbName) ->
get_security(DbName, [?ADMIN_CTX]).
@@ -267,8 +280,24 @@ update_docs(DbName, Docs, Options) ->
{aborted, PreCommitFailures}
end.
-purge_docs(_DbName, _IdsRevs) ->
- not_implemented.
+
+%% @doc purge revisions for a list '{Id, Revs}'
+%% returns {ok, {PurgeSeq, Results}}
+-spec purge_docs(dbname(), [{docid(), [revision()]}], [option()]) ->
+ {Health, {PurgeSeq, [{Health, [revision()]}] }} when
+ Health :: ok | accepted,
+ PurgeSeq :: any().
+purge_docs(DbName, IdsRevs, Options) when is_list(IdsRevs) ->
+ IdsRevs2 = [idrevs(IdRs) || IdRs <- IdsRevs],
+ case fabric_doc_purge:go(dbname(DbName), IdsRevs2, opts(Options)) of
+ {ok, Results} ->
+ {ok, Results};
+ {accepted, Results} ->
+ {accepted, Results};
+ Error ->
+ throw(Error)
+ end.
+
%% @doc spawns a process to upload attachment data and
%% returns a function that shards can use to communicate
diff --git a/src/fabric/src/fabric_db_info.erl b/src/fabric/src/fabric_db_info.erl
index 98e8e52..97a31c2 100644
--- a/src/fabric/src/fabric_db_info.erl
+++ b/src/fabric/src/fabric_db_info.erl
@@ -23,10 +23,12 @@ go(DbName) ->
RexiMon = fabric_util:create_monitors(Shards),
Fun = fun handle_message/3,
{ok, ClusterInfo} = get_cluster_info(Shards),
- Acc0 = {fabric_dict:init(Workers, nil), [{cluster, ClusterInfo}]},
+ Acc0 = {fabric_dict:init(Workers, nil), [], [{cluster, ClusterInfo}]},
try
case fabric_util:recv(Workers, #shard.ref, Fun, Acc0) of
- {ok, Acc} -> {ok, Acc};
+
+ {ok, Acc} ->
+ {ok, Acc};
{timeout, {WorkersDict, _}} ->
DefunctWorkers = fabric_util:remove_done_workers(
WorkersDict,
@@ -37,44 +39,49 @@ go(DbName) ->
"get_db_info"
),
{error, timeout};
- {error, Error} -> throw(Error)
+ {error, Error} ->
+ throw(Error)
end
after
rexi_monitor:stop(RexiMon)
end.
-handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Shard, {Counters, Acc}) ->
+handle_message({rexi_DOWN,
+ _, {_,NodeRef},_}, _Shard, {Counters, PseqAcc, Acc}) ->
case fabric_util:remove_down_workers(Counters, NodeRef) of
{ok, NewCounters} ->
- {ok, {NewCounters, Acc}};
+ {ok, {NewCounters, PseqAcc, Acc}};
error ->
{error, {nodedown, <<"progress not possible">>}}
end;
-handle_message({rexi_EXIT, Reason}, Shard, {Counters, Acc}) ->
+handle_message({rexi_EXIT, Reason}, Shard, {Counters, PseqAcc, Acc}) ->
NewCounters = fabric_dict:erase(Shard, Counters),
case fabric_view:is_progress_possible(NewCounters) of
true ->
- {ok, {NewCounters, Acc}};
+ {ok, {NewCounters, PseqAcc, Acc}};
false ->
{error, Reason}
end;
-handle_message({ok, Info}, #shard{dbname=Name} = Shard, {Counters, Acc}) ->
+handle_message({ok, Info}, #shard{dbname=Name} = Shard, {Counters, PseqAcc, Acc}) ->
case fabric_dict:lookup_element(Shard, Counters) of
undefined ->
% already heard from someone else in this range
- {ok, {Counters, Acc}};
+ {ok, {Counters, PseqAcc, Acc}};
nil ->
Seq = couch_util:get_value(update_seq, Info),
C1 = fabric_dict:store(Shard, Seq, Counters),
C2 = fabric_view:remove_overlapping_shards(Shard, C1),
+ PSeq = couch_util:get_value(purge_seq, Info),
+ NewPseqAcc = [{Shard, PSeq}|PseqAcc],
case fabric_dict:any(nil, C2) of
true ->
- {ok, {C2, [Info|Acc]}};
+ {ok, {C2, NewPseqAcc, [Info|Acc]}};
false ->
{stop, [
{db_name,Name},
+ {purge_seq, fabric_view_changes:pack_seqs(NewPseqAcc)},
{update_seq, fabric_view_changes:pack_seqs(C2)} |
merge_results(lists:flatten([Info|Acc]))
]}
@@ -91,8 +98,6 @@ merge_results(Info) ->
[{doc_count, lists:sum(X)} | Acc];
(doc_del_count, X, Acc) ->
[{doc_del_count, lists:sum(X)} | Acc];
- (purge_seq, X, Acc) ->
- [{purge_seq, lists:sum(X)} | Acc];
(compact_running, X, Acc) ->
[{compact_running, lists:member(true, X)} | Acc];
(disk_size, X, Acc) -> % legacy
diff --git a/src/fabric/src/fabric_db_meta.erl b/src/fabric/src/fabric_db_meta.erl
index 367ef06..4fd9365 100644
--- a/src/fabric/src/fabric_db_meta.erl
+++ b/src/fabric/src/fabric_db_meta.erl
@@ -12,7 +12,8 @@
-module(fabric_db_meta).
--export([set_revs_limit/3, set_security/3, get_all_security/2]).
+-export([set_revs_limit/3, set_security/3, get_all_security/2,
+ set_purged_docs_limit/3]).
-include_lib("fabric/include/fabric.hrl").
-include_lib("mem3/include/mem3.hrl").
@@ -48,6 +49,29 @@ handle_revs_message(Error, _, _Acc) ->
{error, Error}.
+set_purged_docs_limit(DbName, Limit, Options) ->
+ Shards = mem3:shards(DbName),
+ Workers = fabric_util:submit_jobs(Shards, set_purged_docs_limit, [Limit, Options]),
+ Handler = fun handle_purge_message/3,
+ Acc0 = {Workers, length(Workers) - 1},
+ case fabric_util:recv(Workers, #shard.ref, Handler, Acc0) of
+ {ok, ok} ->
+ ok;
+ {timeout, {DefunctWorkers, _}} ->
+ fabric_util:log_timeout(DefunctWorkers, "set_purged_docs_limit"),
+ {error, timeout};
+ Error ->
+ Error
+ end.
+
+handle_purge_message(ok, _, {_Workers, 0}) ->
+ {stop, ok};
+handle_purge_message(ok, Worker, {Workers, Waiting}) ->
+ {ok, {lists:delete(Worker, Workers), Waiting - 1}};
+handle_purge_message(Error, _, _Acc) ->
+ {error, Error}.
+
+
set_security(DbName, SecObj, Options) ->
Shards = mem3:shards(DbName),
RexiMon = fabric_util:create_monitors(Shards),
diff --git a/src/fabric/src/fabric_doc_open.erl b/src/fabric/src/fabric_doc_open.erl
index 9c45bd9..b974880 100644
--- a/src/fabric/src/fabric_doc_open.erl
+++ b/src/fabric/src/fabric_doc_open.erl
@@ -25,7 +25,8 @@
r,
state,
replies,
- q_reply
+ q_reply,
+ replies_by_node=[] %[{Node, Reply}] used for checking if a doc is purged
}).
@@ -83,7 +84,8 @@ handle_message({rexi_EXIT, _Reason}, Worker, Acc) ->
end;
handle_message(Reply, Worker, Acc) ->
NewReplies = fabric_util:update_counter(Reply, 1, Acc#acc.replies),
- NewAcc = Acc#acc{replies = NewReplies},
+ NewNReplies = [{Worker#shard.node, Reply}|Acc#acc.replies_by_node],
+ NewAcc = Acc#acc{replies = NewReplies, replies_by_node = NewNReplies},
case is_r_met(Acc#acc.workers, NewReplies, Acc#acc.r) of
{true, QuorumReply} ->
fabric_util:cleanup(lists:delete(Worker, Acc#acc.workers)),
@@ -122,14 +124,15 @@ is_r_met(Workers, Replies, R) ->
no_more_workers
end.
-read_repair(#acc{dbname=DbName, replies=Replies}) ->
+read_repair(#acc{dbname=DbName, replies=Replies, replies_by_node=NReplies0}) ->
Docs = [Doc || {_, {{ok, #doc{}=Doc}, _}} <- Replies],
+ NReplies = [{Node, Doc} || {Node, {ok, #doc{}=Doc}} <- NReplies0],
case Docs of
% omit local docs from read repair
[#doc{id = <<?LOCAL_DOC_PREFIX, _/binary>>} | _] ->
choose_reply(Docs);
[#doc{id=Id} | _] ->
- Opts = [replicated_changes, ?ADMIN_CTX],
+ Opts = [replicated_changes, ?ADMIN_CTX, {read_repair, NReplies}],
Res = fabric:update_docs(DbName, Docs, Opts),
case Res of
{ok, []} ->
@@ -319,7 +322,8 @@ handle_message_reply_test() ->
?assertEqual(
{ok, Acc0#acc{
workers=[Worker0, Worker1],
- replies=[fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(foo,1)],
+ replies_by_node=[{undefined, foo}]
}},
handle_message(foo, Worker2, Acc0)
),
@@ -327,7 +331,8 @@ handle_message_reply_test() ->
?assertEqual(
{ok, Acc0#acc{
workers=[Worker0, Worker1],
- replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)],
+ replies_by_node=[{undefined, bar}]
}},
handle_message(bar, Worker2, Acc0#acc{
replies=[fabric_util:kv(foo,1)]
@@ -339,18 +344,21 @@ handle_message_reply_test() ->
% is returned. Bit subtle on the assertions here.
?assertEqual(
- {stop, Acc0#acc{workers=[],replies=[fabric_util:kv(foo,1)]}},
+ {stop, Acc0#acc{workers=[],replies=[fabric_util:kv(foo,1)],
+ replies_by_node=[{undefined, foo}]}},
handle_message(foo, Worker0, Acc0#acc{workers=[Worker0]})
),
?assertEqual(
{stop, Acc0#acc{
workers=[],
- replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)],
+ replies_by_node =[{undefined, bar}, {undefined, foo}]
}},
handle_message(bar, Worker0, Acc0#acc{
workers=[Worker0],
- replies=[fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(foo,1)],
+ replies_by_node=[{undefined, foo}]
})
),
@@ -362,11 +370,13 @@ handle_message_reply_test() ->
workers=[],
replies=[fabric_util:kv(foo,2)],
state=r_met,
- q_reply=foo
+ q_reply=foo,
+ replies_by_node =[{undefined, foo}, {undefined, foo}]
}},
handle_message(foo, Worker1, Acc0#acc{
workers=[Worker0, Worker1],
- replies=[fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(foo,1)],
+ replies_by_node =[{undefined, foo}]
})
),
@@ -376,7 +386,8 @@ handle_message_reply_test() ->
r=1,
replies=[fabric_util:kv(foo,1)],
state=r_met,
- q_reply=foo
+ q_reply=foo,
+ replies_by_node =[{undefined, foo}]
}},
handle_message(foo, Worker0, Acc0#acc{r=1})
),
@@ -386,11 +397,14 @@ handle_message_reply_test() ->
workers=[],
replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,2)],
state=r_met,
- q_reply=foo
+ q_reply=foo,
+ replies_by_node =[{undefined, foo}, {undefined, foo},
+ {undefined, bar}]
}},
handle_message(foo, Worker0, Acc0#acc{
workers=[Worker0],
- replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)]
+ replies=[fabric_util:kv(bar,1), fabric_util:kv(foo,1)],
+ replies_by_node =[{undefined, foo}, {undefined, bar}]
})
),
diff --git a/src/fabric/src/fabric_doc_purge.erl b/src/fabric/src/fabric_doc_purge.erl
new file mode 100644
index 0000000..24e8c66
--- /dev/null
+++ b/src/fabric/src/fabric_doc_purge.erl
@@ -0,0 +1,414 @@
+% 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(fabric_doc_purge).
+
+-export([go/3]).
+
+-include_lib("fabric/include/fabric.hrl").
+-include_lib("mem3/include/mem3.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+go(_, [], _) ->
+ {ok, []};
+go(DbName, AllIdsRevs, Opts) ->
+ % tag each purge request with UUId
+ {AllUUIDs, AllUUIDsIdsRevs, DocCount} = tag_docs(AllIdsRevs),
+
+ Options = lists:delete(all_or_nothing, Opts),
+ % Counters -> [{Worker, UUIDs}]
+ {Counters, Workers} = dict:fold(fun(Shard, UUIDsIdsRevs, {Cs,Ws}) ->
+ UUIDs = [UUID || {UUID, _Id, _Revs} <-UUIDsIdsRevs],
+ #shard{name=Name, node=Node} = Shard,
+ Ref = rexi:cast(Node,
+ {fabric_rpc, purge_docs, [Name, UUIDsIdsRevs, Options]}),
+ Worker = Shard#shard{ref=Ref},
+ {[{Worker, UUIDs}|Cs], [Worker|Ws]}
+ end, {[], []}, group_idrevs_by_shard(DbName, AllUUIDsIdsRevs)),
+
+ RexiMon = fabric_util:create_monitors(Workers),
+ W = couch_util:get_value(w, Options, integer_to_list(mem3:quorum(DbName))),
+ Acc = {length(Workers), DocCount, list_to_integer(W), Counters, dict:new()},
+ Timeout = fabric_util:request_timeout(),
+ try rexi_utils:recv(Workers, #shard.ref,
+ fun handle_message/3, Acc, infinity, Timeout) of
+ {ok, {Health, Results}} when Health =:= ok; Health =:= accepted ->
+ % Results-> [{UUID, {ok, Revs}}]
+ {Health, [R || R <-
+ couch_util:reorder_results(AllUUIDs, Results)]};
+ {timeout, Acc1} ->
+ {_, _, W1, Counters1, DocReplDict0} = Acc1,
+ {DefunctWorkers, _} = lists:unzip(Counters1),
+ fabric_util:log_timeout(DefunctWorkers, "purge_docs"),
+ DocReplDict = lists:foldl(fun({_W, Docs}, Dict) ->
+ Replies = [{error, timeout} || _D <- Docs],
+ append_purge_replies(Docs, Replies, Dict)
+ end, DocReplDict0, Counters1),
+ {Health, _, Resp} = dict:fold(
+ fun force_reply/3, {ok, W1, []}, DocReplDict),
+ case Health of
+ error -> timeout;
+ _ -> {Health, [R || R <-
+ couch_util:reorder_results(AllUUIDs, Resp)]}
+
+ end;
+ Else ->
+ Else
+ after
+ rexi_monitor:stop(RexiMon)
+ end.
+
+handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, Acc0) ->
+ {_, DocCount, W, Counters, DocsDict0} = Acc0,
+ {FailCounters, NewCounters} = lists:partition(fun({#shard{node=N}, _}) ->
+ N == NodeRef
+ end, Counters),
+ % fill DocsDict with error messages for relevant Docs
+ DocsDict = lists:foldl(fun({_W, Docs}, CDocsDict) ->
+ Replies = [{error, internal_server_error} || _D <- Docs],
+ append_purge_replies(Docs, Replies, CDocsDict)
+ end, DocsDict0, FailCounters),
+ skip_message({length(NewCounters), DocCount, W, NewCounters, DocsDict});
+handle_message({rexi_EXIT, _}, Worker, Acc0) ->
+ {WC, DocCount , W, Counters, DocsDict0} = Acc0,
+ % fill DocsDict with error messages for relevant Docs
+ {value, {_W, Docs}, NewCounters} = lists:keytake(Worker, 1, Counters),
+ Replies = [{error, internal_server_error} || _D <- Docs],
+ DocsDict = append_purge_replies(Docs, Replies, DocsDict0),
+ skip_message({WC-1, DocCount, W, NewCounters, DocsDict});
+handle_message({ok, Replies0}, Worker, Acc0) ->
+ {WCount, DocCount, W, Counters, DocsDict0} = Acc0,
+ {value, {_W, Docs}, NewCounters} = lists:keytake(Worker, 1, Counters),
+ DocsDict = append_purge_replies(Docs, Replies0, DocsDict0),
+ case {WCount, dict:size(DocsDict)} of
+ {1, _} ->
+ % last message has arrived, we need to conclude things
+ {Health, W, Replies} = dict:fold(fun force_reply/3, {ok, W, []},
+ DocsDict),
+ {stop, {Health, Replies}};
+ {_, DocCount} ->
+ % we've got at least one reply for each document, let's take a look
+ case dict:fold(fun maybe_reply/3, {stop,W,[]}, DocsDict) of
+ continue ->
+ {ok, {WCount - 1, DocCount, W, NewCounters, DocsDict}};
+ {stop, W, Replies} ->
+ {stop, {ok, Replies}}
+ end;
+ _ ->
+ {ok, {WCount - 1, DocCount, W, NewCounters, DocsDict}}
+ end;
+handle_message({error, purged_docs_limit_exceeded}=Error, Worker, Acc0) ->
+ {WC, DocCount , W, Counters, DocsDict0} = Acc0,
+ % fill DocsDict with error messages for relevant Docs
+ {value, {_W, Docs}, NewCounters} = lists:keytake(Worker, 1, Counters),
+ Replies = [Error || _D <- Docs],
+ DocsDict = append_purge_replies(Docs, Replies, DocsDict0),
+ skip_message({WC-1, DocCount, W, NewCounters, DocsDict});
+handle_message({bad_request, Msg}, _, _) ->
+ throw({bad_request, Msg}).
+
+
+tag_docs(AllIdsRevs) ->
+ {UUIDs, UUIDsIdsRevs, DocCount} = lists:foldl(fun(
+ {Id, Revs}, {UAcc, UIRAcc, C}) ->
+ UUID = couch_uuids:new(),
+ {[UUID|UAcc], [{UUID, Id, Revs}|UIRAcc], C+1}
+ end, {[], [], 0}, AllIdsRevs),
+ {lists:reverse(UUIDs), lists:reverse(UUIDsIdsRevs), DocCount}.
+
+
+force_reply(Doc, Replies, {Health, W, Acc}) ->
+ case update_quorum_met(W, Replies) of
+ {true, FinalReply} ->
+ {Health, W, [{Doc, FinalReply} | Acc]};
+ false ->
+ case [Reply || {ok, Reply} <- Replies] of
+ [] ->
+ UReplies = lists:usort(Replies),
+ case UReplies of
+ [{error, internal_server_error}] ->
+ {error, W, [{Doc, {error, internal_server_error}} | Acc]};
+ [{error, timeout}] ->
+ {error, W, [{Doc, {error, timeout}} | Acc]};
+ [FirstReply|[]] ->
+ % check if all errors are identical, if so inherit health
+ {Health, W, [{Doc, FirstReply} | Acc]};
+ _ ->
+ {error, W, [{Doc, UReplies} | Acc]}
+ end;
+ AcceptedReplies0 ->
+ NewHealth = case Health of ok -> accepted; _ -> Health end,
+ AcceptedReplies = lists:usort(lists:flatten(AcceptedReplies0)),
+ {NewHealth, W, [{Doc, {accepted, AcceptedReplies}} | Acc]}
+ end
+ end.
+
+
+maybe_reply(_, _, continue) ->
+ % we didn't meet quorum for all docs, so we're fast-forwarding the fold
+ continue;
+maybe_reply(Doc, Replies, {stop, W, Acc}) ->
+ case update_quorum_met(W, Replies) of
+ {true, Reply} ->
+ {stop, W, [{Doc, Reply} | Acc]};
+ false ->
+ continue
+ end.
+
+update_quorum_met(W, Replies) ->
+ OkReplies = lists:foldl(fun(Reply, PrevsAcc) ->
+ case Reply of
+ {ok, PurgedRevs} -> [PurgedRevs | PrevsAcc];
+ _ -> PrevsAcc
+ end
+ end, [], Replies),
+ if length(OkReplies) < W -> false; true ->
+ % make a union of PurgedRevs
+ FinalReply = {ok, lists:usort(lists:flatten(OkReplies))},
+ {true, FinalReply}
+ end.
+
+
+group_idrevs_by_shard(DbName, UUIDsIdsRevs) ->
+ lists:foldl(fun({_UUID, Id, _Revs} = UUIDIdRevs, D0) ->
+ lists:foldl(fun(Shard, D1) ->
+ dict:append(Shard, UUIDIdRevs, D1)
+ end, D0, mem3:shards(DbName, Id))
+ end, dict:new(), UUIDsIdsRevs).
+
+
+append_purge_replies([], [], DocReplyDict) ->
+ DocReplyDict;
+append_purge_replies([Doc|Rest1], [Reply|Rest2], Dict0) ->
+ append_purge_replies(Rest1, Rest2, dict:append(Doc, Reply, Dict0)).
+
+
+skip_message({0, _, W, _, DocsDict}) ->
+ {Health, W, Reply} = dict:fold(fun force_reply/3, {ok, W, []}, DocsDict),
+ {stop, {Health, Reply}};
+skip_message(Acc0) ->
+ {ok, Acc0}.
+
+
+% eunits
+doc_purge_ok_test() ->
+ meck:new(couch_log),
+ meck:expect(couch_log, warning, fun(_,_) -> ok end),
+ meck:expect(couch_log, notice, fun(_,_) -> ok end),
+
+ Revs1 = [{1, <<"rev11">>}], UUID1 = <<"3de03c5f4c2cd34cc515a9d1ea000abd">>,
+ UUIDIdRevs1 = {UUID1, <<"id1">>, Revs1},
+ Revs2 = [{1, <<"rev12">>}], UUID2 = <<"4de03c5f4c2cd34cc515a9d1ea000abc">>,
+ UUIDIdRevs2 = {UUID2, <<"id2">>, Revs2},
+ UUIDsIDdsRevs = [UUIDIdRevs1, UUIDIdRevs2],
+ Shards =
+ mem3_util:create_partition_map("foo",3,1,["node1","node2","node3"]),
+ Counters = dict:to_list(
+ group_idrevs_by_shard_hack(<<"foo">>, Shards, UUIDsIDdsRevs)),
+ DocsDict = dict:new(),
+
+ % ***test for W = 2
+ AccW2 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("2"),
+ Counters, DocsDict},
+ {ok, {WaitingCountW2_1,_,_,_,_} = AccW2_1} =
+ handle_message({ok,[{ok, Revs1}, {ok, Revs2}]}, hd(Shards), AccW2),
+ ?assertEqual(2, WaitingCountW2_1),
+ {stop, FinalReplyW2 } =
+ handle_message({ok, [{ok, Revs1}, {ok, Revs2}]},
+ lists:nth(2,Shards), AccW2_1),
+ ?assertEqual(
+ {ok, [{UUID1, {ok, Revs1}}, {UUID2, {ok, Revs2}}]},
+ FinalReplyW2
+ ),
+
+ % ***test for W = 3
+ AccW3 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("3"),
+ Counters, DocsDict},
+ {ok, {WaitingCountW3_1,_,_,_,_} = AccW3_1} =
+ handle_message({ok, [{ok, Revs1}, {ok, Revs2}]}, hd(Shards), AccW3),
+ ?assertEqual(2, WaitingCountW3_1),
+ {ok, {WaitingCountW3_2,_,_,_,_} = AccW3_2} =
+ handle_message({ok,[{ok, Revs1}, {ok, Revs2}]},
+ lists:nth(2,Shards), AccW3_1),
+ ?assertEqual(1, WaitingCountW3_2),
+ {stop, FinalReplyW3 } =
+ handle_message({ok, [{ok, Revs1}, {ok, Revs2}]},
+ lists:nth(3,Shards), AccW3_2),
+ ?assertEqual(
+ {ok, [{UUID1, {ok, Revs1}}, {UUID2, {ok, Revs2}}]},
+ FinalReplyW3
+ ),
+
+ % *** test rexi_exit on 1 node
+ Acc0 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("2"),
+ Counters, DocsDict},
+ {ok, {WaitingCount1,_,_,_,_} = Acc1} =
+ handle_message({ok, [{ok, Revs1}, {ok, Revs2}]}, hd(Shards), Acc0),
+ ?assertEqual(2, WaitingCount1),
+ {ok, {WaitingCount2,_,_,_,_} = Acc2} =
+ handle_message({rexi_EXIT, nil}, lists:nth(2,Shards), Acc1),
+ ?assertEqual(1, WaitingCount2),
+ {stop, Reply} =
+ handle_message({ok, [{ok, Revs1}, {ok, Revs2}]},
+ lists:nth(3,Shards), Acc2),
+ ?assertEqual(
+ {ok,[{UUID1, {ok, Revs1}}, {UUID2, {ok, Revs2}}]},
+ Reply
+ ),
+
+ % *** test {error, purge_during_compaction_exceeded_limit} on all nodes
+ % *** still should return ok reply for the request
+ ErrPDCEL = {error, purge_during_compaction_exceeded_limit},
+ Acc20 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("3"),
+ Counters, DocsDict},
+ {ok, {WaitingCount21,_,_,_,_} = Acc21} =
+ handle_message({ok, [ErrPDCEL, ErrPDCEL]}, hd(Shards), Acc20),
+ ?assertEqual(2, WaitingCount21),
+ {ok, {WaitingCount22,_,_,_,_} = Acc22} =
+ handle_message({ok, [ErrPDCEL, ErrPDCEL]}, lists:nth(2,Shards), Acc21),
+ ?assertEqual(1, WaitingCount22),
+ {stop, Reply2 } =
+ handle_message({ok, [ErrPDCEL, ErrPDCEL]}, lists:nth(3,Shards), Acc22),
+ ?assertEqual(
+ {ok, [{UUID1, ErrPDCEL}, {UUID2, ErrPDCEL}]},
+ Reply2
+ ),
+
+ % *** test {error, purged_docs_limit_exceeded} on all nodes
+ % *** still should return ok reply for the request
+ ErrPDLE = {error, purged_docs_limit_exceeded},
+ Acc30 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("3"),
+ Counters, DocsDict},
+ {ok, {WaitingCount31,_,_,_,_} = Acc31} =
+ handle_message({ok, [ErrPDLE, ErrPDLE]}, hd(Shards), Acc30),
+ ?assertEqual(2, WaitingCount31),
+ {ok, {WaitingCount32,_,_,_,_} = Acc32} =
+ handle_message({ok, [ErrPDLE, ErrPDLE]}, lists:nth(2,Shards), Acc31),
+ ?assertEqual(1, WaitingCount32),
+ {stop, Reply3 } =
+ handle_message({ok, [ErrPDLE, ErrPDLE]},lists:nth(3,Shards), Acc32),
+ ?assertEqual(
+ {ok, [{UUID1, ErrPDLE}, {UUID2, ErrPDLE}]},
+ Reply3
+ ),
+ meck:unload(couch_log).
+
+
+doc_purge_accepted_test() ->
+ meck:new(couch_log),
+ meck:expect(couch_log, warning, fun(_,_) -> ok end),
+ meck:expect(couch_log, notice, fun(_,_) -> ok end),
+
+ Revs1 = [{1, <<"rev11">>}], UUID1 = <<"3de03c5f4c2cd34cc515a9d1ea000abd">>,
+ UUIDIdRevs1 = {UUID1, <<"id1">>, Revs1},
+ Revs2 = [{1, <<"rev12">>}], UUID2 = <<"4de03c5f4c2cd34cc515a9d1ea000abc">>,
+ UUIDIdRevs2 = {UUID2, <<"id2">>, Revs2},
+ UUIDsIDdsRevs = [UUIDIdRevs1, UUIDIdRevs2],
+ Shards =
+ mem3_util:create_partition_map("foo",3,1,["node1","node2","node3"]),
+ Counters = dict:to_list(
+ group_idrevs_by_shard_hack(<<"foo">>, Shards, UUIDsIDdsRevs)),
+ DocsDict = dict:new(),
+
+ % *** test rexi_exit on 2 nodes
+ Acc0 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("2"),
+ Counters, DocsDict},
+ {ok, {WaitingCount1,_,_,_,_} = Acc1} =
+ handle_message({ok, [{ok, Revs1}, {ok, Revs2}]}, hd(Shards), Acc0),
+ ?assertEqual(2, WaitingCount1),
+ {ok, {WaitingCount2,_,_,_,_} = Acc2} =
+ handle_message({rexi_EXIT, nil}, lists:nth(2, Shards), Acc1),
+ ?assertEqual(1, WaitingCount2),
+ {stop, Reply} =
+ handle_message({rexi_EXIT, nil}, lists:nth(3, Shards), Acc2),
+ ?assertEqual(
+ {accepted, [{UUID1, {accepted, Revs1}}, {UUID2, {accepted, Revs2}}]},
+ Reply
+ ),
+ meck:unload(couch_log).
+
+
+doc_purge_error_test() ->
+ meck:new(couch_log),
+ meck:expect(couch_log, warning, fun(_,_) -> ok end),
+ meck:expect(couch_log, notice, fun(_,_) -> ok end),
+
+ Revs1 = [{1, <<"rev11">>}], UUID1 = <<"3de03c5f4c2cd34cc515a9d1ea000abd">>,
+ UUIDIdRevs1 = {UUID1, <<"id1">>, Revs1},
+ Revs2 = [{1, <<"rev12">>}], UUID2 = <<"4de03c5f4c2cd34cc515a9d1ea000abc">>,
+ UUIDIdRevs2 = {UUID2, <<"id2">>, Revs2},
+ UUIDsIDdsRevs = [UUIDIdRevs1, UUIDIdRevs2],
+ Shards =
+ mem3_util:create_partition_map("foo",3,1,["node1","node2","node3"]),
+ Counters = dict:to_list(
+ group_idrevs_by_shard_hack(<<"foo">>, Shards, UUIDsIDdsRevs)),
+ DocsDict = dict:new(),
+
+ % *** test rexi_exit on all 3 nodes
+ Acc0 = {length(Shards), length(UUIDsIDdsRevs), list_to_integer("2"),
+ Counters, DocsDict},
+ {ok, {WaitingCount1,_,_,_,_} = Acc1} =
+ handle_message({rexi_EXIT, nil}, hd(Shards), Acc0),
+ ?assertEqual(2, WaitingCount1),
+ {ok, {WaitingCount2,_,_,_,_} = Acc2} =
+ handle_message({rexi_EXIT, nil}, lists:nth(2,Shards), Acc1),
+ ?assertEqual(1, WaitingCount2),
+ {stop, Reply} =
+ handle_message({rexi_EXIT, nil}, lists:nth(3,Shards), Acc2),
+ ?assertEqual(
+ {error, [{UUID1, {error, internal_server_error}},
+ {UUID2, {error, internal_server_error}}]},
+ Reply
+ ),
+
+ % ***test w quorum > # shards, which should fail immediately
+ Shards2 = mem3_util:create_partition_map("foo",1,1,["node1"]),
+ Counters2 = dict:to_list(
+ group_idrevs_by_shard_hack(<<"foo">>, Shards2, UUIDsIDdsRevs)),
+ AccW4 = {length(Shards2), length(UUIDsIDdsRevs), list_to_integer("2"),
+ Counters2, DocsDict},
+ Bool =
+ case handle_message({ok, [{ok, Revs1}, {ok, Revs2}]},
+ hd(Shards), AccW4) of
+ {stop, _Reply} ->
+ true;
+ _ -> false
+ end,
+ ?assertEqual(true, Bool),
+
+ % *** test Docs with no replies should end up as {error, internal_server_error}
+ SA1 = #shard{node = a, range = [1]},
+ SA2 = #shard{node = a, range = [2]},
+ SB1 = #shard{node = b, range = [1]},
+ SB2 = #shard{node = b, range = [2]},
+ Counters3 = [{SA1,[UUID1]}, {SB1,[UUID1]},
+ {SA2,[UUID2]}, {SB2,[UUID2]}],
+ Acc30 = {length(Counters3), length(UUIDsIDdsRevs), 2, Counters3, DocsDict},
+ {ok, Acc31} = handle_message({ok, [{ok, Revs1}]}, SA1, Acc30),
+ {ok, Acc32} = handle_message({rexi_EXIT, nil}, SB1, Acc31),
+ {ok, Acc33} = handle_message({rexi_EXIT, nil}, SA2, Acc32),
+ {stop, Acc34} = handle_message({rexi_EXIT, nil}, SB2, Acc33),
+ ?assertEqual(
+ {error, [{UUID1, {accepted, Revs1}},
+ {UUID2, {error, internal_server_error}}]},
+ Acc34
+ ),
+ meck:unload(couch_log).
+
+
+% needed for testing to avoid having to start the mem3 application
+group_idrevs_by_shard_hack(_DbName, Shards, UUIDsIdsRevs) ->
+ lists:foldl(fun({UUID, _Id, _Revs}, Dict0) ->
+ lists:foldl(fun(Shard, Dict1) ->
+ dict:append(Shard, UUID, Dict1)
+ end, Dict0, Shards)
+ end, dict:new(), UUIDsIdsRevs).
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 4a69e7e..6e2c05f 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -21,6 +21,7 @@
delete_shard_db_doc/2]).
-export([get_all_security/2, open_shard/2]).
-export([compact/1, compact/2]).
+-export([get_purge_seq/2, purge_docs/3, set_purged_docs_limit/3]).
-export([get_db_info/2, get_doc_count/2, get_update_seq/2,
changes/4, map_view/5, reduce_view/5, group_info/3, update_mrview/4]).
@@ -201,6 +202,9 @@ get_all_security(DbName, Options) ->
set_revs_limit(DbName, Limit, Options) ->
with_db(DbName, Options, {couch_db, set_revs_limit, [Limit]}).
+set_purged_docs_limit(DbName, Limit, Options) ->
+ with_db(DbName, Options, {couch_db, set_purged_docs_limit, [Limit]}).
+
open_doc(DbName, DocId, Options) ->
with_db(DbName, Options, {couch_db, open_doc, [DocId, Options]}).
@@ -236,14 +240,31 @@ get_missing_revs(DbName, IdRevsList, Options) ->
end).
update_docs(DbName, Docs0, Options) ->
+ X = case proplists:get_value(replicated_changes, Options) of
+ true -> replicated_changes;
+ _ -> interactive_edit
+ end,
+ DocsByNode = couch_util:get_value(read_repair, Options),
+ case {X, DocsByNode} of
+ {_, undefined} ->
+ Docs = make_att_readers(Docs0),
+ with_db(DbName, Options,
+ {couch_db, update_docs, [Docs, Options, X]});
+ {replicated_changes, _} ->
+ update_docs_read_repair(DbName, DocsByNode, Options)
+ end.
+
+get_purge_seq(DbName, Options) ->
+ with_db(DbName, Options, {couch_db, get_purge_seq, []}).
+
+purge_docs(DbName, UUIdsIdsRevs, Options) ->
case proplists:get_value(replicated_changes, Options) of
- true ->
- X = replicated_changes;
- _ ->
- X = interactive_edit
+ true ->
+ X = replicated_changes;
+ _ ->
+ X = interactive_edit
end,
- Docs = make_att_readers(Docs0),
- with_db(DbName, Options, {couch_db, update_docs, [Docs, Options, X]}).
+ with_db(DbName, Options, {couch_db, purge_docs, [UUIdsIdsRevs, X]}).
%% @equiv group_info(DbName, DDocId, [])
group_info(DbName, DDocId) ->
@@ -298,6 +319,75 @@ with_db(DbName, Options, {M,F,A}) ->
rexi:reply(Error)
end.
+
+update_docs_read_repair(DbName, DocsByNode, Options) ->
+ set_io_priority(DbName, Options),
+ case get_or_create_db(DbName, Options) of
+ {ok, Db} ->
+ % omit Revisions that have been purged
+ Docs = filter_purged_revs(Db, DocsByNode),
+ Docs2 = make_att_readers(Docs),
+ {M,F,A} = {couch_db, update_docs, [Docs2, Options, replicated_changes]},
+ rexi:reply(try
+ apply(M, F, [Db | A])
+ catch Exception ->
+ Exception;
+ error:Reason ->
+ couch_log:error("rpc ~p:~p/~p ~p ~p", [M, F, length(A)+1, Reason,
+ clean_stack()]),
+ {error, Reason}
+ end);
+ Error ->
+ rexi:reply(Error)
+ end.
+
+
+% given [{Node, Doc}] diff revs of the same DocID from diff nodes
+% returns [Doc] filtering out purged docs.
+% This is done for read-repair from fabric_doc_open,
+% so that not to recreate Docs that have been purged before
+% on this node() from Nodes that are out of sync.
+filter_purged_revs(Db, DocsByNode) ->
+ AllowedPSeqLag = config:get_integer("purge", "allowed_purge_seq_lag", 100),
+ {ok, DbPSeq} = couch_db:get_purge_seq(Db),
+ PurgeFoldFun = fun({_P,_U, Id, Revs}, Acc) -> [{Id, Revs}|Acc] end,
+ V = "v" ++ config:get("purge", "version", "1") ++ "-",
+ StartKey = ?l2b(?LOCAL_DOC_PREFIX ++ "purge-" ++ V ++ "mem3-"),
+ EndKey = ?l2b(?LOCAL_DOC_PREFIX ++ "purge-" ++ V ++ "mem31"),
+ Opts = [{start_key, StartKey}, {end_key_gt, EndKey}],
+ % go through _local/purge-mem3-.. docs
+ % find Node that this LDoc corresponds to
+ % check if update from Node has not been recently purged on current node
+ LDocsFoldFun = fun(#doc{body={Props}}, Acc) ->
+ {VOps} = couch_util:get_value(<<"verify_options">>, Props),
+ Node = couch_util:get_value(<<"node">>, VOps),
+ Result = lists:keyfind(Node, 1, DocsByNode),
+ NewAcc = if not Result -> Acc; true ->
+ {Node, Doc} = Result,
+ NodePSeq = couch_util:get_value(<<"purge_seq">>, Props),
+ if NodePSeq == DbPSeq ->
+ [Doc|Acc];
+ (NodePSeq+AllowedPSeqLag) < DbPSeq ->
+ % Node is very out of sync, ignore updates from it
+ Acc;
+ true -> %(NodePSeq+ClientAllowedPSeqLag) >= DbPSeq
+ % if Doc has been purged recently, than ignore it
+ {ok, PurgedIdsRevs} = couch_db:fold_purged_docs(Db,
+ NodePSeq, PurgeFoldFun, [], []),
+ {Start, [FirstRevId|_]} = Doc#doc.revs,
+ DocIdRevs = {Doc#doc.id, [{Start, FirstRevId}]},
+ case lists:member(DocIdRevs, PurgedIdsRevs) of
+ true -> Acc;
+ false -> [Doc|Acc]
+ end
+ end
+ end,
+ {ok, NewAcc}
+ end,
+ {ok, Docs} = couch_db_engine:fold_local_docs(Db, LDocsFoldFun, [], Opts),
+ Docs.
+
+
get_or_create_db(DbName, Options) ->
couch_db:open_int(DbName, [{create_if_missing, true} | Options]).
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 08/20: WIP - couch_db.erl
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 334af0165dc864b8aa235145cadee4fd465b0981
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Fri Mar 16 14:55:51 2018 -0500
WIP - couch_db.erl
---
src/couch/src/couch_db.erl | 105 ++++++++++++++++++++++++++++++++++++++++++---
1 file changed, 99 insertions(+), 6 deletions(-)
diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index e27f632..3ef6ab0 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -50,6 +50,7 @@
get_user_ctx/1,
get_uuid/1,
get_purge_seq/1,
+ get_oldest_purge_seq/1,
get_purge_infos_limit/1,
is_db/1,
@@ -76,7 +77,9 @@
get_full_doc_infos/2,
get_missing_revs/2,
get_design_docs/1,
- load_purge_infos/2,
+ get_purge_infos/2,
+
+ get_minimum_purge_seq/1,
update_doc/3,
update_doc/4,
@@ -381,15 +384,107 @@ get_full_doc_infos(Db, Ids) ->
purge_docs(#db{main_pid = Pid}, UUIdsIdsRevs) ->
gen_server:call(Pid, {purge_docs, UUIdsIdsRevs});
--spec load_purge_infos(#db{}, [UUId]) -> [PurgeInfo] when
+-spec get_purge_infos(#db{}, [UUId]) -> [PurgeInfo] when
UUId :: binary(),
PurgeInfo :: {PurgeSeq, UUId, Id, [Rev]} | not_found,
Id :: binary(),
Rev :: {non_neg_integer(), binary()}.
-load_purge_infos(Db, UUIDs) ->
+get_purge_infos(Db, UUIDs) ->
couch_db_engine:load_purge_infos(Db, UUIDs).
+get_minimum_purge_seq(#db{} = Db) ->
+ PurgeSeq = couch_db:get_purge_seq(Db),
+ OldestPurgeSeq = couch_db:get_oldest_purge_seq(Db),
+ PurgeInfosLimit = couch_db:get_purge_infos_limit(Db),
+
+ FoldFun = fun(#doc{id = DocId, body = {Props}}, SeqAcc) ->
+ ClientSeq = couch_util:get_value(<<"purge_seq">>, Props),
+ case ClientSeq of
+ CS when is_integer(CS), CS >= PurgeSeq - PurgeInfosLimit ->
+ {ok, SeqAcc};
+ CS when is_integer(CS) ->
+ case purge_client_exists(Db, DocId, Props) of
+ true -> {ok, erlang:min(CS, SeqAcc)};
+ false -> {ok, SeqAcc}
+ end;
+ _ ->
+ % If there's a broken doc we have to keep every
+ % purge info until the doc is fixed or removed.
+ Fmt = "Invalid purge doc '~s' with purge_seq '~w'",
+ couch_log:error(Fmt, [DocId, ClientSeq]),
+ {ok, erlang:min(OldestPurgeSeq, SeqAcc)}
+ end
+ end,
+ InitMinSeq = PurgeSeq - PurgeInfosLimit,
+ Opts = [
+ {start_key, list_to_binary(?LOCAL_DOC_PREFIX + "purge-")},
+ {end_key_gt, list_to_binary(?LOCAL_DOC_PREFIX + "purge.")}
+ ],
+ {ok, MinIdxSeq} = couch_db:fold_local_docs(Db, FoldFun, InitSeq, Opts),
+ FinalSeq = case MinIdxSeq < PurgeSeq - PurgeInfosLimit of
+ true -> MinIdxSeq;
+ false -> erlang:max(0, PurgeSeq - PurgeInfosLimit)
+ end,
+ % Log a warning if we've got a purge sequence exceeding the
+ % configured threshold.
+ if FinalSeq < (PurgeSeq - PurgeInfosLimit) ->
+ Fmt = "The purge sequence for '~s' exceeds configured threshold",
+ couch_log:warning(Fmt, [couch_db:name(Db)])
+ end,
+ FinalSeq.
+
+
+purge_client_exists(DbName, DocID, Props) ->
+ % Warn about clients that have not updated their purge
+ % checkpoints in the last "index_lag_warn_seconds"
+ LagWindow = config:get_integer(
+ "purge", "index_lag_warn_seconds", 86400), % Default 24 hours
+
+ {Mega, Secs, _} = os:timestamp(),
+ NowSecs = Mega * 1000000 + Secs,
+ LagThreshold = NowSecs - LagWindow,
+
+ try
+ CheckFun = get_purge_client_fun(Props),
+ Exists = CheckFun(DbName, DocId, Props),
+ if not Exists -> ok; true ->
+ Updated = couch_util:get_value(<<"updated_on">>, Props),
+ if is_integer(Updated) and Updated > LagThreshold -> ok; true ->
+ Diff = NowSecs - LU,
+ Fmt = "Purge checkpint '~s' not updated in ~p seconds",
+ couch_log:error(Fmt, [DocId, NowSecs - LU])
+ end
+ end,
+ Exists
+ catch _:_ ->
+ % If we fail to check for a client we have to assume that
+ % it exists.
+ true
+ end.
+
+
+get_purge_client_fun(DocId, Props) -
+ M0 = couch_util:get_value(<<"verify_module">>, Props),
+ try
+ M = binary_to_existing_atom(M0, latin1)
+ catch error:badarg ->
+ Fmt = "Missing index module '~s' for purge checkpoint '~s'",
+ couch_log:error(Fmt, [M0, DocId]),
+ throw(failed)
+ end,
+
+ F0 = couch_util:get_value(<<"verify_function">>, Props),
+ try
+ F = binary_to_existing_atom(F0, latin1),
+ fun M:F/2
+ catch error:badarg ->
+ Fmt = "Missing function '~s' in '~s' for purge checkpoint '~s'",
+ couch_log:error(Fmt, [F0, M0, DocId]),
+ throw(failed)
+ end.
+
+
set_purge_infos_limit(#db{main_pid=Pid}=Db, Limit) when Limit > 0 ->
check_is_admin(Db),
gen_server:call(Pid, {set_purge_infos_limit, Limit}, infinity);
@@ -418,9 +513,7 @@ get_purge_seq(#db{}=Db) ->
{ok, couch_db_engine:get_purge_seq(Db)}.
get_oldest_purge_seq(#db{}=Db) ->
- {ok, StartSeq} = get_purge_seq(Db),
- FoldFun = fun({_UUId, PurgeSeq, _, _}, _) -> {stop, PurgeSeq} end,
- fold_purge_infos(Db, StartSeq, FoldFun, StartSeq).
+ {ok, couch_db_engine:get_oldest_purge_seq(Db)}.
get_purge_infos_limit(#db{}=Db) ->
couch_db_engine:get_purge_infos_limit(Db).
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 20/20: Temporarily disable
should_compare_compression_methods/1
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 98be828165142c882845f9d5db2a60c2be6826b3
Author: jiangphcn <ji...@cn.ibm.com>
AuthorDate: Tue May 16 09:50:19 2017 +0800
Temporarily disable should_compare_compression_methods/1
COUCHDB-3326
---
src/couch/test/couchdb_file_compression_tests.erl | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/src/couch/test/couchdb_file_compression_tests.erl b/src/couch/test/couchdb_file_compression_tests.erl
index 8f0fe5b..e5ef74c 100644
--- a/src/couch/test/couchdb_file_compression_tests.erl
+++ b/src/couch/test/couchdb_file_compression_tests.erl
@@ -57,8 +57,8 @@ couch_file_compression_test_() ->
fun should_use_none/1,
fun should_use_deflate_1/1,
fun should_use_deflate_9/1,
- fun should_use_snappy/1,
- fun should_compare_compression_methods/1
+ fun should_use_snappy/1
+ %fun should_compare_compression_methods/1
]
}
}
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 16/20: Update fabric_doc_open_revs to handle purges
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit da8f86ccc6c8219ab111fb361a9ef8ee61b3db7b
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Tue Oct 3 15:56:55 2017 -0400
Update fabric_doc_open_revs to handle purges
We need to account for the possibility that a document is opened while a
purge is still propogating between shards. This means we need to inform
the read-repair algorithms that a difference in revisions may be due to
a purge request in progress. If we don't do this its possible that a
read-repair may race the purge request and effectively undo the purge.
---
src/fabric/src/fabric_doc_open_revs.erl | 148 ++++++++++++++++++++++----------
src/fabric/src/fabric_rpc.erl | 67 ++++++++-------
2 files changed, 137 insertions(+), 78 deletions(-)
diff --git a/src/fabric/src/fabric_doc_open_revs.erl b/src/fabric/src/fabric_doc_open_revs.erl
index 096722f..dbe02bf 100644
--- a/src/fabric/src/fabric_doc_open_revs.erl
+++ b/src/fabric/src/fabric_doc_open_revs.erl
@@ -29,7 +29,8 @@
revs,
latest,
replies = [],
- repair = false
+ repair = false,
+ replies_by_node=[] %[{Node, Reply}] used for read_repair
}).
go(DbName, Id, Revs, Options) ->
@@ -82,6 +83,7 @@ handle_message({ok, RawReplies}, Worker, State) ->
worker_count = WorkerCount,
workers = Workers,
replies = PrevReplies,
+ replies_by_node = PrevNReplies,
r = R,
revs = Revs,
latest = Latest,
@@ -92,13 +94,14 @@ handle_message({ok, RawReplies}, Worker, State) ->
IsTree = Revs == all orelse Latest,
% Do not count error replies when checking quorum
-
RealReplyCount = ReplyCount + 1 - ReplyErrorCount,
QuorumReplies = RealReplyCount >= R,
{NewReplies, QuorumMet, Repair} = case IsTree of
true ->
- {NewReplies0, AllInternal, Repair0} =
+ {NewReplies0, AllInternal, Repair00} =
tree_replies(PrevReplies, tree_sort(RawReplies)),
+ % don't set Repair=true on the first reply
+ Repair0 = (ReplyCount > 0) and Repair00,
NumLeafs = couch_key_tree:count_leafs(PrevReplies),
SameNumRevs = length(RawReplies) == NumLeafs,
QMet = AllInternal andalso SameNumRevs andalso QuorumReplies,
@@ -107,6 +110,10 @@ handle_message({ok, RawReplies}, Worker, State) ->
{NewReplies0, MinCount} = dict_replies(PrevReplies, RawReplies),
{NewReplies0, MinCount >= R, false}
end,
+ NewNReplies = case Worker of
+ nil -> PrevNReplies;
+ _ -> [{Worker#shard.node, RawReplies}|PrevNReplies]
+ end,
Complete = (ReplyCount =:= (WorkerCount - 1)),
@@ -117,6 +124,7 @@ handle_message({ok, RawReplies}, Worker, State) ->
DbName,
IsTree,
NewReplies,
+ NewNReplies,
ReplyCount + 1,
InRepair orelse Repair
),
@@ -124,6 +132,7 @@ handle_message({ok, RawReplies}, Worker, State) ->
false ->
{ok, State#state{
replies = NewReplies,
+ replies_by_node = NewNReplies,
reply_count = ReplyCount + 1,
workers = lists:delete(Worker, Workers),
repair = InRepair orelse Repair
@@ -180,7 +189,7 @@ dict_replies(Dict, [Reply | Rest]) ->
dict_replies(NewDict, Rest).
-maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
+maybe_read_repair(Db, IsTree, Replies, NodeReplies0, ReplyCount, DoRepair) ->
Docs = case IsTree of
true -> tree_repair_docs(Replies, DoRepair);
false -> dict_repair_docs(Replies, ReplyCount)
@@ -189,7 +198,11 @@ maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
[] ->
ok;
_ ->
- erlang:spawn(fun() -> read_repair(Db, Docs) end)
+ NodeReplies = lists:foldl(fun({Node, NReplies}, Acc) ->
+ NewAcc = [{Node, Doc} || {ok, Doc} <- NReplies],
+ NewAcc ++ Acc
+ end, [], NodeReplies0),
+ erlang:spawn(fun() -> read_repair(Db, Docs, NodeReplies) end)
end.
@@ -208,8 +221,9 @@ dict_repair_docs(Replies, ReplyCount) ->
end.
-read_repair(Db, Docs) ->
- Res = fabric:update_docs(Db, Docs, [replicated_changes, ?ADMIN_CTX]),
+read_repair(Db, Docs, NodeReplies) ->
+ Opts = [replicated_changes, ?ADMIN_CTX, {read_repair, NodeReplies}],
+ Res = fabric:update_docs(Db, Docs, Opts),
case Res of
{ok, []} ->
couch_stats:increment_counter([fabric, read_repairs, success]);
@@ -268,20 +282,24 @@ filter_reply(Replies) ->
setup() ->
config:start_link([]),
meck:new([fabric, couch_stats, couch_log]),
+ meck:new(fabric_util, [passthrough]),
meck:expect(fabric, update_docs, fun(_, _, _) -> {ok, nil} end),
meck:expect(couch_stats, increment_counter, fun(_) -> ok end),
- meck:expect(couch_log, notice, fun(_, _) -> ok end).
+ meck:expect(couch_log, notice, fun(_, _) -> ok end),
+ meck:expect(fabric_util, cleanup, fun(_) -> ok end).
+
teardown(_) ->
- (catch meck:unload([fabric, couch_stats, couch_log])),
+ (catch meck:unload([fabric, couch_stats, couch_log, fabric_util])),
config:stop().
state0(Revs, Latest) ->
#state{
worker_count = 3,
- workers = [w1, w2, w3],
+ workers =
+ [#shard{node='node1'}, #shard{node='node2'}, #shard{node='node3'}],
r = 2,
revs = Revs,
latest = Latest
@@ -334,27 +352,35 @@ open_doc_revs_test_() ->
check_empty_response_not_quorum() ->
% Simple smoke test that we don't think we're
% done with a first empty response
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
?_assertMatch(
- {ok, #state{workers = [w2, w3]}},
- handle_message({ok, []}, w1, state0(all, false))
+ {ok, #state{workers = [W2, W3]}},
+ handle_message({ok, []}, W1, state0(all, false))
).
check_basic_response() ->
% Check that we've handle a response
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
?_assertMatch(
- {ok, #state{reply_count = 1, workers = [w2, w3]}},
- handle_message({ok, [foo1(), bar1()]}, w1, state0(all, false))
+ {ok, #state{reply_count = 1, workers = [W2, W3]}},
+ handle_message({ok, [foo1(), bar1()]}, W1, state0(all, false))
).
check_finish_quorum() ->
% Two messages with the same revisions means we're done
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(all, false),
- {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, W1, S0),
Expect = {stop, [bar1(), foo1()]},
- ?assertEqual(Expect, handle_message({ok, [foo1(), bar1()]}, w2, S1))
+ ?assertEqual(Expect, handle_message({ok, [foo1(), bar1()]}, W2, S1))
end).
@@ -363,11 +389,13 @@ check_finish_quorum_newer() ->
% foo1 should count for foo2 which means we're finished.
% We also validate that read_repair was triggered.
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(all, false),
- {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, W1, S0),
Expect = {stop, [bar1(), foo2()]},
ok = meck:reset(fabric),
- ?assertEqual(Expect, handle_message({ok, [foo2(), bar1()]}, w2, S1)),
+ ?assertEqual(Expect, handle_message({ok, [foo2(), bar1()]}, W2, S1)),
ok = meck:wait(fabric, update_docs, '_', 5000),
?assertMatch(
[{_, {fabric, update_docs, [_, _, _]}, _}],
@@ -380,11 +408,14 @@ check_no_quorum_on_second() ->
% Quorum not yet met for the foo revision so we
% would wait for w3
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(all, false),
- {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, W1, S0),
?assertMatch(
- {ok, #state{workers = [w3]}},
- handle_message({ok, [bar1()]}, w2, S1)
+ {ok, #state{workers = [W3]}},
+ handle_message({ok, [bar1()]}, W2, S1)
)
end).
@@ -394,11 +425,14 @@ check_done_on_third() ->
% what. Every revision seen in this pattern should be
% included.
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(all, false),
- {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
- {ok, S2} = handle_message({ok, [bar1()]}, w2, S1),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, W1, S0),
+ {ok, S2} = handle_message({ok, [bar1()]}, W2, S1),
Expect = {stop, [bar1(), foo1()]},
- ?assertEqual(Expect, handle_message({ok, [bar1()]}, w3, S2))
+ ?assertEqual(Expect, handle_message({ok, [bar1()]}, W3, S2))
end).
@@ -407,108 +441,128 @@ check_done_on_third() ->
check_specific_revs_first_msg() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(revs(), false),
?assertMatch(
- {ok, #state{reply_count = 1, workers = [w2, w3]}},
- handle_message({ok, [foo1(), bar1(), bazNF()]}, w1, S0)
+ {ok, #state{reply_count = 1, workers = [W2, W3]}},
+ handle_message({ok, [foo1(), bar1(), bazNF()]}, W1, S0)
)
end).
check_revs_done_on_agreement() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(revs(), false),
Msg = {ok, [foo1(), bar1(), bazNF()]},
- {ok, S1} = handle_message(Msg, w1, S0),
+ {ok, S1} = handle_message(Msg, W1, S0),
Expect = {stop, [bar1(), foo1(), bazNF()]},
- ?assertEqual(Expect, handle_message(Msg, w2, S1))
+ ?assertEqual(Expect, handle_message(Msg, W2, S1))
end).
check_latest_true() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(revs(), true),
Msg1 = {ok, [foo2(), bar1(), bazNF()]},
Msg2 = {ok, [foo2(), bar1(), bazNF()]},
- {ok, S1} = handle_message(Msg1, w1, S0),
+ {ok, S1} = handle_message(Msg1, W1, S0),
Expect = {stop, [bar1(), foo2(), bazNF()]},
- ?assertEqual(Expect, handle_message(Msg2, w2, S1))
+ ?assertEqual(Expect, handle_message(Msg2, W2, S1))
end).
check_ancestor_counted_in_quorum() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(revs(), true),
Msg1 = {ok, [foo1(), bar1(), bazNF()]},
Msg2 = {ok, [foo2(), bar1(), bazNF()]},
Expect = {stop, [bar1(), foo2(), bazNF()]},
% Older first
- {ok, S1} = handle_message(Msg1, w1, S0),
- ?assertEqual(Expect, handle_message(Msg2, w2, S1)),
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ ?assertEqual(Expect, handle_message(Msg2, W2, S1)),
% Newer first
- {ok, S2} = handle_message(Msg2, w2, S0),
- ?assertEqual(Expect, handle_message(Msg1, w1, S2))
+ {ok, S2} = handle_message(Msg2, W2, S0),
+ ?assertEqual(Expect, handle_message(Msg1, W1, S2))
end).
check_not_found_counts_for_descendant() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
S0 = state0(revs(), true),
Msg1 = {ok, [foo1(), bar1(), bazNF()]},
Msg2 = {ok, [foo1(), bar1(), baz1()]},
Expect = {stop, [bar1(), baz1(), foo1()]},
% not_found first
- {ok, S1} = handle_message(Msg1, w1, S0),
- ?assertEqual(Expect, handle_message(Msg2, w2, S1)),
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ ?assertEqual(Expect, handle_message(Msg2, W2, S1)),
% not_found second
- {ok, S2} = handle_message(Msg2, w2, S0),
- ?assertEqual(Expect, handle_message(Msg1, w1, S2))
+ {ok, S2} = handle_message(Msg2, W2, S0),
+ ?assertEqual(Expect, handle_message(Msg1, W1, S2))
end).
check_worker_error_skipped() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(revs(), true),
Msg1 = {ok, [foo1(), bar1(), baz1()]},
Msg2 = {rexi_EXIT, reason},
Msg3 = {ok, [foo1(), bar1(), baz1()]},
Expect = {stop, [bar1(), baz1(), foo1()]},
- {ok, S1} = handle_message(Msg1, w1, S0),
- {ok, S2} = handle_message(Msg2, w2, S1),
- ?assertEqual(Expect, handle_message(Msg3, w3, S2))
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ {ok, S2} = handle_message(Msg2, W2, S1),
+ ?assertEqual(Expect, handle_message(Msg3, W3, S2))
end).
check_quorum_only_counts_valid_responses() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(revs(), true),
Msg1 = {rexi_EXIT, reason},
Msg2 = {rexi_EXIT, reason},
Msg3 = {ok, [foo1(), bar1(), baz1()]},
Expect = {stop, [bar1(), baz1(), foo1()]},
- {ok, S1} = handle_message(Msg1, w1, S0),
- {ok, S2} = handle_message(Msg2, w2, S1),
- ?assertEqual(Expect, handle_message(Msg3, w3, S2))
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ {ok, S2} = handle_message(Msg2, W2, S1),
+ ?assertEqual(Expect, handle_message(Msg3, W3, S2))
end).
check_empty_list_when_no_workers_reply() ->
?_test(begin
+ W1 = #shard{node='node1'},
+ W2 = #shard{node='node2'},
+ W3 = #shard{node='node3'},
S0 = state0(revs(), true),
Msg1 = {rexi_EXIT, reason},
Msg2 = {rexi_EXIT, reason},
Msg3 = {rexi_DOWN, nodedown, {nil, node()}, nil},
Expect = {stop, all_workers_died},
- {ok, S1} = handle_message(Msg1, w1, S0),
- {ok, S2} = handle_message(Msg2, w2, S1),
- ?assertEqual(Expect, handle_message(Msg3, w3, S2))
+ {ok, S1} = handle_message(Msg1, W1, S0),
+ {ok, S2} = handle_message(Msg2, W2, S1),
+ ?assertEqual(Expect, handle_message(Msg3, W3, S2))
end).
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 6e2c05f..2c4d5f4 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -344,48 +344,53 @@ update_docs_read_repair(DbName, DocsByNode, Options) ->
% given [{Node, Doc}] diff revs of the same DocID from diff nodes
% returns [Doc] filtering out purged docs.
-% This is done for read-repair from fabric_doc_open,
+% This is done for read-repair from fabric_doc_open or fabric_doc_open_revs,
% so that not to recreate Docs that have been purged before
% on this node() from Nodes that are out of sync.
filter_purged_revs(Db, DocsByNode) ->
- AllowedPSeqLag = config:get_integer("purge", "allowed_purge_seq_lag", 100),
- {ok, DbPSeq} = couch_db:get_purge_seq(Db),
- PurgeFoldFun = fun({_P,_U, Id, Revs}, Acc) -> [{Id, Revs}|Acc] end,
+ % go through _local/purge-mem3-.. docs
+ % and assemble NodePSeqs = [{Node1, NodePSeq1}, ...]
+ % NodePSeq1 - purge_seq of this node known to Node1
V = "v" ++ config:get("purge", "version", "1") ++ "-",
StartKey = ?l2b(?LOCAL_DOC_PREFIX ++ "purge-" ++ V ++ "mem3-"),
EndKey = ?l2b(?LOCAL_DOC_PREFIX ++ "purge-" ++ V ++ "mem31"),
Opts = [{start_key, StartKey}, {end_key_gt, EndKey}],
- % go through _local/purge-mem3-.. docs
- % find Node that this LDoc corresponds to
- % check if update from Node has not been recently purged on current node
LDocsFoldFun = fun(#doc{body={Props}}, Acc) ->
{VOps} = couch_util:get_value(<<"verify_options">>, Props),
Node = couch_util:get_value(<<"node">>, VOps),
- Result = lists:keyfind(Node, 1, DocsByNode),
- NewAcc = if not Result -> Acc; true ->
- {Node, Doc} = Result,
- NodePSeq = couch_util:get_value(<<"purge_seq">>, Props),
- if NodePSeq == DbPSeq ->
- [Doc|Acc];
- (NodePSeq+AllowedPSeqLag) < DbPSeq ->
- % Node is very out of sync, ignore updates from it
- Acc;
- true -> %(NodePSeq+ClientAllowedPSeqLag) >= DbPSeq
- % if Doc has been purged recently, than ignore it
- {ok, PurgedIdsRevs} = couch_db:fold_purged_docs(Db,
- NodePSeq, PurgeFoldFun, [], []),
- {Start, [FirstRevId|_]} = Doc#doc.revs,
- DocIdRevs = {Doc#doc.id, [{Start, FirstRevId}]},
- case lists:member(DocIdRevs, PurgedIdsRevs) of
- true -> Acc;
- false -> [Doc|Acc]
- end
- end
- end,
- {ok, NewAcc}
+ NodePSeq = couch_util:get_value(<<"purge_seq">>, Props),
+ {ok, [{Node, NodePSeq} | Acc]}
end,
- {ok, Docs} = couch_db_engine:fold_local_docs(Db, LDocsFoldFun, [], Opts),
- Docs.
+ {ok, NodePSeqs} =
+ couch_db_engine:fold_local_docs(Db, LDocsFoldFun, [], Opts),
+
+ % go through all doc_updates and
+ % filter out updates from nodes that are behind in purges synchronization
+ AllowedPSeqLag = config:get_integer("purge", "allowed_purge_seq_lag", 100),
+ {ok, DbPSeq} = couch_db:get_purge_seq(Db),
+ PurgeFoldFun = fun({_P,_U, Id, Revs}, Acc) -> [{Id, Revs}|Acc] end,
+ lists:foldl(fun({Node, Doc}, Docs) ->
+ NodePSeq = case lists:keyfind(Node, 1, NodePSeqs) of
+ {Node, NodePSeq0} -> NodePSeq0;
+ false -> 0
+ end,
+ if NodePSeq == DbPSeq ->
+ [Doc|Docs];
+ (NodePSeq+AllowedPSeqLag) < DbPSeq ->
+ % Node is very out of sync, ignore updates from it
+ Docs;
+ true -> %(NodePSeq+ClientAllowedPSeqLag) >= DbPSeq
+ % if Doc has been purged recently -> ignore it
+ {ok, PurgedIdsRevs} = couch_db:fold_purged_docs(Db,
+ NodePSeq, PurgeFoldFun, [], []),
+ {Start, [FirstRevId|_]} = Doc#doc.revs,
+ DocIdRevs = {Doc#doc.id, [{Start, FirstRevId}]},
+ case lists:member(DocIdRevs, PurgedIdsRevs) of
+ true -> Docs;
+ false -> [Doc|Docs]
+ end
+ end
+ end, [], DocsByNode).
get_or_create_db(DbName, Options) ->
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.
[couchdb] 03/20: WIP - couch_db_updater
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 63d39c2ef66f83adc96f5c3133f1893014227a33
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Wed Mar 14 13:17:58 2018 -0500
WIP - couch_db_updater
---
src/couch/src/couch_db_updater.erl | 243 ++++++++++++++++++++++++-------------
1 file changed, 162 insertions(+), 81 deletions(-)
diff --git a/src/couch/src/couch_db_updater.erl b/src/couch/src/couch_db_updater.erl
index 79567e9..ce0d45f 100644
--- a/src/couch/src/couch_db_updater.erl
+++ b/src/couch/src/couch_db_updater.erl
@@ -93,79 +93,37 @@ handle_call({set_revs_limit, Limit}, _From, Db) ->
ok = gen_server:call(couch_server, {db_updated, Db2}, infinity),
{reply, ok, Db2, idle_limit()};
-handle_call({purge_docs, _IdRevs}, _From,
- #db{compactor_pid=Pid}=Db) when Pid /= nil ->
- {reply, {error, purge_during_compaction}, Db, idle_limit()};
-handle_call({purge_docs, IdRevs}, _From, Db) ->
- DocIds = [Id || {Id, _Revs} <- IdRevs],
- OldDocInfos = couch_db_engine:open_docs(Db, DocIds),
-
- NewDocInfos = lists:flatmap(fun
- ({{Id, Revs}, #full_doc_info{id = Id, rev_tree = Tree} = FDI}) ->
- case couch_key_tree:remove_leafs(Tree, Revs) of
- {_, [] = _RemovedRevs} -> % no change
- [];
- {NewTree, RemovedRevs} ->
- NewFDI = FDI#full_doc_info{rev_tree = NewTree},
- [{FDI, NewFDI, RemovedRevs}]
- end;
- ({_, not_found}) ->
- []
- end, lists:zip(IdRevs, OldDocInfos)),
-
- InitUpdateSeq = couch_db_engine:get_update_seq(Db),
- InitAcc = {InitUpdateSeq, [], []},
- FinalAcc = lists:foldl(fun({_, #full_doc_info{} = OldFDI, RemRevs}, Acc) ->
- #full_doc_info{
- id = Id,
- rev_tree = OldTree
- } = OldFDI,
- {SeqAcc0, FDIAcc, IdRevsAcc} = Acc,
-
- {NewFDIAcc, NewSeqAcc} = case OldTree of
- [] ->
- % If we purged every #leaf{} in the doc record
- % then we're removing it completely from the
- % database.
- FDIAcc;
- _ ->
- % Its possible to purge the #leaf{} that contains
- % the update_seq where this doc sits in the update_seq
- % sequence. Rather than do a bunch of complicated checks
- % we just re-label every #leaf{} and reinsert it into
- % the update_seq sequence.
- {NewTree, SeqAcc1} = couch_key_tree:mapfold(fun
- (_RevId, Leaf, leaf, InnerSeqAcc) ->
- {Leaf#leaf{seq = InnerSeqAcc + 1}, InnerSeqAcc + 1};
- (_RevId, Value, _Type, InnerSeqAcc) ->
- {Value, InnerSeqAcc}
- end, SeqAcc0, OldTree),
-
- NewFDI = OldFDI#full_doc_info{
- update_seq = SeqAcc1,
- rev_tree = NewTree
- },
-
- {[NewFDI | FDIAcc], SeqAcc1}
- end,
- NewIdRevsAcc = [{Id, RemRevs} | IdRevsAcc],
- {NewSeqAcc, NewFDIAcc, NewIdRevsAcc}
- end, InitAcc, NewDocInfos),
-
- {_FinalSeq, FDIs, PurgedIdRevs} = FinalAcc,
-
- % We need to only use the list of #full_doc_info{} records
- % that we have actually changed due to a purge.
- PreviousFDIs = [PrevFDI || {PrevFDI, _, _} <- NewDocInfos],
- Pairs = pair_purge_info(PreviousFDIs, FDIs),
-
- {ok, Db2} = couch_db_engine:write_doc_infos(Db, Pairs, [], PurgedIdRevs),
-
+handle_call({set_purge_infos_limit, Limit}, _From, Db) ->
+ {ok, Db2} = couch_db_engine:set_purge_infos_limit(Db, Limit),
ok = gen_server:call(couch_server, {db_updated, Db2}, infinity),
- couch_event:notify(Db#db.name, updated),
+ {reply, ok, Db2};
+
+handle_call({purge_docs, PurgeReqs0}, _From, Db) ->
+ % First filter out any purge requests we've already
+ % processed.
+ UUIDs = [UUID || {UUID, _Id, _Revs} <- PurgeReqs0],
+ {ok, PurgeInfos} = couch_db:load_purge_infos(Db, UUIDs),
+ PurgeReqs = lists:foldr(fun
+ ({not_found, PReq}, Acc) -> [PReq | Acc];
+ ({{_, _, _, _}, _}, Acc) -> Acc
+ end, lists:zip(PurgeInfos, PurgeReqs0)),
+
+ % Processing any remaining purge requests
+ Ids = [Id || {_UUID, Id, _Revs} <- PurgeReqs],
+ DocInfos = couch_db_engine:open_docs(Db, Ids),
+ UpdateSeq = couch_db_engine:get_update_seq(Db),
+ PurgeSeq = couch_db_engine:get_purge_seq(Db),
+
+ InitAcc = {[], [], []},
+ {Pairs, PInfos, Replies} = purge_docs(
+ Db, PurgeReqs, DocInfos, UpdateSeq, PurgeSeq, InitAcc),
- PurgeSeq = couch_db_engine:get_purge_seq(Db2),
- {reply, {ok, PurgeSeq, PurgedIdRevs}, Db2, idle_limit()};
+ Db2 = if Pairs == [] -> Db; true ->
+ {ok, Db1} = couch_db_engine:purge_docs(Db, Pairs, PInfos),
+ ok = gen_server:call(couch_server, {db_updated, Db1}, infinity),
+ couch_event:notify(Db1#db.name, updated)
+ end,
+ {reply, {ok, Replies}, Db2};
handle_call(Msg, From, Db) ->
case couch_db_engine:handle_db_updater_call(Msg, From, Db) of
@@ -645,7 +603,7 @@ update_docs_int(Db, DocsList, LocalDocs, MergeConflicts, FullCommit) ->
Pairs = pair_write_info(OldDocLookups, IndexFDIs),
LocalDocs2 = update_local_doc_revs(LocalDocs),
- {ok, Db1} = couch_db_engine:write_doc_infos(Db, Pairs, LocalDocs2, []),
+ {ok, Db1} = couch_db_engine:write_doc_infos(Db, Pairs, LocalDocs2),
WriteCount = length(IndexFDIs),
couch_stats:increment_counter([couchdb, document_inserts],
@@ -691,6 +649,138 @@ update_local_doc_revs(Docs) ->
end, Docs).
+purge_docs([], [], _USeq, _PSeq, {Pairs, PInfos, Replies}) ->
+ {lists:reverse(Pairs), lists:reverse(PInfos), lists:reverse(Replies)};
+
+purge_docs([Req | RestReqs], [FDI | RestInfos], USeq, PSeq, Acc) ->
+ {UUID, DocId, Revs} = Req,
+ {Pair, RemovedRevs, NewUSeq} = case FDI of
+ #full_doc_info{rev_tree = Tree} ->
+ case couch_key_tree:remove_leafs(Tree, Revs) of
+ {_, []} ->
+ % No change
+ {{not_found, not_found}, [], USeq};
+ {[], Removed} ->
+ % Completely purged
+ {{FDI, not_found}, Removed, USeq};
+ {NewTree, Removed} ->
+ % Its possible to purge the #leaf{} that contains
+ % the update_seq where this doc sits in the
+ % update_seq sequence. Rather than do a bunch of
+ % complicated checks we just re-label every #leaf{}
+ % and reinsert it into the update_seq sequence.
+ {NewTree2, NewUpdateSeq} = couch_key_tree:mapfold(fun
+ (_RevId, Leaf, leaf, SeqAcc) ->
+ {Leaf#leaf{seq = SeqAcc + 1},
+ SeqAcc + 1};
+ (_RevId, Value, _Type, SeqAcc) ->
+ {Value, SeqAcc}
+ end, USeq, NewTree),
+
+ NewFDI = FDI#full_doc_info{
+ update_seq = NewUpdateSeq,
+ rev_tree = NewTree2
+ },
+ {{FDI, NewFDI}, Removed, NewUpdateSeq}
+ end;
+ not_found ->
+ {{not_found, not_found}, [], USeq}
+ end,
+ {Pairs, PInfos, Replies} = Acc,
+ NewAcc = {
+ [Pair | Pairs],
+ [{PSeq, UUID, DocId, Revs} | PInfos],
+ [{ok, RemovedRevs} | Replies]
+ },
+ purge_docs(RestReqs, RestInfos, NewUSeq, PSeq + 1, NewAcc).
+
+
+% find purge seq such that all purge requests that happen before or
+% during it can be removed from purge trees
+get_disposable_purge_seq(#db{name=DbName} = Db) ->
+ PSeq = couch_db_engine:get_purge_seq(Db),
+ OldestPSeq = couch_db_engine:get_oldest_purge_seq(Db),
+ PDocsLimit = couch_db_engine:get_purged_docs_limit(Db),
+ ExpectedDispPSeq = PSeq - PDocsLimit,
+ % client's purge_seq can be up to "allowed_purge_seq_lag"
+ % behind ExpectedDispPSeq
+ AllowedPSeqLag = config:get_integer("purge", "allowed_purge_seq_lag", 100),
+ ClientAllowedMinPSeq = ExpectedDispPSeq - AllowedPSeqLag,
+ DisposablePSeq = if OldestPSeq > ClientAllowedMinPSeq ->
+ % DisposablePSeq is the last pseq we can remove;
+ % it should be one less than OldestPSeq when #purges is within limit
+ OldestPSeq - 1;
+ true ->
+ % Find the smallest checkpointed purge_seq among clients
+ V = "v" ++ config:get("purge", "version", "1") ++ "-",
+ Opts = [
+ {start_key, list_to_binary(?LOCAL_DOC_PREFIX ++ "purge-" ++ V)},
+ {end_key_gt, list_to_binary(?LOCAL_DOC_PREFIX ++ "purge1")}
+ ],
+ FoldFun = fun(#doc{id=DocID, body={Props}}, MinPSeq) ->
+ ClientPSeq = couch_util:get_value(<<"purge_seq">>, Props),
+ MinPSeq2 = if ClientPSeq >= ClientAllowedMinPSeq ->
+ erlang:min(MinPSeq, ClientPSeq);
+ true ->
+ case check_client_exists(DbName, DocID, Props) of
+ true -> erlang:min(MinPSeq, ClientPSeq);
+ false -> MinPSeq % ignore nonexisting clients
+ end
+ end,
+ {ok, MinPSeq2}
+ end,
+ {ok, ClientPSeq} = couch_db_engine:fold_local_docs(
+ Db, FoldFun, PSeq, Opts),
+ erlang:min(ClientPSeq, ExpectedDispPSeq)
+ end,
+ DisposablePSeq.
+
+
+check_client_exists(DbName, DocID, Props) ->
+ % will warn about clients that have not
+ % checkpointed more than "allowed_purge_time_lag"
+ AllowedPTimeLag = config:get_integer("purge",
+ "allowed_purge_time_lag", 86400), % secs in 1 day
+ M0 = couch_util:get_value(<<"verify_module">>, Props),
+ F0 = couch_util:get_value(<<"verify_function">>, Props),
+ M = binary_to_atom(M0, latin1),
+ F = binary_to_atom(F0, latin1),
+ {A} = couch_util:get_value(<<"verify_options">>, Props),
+ ClientExists = try erlang:apply(M, F, [A]) of
+ true ->
+ % warn if we haven't heard of this client more than AllowedPTimeLag
+ ClientTime = ?b2l(couch_util:get_value(<<"timestamp_utc">>, Props)),
+ {ok, [Y, Mon, D, H, Min, S], [] }=
+ io_lib:fread("~4d-~2d-~2dT~2d:~2d:~2dZ", ClientTime),
+ SecsClient = calendar:datetime_to_gregorian_seconds(
+ {{Y, Mon, D}, {H, Min, S}}),
+ SecsNow = calendar:datetime_to_gregorian_seconds(
+ calendar:now_to_universal_time(os:timestamp())),
+ if SecsClient + AllowedPTimeLag > SecsNow -> ok; true ->
+ couch_log:warning(
+ "Client: ~p hasn't processed purge requests for more than"
+ " ~p secs. Check this client, as it prevents compaction of "
+ "purge trees on db:~p.", [A, AllowedPTimeLag, DbName]
+ )
+ end,
+ true;
+ false ->
+ couch_log:warning(
+ "Client ~p doesn't exist, "
+ "but its checkpoint purge doc: ~p is still available. "
+ "Remove this doc from: ~p", [A, DocID, DbName]
+ ),
+ false
+ catch
+ error:Error ->
+ couch_log:error(
+ "error in evaluating if client: ~p exists: ~p", [A, Error]
+ ),
+ false
+ end,
+ ClientExists.
+
+
commit_data(Db) ->
commit_data(Db, false).
@@ -720,15 +810,6 @@ pair_write_info(Old, New) ->
end, New).
-pair_purge_info(Old, New) ->
- lists:map(fun(OldFDI) ->
- case lists:keyfind(OldFDI#full_doc_info.id, #full_doc_info.id, New) of
- #full_doc_info{} = NewFDI -> {OldFDI, NewFDI};
- false -> {OldFDI, not_found}
- end
- end, Old).
-
-
get_meta_body_size(Meta) ->
{ejson_size, ExternalSize} = lists:keyfind(ejson_size, 1, Meta),
ExternalSize.
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.