nickva commented on code in PR #6138:
URL: https://github.com/apache/couchdb/pull/6138#discussion_r4163507823
##########
src/nouveau/src/nouveau_index_updater.erl:
##########
@@ -251,53 +238,44 @@ index_definition(#index{} = Index, InitialPurgeSeq) ->
<<"initial_purge_seq">> => InitialPurgeSeq
}.
-purge_index(Db, Index, #purge_acc{} = PurgeAcc0) ->
- Proc = get_os_process(Index#index.def_lang),
- try
- true = proc_prompt(Proc, [<<"add_fun">>, Index#index.def,
<<"nouveau">>]),
- FoldFun = fun({PurgeSeq, _UUID, Id, _Revs}, #purge_acc{} = PurgeAcc1)
->
- PurgeAcc3 =
- case couch_db:get_full_doc_info(Db, Id) of
- not_found ->
- Item = nouveau_api:make_purge(
- Id, PurgeAcc1#purge_acc.index_purge_seq, PurgeSeq
- ),
- PurgeAcc2 = PurgeAcc1#purge_acc{batch = [Item |
PurgeAcc1#purge_acc.batch]},
- PurgeAcc2#purge_acc{index_purge_seq = PurgeSeq};
- FDI ->
- DI = couch_doc:to_doc_info(FDI),
- #doc_info{id = Id, high_seq = Seq, revs =
[#rev_info{rev = Rev} | _]} = DI,
- case lists:member({Id, Rev},
PurgeAcc1#purge_acc.exclude_list) of
- true ->
- PurgeAcc1;
- false ->
- Item = update_or_delete_index(
- Db, PurgeAcc1#purge_acc.index_update_seq,
DI, Proc
- ),
- PurgeAcc1#purge_acc{
- batch = [Item | PurgeAcc1#purge_acc.batch],
- exclude_list = [{Id, Rev} |
PurgeAcc1#purge_acc.exclude_list],
- index_update_seq = Seq
- }
- end
- end,
- update_task(1),
- maybe_flush_batch(PurgeAcc3)
- end,
+purge_index(Db, Index, StartUpdateSeq, #purge_acc{} = PurgeAcc0) ->
+ FoldFun = fun({PurgeSeq, _UUID, Id, _Revs}, #purge_acc{} = PurgeAcc1) ->
+ PurgeAcc2 =
+ case couch_db:get_full_doc_info(Db, Id) of
+ not_found ->
+ queue_purge(Id, PurgeSeq, PurgeAcc1);
+ #full_doc_info{} = FDI ->
+ #doc_info{high_seq = Seq} = couch_doc:to_doc_info(FDI),
+ case Seq > StartUpdateSeq of
+ true ->
+ queue_purge(Id, PurgeSeq, PurgeAcc1);
+ false ->
+ PurgeAcc1
+ end
+ end,
+ update_task(1),
+ maybe_flush_batch(PurgeAcc2)
Review Comment:
This returns an `{error, _}` but b-tree traversal code doesn't expect an
error, only `{ok, _}` or `{stop, _}` so maybe we can exit:
```erlang
case maybe_flush_batch(PurgeAcc2) of
{ok, PurgeAcc3} -> {ok, PurgeAcc3};
{error, Reason} -> exit({error, Reason})
end
```
##########
src/nouveau/src/nouveau_index_updater.erl:
##########
@@ -251,53 +238,44 @@ index_definition(#index{} = Index, InitialPurgeSeq) ->
<<"initial_purge_seq">> => InitialPurgeSeq
}.
-purge_index(Db, Index, #purge_acc{} = PurgeAcc0) ->
- Proc = get_os_process(Index#index.def_lang),
- try
- true = proc_prompt(Proc, [<<"add_fun">>, Index#index.def,
<<"nouveau">>]),
- FoldFun = fun({PurgeSeq, _UUID, Id, _Revs}, #purge_acc{} = PurgeAcc1)
->
- PurgeAcc3 =
- case couch_db:get_full_doc_info(Db, Id) of
- not_found ->
- Item = nouveau_api:make_purge(
- Id, PurgeAcc1#purge_acc.index_purge_seq, PurgeSeq
- ),
- PurgeAcc2 = PurgeAcc1#purge_acc{batch = [Item |
PurgeAcc1#purge_acc.batch]},
- PurgeAcc2#purge_acc{index_purge_seq = PurgeSeq};
- FDI ->
- DI = couch_doc:to_doc_info(FDI),
- #doc_info{id = Id, high_seq = Seq, revs =
[#rev_info{rev = Rev} | _]} = DI,
- case lists:member({Id, Rev},
PurgeAcc1#purge_acc.exclude_list) of
- true ->
- PurgeAcc1;
- false ->
- Item = update_or_delete_index(
- Db, PurgeAcc1#purge_acc.index_update_seq,
DI, Proc
- ),
- PurgeAcc1#purge_acc{
- batch = [Item | PurgeAcc1#purge_acc.batch],
- exclude_list = [{Id, Rev} |
PurgeAcc1#purge_acc.exclude_list],
- index_update_seq = Seq
- }
- end
- end,
- update_task(1),
- maybe_flush_batch(PurgeAcc3)
- end,
+purge_index(Db, Index, StartUpdateSeq, #purge_acc{} = PurgeAcc0) ->
+ FoldFun = fun({PurgeSeq, _UUID, Id, _Revs}, #purge_acc{} = PurgeAcc1) ->
+ PurgeAcc2 =
+ case couch_db:get_full_doc_info(Db, Id) of
+ not_found ->
+ queue_purge(Id, PurgeSeq, PurgeAcc1);
+ #full_doc_info{} = FDI ->
+ #doc_info{high_seq = Seq} = couch_doc:to_doc_info(FDI),
Review Comment:
Let's add a comment here explaining that the doc still exists. If it changed
after the index last saw (doc recreated or a leaf was purged and it was
relabeled) we remove it during the purge phase then let the the regular index
updater index the winner.
##########
src/nouveau/src/nouveau_index_updater.erl:
##########
@@ -251,53 +238,44 @@ index_definition(#index{} = Index, InitialPurgeSeq) ->
<<"initial_purge_seq">> => InitialPurgeSeq
}.
-purge_index(Db, Index, #purge_acc{} = PurgeAcc0) ->
- Proc = get_os_process(Index#index.def_lang),
- try
- true = proc_prompt(Proc, [<<"add_fun">>, Index#index.def,
<<"nouveau">>]),
- FoldFun = fun({PurgeSeq, _UUID, Id, _Revs}, #purge_acc{} = PurgeAcc1)
->
- PurgeAcc3 =
- case couch_db:get_full_doc_info(Db, Id) of
- not_found ->
- Item = nouveau_api:make_purge(
- Id, PurgeAcc1#purge_acc.index_purge_seq, PurgeSeq
- ),
- PurgeAcc2 = PurgeAcc1#purge_acc{batch = [Item |
PurgeAcc1#purge_acc.batch]},
- PurgeAcc2#purge_acc{index_purge_seq = PurgeSeq};
- FDI ->
- DI = couch_doc:to_doc_info(FDI),
- #doc_info{id = Id, high_seq = Seq, revs =
[#rev_info{rev = Rev} | _]} = DI,
- case lists:member({Id, Rev},
PurgeAcc1#purge_acc.exclude_list) of
- true ->
- PurgeAcc1;
- false ->
- Item = update_or_delete_index(
- Db, PurgeAcc1#purge_acc.index_update_seq,
DI, Proc
- ),
- PurgeAcc1#purge_acc{
- batch = [Item | PurgeAcc1#purge_acc.batch],
- exclude_list = [{Id, Rev} |
PurgeAcc1#purge_acc.exclude_list],
- index_update_seq = Seq
- }
- end
- end,
- update_task(1),
- maybe_flush_batch(PurgeAcc3)
- end,
+purge_index(Db, Index, StartUpdateSeq, #purge_acc{} = PurgeAcc0) ->
+ FoldFun = fun({PurgeSeq, _UUID, Id, _Revs}, #purge_acc{} = PurgeAcc1) ->
+ PurgeAcc2 =
+ case couch_db:get_full_doc_info(Db, Id) of
+ not_found ->
+ queue_purge(Id, PurgeSeq, PurgeAcc1);
+ #full_doc_info{} = FDI ->
+ #doc_info{high_seq = Seq} = couch_doc:to_doc_info(FDI),
Review Comment:
Maybe don't bother getting a doc_info{}. Use `FDI` `update_seq` instead as
the sequence.
```
case ... of
#full_doc_info{update_seq = Seq} ->
...
```
##########
src/nouveau/src/nouveau_index_updater.erl:
##########
@@ -251,53 +238,44 @@ index_definition(#index{} = Index, InitialPurgeSeq) ->
<<"initial_purge_seq">> => InitialPurgeSeq
}.
-purge_index(Db, Index, #purge_acc{} = PurgeAcc0) ->
- Proc = get_os_process(Index#index.def_lang),
- try
- true = proc_prompt(Proc, [<<"add_fun">>, Index#index.def,
<<"nouveau">>]),
- FoldFun = fun({PurgeSeq, _UUID, Id, _Revs}, #purge_acc{} = PurgeAcc1)
->
- PurgeAcc3 =
- case couch_db:get_full_doc_info(Db, Id) of
- not_found ->
- Item = nouveau_api:make_purge(
- Id, PurgeAcc1#purge_acc.index_purge_seq, PurgeSeq
- ),
- PurgeAcc2 = PurgeAcc1#purge_acc{batch = [Item |
PurgeAcc1#purge_acc.batch]},
- PurgeAcc2#purge_acc{index_purge_seq = PurgeSeq};
- FDI ->
- DI = couch_doc:to_doc_info(FDI),
- #doc_info{id = Id, high_seq = Seq, revs =
[#rev_info{rev = Rev} | _]} = DI,
- case lists:member({Id, Rev},
PurgeAcc1#purge_acc.exclude_list) of
- true ->
- PurgeAcc1;
- false ->
- Item = update_or_delete_index(
- Db, PurgeAcc1#purge_acc.index_update_seq,
DI, Proc
- ),
- PurgeAcc1#purge_acc{
- batch = [Item | PurgeAcc1#purge_acc.batch],
- exclude_list = [{Id, Rev} |
PurgeAcc1#purge_acc.exclude_list],
- index_update_seq = Seq
- }
- end
- end,
- update_task(1),
- maybe_flush_batch(PurgeAcc3)
- end,
+purge_index(Db, Index, StartUpdateSeq, #purge_acc{} = PurgeAcc0) ->
+ FoldFun = fun({PurgeSeq, _UUID, Id, _Revs}, #purge_acc{} = PurgeAcc1) ->
+ PurgeAcc2 =
+ case couch_db:get_full_doc_info(Db, Id) of
+ not_found ->
+ queue_purge(Id, PurgeSeq, PurgeAcc1);
+ #full_doc_info{} = FDI ->
+ #doc_info{high_seq = Seq} = couch_doc:to_doc_info(FDI),
+ case Seq > StartUpdateSeq of
+ true ->
+ queue_purge(Id, PurgeSeq, PurgeAcc1);
+ false ->
+ PurgeAcc1
+ end
+ end,
+ update_task(1),
+ maybe_flush_batch(PurgeAcc2)
+ end,
- {ok, #purge_acc{} = PurgeAcc3} = couch_db:fold_purge_infos(
- Db, PurgeAcc0#purge_acc.index_purge_seq, FoldFun, PurgeAcc0, []
- ),
- {ok, PurgeAcc4} = flush_batch(PurgeAcc3),
- DbPurgeSeq = couch_db:get_purge_seq(Db),
- ok = nouveau_api:set_purge_seq(
- Index, PurgeAcc4#purge_acc.index_purge_seq, DbPurgeSeq
- ),
- update_local_doc(Db, Index, DbPurgeSeq),
- {ok, PurgeAcc4}
- after
- ret_os_process(Proc)
- end.
+ {ok, #purge_acc{} = PurgeAcc3} = couch_db:fold_purge_infos(
+ Db, PurgeAcc0#purge_acc.index_purge_seq, FoldFun, PurgeAcc0, []
+ ),
+ {ok, PurgeAcc4} = flush_batch(PurgeAcc3),
Review Comment:
Maybe the same case and exit as above. So we get a better error instead of
a badmatch?
##########
test/elixir/test/nouveau_test.exs:
##########
@@ -753,6 +753,59 @@ defmodule NouveauTest do
assert seq(db_info["purge_seq"]) == resp.body["search_index"]["purge_seq"]
end
+ @tag :with_db
+ @tag query: %{q: 1}
+ test "purge with out of order edits", context do
+ db_name = context[:db_name]
+ create_resp = create_search_docs(db_name)
+ create_ddoc(db_name)
+
+ search_url = "/#{db_name}/_design/foo/_nouveau/bar"
+
+ # confirm all hits
+ resp = Couch.get(search_url, query: %{q: "*:*", include_docs: true})
+ assert_status_code(resp, 200)
+ assert get_total_hits(resp) == 4
+
+ # purge two docs in separate requests
+ doc_a = (Enum.at(create_resp.body, 0))
+ resp =
+ Couch.post("/#{db_name}/_purge",
+ body: %{doc_a["id"] => [doc_a["rev"]]}
+ )
+ assert_status_code(resp, 201)
+
+ doc_b = (Enum.at(create_resp.body, 1))
+ resp =
+ Couch.post("/#{db_name}/_purge",
+ body: %{doc_b["id"] => [doc_b["rev"]]}
+ )
+ assert_status_code(resp, 201)
+
+ # add another doc to catch if update seq has skipped docs
+ resp = Couch.put("/#{db_name}/docC", body: %{"foo" => "c", "bar" => 1})
+ assert_status_code(resp, 201)
+
+ # recreate the purged docs in a different order
+ resp = Couch.put("/#{db_name}/#{doc_b["id"]}", body: %{"foo" => "foo",
"bar" => 42})
+ assert_status_code(resp, 201)
+
+ resp = Couch.put("/#{db_name}/#{doc_a["id"]}", body: %{"foo" => "foo",
"bar" => 42})
+ assert_status_code(resp, 201)
+
+ # confirm a and b are back
+ resp = Couch.get(search_url, query: %{q: "*:*", include_docs: true})
+ assert_status_code(resp, 200)
+ assert get_total_hits(resp) == 5
Review Comment:
We should assert content too if possible.
doc_b is doc3, whose old foo was bar and now foo is foo so something like
this maybe
```elixir
resp = Couch.get(search_url, query: %{q: "foo:bar"})
assert_status_code(resp, 200)
assert get_total_hits(resp) == 0
resp = Couch.get(search_url, query: %{q: "foo:foo"})
assert_status_code(resp, 200)
assert get_total_hits(resp) == 2
```
##########
test/elixir/test/nouveau_test.exs:
##########
@@ -753,6 +753,59 @@ defmodule NouveauTest do
assert seq(db_info["purge_seq"]) == resp.body["search_index"]["purge_seq"]
end
+ @tag :with_db
+ @tag query: %{q: 1}
+ test "purge with out of order edits", context do
+ db_name = context[:db_name]
+ create_resp = create_search_docs(db_name)
+ create_ddoc(db_name)
+
+ search_url = "/#{db_name}/_design/foo/_nouveau/bar"
+
+ # confirm all hits
+ resp = Couch.get(search_url, query: %{q: "*:*", include_docs: true})
+ assert_status_code(resp, 200)
+ assert get_total_hits(resp) == 4
+
+ # purge two docs in separate requests
+ doc_a = (Enum.at(create_resp.body, 0))
Review Comment:
extra parantheses?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
