Add couch_views_updater interactive indexer

This adds the ability for couch_views to index an index in the docs
update transaction. This only happens if a design doc has the
field <<"interactive">> = true.

9 files changed, 470 insertions, 5 deletions

diff --git a/rel/apps/couch_epi.config b/rel/apps/couch_epi.config
index 0f3d2da55..d3711636f 100644
--- a/rel/apps/couch_epi.config
+++ b/rel/apps/couch_epi.config
@@ -15,6 +15,7 @@
     fabric2_epi,
     chttpd_epi,
     couch_index_epi,
+    couch_views_epi,
     dreyfus_epi,
     global_changes_epi,
     mango_epi,
diff --git a/src/couch_views/src/couch_views.app.src b/src/couch_views/src/couch_views.app.src
index b704c9745..cb8285ac2 100644
--- a/src/couch_views/src/couch_views.app.src
+++ b/src/couch_views/src/couch_views.app.src
@@ -22,6 +22,7 @@
         kernel,
         stdlib,
         erlfdb,
+        couch_epi,
         couch_log,
         config,
         couch_stats,
diff --git a/src/couch_views/src/couch_views.erl b/src/couch_views/src/couch_views.erl
index 2268052f8..2acba00a6 100644
--- a/src/couch_views/src/couch_views.erl
+++ b/src/couch_views/src/couch_views.erl
@@ -37,6 +37,7 @@ query(Db, DDoc, ViewName, Callback, Acc0, Args0) ->
     end,
 
     DbName = fabric2_db:name(Db),
+    IsInteractive = couch_views_ddoc:is_interactive(DDoc),
     {ok, Mrst} = couch_views_util:ddoc_to_mrst(DbName, DDoc),
 
     #mrst{
@@ -54,7 +55,7 @@ query(Db, DDoc, ViewName, Callback, Acc0, Args0) ->
 
     try
         fabric2_fdb:transactional(Db, fun(TxDb) ->
-            ok = maybe_update_view(TxDb, Mrst, Args3),
+            ok = maybe_update_view(TxDb, Mrst, IsInteractive, Args3),
             read_view(TxDb, Mrst, ViewName, Callback, Acc0, Args3)
         end)
     catch throw:{build_view, WaitSeq} ->
@@ -127,13 +128,20 @@ read_view(Db, Mrst, ViewName, Callback, Acc0, Args) ->
     end).
 
 
-maybe_update_view(_Db, _Mrst, #mrargs{update = false}) ->
+maybe_update_view(_Db, _Mrst, _, #mrargs{update = false}) ->
     ok;
 
-maybe_update_view(_Db, _Mrst, #mrargs{update = lazy}) ->
+maybe_update_view(_Db, _Mrst, _, #mrargs{update = lazy}) ->
     ok;
 
-maybe_update_view(TxDb, Mrst, _Args) ->
+maybe_update_view(TxDb, Mrst, true, _Args) ->
+    BuildState = couch_views_fdb:get_build_status(TxDb, Mrst),
+    if BuildState == ?INDEX_READY -> ok; true ->
+        VS = couch_views_fdb:get_creation_vs(TxDb, Mrst),
+        throw({build_view, fabric2_fdb:vs_to_seq(VS)})
+    end;
+
+maybe_update_view(TxDb, Mrst, false, _Args) ->
     DbSeq = fabric2_db:get_update_seq(TxDb),
     ViewSeq = couch_views_fdb:get_update_seq(TxDb, Mrst),
     case DbSeq == ViewSeq of
diff --git a/src/couch_views/src/couch_views_ddoc.erl b/src/couch_views/src/couch_views_ddoc.erl
new file mode 100644
index 000000000..fae4a3433
--- /dev/null
+++ b/src/couch_views/src/couch_views_ddoc.erl
@@ -0,0 +1,42 @@
+% 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_views_ddoc).
+
+
+-export([
+    get_interactive_list/1,
+    get_mango_list/1,
+    is_interactive/1
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+
+
+% TODO: build a ddoc cache that checks the md_version
+get_interactive_list(Db) ->
+    DDocs = fabric2_db:get_design_docs(Db),
+    lists:filter(fun is_interactive/1, DDocs).
+
+
+get_mango_list(Db) ->
+    DDocs = fabric2_db:get_design_docs(Db),
+    lists:filter(fun (DDoc) ->
+        {Props} = couch_doc:to_json_obj(DDoc, []),
+        fabric2_util:get_value(<<"language">>, Props) == <<"query">>
+    end, DDocs).
+
+
+is_interactive(#doc{} = DDoc) ->
+    {Props} = couch_doc:to_json_obj(DDoc, []),
+    {Opts} = fabric2_util:get_value(<<"options">>, Props, {[]}),
+    fabric2_util:get_value(<<"interactive">>, Opts, false).
diff --git a/src/couch_views/src/couch_views_epi.erl b/src/couch_views/src/couch_views_epi.erl
new file mode 100644
index 000000000..6d39d9a5e
--- /dev/null
+++ b/src/couch_views/src/couch_views_epi.erl
@@ -0,0 +1,58 @@
+% 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_views_epi).
+
+
+-behaviour(couch_epi_plugin).
+
+
+-export([
+    app/0,
+    providers/0,
+    services/0,
+    data_subscriptions/0,
+    data_providers/0,
+    processes/0,
+    notify/3
+]).
+
+
+app() ->
+    couch_views.
+
+
+providers() ->
+    [
+        {fabric2_db, couch_views_fabric2_plugin}
+    ].
+
+
+services() ->
+    [].
+
+
+data_subscriptions() ->
+    [].
+
+
+data_providers() ->
+    [].
+
+
+processes() ->
+    [].
+
+
+notify(_Key, _Old, _New) ->
+    ok.
diff --git a/src/couch_views/src/couch_views_fabric2_plugin.erl b/src/couch_views/src/couch_views_fabric2_plugin.erl
new file mode 100644
index 000000000..cae0e1f75
--- /dev/null
+++ b/src/couch_views/src/couch_views_fabric2_plugin.erl
@@ -0,0 +1,24 @@
+% 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_views_fabric2_plugin).
+
+
+-export([
+    after_doc_write/6
+]).
+
+
+after_doc_write(Db, Doc, NewWinner, OldWinner, NewRevId, Seq)->
+    couch_views_updater:index(Db, Doc, NewWinner, OldWinner, NewRevId, Seq),
+    [Db, Doc, NewWinner, OldWinner, NewRevId, Seq].
diff --git a/src/couch_views/src/couch_views_sup.erl b/src/couch_views/src/couch_views_sup.erl
index 2a40f0a79..94531893d 100644
--- a/src/couch_views/src/couch_views_sup.erl
+++ b/src/couch_views/src/couch_views_sup.erl
@@ -42,7 +42,7 @@ init(normal) ->
             id => couch_views_server,
             start => {couch_views_server, start_link, []}
         }
-    ],
+    ] ++ couch_epi:register_service(couch_views_epi, []),
     {ok, {flags(), Children}};
 
 init(builds_disabled) ->
diff --git a/src/couch_views/src/couch_views_updater.erl b/src/couch_views/src/couch_views_updater.erl
new file mode 100644
index 000000000..f405123fa
--- /dev/null
+++ b/src/couch_views/src/couch_views_updater.erl
@@ -0,0 +1,101 @@
+% 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_views_updater).
+
+-export([
+    index/6
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+
+% If the doc revision doesn't not match the NewRevId passed here we can ignore
+% the document since it is then a conflict document and it doesn't need
+% to be indexed.
+index(Db, #doc{id = Id, revs = Revs} = Doc, _NewWinner, _OldWinner, NewRevId,
+        Seq) ->
+    try
+        {Depth, [FirstRev | _]} = Revs,
+        DocRev = {Depth, FirstRev},
+        if DocRev /= NewRevId -> ok; true ->
+            index_int(Db, Doc, Seq)
+        end
+    catch
+        Error:Reason ->
+            DbName = fabric2_db:name(Db),
+            couch_log:error("Mango index error for Db ~s Doc ~p ~p ~p",
+                [DbName, Id, Error, Reason])
+    end.
+
+
+% Check if design doc is an interactive index and kick off background worker
+% to build the new index up to the creation_vs
+index_int(Db, #doc{id = <<?DESIGN_DOC_PREFIX, _/binary>>,
+        deleted = false} = DDoc, Seq) ->
+    DbName = fabric2_db:name(Db),
+
+    case couch_views_ddoc:is_interactive(DDoc) of
+        true ->
+            {ok, Mrst} = couch_mrview_util:ddoc_to_mrst(DbName, DDoc),
+            case couch_views_fdb:get_creation_vs(Db, Mrst) of
+                not_found ->
+                    couch_views_fdb:new_interactive_index(Db, Mrst, Seq),
+                    {ok, _} = couch_views_jobs:build_view_async(Db, Mrst);
+                _ ->
+                    ok
+            end;
+        false ->
+            ok
+    end,
+    write_doc(Db, DDoc);
+
+
+index_int(Db, #doc{} = Doc, _Seq) ->
+    write_doc(Db, Doc).
+
+
+write_doc(Db, #doc{deleted = Deleted} = Doc) ->
+    DbName = fabric2_db:name(Db),
+    DDocs = couch_views_ddoc:get_interactive_list(Db),
+
+    Result0 = [#{
+        id => Doc#doc.id,
+        results => [],
+        deleted => Deleted,
+        doc => Doc
+    }],
+
+    %% Interactive updates do not update the views update_seq
+    State = #{
+        last_seq => false
+    },
+
+    lists:foreach(fun(DDoc) ->
+        {ok, Mrst} = couch_mrview_util:ddoc_to_mrst(DbName, DDoc),
+
+        case should_index_doc(Doc, Mrst) of
+            true ->
+                {Mrst1, Result1} = couch_views_indexer:map_docs(Mrst, Result0),
+                couch_views_indexer:write_docs(Db, Mrst1, Result1, State),
+                couch_eval:release_map_context(Mrst1#mrst.qserver);
+            false ->
+                ok
+        end
+    end, DDocs).
+
+
+should_index_doc(<<?DESIGN_DOC_PREFIX, _/binary>>, Mrst) ->
+    lists:keymember(<<"include_design">>, 1, Mrst#mrst.design_opts);
+
+should_index_doc(_,  _) ->
+    true.
diff --git a/src/couch_views/test/couch_views_updater_test.erl b/src/couch_views/test/couch_views_updater_test.erl
new file mode 100644
index 000000000..e45622512
--- /dev/null
+++ b/src/couch_views/test/couch_views_updater_test.erl
@@ -0,0 +1,230 @@
+% 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_views_updater_test).
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
+-include_lib("mango/src/mango_idx.hrl").
+-include_lib("couch_views/include/couch_views.hrl").
+
+
+indexer_test_() ->
+    {
+        "Test indexing",
+        {
+            setup,
+            fun setup/0,
+            fun cleanup/1,
+            {
+                foreach,
+                fun foreach_setup/0,
+                fun foreach_teardown/1,
+                [
+                    ?TDEF_FE(index_docs),
+                    ?TDEF_FE(update_doc),
+                    ?TDEF_FE(delete_doc),
+                    ?TDEF_FE(includes_design_docs)
+                ]
+            }
+        }
+    }.
+
+
+setup() ->
+    Ctx = test_util:start_couch([
+        fabric,
+        couch_jobs,
+        couch_js,
+        couch_views,
+        mango
+    ]),
+    Ctx.
+
+
+cleanup(Ctx) ->
+    test_util:stop_couch(Ctx).
+
+
+foreach_setup() ->
+    {ok, Db} = fabric2_db:create(?tempdb(), [{user_ctx, ?ADMIN_USER}]),
+
+    DDoc = create_idx_ddoc(),
+    fabric2_db:update_docs(Db, [DDoc]),
+    % make sure the index is built for the first time so the background
+    % indexer doesn't build the index
+    wait_while_ddoc_builds(Db),
+
+    Docs = make_docs(3),
+    fabric2_db:update_docs(Db, Docs),
+    {Db, DDoc}.
+
+
+foreach_teardown({Db, _}) ->
+    ok = fabric2_db:delete(fabric2_db:name(Db), []).
+
+
+index_docs({Db, DDoc}) ->
+    Docs = run_query(Db, DDoc),
+    ?assertEqual([
+        [{id, <<"1">>}, {value, 1}],
+        [{id, <<"2">>}, {value, 2}],
+        [{id, <<"3">>}, {value, 3}]
+    ], Docs).
+
+
+update_doc({Db, DDoc}) ->
+    {ok, Doc} = fabric2_db:open_doc(Db, <<"2">>),
+    JsonDoc = couch_doc:to_json_obj(Doc, []),
+    JsonDoc2 = couch_util:json_apply_field({<<"value">>, 4}, JsonDoc),
+    Doc2 = couch_doc:from_json_obj(JsonDoc2),
+    fabric2_db:update_doc(Db, Doc2),
+
+    Docs = run_query(Db, DDoc),
+    ?assertEqual([
+        [{id, <<"1">>}, {value, 1}],
+        [{id, <<"3">>}, {value, 3}],
+        [{id, <<"2">>}, {value, 4}]
+    ], Docs).
+
+
+delete_doc({Db, DDoc}) ->
+    {ok, Doc} = fabric2_db:open_doc(Db, <<"2">>),
+    JsonDoc = couch_doc:to_json_obj(Doc, []),
+    JsonDoc2 = couch_util:json_apply_field({<<"_deleted">>, true}, JsonDoc),
+    Doc2 = couch_doc:from_json_obj(JsonDoc2),
+    fabric2_db:update_doc(Db, Doc2),
+
+    Docs = run_query(Db, DDoc),
+    ?assertEqual([
+        [{id, <<"1">>}, {value, 1}],
+        [{id, <<"3">>}, {value, 3}]
+    ], Docs).
+
+
+includes_design_docs({Db, _}) ->
+    DDoc = create_idx_include_ddocs(),
+    fabric2_db:update_docs(Db, [DDoc]),
+
+    IndexDDoc0 = create_idx_ddoc(),
+    IndexDDoc = IndexDDoc0#doc{
+        id = <<"_design/to_be_indexed">>
+    },
+
+    fabric2_db:update_docs(Db, [IndexDDoc]),
+
+    Docs = run_query(Db, DDoc),
+    ?assertEqual([
+        [{id, <<"_design/ddoc_that_indexes_ddocs">>}, {value, 1}],
+        [{id, <<"_design/to_be_indexed">>}, {value, 1}]
+    ], Docs).
+
+
+run_query(Db, DDoc) ->
+    Args = #mrargs{
+        view_type = map,
+        reduce = false,
+        include_docs = true,
+        update = false
+    },
+    CB = fun query_cb/2,
+    {ok, Acc} = couch_views:query(Db, DDoc, <<"idx_01">>, CB, [], Args),
+    lists:map(fun ({Props}) ->
+        [
+            {id, couch_util:get_value(<<"_id">>, Props)},
+            {value, couch_util:get_value(<<"value">>, Props, 1)}
+        ]
+
+    end, Acc).
+
+
+create_idx_ddoc() ->
+    couch_doc:from_json_obj({[
+        {<<"_id">>, <<"_design/ddoc1">>},
+        {<<"language">>, <<"query">>},
+        {<<"views">>, {[
+            {<<"idx_01">>, {[
+                {<<"map">>, {[
+                    {<<"fields">>, {[{<<"value">>, <<"asc">>}]}}
+                ]}},
+                {<<"reduce">>, <<"_count">>},
+                {<<"options">>, {[
+                        {<<"def">>,
+                        {[{<<"fields">>,
+                            {[{<<"value">>, <<"asc">>}]}}]}}
+                    ]}}
+            ]}}
+        ]}
+        },
+        {<<"autoupdate">>, false},
+        {<<"options">>, {[{<<"interactive">>, true}]}}
+    ]}).
+
+
+create_idx_include_ddocs() ->
+    couch_doc:from_json_obj({[
+        {<<"_id">>, <<"_design/ddoc_that_indexes_ddocs">>},
+        {<<"language">>, <<"javascript">>},
+        {<<"views">>, {[
+            {<<"idx_01">>, {[
+                {<<"map">>, <<
+                    "function(doc) {"
+                        "if (doc.language) {"
+                            "emit(doc.language, 1);"
+                        "}"
+                    "}">>}
+            ]}}
+        ]}},
+        {<<"autoupdate">>, false},
+        {<<"options">>, {[
+            {<<"include_design">>, true},
+            {<<"interactive">>, true}
+        ]}}
+    ]}).
+
+
+wait_while_ddoc_builds(Db) ->
+    Fun = fun () ->
+        fabric2_fdb:transactional(Db, fun(TxDb) ->
+            Ready = lists:filter(fun (Idx) ->
+                Idx#idx.build_status == ?INDEX_READY
+            end, mango_idx:list(TxDb)),
+
+            if length(Ready) > 1 -> ok; true ->
+                wait
+            end
+        end)
+    end,
+    test_util:wait(Fun).
+
+
+
+make_docs(Count) ->
+    [doc(I) || I <- lists:seq(1, Count)].
+
+
+doc(Id) ->
+    couch_doc:from_json_obj({[
+        {<<"_id">>, list_to_binary(integer_to_list(Id))},
+        {<<"value">>, Id}
+    ]}).
+
+
+query_cb({row, Props},  Acc) ->
+    Doc = couch_util:get_value(doc, Props),
+    {ok, Acc ++ [Doc]};
+
+query_cb(_, Acc) ->
+    {ok, Acc}.
+