1#!/usr/bin/env escript
2%% -*- Mode: Erlang; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
3%%! -smp enable
4
5% Licensed under the Apache License, Version 2.0 (the "License"); you may not
6% use this file except in compliance with the License. You may obtain a copy of
7% the License at
8%
9%   http://www.apache.org/licenses/LICENSE-2.0
10%
11% Unless required by applicable law or agreed to in writing, software
12% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14% License for the specific language governing permissions and limitations under
15% the License.
16
17-define(JSON_ENCODE(V), ejson:encode(V)). % couch_db.hrl
18-define(MAX_WAIT_TIME, 900 * 1000).
19
20-include_lib("couch_set_view/include/couch_set_view.hrl").
21
22test_set_name() -> <<"couch_test_set_index_updater_cleanup">>.
23num_set_partitions() -> 64.
24ddoc_id() -> <<"_design/test">>.
25num_docs_0() -> 42624.  % keep it a multiple of num_set_partitions()
26
27
28main(_) ->
29    test_util:init_code_path(),
30
31    etap:plan(31),
32    case (catch test()) of
33        ok ->
34            etap:end_tests();
35        Other ->
36            etap:diag(io_lib:format("Test died abnormally: ~p", [Other])),
37            etap:bail(Other)
38    end,
39    ok.
40
41
42test() ->
43    couch_set_view_test_util:start_server(test_set_name()),
44
45    couch_set_view_test_util:delete_set_dbs(test_set_name(), num_set_partitions()),
46    couch_set_view_test_util:create_set_dbs(test_set_name(), num_set_partitions()),
47
48    create_set(),
49    GroupPid = couch_set_view:get_group_pid(
50        mapreduce_view, test_set_name(), ddoc_id(), prod),
51    ok = gen_server:call(GroupPid, {set_auto_cleanup, false}, infinity),
52
53    ValueGenFun1 = fun(I) -> I end,
54    update_documents(0, num_docs_0(), ValueGenFun1),
55
56    % build index
57    Group0 = get_group_snapshot(),
58
59    verify_btrees(ValueGenFun1, num_docs_0(), []),
60
61    % Add a few new documents
62    update_documents(num_docs_0(), 8192, ValueGenFun1),
63
64    verify_btrees(ValueGenFun1, num_docs_0() + 8192, []),
65
66    % Mark some partitions for cleanup
67    etap:diag("Marking partitions for cleanup"),
68    CleanupPartitions = lists:seq(1, num_set_partitions() - 1, 2),
69    ok = couch_set_view:set_partition_states(
70        mapreduce_view, test_set_name(), ddoc_id(), [], [], CleanupPartitions),
71
72    ValueGenFun2 = fun(I) ->
73        case I < num_docs_0() of
74        true ->
75            I * 3;
76        false ->
77            I
78        end
79    end,
80    update_documents(0, num_docs_0(), ValueGenFun2),
81
82    [StatsBefore] = ets:lookup(Group0#set_view_group.stats_ets,
83        ?set_view_group_stats_key(Group0)),
84
85    verify_btrees(ValueGenFun2, num_docs_0() + 8192, CleanupPartitions),
86
87    [StatsAfter] = ets:lookup(Group0#set_view_group.stats_ets,
88        ?set_view_group_stats_key(Group0)),
89
90    etap:is(StatsBefore#set_view_group_stats.cleanups, 0,
91            "# of cleanups before was 0"),
92    etap:is(StatsAfter#set_view_group_stats.cleanups, 1,
93            "# of cleanups after is 1"),
94    etap:is(StatsBefore#set_view_group_stats.updater_cleanups, 0,
95            "# of updater_cleanups before is 0"),
96    etap:is(StatsAfter#set_view_group_stats.updater_cleanups, 1,
97            "# of updater_cleanups after is 1"),
98
99    couch_set_view_test_util:delete_set_dbs(test_set_name(), num_set_partitions()),
100    couch_set_view_test_util:stop_server(),
101    ok.
102
103
104get_group_snapshot() ->
105    GroupPid = couch_set_view:get_group_pid(
106        mapreduce_view, test_set_name(), ddoc_id(), prod),
107    {ok, UpPid} = gen_server:call(GroupPid, {start_updater, []}, infinity),
108    Ref = erlang:monitor(process, UpPid),
109    receive
110    {'DOWN', Ref, _, _, _} ->
111        {ok, Group, 0} = gen_server:call(
112        GroupPid, #set_view_group_req{stale = ok, debug = true}, infinity),
113        Group
114    after 600000 ->
115        etap:bail("Timeout waiting for updater to finish")
116    end.
117
118
119create_set() ->
120    couch_set_view_test_util:delete_set_dbs(test_set_name(), num_set_partitions()),
121    couch_set_view_test_util:create_set_dbs(test_set_name(), num_set_partitions()),
122    couch_set_view:cleanup_index_files(mapreduce_view, test_set_name()),
123    etap:diag("Creating the set databases (# of partitions: " ++
124        integer_to_list(num_set_partitions()) ++ ")"),
125    DDoc = {[
126        {<<"meta">>, {[{<<"id">>, ddoc_id()}]}},
127        {<<"json">>, {[
128        {<<"language">>, <<"javascript">>},
129        {<<"views">>, {[
130            {<<"view_1">>, {[
131                {<<"map">>, <<"function(doc, meta) { emit(meta.id, doc.value); }">>},
132                {<<"reduce">>, <<"_count">>}
133            ]}},
134            {<<"view_2">>, {[
135                {<<"map">>, <<"function(doc, meta) { emit(meta.id, meta.id); }">>},
136                {<<"reduce">>, <<"_count">>}
137            ]}}
138        ]}}
139        ]}}
140    ]},
141    ok = couch_set_view_test_util:update_ddoc(test_set_name(), DDoc),
142    etap:diag("Configuring set view with partitions [0 .. 63] as active"),
143    Params = #set_view_params{
144        max_partitions = num_set_partitions(),
145        active_partitions = lists:seq(0, num_set_partitions() - 1),
146        passive_partitions = [],
147        use_replica_index = false
148    },
149    ok = couch_set_view:define_group(
150        mapreduce_view, test_set_name(), ddoc_id(), Params).
151
152
153update_documents(StartId, Count, ValueGenFun) ->
154    etap:diag("Updating " ++ integer_to_list(Count) ++ " documents"),
155    DocList0 = lists:map(
156        fun(I) ->
157            {I rem num_set_partitions(), {[
158                {<<"meta">>, {[{<<"id">>, doc_id(I)}]}},
159                {<<"json">>, {[
160                    {<<"value">>, ValueGenFun(I)}
161                ]}}
162            ]}}
163        end,
164        lists:seq(StartId, StartId + Count - 1)),
165    DocList = [Doc || {_, Doc} <- lists:keysort(1, DocList0)],
166    ok = couch_set_view_test_util:populate_set_sequentially(
167        test_set_name(),
168        lists:seq(0, num_set_partitions() - 1),
169        DocList).
170
171
172doc_id(I) ->
173    iolist_to_binary(io_lib:format("doc_~8..0b", [I])).
174
175
176verify_btrees(ValueGenFun, NumDocs, CleanupParts) ->
177    Group = get_group_snapshot(),
178    #set_view_group{
179        id_btree = IdBtree,
180        views = [View1, View2],
181        index_header = #set_view_index_header{
182            seqs = HeaderUpdateSeqs,
183            abitmask = Abitmask,
184            pbitmask = Pbitmask,
185            cbitmask = Cbitmask
186        }
187    } = Group,
188    #set_view{
189        indexer = #mapreduce_view{
190            btree = View1Btree
191        }
192    } = View1,
193    #set_view{
194        indexer = #mapreduce_view{
195            btree = View2Btree
196        }
197    } = View2,
198    ActiveParts = ordsets:subtract(lists:seq(0, num_set_partitions() - 1), CleanupParts),
199    ExpectedBitmask = couch_set_view_util:build_bitmask(ActiveParts),
200    DbSeqs = couch_set_view_test_util:get_db_seqs(test_set_name(), ActiveParts),
201    ExpectedKVCount = length([I || I <- lists:seq(0, NumDocs - 1),
202        ordsets:is_element((I rem num_set_partitions()), ActiveParts)]),
203
204    etap:is(
205        couch_set_view_test_util:full_reduce_id_btree(Group, IdBtree),
206        {ok, {ExpectedKVCount, ExpectedBitmask}},
207        "Id Btree has the right reduce value"),
208    etap:is(
209        couch_set_view_test_util:full_reduce_view_btree(Group, View1Btree),
210        {ok, {ExpectedKVCount, [ExpectedKVCount], ExpectedBitmask}},
211        "View1 Btree has the right reduce value"),
212
213    etap:is(HeaderUpdateSeqs, DbSeqs, "Header has right update seqs list"),
214    etap:is(Abitmask, ExpectedBitmask, "Header has right active bitmask"),
215    etap:is(Pbitmask, 0, "Header has right passive bitmask"),
216    etap:is(Cbitmask, 0, "Header has right cleanup bitmask"),
217
218    etap:diag("Verifying the Id Btree"),
219    MaxPerPart = NumDocs div num_set_partitions(),
220    {ok, _, {_, _, _, IdBtreeFoldResult}} = couch_set_view_test_util:fold_id_btree(
221        Group,
222        IdBtree,
223        fun(Kv, _, {Parts, I0, C0, It}) ->
224            case C0 >= MaxPerPart of
225            true ->
226                [_ | RestParts] = Parts,
227                [P | _] = RestParts,
228                I = P,
229                C = 1;
230            false ->
231                RestParts = Parts,
232                [P | _] = RestParts,
233                I = I0,
234                C = C0 + 1
235            end,
236            true = (P < num_set_partitions()),
237            DocId = doc_id(I),
238            Value = [{View1#set_view.id_num, DocId}, {View2#set_view.id_num, DocId}],
239            ExpectedKv = {<<P:16, DocId/binary>>, {P, Value}},
240            case ExpectedKv =:= Kv of
241            true ->
242                ok;
243            false ->
244                etap:bail("Id Btree has an unexpected KV at iteration " ++ integer_to_list(It))
245            end,
246            {ok, {RestParts, I + num_set_partitions(), C, It + 1}}
247        end,
248        {ActiveParts, hd(ActiveParts), 0, 0}, []),
249    etap:is(IdBtreeFoldResult, ExpectedKVCount,
250        "Id Btree has " ++ integer_to_list(ExpectedKVCount) ++ " entries"),
251
252    etap:diag("Verifying the View1 Btree"),
253    {ok, _, {_, View1BtreeFoldResult}} = couch_set_view_test_util:fold_view_btree(
254        Group,
255        View1Btree,
256        fun(Kv, _, {I, Count}) ->
257            PartId = I rem num_set_partitions(),
258            DocId = doc_id(I),
259            ExpectedKv = {{DocId, DocId}, {PartId, ValueGenFun(I)}},
260            case ExpectedKv =:= Kv of
261            true ->
262                ok;
263            false ->
264                etap:bail("View1 Btree has an unexpected KV at iteration " ++ integer_to_list(Count + 1))
265            end,
266            {ok, {next_i(I, ActiveParts), Count + 1}}
267        end,
268        {hd(ActiveParts), 0}, []),
269    etap:is(View1BtreeFoldResult, ExpectedKVCount,
270        "View1 Btree has " ++ integer_to_list(ExpectedKVCount) ++ " entries"),
271
272    etap:diag("Verifying the View2 Btree"),
273    {ok, _, {_, View2BtreeFoldResult}} = couch_set_view_test_util:fold_view_btree(
274        Group,
275        View2Btree,
276        fun(Kv, _, {I, Count}) ->
277            PartId = I rem num_set_partitions(),
278            DocId = doc_id(I),
279            ExpectedKv = {{DocId, DocId}, {PartId, DocId}},
280            case ExpectedKv =:= Kv of
281            true ->
282                ok;
283            false ->
284                etap:bail("View2 Btree has an unexpected KV at iteration " ++ integer_to_list(Count + 1))
285            end,
286            {ok, {next_i(I, ActiveParts), Count + 1}}
287        end,
288        {hd(ActiveParts), 0}, []),
289    etap:is(View2BtreeFoldResult, ExpectedKVCount,
290        "View2 Btree has " ++ integer_to_list(ExpectedKVCount) ++ " entries"),
291    ok.
292
293
294next_i(I, ActiveParts) ->
295    case ordsets:is_element((I + 1) rem num_set_partitions(), ActiveParts) of
296    true ->
297        I + 1;
298    false ->
299        next_i(I + 1, ActiveParts)
300    end.
301