You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by ja...@apache.org on 2018/11/28 16:46:49 UTC
[couchdb] 01/01: fix: re-enable _doc_id filter optimisation path
This is an automated email from the ASF dual-hosted git repository.
jan pushed a commit to branch fix/1737/changes-docid-filter
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 7efb79d9bd0261324694dd8dcef51e9996265bad
Author: Jan Lehnardt <ja...@apache.org>
AuthorDate: Wed Nov 28 16:56:51 2018 +0100
fix: re-enable _doc_id filter optimisation path
---
src/chttpd/src/chttpd_db.erl | 80 ++++++++++++++++++-------------------
src/couch/include/couch_db.hrl | 8 ++++
src/couch/src/couch_changes.erl | 12 +++++-
src/fabric/src/fabric_rpc.erl | 29 +++++++++-----
src/mango/src/mango_cursor_text.erl | 42 +++++++++----------
5 files changed, 98 insertions(+), 73 deletions(-)
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index 7590467..40b820e 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -38,7 +38,7 @@
}).
% Accumulator for changes_callback function
--record(cacc, {
+-record(chacc, {
etag,
feed,
mochi,
@@ -99,7 +99,7 @@ handle_changes_req1(#httpd{}=Req, Db) ->
DeltaT = timer:now_diff(os:timestamp(), T0) / 1000,
couch_stats:update_histogram([couchdb, dbinfo], DeltaT),
chttpd:etag_respond(Req, Etag, fun() ->
- Acc0 = #cacc{
+ Acc0 = #chacc{
feed = normal,
etag = Etag,
mochi = Req,
@@ -109,7 +109,7 @@ handle_changes_req1(#httpd{}=Req, Db) ->
end);
Feed when Feed =:= "continuous"; Feed =:= "longpoll"; Feed =:= "eventsource" ->
couch_stats:increment_counter([couchdb, httpd, clients_requesting_changes]),
- Acc0 = #cacc{
+ Acc0 = #chacc{
feed = list_to_atom(Feed),
mochi = Req,
threshold = Max
@@ -125,15 +125,15 @@ handle_changes_req1(#httpd{}=Req, Db) ->
end.
% callbacks for continuous feed (newline-delimited JSON Objects)
-changes_callback(start, #cacc{feed = continuous} = Acc) ->
- {ok, Resp} = chttpd:start_delayed_json_response(Acc#cacc.mochi, 200),
- {ok, Acc#cacc{mochi = Resp, responding = true}};
-changes_callback({change, Change}, #cacc{feed = continuous} = Acc) ->
+changes_callback(start, #chacc{feed = continuous} = Acc) ->
+ {ok, Resp} = chttpd:start_delayed_json_response(Acc#chacc.mochi, 200),
+ {ok, Acc#chacc{mochi = Resp, responding = true}};
+changes_callback({change, Change}, #chacc{feed = continuous} = Acc) ->
Data = [?JSON_ENCODE(Change) | "\n"],
Len = iolist_size(Data),
maybe_flush_changes_feed(Acc, Data, Len);
-changes_callback({stop, EndSeq, Pending}, #cacc{feed = continuous} = Acc) ->
- #cacc{mochi = Resp, buffer = Buf} = Acc,
+changes_callback({stop, EndSeq, Pending}, #chacc{feed = continuous} = Acc) ->
+ #chacc{mochi = Resp, buffer = Buf} = Acc,
Row = {[
{<<"last_seq">>, EndSeq},
{<<"pending">>, Pending}
@@ -143,15 +143,15 @@ changes_callback({stop, EndSeq, Pending}, #cacc{feed = continuous} = Acc) ->
chttpd:end_delayed_json_response(Resp1);
% callbacks for eventsource feed (newline-delimited eventsource Objects)
-changes_callback(start, #cacc{feed = eventsource} = Acc) ->
- #cacc{mochi = Req} = Acc,
+changes_callback(start, #chacc{feed = eventsource} = Acc) ->
+ #chacc{mochi = Req} = Acc,
Headers = [
{"Content-Type", "text/event-stream"},
{"Cache-Control", "no-cache"}
],
{ok, Resp} = chttpd:start_delayed_json_response(Req, 200, Headers),
- {ok, Acc#cacc{mochi = Resp, responding = true}};
-changes_callback({change, {ChangeProp}=Change}, #cacc{feed = eventsource} = Acc) ->
+ {ok, Acc#chacc{mochi = Resp, responding = true}};
+changes_callback({change, {ChangeProp}=Change}, #chacc{feed = eventsource} = Acc) ->
Seq = proplists:get_value(seq, ChangeProp),
Chunk = [
"data: ", ?JSON_ENCODE(Change),
@@ -160,34 +160,34 @@ changes_callback({change, {ChangeProp}=Change}, #cacc{feed = eventsource} = Acc)
],
Len = iolist_size(Chunk),
maybe_flush_changes_feed(Acc, Chunk, Len);
-changes_callback(timeout, #cacc{feed = eventsource} = Acc) ->
- #cacc{mochi = Resp} = Acc,
+changes_callback(timeout, #chacc{feed = eventsource} = Acc) ->
+ #chacc{mochi = Resp} = Acc,
Chunk = "event: heartbeat\ndata: \n\n",
{ok, Resp1} = chttpd:send_delayed_chunk(Resp, Chunk),
- {ok, Acc#cacc{mochi = Resp1}};
-changes_callback({stop, _EndSeq}, #cacc{feed = eventsource} = Acc) ->
- #cacc{mochi = Resp, buffer = Buf} = Acc,
+ {ok, Acc#chacc{mochi = Resp1}};
+changes_callback({stop, _EndSeq}, #chacc{feed = eventsource} = Acc) ->
+ #chacc{mochi = Resp, buffer = Buf} = Acc,
{ok, Resp1} = chttpd:send_delayed_chunk(Resp, Buf),
chttpd:end_delayed_json_response(Resp1);
% callbacks for longpoll and normal (single JSON Object)
-changes_callback(start, #cacc{feed = normal} = Acc) ->
- #cacc{etag = Etag, mochi = Req} = Acc,
+changes_callback(start, #chacc{feed = normal} = Acc) ->
+ #chacc{etag = Etag, mochi = Req} = Acc,
FirstChunk = "{\"results\":[\n",
{ok, Resp} = chttpd:start_delayed_json_response(Req, 200,
[{"ETag",Etag}], FirstChunk),
- {ok, Acc#cacc{mochi = Resp, responding = true}};
+ {ok, Acc#chacc{mochi = Resp, responding = true}};
changes_callback(start, Acc) ->
- #cacc{mochi = Req} = Acc,
+ #chacc{mochi = Req} = Acc,
FirstChunk = "{\"results\":[\n",
{ok, Resp} = chttpd:start_delayed_json_response(Req, 200, [], FirstChunk),
- {ok, Acc#cacc{mochi = Resp, responding = true}};
+ {ok, Acc#chacc{mochi = Resp, responding = true}};
changes_callback({change, Change}, Acc) ->
- Data = [Acc#cacc.prepend, ?JSON_ENCODE(Change)],
+ Data = [Acc#chacc.prepend, ?JSON_ENCODE(Change)],
Len = iolist_size(Data),
maybe_flush_changes_feed(Acc, Data, Len);
changes_callback({stop, EndSeq, Pending}, Acc) ->
- #cacc{buffer = Buf, mochi = Resp, threshold = Max} = Acc,
+ #chacc{buffer = Buf, mochi = Resp, threshold = Max} = Acc,
Terminator = [
"\n],\n\"last_seq\":",
?JSON_ENCODE(EndSeq),
@@ -198,32 +198,32 @@ changes_callback({stop, EndSeq, Pending}, Acc) ->
{ok, Resp1} = chttpd:close_delayed_json_object(Resp, Buf, Terminator, Max),
chttpd:end_delayed_json_response(Resp1);
-changes_callback(waiting_for_updates, #cacc{buffer = []} = Acc) ->
+changes_callback(waiting_for_updates, #chacc{buffer = []} = Acc) ->
{ok, Acc};
changes_callback(waiting_for_updates, Acc) ->
- #cacc{buffer = Buf, mochi = Resp} = Acc,
+ #chacc{buffer = Buf, mochi = Resp} = Acc,
{ok, Resp1} = chttpd:send_delayed_chunk(Resp, Buf),
- {ok, Acc#cacc{buffer = [], bufsize = 0, mochi = Resp1}};
+ {ok, Acc#chacc{buffer = [], bufsize = 0, mochi = Resp1}};
changes_callback(timeout, Acc) ->
- {ok, Resp1} = chttpd:send_delayed_chunk(Acc#cacc.mochi, "\n"),
- {ok, Acc#cacc{mochi = Resp1}};
-changes_callback({error, Reason}, #cacc{mochi = #httpd{}} = Acc) ->
- #cacc{mochi = Req} = Acc,
+ {ok, Resp1} = chttpd:send_delayed_chunk(Acc#chacc.mochi, "\n"),
+ {ok, Acc#chacc{mochi = Resp1}};
+changes_callback({error, Reason}, #chacc{mochi = #httpd{}} = Acc) ->
+ #chacc{mochi = Req} = Acc,
chttpd:send_error(Req, Reason);
-changes_callback({error, Reason}, #cacc{feed = normal, responding = false} = Acc) ->
- #cacc{mochi = Req} = Acc,
+changes_callback({error, Reason}, #chacc{feed = normal, responding = false} = Acc) ->
+ #chacc{mochi = Req} = Acc,
chttpd:send_error(Req, Reason);
changes_callback({error, Reason}, Acc) ->
- chttpd:send_delayed_error(Acc#cacc.mochi, Reason).
+ chttpd:send_delayed_error(Acc#chacc.mochi, Reason).
-maybe_flush_changes_feed(#cacc{bufsize=Size, threshold=Max} = Acc, Data, Len)
+maybe_flush_changes_feed(#chacc{bufsize=Size, threshold=Max} = Acc, Data, Len)
when Size > 0 andalso (Size + Len) > Max ->
- #cacc{buffer = Buffer, mochi = Resp} = Acc,
+ #chacc{buffer = Buffer, mochi = Resp} = Acc,
{ok, R1} = chttpd:send_delayed_chunk(Resp, Buffer),
- {ok, Acc#cacc{prepend = ",\r\n", buffer = Data, bufsize=Len, mochi = R1}};
+ {ok, Acc#chacc{prepend = ",\r\n", buffer = Data, bufsize=Len, mochi = R1}};
maybe_flush_changes_feed(Acc0, Data, Len) ->
- #cacc{buffer = Buf, bufsize = Size} = Acc0,
- Acc = Acc0#cacc{
+ #chacc{buffer = Buf, bufsize = Size} = Acc0,
+ Acc = Acc0#chacc{
prepend = ",\r\n",
buffer = [Buf | Data],
bufsize = Size + Len
diff --git a/src/couch/include/couch_db.hrl b/src/couch/include/couch_db.hrl
index feaa640..cdfbe67 100644
--- a/src/couch/include/couch_db.hrl
+++ b/src/couch/include/couch_db.hrl
@@ -203,6 +203,14 @@
atts = []
}).
+-record (cacc, {
+ db,
+ seq,
+ args,
+ options,
+ pending,
+ epochs
+}).
-type doc() :: #doc{}.
-type ddoc() :: #doc{}.
diff --git a/src/couch/src/couch_changes.erl b/src/couch/src/couch_changes.erl
index db465ee..99a3590 100644
--- a/src/couch/src/couch_changes.erl
+++ b/src/couch/src/couch_changes.erl
@@ -24,7 +24,8 @@
filter/3,
handle_db_event/3,
handle_view_event/3,
- view_filter/3
+ view_filter/3,
+ send_changes_doc_ids/6
]).
-export([changes_enumerator/2]).
@@ -627,7 +628,14 @@ send_lookup_changes(FullDocInfos, StartSeq, Dir, Db, Fun, Acc0) ->
{stop, Acc} -> Acc
end,
case Dir of
- fwd -> {ok, FinalAcc#changes_acc{seq = couch_db:get_update_seq(Db)}};
+ fwd ->
+ FinalAcc0 = case element(1, FinalAcc) of
+ changes_acc -> % we came here via couch_http or internal call
+ FinalAcc#changes_acc{seq = couch_db:get_update_seq(Db)};
+ cacc -> % we came here via chttpd / fabric / rexi
+ FinalAcc#cacc{seq = couch_db:get_update_seq(Db)}
+ end,
+ {ok, FinalAcc0};
rev -> {ok, FinalAcc}
end.
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index c8aa19e..9ded83d 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -32,15 +32,6 @@
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
--record (cacc, {
- db,
- seq,
- args,
- options,
- pending,
- epochs
-}).
-
%% rpc endpoints
%% call to with_db will supply your M:F with a Db instance
%% and then remaining args
@@ -85,7 +76,7 @@ changes(DbName, Options, StartVector, DbOptions) ->
},
try
{ok, #cacc{seq=LastSeq, pending=Pending, epochs=Epochs}} =
- couch_db:fold_changes(Db, StartSeq, Enum, Acc0, Opts),
+ do_changes(Db, StartSeq, Enum, Acc0, Opts),
rexi:stream_last({complete, [
{seq, {LastSeq, uuid(Db), couch_db:owner_of(Epochs, LastSeq)}},
{pending, Pending}
@@ -97,6 +88,24 @@ changes(DbName, Options, StartVector, DbOptions) ->
rexi:stream_last(Error)
end.
+do_changes(Db, StartSeq, Enum, Acc0, Opts) ->
+ #cacc {
+ args = Args
+ } = Acc0,
+ #changes_args {
+ filter = Filter
+ } = Args,
+ case Filter of
+ "_doc_ids" ->
+ #changes_args {
+ filter_fun = {doc_ids, Style, DocIds},
+ dir = Dir
+ } = Args,
+ couch_changes:send_changes_doc_ids(Db, StartSeq, Dir, Enum, Acc0, {doc_ids, Style, DocIds});
+ _ ->
+ couch_db:fold_changes(Db, StartSeq, Enum, Acc0, Opts)
+ end.
+
all_docs(DbName, Options, Args0) ->
case fabric_util:upgrade_mrargs(Args0) of
#mrargs{keys=undefined} = Args1 ->
diff --git a/src/mango/src/mango_cursor_text.erl b/src/mango/src/mango_cursor_text.erl
index 3883bc8..848d4d3 100644
--- a/src/mango/src/mango_cursor_text.erl
+++ b/src/mango/src/mango_cursor_text.erl
@@ -27,7 +27,7 @@
-include("mango.hrl").
--record(cacc, {
+-record(cuacc, {
selector,
dbname,
ddocid,
@@ -96,7 +96,7 @@ execute(Cursor, UserFun, UserAcc) ->
sort = sort_query(Opts, Selector),
raw_bookmark = true
},
- CAcc = #cacc{
+ CAcc = #cuacc{
selector = Selector,
dbname = couch_db:name(Db),
ddocid = ddocid(Idx),
@@ -114,7 +114,7 @@ execute(Cursor, UserFun, UserAcc) ->
execute(CAcc)
catch
throw:{stop, FinalCAcc} ->
- #cacc{
+ #cuacc{
bookmark = FinalBM,
user_fun = UserFun,
user_acc = LastUserAcc,
@@ -135,18 +135,18 @@ execute(CAcc) ->
% If we don't have any results from the
% query it means the request has paged through
% all possible results and the request is over.
- NewCAcc = CAcc#cacc{bookmark = Bookmark},
+ NewCAcc = CAcc#cuacc{bookmark = Bookmark},
throw({stop, NewCAcc});
{ok, Bookmark, Hits} ->
- NewCAcc = CAcc#cacc{bookmark = Bookmark},
- HitDocs = get_json_docs(CAcc#cacc.dbname, Hits),
+ NewCAcc = CAcc#cuacc{bookmark = Bookmark},
+ HitDocs = get_json_docs(CAcc#cuacc.dbname, Hits),
{ok, FinalCAcc} = handle_hits(NewCAcc, HitDocs),
execute(FinalCAcc)
end.
search_docs(CAcc) ->
- #cacc{
+ #cuacc{
dbname = DbName,
ddocid = DDocId,
idx_name = IdxName
@@ -169,17 +169,17 @@ handle_hits(CAcc0, [{Sort, Doc} | Rest]) ->
handle_hit(CAcc0, Sort, Doc) ->
- #cacc{
+ #cuacc{
limit = Limit,
skip = Skip,
execution_stats = Stats
} = CAcc0,
CAcc1 = update_bookmark(CAcc0, Sort),
Stats1 = mango_execution_stats:incr_docs_examined(Stats),
- CAcc2 = CAcc1#cacc{execution_stats = Stats1},
- case mango_selector:match(CAcc2#cacc.selector, Doc) of
+ CAcc2 = CAcc1#cuacc{execution_stats = Stats1},
+ case mango_selector:match(CAcc2#cuacc.selector, Doc) of
true when Skip > 0 ->
- CAcc2#cacc{skip = Skip - 1};
+ CAcc2#cuacc{skip = Skip - 1};
true when Limit == 0 ->
% We hit this case if the user spcified with a
% zero limit. Notice that in this case we need
@@ -190,15 +190,15 @@ handle_hit(CAcc0, Sort, Doc) ->
throw({stop, NewCAcc});
true when Limit > 1 ->
NewCAcc = apply_user_fun(CAcc2, Doc),
- NewCAcc#cacc{limit = Limit - 1};
+ NewCAcc#cuacc{limit = Limit - 1};
false ->
CAcc2
end.
apply_user_fun(CAcc, Doc) ->
- FinalDoc = mango_fields:extract(Doc, CAcc#cacc.fields),
- #cacc{
+ FinalDoc = mango_fields:extract(Doc, CAcc#cuacc.fields),
+ #cuacc{
user_fun = UserFun,
user_acc = UserAcc,
execution_stats = Stats
@@ -206,9 +206,9 @@ apply_user_fun(CAcc, Doc) ->
Stats0 = mango_execution_stats:incr_results_returned(Stats),
case UserFun({row, FinalDoc}, UserAcc) of
{ok, NewUserAcc} ->
- CAcc#cacc{user_acc = NewUserAcc, execution_stats = Stats0};
+ CAcc#cuacc{user_acc = NewUserAcc, execution_stats = Stats0};
{stop, NewUserAcc} ->
- throw({stop, CAcc#cacc{user_acc = NewUserAcc, execution_stats = Stats0}})
+ throw({stop, CAcc#cuacc{user_acc = NewUserAcc, execution_stats = Stats0}})
end.
@@ -247,11 +247,11 @@ get_bookmark(Opts) ->
update_bookmark(CAcc, Sortable) ->
- BM = CAcc#cacc.bookmark,
- QueryArgs = CAcc#cacc.query_args,
+ BM = CAcc#cuacc.bookmark,
+ QueryArgs = CAcc#cuacc.query_args,
Sort = QueryArgs#index_query_args.sort,
NewBM = dreyfus_bookmark:update(Sort, BM, [Sortable]),
- CAcc#cacc{bookmark = NewBM}.
+ CAcc#cuacc{bookmark = NewBM}.
pack_bookmark(Bookmark) ->
@@ -285,7 +285,7 @@ ddocid(Idx) ->
update_query_args(CAcc) ->
- #cacc{
+ #cuacc{
bookmark = Bookmark,
query_args = QueryArgs
} = CAcc,
@@ -296,7 +296,7 @@ update_query_args(CAcc) ->
get_limit(CAcc) ->
- erlang:min(get_dreyfus_limit(), CAcc#cacc.limit + CAcc#cacc.skip).
+ erlang:min(get_dreyfus_limit(), CAcc#cuacc.limit + CAcc#cuacc.skip).
get_dreyfus_limit() ->