You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2023/01/23 18:23:01 UTC
[couchdb] branch enable-doc-ids-limit-filter created (now b9b8d490d)
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a change to branch enable-doc-ids-limit-filter
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at b9b8d490d Enforce docs ids _changes filter optimization limit
This branch includes the following new commits:
new b9b8d490d Enforce docs ids _changes filter optimization limit
The 1 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.
[couchdb] 01/01: Enforce docs ids _changes filter optimization limit
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch enable-doc-ids-limit-filter
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit b9b8d490dc922b167a94807b4ec6ade0d1db1ae8
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Mon Jan 23 13:00:43 2023 -0500
Enforce docs ids _changes filter optimization limit
It turns out `changes_doc_ids_optimization_threshold` limit has never been
applied for the clustered changes feeds. So it was effectively unlimited. This
commit enables it, and also adds tests to ensure the limit works.
Since we didn't have a good Erlang integration test suite for clustered changes
feeds, which allowed this case to slip through the cracks, add a few more tests
along the way to test the majority of parameter combinations which might
interact: sharding single shards vs multiple, continuous vs normal, reverse,
row limits etc.
The previous limit was 100, but since it was never actually applied it's
equivalent not having one at all, so let's pick a new one. I chose 1000
noticing that at Cloudant, close to 3000 we had fabric timeouts on a busy
cluster, so that seemed too high. And 1000 seemed about the ballpark of the
what size of _bulk_get batch might be. Adding a benchmarking eunit test
https://gist.github.com/nickva/a21ef04b7e4bdbed5fdeb708f1d613b5 showed about
50-75 msec to query batches of 1000 random (uuid) doc_ids for Q values 1
through 8.
---
rel/overlay/etc/default.ini | 5 +-
src/chttpd/test/eunit/chttpd_changes_test.erl | 663 ++++++++++++++++++++++++++
src/couch/src/couch_changes.erl | 21 +-
src/fabric/src/fabric_rpc.erl | 41 +-
4 files changed, 691 insertions(+), 39 deletions(-)
diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini
index ae691bb8d..40f0ccd7a 100644
--- a/rel/overlay/etc/default.ini
+++ b/rel/overlay/etc/default.ini
@@ -45,7 +45,10 @@ view_index_dir = {{view_index_dir}}
; The speed of processing the _changes feed with doc_ids filter can be
; influenced directly with this setting - increase for faster processing at the
; expense of more memory usage.
-;changes_doc_ids_optimization_threshold = 100
+;
+; NOTE: Up until version 3.3.1 this setting did not actually take effect for
+; clustered _changes operations. It was applied to node local _changes feeds only.
+;changes_doc_ids_optimization_threshold = 1000
; Maximum document ID length. Can be set to an integer or 'infinity'.
;max_document_id_length = infinity
diff --git a/src/chttpd/test/eunit/chttpd_changes_test.erl b/src/chttpd/test/eunit/chttpd_changes_test.erl
new file mode 100644
index 000000000..fd68aab39
--- /dev/null
+++ b/src/chttpd/test/eunit/chttpd_changes_test.erl
@@ -0,0 +1,663 @@
+% 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_changes_test).
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch/include/couch_eunit.hrl").
+
+-define(USER, "chttpd_changes_test_admin").
+-define(PASS, "pass").
+-define(AUTH, {basic_auth, {?USER, ?PASS}}).
+-define(JSON, {"Content-Type", "application/json"}).
+
+-define(DOC1, <<"doc1">>).
+-define(DDC2, <<"_design/doc2">>).
+-define(DOC3, <<"doc3">>).
+-define(REVA, <<"a">>).
+-define(REVB, <<"b">>).
+-define(REVC, <<"c">>).
+-define(DELETED, true).
+-define(LEAFREV, false).
+
+% doc1 starts as rev-a, then gets 2 conflicting revisions b and c
+% ddoc2 starts as deleted at rev-a, then gets re-created as rev-c
+% doc3 starts as rev-a, then gets deleted as rev-c
+%
+test_docs() ->
+ [
+ {?DOC1, [?REVA], ?LEAFREV},
+ {?DDC2, [?REVA], ?DELETED},
+ {?DOC3, [?REVA], ?LEAFREV},
+ {?DOC1, [?REVB, ?REVA], ?LEAFREV},
+ {?DOC1, [?REVC, ?REVA], ?LEAFREV},
+ {?DOC3, [?REVB, ?REVA], ?DELETED},
+ {?DDC2, [?REVC, ?REVA], ?LEAFREV}
+ ].
+
+% Thesa are run against a Q=1, N=1 db, so we can make
+% some stronger assumptions about the exact Seq prefixes
+% returned sequences will have
+%
+changes_test_() ->
+ {
+ setup,
+ fun setup_basic/0,
+ fun teardown_basic/1,
+ with([
+ ?TDEF(t_basic),
+ ?TDEF(t_basic_post),
+ ?TDEF(t_continuous),
+ ?TDEF(t_continuous_zero_timeout),
+ ?TDEF(t_longpoll),
+ ?TDEF(t_limit_zero),
+ ?TDEF(t_continuous_limit_zero),
+ ?TDEF(t_limit_one),
+ ?TDEF(t_since_now),
+ ?TDEF(t_continuous_since_now),
+ ?TDEF(t_longpoll_since_now),
+ ?TDEF(t_style_all_docs),
+ ?TDEF(t_reverse),
+ ?TDEF(t_continuous_reverse),
+ ?TDEF(t_reverse_limit_zero),
+ ?TDEF(t_reverse_limit_one),
+ ?TDEF(t_seq_interval),
+ ?TDEF(t_selector_filter),
+ ?TDEF(t_design_filter),
+ ?TDEF(t_docs_id_filter),
+ ?TDEF(t_docs_id_filter_over_limit)
+ ])
+ }.
+
+% For Q=8 sharded dbs, unlike Q=1, we cannot make strong
+% assumptions about the exact sequence IDs for each row
+% so we'll test all the changes return and that the sequences
+% are increasing.
+%
+changes_q8_test_() ->
+ {
+ setup,
+ fun setup_q8/0,
+ fun teardown_basic/1,
+ with([
+ ?TDEF(t_basic_q8),
+ ?TDEF(t_continuous_q8),
+ ?TDEF(t_limit_zero),
+ ?TDEF(t_limit_one_q8),
+ ?TDEF(t_since_now),
+ ?TDEF(t_longpoll_since_now),
+ ?TDEF(t_reverse_q8),
+ ?TDEF(t_reverse_limit_zero),
+ ?TDEF(t_reverse_limit_one_q8),
+ ?TDEF(t_selector_filter),
+ ?TDEF(t_design_filter),
+ ?TDEF(t_docs_id_filter_q8)
+ ])
+ }.
+
+% These tests are separate as they create aditional design docs
+% as they so technically would be order dependent as the sequence
+% would keep climbing up from test to test. To avoid that run them
+% in a foreach context so setup/teardown happens for each test case.
+%
+changes_js_filters_test_() ->
+ {
+ foreach,
+ fun setup_basic/0,
+ fun teardown_basic/1,
+ [
+ ?TDEF_FE(t_js_filter),
+ ?TDEF_FE(t_js_filter_no_match),
+ ?TDEF_FE(t_js_filter_with_query_param),
+ ?TDEF_FE(t_view_filter),
+ ?TDEF_FE(t_view_filter_no_match)
+ ]
+ }.
+
+t_basic({_, DbUrl}) ->
+ Res = {Seq, Pending, Rows} = changes(DbUrl),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ ?assertEqual(
+ [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ),
+ % since=0 is the default, so it should look exactly the same
+ ?assertEqual(Res, changes(DbUrl, "?since=0")).
+
+t_basic_q8({_, DbUrl}) ->
+ {Seq, Pending, Rows} = changes(DbUrl),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ {Seqs, Revs, _Deleted} = lists:unzip3(Rows),
+ ?assertEqual(
+ [
+ {?DDC2, <<"2-c">>},
+ {?DOC1, <<"2-c">>},
+ {?DOC3, <<"2-b">>}
+ ],
+ lists:sort(Revs)
+ ),
+ ?assertEqual(Seqs, lists:sort(Seqs)).
+
+t_basic_post({_, DbUrl}) ->
+ {Seq, Pending, Rows} = changes_post(DbUrl, #{}),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ ?assertEqual(
+ [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ).
+
+t_continuous({_, DbUrl}) ->
+ Params = "?feed=continuous&timeout=10",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ ?assertEqual(
+ [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ).
+
+t_continuous_q8({_, DbUrl}) ->
+ Params = "?feed=continuous&timeout=10",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ {Seqs, Revs, _Deleted} = lists:unzip3(Rows),
+ ?assertEqual(
+ [
+ {?DDC2, <<"2-c">>},
+ {?DOC1, <<"2-c">>},
+ {?DOC3, <<"2-b">>}
+ ],
+ lists:sort(Revs)
+ ),
+ ?assertEqual(Seqs, lists:sort(Seqs)).
+
+t_continuous_zero_timeout({_, DbUrl}) ->
+ Params = "?feed=continuous&timeout=0",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ ?assertEqual(
+ [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ).
+
+t_longpoll({_, DbUrl}) ->
+ Params = "?feed=longpoll",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ ?assertEqual(
+ [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ).
+
+t_limit_zero({_, DbUrl}) ->
+ Params = "?limit=0",
+ ?assertEqual({0, 3, []}, changes(DbUrl, Params)).
+
+t_continuous_limit_zero({_, DbUrl}) ->
+ Params = "?feed=continuous&timeout=10&limit=0",
+ ?assertEqual({0, 3, []}, changes(DbUrl, Params)).
+
+t_limit_one({_, DbUrl}) ->
+ Params = "?limit=1",
+ ?assertEqual(
+ {5, 2, [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV}
+ ]},
+ changes(DbUrl, Params)
+ ).
+
+t_limit_one_q8({_, DbUrl}) ->
+ Params = "?limit=1",
+ ?assertMatch(
+ {_, _, [
+ {_, {<<_/binary>>, <<_/binary>>}, _}
+ ]},
+ changes(DbUrl, Params)
+ ).
+
+t_style_all_docs({_, DbUrl}) ->
+ Params = "?style=all_docs",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ ?assertEqual(
+ [
+ {5, {?DOC1, [<<"2-c">>, <<"2-b">>]}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ).
+
+t_since_now({_, DbUrl}) ->
+ Params = "?since=now",
+ ?assertEqual({7, 0, []}, changes(DbUrl, Params)).
+
+t_continuous_since_now({_, DbUrl}) ->
+ Params = "?feed=continuous&timeout=10&since=now",
+ ?assertEqual({7, 0, []}, changes(DbUrl, Params)).
+
+t_longpoll_since_now({_, DbUrl}) ->
+ Params = "?feed=longpoll&timeout=10&since=now",
+ ?assertEqual({7, 0, []}, changes(DbUrl, Params)).
+
+t_reverse({_, DbUrl}) ->
+ Params = "?descending=true",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(5, Seq),
+ ?assertEqual(-3, Pending),
+ ?assertEqual(
+ [
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ).
+
+t_continuous_reverse({_, DbUrl}) ->
+ Params = "?feed=continuous&timeout=10&descending=true",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(5, Seq),
+ ?assertEqual(-3, Pending),
+ ?assertEqual(
+ [
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ).
+
+t_reverse_q8({_, DbUrl}) ->
+ Params = "?descending=true",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(7, Seq),
+ ?assertEqual(-3, Pending),
+ {Seqs, Revs, _Deleted} = lists:unzip3(Rows),
+ ?assertEqual(
+ [
+ {?DDC2, <<"2-c">>},
+ {?DOC1, <<"2-c">>},
+ {?DOC3, <<"2-b">>}
+ ],
+ lists:sort(Revs)
+ ),
+ ?assertEqual(Seqs, lists:sort(Seqs)).
+
+t_reverse_limit_zero({_, DbUrl}) ->
+ Params = "?descending=true&limit=0",
+ ?assertEqual({7, 0, []}, changes(DbUrl, Params)).
+
+t_reverse_limit_one({_, DbUrl}) ->
+ Params = "?descending=true&limit=1",
+ ?assertEqual(
+ {7, -1, [
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV}
+ ]},
+ changes(DbUrl, Params)
+ ).
+
+t_reverse_limit_one_q8({_, DbUrl}) ->
+ Params = "?descending=true&limit=1",
+ ?assertMatch(
+ {7, -1, [
+ {_, {<<_/binary>>, <<_/binary>>}, _}
+ ]},
+ changes(DbUrl, Params)
+ ).
+
+t_seq_interval({_, DbUrl}) ->
+ Params = "?seq_interval=3",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ ?assertEqual(
+ [
+ {null, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {null, {?DDC2, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ).
+
+t_selector_filter({_, DbUrl}) ->
+ Params = "?filter=_selector",
+ Body = #{<<"selector">> => #{<<"_id">> => ?DOC1}},
+ {Seq, Pending, Rows} = changes_post(DbUrl, Body, Params),
+ ?assertEqual(7, Seq),
+ ?assertEqual(0, Pending),
+ ?assertMatch([{_, {?DOC1, <<"2-c">>}, ?LEAFREV}], Rows).
+
+t_design_filter({_, DbUrl}) ->
+ Params = "?filter=_design",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(7, Seq),
+ ?assertEqual(2, Pending),
+ ?assertMatch([{_, {?DDC2, <<"2-c">>}, ?LEAFREV}], Rows).
+
+t_docs_id_filter({_, DbUrl}) ->
+ Params = "?filter=_doc_ids",
+ Body = #{<<"doc_ids">> => [?DOC3, ?DOC1]},
+ meck:reset(couch_changes),
+ {_, _, Rows} = changes_post(DbUrl, Body, Params),
+ ?assertEqual(1, meck:num_calls(couch_changes, send_changes_doc_ids, 6)),
+ ?assertEqual(
+ [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED}
+ ],
+ Rows
+ ).
+
+t_docs_id_filter_q8({_, DbUrl}) ->
+ Params = "?filter=_doc_ids",
+ Body = #{<<"doc_ids">> => [?DOC3, ?DOC1]},
+ {_, _, Rows} = changes_post(DbUrl, Body, Params),
+ {Seqs, Revs, _Deleted} = lists:unzip3(Rows),
+ ?assertEqual(
+ [
+ {?DOC1, <<"2-c">>},
+ {?DOC3, <<"2-b">>}
+ ],
+ lists:sort(Revs)
+ ),
+ ?assertEqual(Seqs, lists:sort(Seqs)).
+
+t_docs_id_filter_over_limit({_, DbUrl}) ->
+ Params = "?filter=_doc_ids",
+ Body = #{<<"doc_ids">> => [<<"missingdoc">>, ?DOC3, <<"notthere">>, ?DOC1]},
+ meck:reset(couch_changes),
+ {_, _, Rows} = changes_post(DbUrl, Body, Params),
+ ?assertEqual(0, meck:num_calls(couch_changes, send_changes_doc_ids, 6)),
+ ?assertEqual(
+ [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED}
+ ],
+ Rows
+ ).
+
+t_js_filter({_, DbUrl}) ->
+ DDocId = "_design/filters",
+ FilterFun = <<"function(doc, req) {return (doc._id == 'doc3')}">>,
+ DDoc = #{<<"filters">> => #{<<"f">> => FilterFun}},
+ DDocUrl = DbUrl ++ "/" ++ DDocId,
+ {_, #{<<"rev">> := Rev, <<"ok">> := true}} = req(put, DDocUrl, DDoc),
+ Params = "?filter=filters/f",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(8, Seq),
+ ?assertEqual(0, Pending),
+ ?assertEqual(
+ [
+ {6, {?DOC3, <<"2-b">>}, ?DELETED}
+ ],
+ Rows
+ ),
+ {200, #{}} = req(delete, DDocUrl ++ "?rev=" ++ binary_to_list(Rev)).
+
+t_js_filter_no_match({_, DbUrl}) ->
+ DDocId = "_design/filters",
+ FilterFun = <<"function(doc, req) {return false}">>,
+ DDoc = #{<<"filters">> => #{<<"f">> => FilterFun}},
+ DDocUrl = DbUrl ++ "/" ++ DDocId,
+ {_, #{<<"rev">> := Rev, <<"ok">> := true}} = req(put, DDocUrl, DDoc),
+ Params = "?filter=filters/f",
+ ?assertEqual({8, 0, []}, changes(DbUrl, Params)),
+ {200, #{}} = req(delete, DDocUrl ++ "?rev=" ++ binary_to_list(Rev)).
+
+t_js_filter_with_query_param({_, DbUrl}) ->
+ DDocId = "_design/filters",
+ FilterFun = <<"function(doc, req) {return (req.query.yup == 1)}">>,
+ DDoc = #{<<"filters">> => #{<<"f">> => FilterFun}},
+ DDocUrl = DbUrl ++ "/" ++ DDocId,
+ {_, #{<<"rev">> := Rev, <<"ok">> := true}} = req(put, DDocUrl, DDoc),
+ Params = "?filter=filters/f&yup=1",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(8, Seq),
+ ?assertEqual(0, Pending),
+ ?assertMatch(
+ [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV},
+ {6, {?DOC3, <<"2-b">>}, ?DELETED},
+ {7, {?DDC2, <<"2-c">>}, ?LEAFREV},
+ {8, {<<"_design/filters">>, <<"1-", _/binary>>}, ?LEAFREV}
+ ],
+ Rows
+ ),
+ {200, #{}} = req(delete, DDocUrl ++ "?rev=" ++ binary_to_list(Rev)).
+
+t_view_filter({_, DbUrl}) ->
+ DDocId = "_design/views",
+ ViewFun = <<"function(doc) {if (doc._id == 'doc1') {emit(1, 1);}}">>,
+ DDoc = #{<<"views">> => #{<<"v">> => #{<<"map">> => ViewFun}}},
+ DDocUrl = DbUrl ++ "/" ++ DDocId,
+ {_, #{<<"rev">> := Rev, <<"ok">> := true}} = req(put, DDocUrl, DDoc),
+ Params = "?filter=_view&view=views/v",
+ {Seq, Pending, Rows} = changes(DbUrl, Params),
+ ?assertEqual(8, Seq),
+ ?assertEqual(0, Pending),
+ ?assertEqual(
+ [
+ {5, {?DOC1, <<"2-c">>}, ?LEAFREV}
+ ],
+ Rows
+ ),
+ {200, #{}} = req(delete, DDocUrl ++ "?rev=" ++ binary_to_list(Rev)).
+
+t_view_filter_no_match({_, DbUrl}) ->
+ DDocId = "_design/views",
+ ViewFun = <<"function(doc) {if (doc._id == 'docX') {emit(1, 1);}}">>,
+ DDoc = #{<<"views">> => #{<<"v">> => #{<<"map">> => ViewFun}}},
+ DDocUrl = DbUrl ++ "/" ++ DDocId,
+ {_, #{<<"rev">> := Rev, <<"ok">> := true}} = req(put, DDocUrl, DDoc),
+ Params = "?filter=_view&view=views/v",
+ ?assertEqual({8, 0, []}, changes(DbUrl, Params)),
+ {200, #{}} = req(delete, DDocUrl ++ "?rev=" ++ binary_to_list(Rev)).
+
+post_doc_ids(DbUrl, Body) ->
+ T0 = erlang:monotonic_time(),
+ {_, _, Rows} = changes_post(DbUrl, Body, "?filter=_doc_ids"),
+ DT = erlang:monotonic_time() - T0,
+ DTMsec = erlang:convert_time_unit(DT, native, millisecond),
+ ?debugVal(DTMsec),
+ Rows.
+
+% Utility functions
+
+setup_ctx(DbCreateParams) ->
+ Ctx = test_util:start_couch([chttpd]),
+ Hashed = couch_passwords:hash_admin_password(?PASS),
+ ok = config:set("admins", ?USER, ?b2l(Hashed), _Persist = false),
+ Addr = config:get("chttpd", "bind_address", "127.0.0.1"),
+ Db = ?b2l(?tempdb()),
+ Port = mochiweb_socket_server:get(chttpd, port),
+ Url = lists:concat(["http://", Addr, ":", Port, "/"]),
+ ok = create_db(Url, Db, DbCreateParams),
+ {Ctx, Url ++ Db}.
+
+teardown_ctx({Ctx, DbUrl}) ->
+ meck:unload(),
+ delete_db(DbUrl),
+ ok = config:delete("admins", ?USER, _Persist = false),
+ test_util:stop_couch(Ctx).
+
+setup_q8() ->
+ {Ctx, DbUrl} = setup_ctx("?q=8"),
+ ok = create_docs(DbUrl, test_docs()),
+ {Ctx, DbUrl}.
+
+setup_basic() ->
+ {Ctx, DbUrl} = setup_ctx("?q=1"),
+ ok = create_docs(DbUrl, test_docs()),
+ CfgKey = "changes_doc_ids_optimization_threshold",
+ ok = config:set("couchdb", CfgKey, "2", _Persist = false),
+ meck:new(couch_changes, [passthrough]),
+ {Ctx, DbUrl}.
+
+teardown_basic({Ctx, DbUrl}) ->
+ CfgKey = "changes_doc_ids_optimization_threshold",
+ ok = config:delete("couchdb", CfgKey, _Persist = false),
+ teardown_ctx({Ctx, DbUrl}).
+
+create_db(Top, Db, Params) ->
+ case req(put, Top ++ Db ++ Params) of
+ {201, #{}} ->
+ ok;
+ Error ->
+ error({failed_to_create_test_db, Db, Error})
+ end.
+
+delete_db(DbUrl) ->
+ case req(delete, DbUrl) of
+ {200, #{}} ->
+ ok;
+ Error ->
+ error({failed_to_delete_test_db, DbUrl, Error})
+ end.
+
+doc_fun({Id, Revs, Deleted}) ->
+ Doc = #{
+ <<"_id">> => Id,
+ <<"_revisions">> => #{
+ <<"ids">> => Revs,
+ <<"start">> => length(Revs)
+ }
+ },
+ case Deleted of
+ true -> Doc#{<<"_deleted">> => true};
+ false -> Doc
+ end.
+
+create_docs(DbUrl, DocRevs) ->
+ lists:foreach(
+ fun(#{<<"_id">> := Id} = Doc) ->
+ ?debugVal(Id),
+ Url = DbUrl ++ "/" ++ binary_to_list(Id),
+ {_, #{<<"ok">> := true}} = req(put, Url ++ "?new_edits=false", Doc)
+ end,
+ lists:map(fun doc_fun/1, DocRevs)
+ ).
+
+changes(DbUrl) ->
+ changes(DbUrl, "").
+
+changes_post(DbUrl, #{} = Body) ->
+ changes_post(DbUrl, Body, "").
+
+changes(DbUrl, Params) when is_list(Params) ->
+ {Code, Res} = reqraw(get, DbUrl ++ "/_changes" ++ Params),
+ ?assertEqual(200, Code),
+ res(Res, Params).
+
+changes_post(DbUrl, #{} = Body, Params) ->
+ {Code, Res} = reqraw(post, DbUrl ++ "/_changes" ++ Params, Body),
+ ?assertEqual(200, Code),
+ res(Res, Params).
+
+req(Method, Url) ->
+ {Code, Res} = reqraw(Method, Url),
+ {Code, json(Res)}.
+
+req(Method, Url, #{} = Body) ->
+ {Code, Res} = reqraw(Method, Url, Body),
+ {Code, json(Res)}.
+
+reqraw(Method, Url) ->
+ Headers = [?JSON, ?AUTH],
+ {ok, Code, _, Res} = test_request:request(Method, Url, Headers),
+ {Code, Res}.
+
+reqraw(Method, Url, #{} = Body) ->
+ reqraw(Method, Url, jiffy:encode(Body));
+reqraw(Method, Url, Body) ->
+ Headers = [?JSON, ?AUTH],
+ {ok, Code, _, Res} = test_request:request(Method, Url, Headers, Body),
+ {Code, Res}.
+
+json(Bin) when is_binary(Bin) ->
+ jiffy:decode(Bin, [return_maps]).
+
+jsonl(Bin) when is_binary(Bin) ->
+ Lines = [string:trim(L) || L <- binary:split(Bin, <<"\n">>, [global])],
+ [json(L) || L <- Lines, L =/= <<>>].
+
+res(<<_/binary>> = Bin, Params) ->
+ Continuous = string:find(Params, "feed=continuous") =/= nomatch,
+ case Continuous of
+ true -> parse_response(jsonl(Bin));
+ false -> parse_response(json(Bin))
+ end.
+
+parse_response(#{} = Resp) ->
+ #{
+ <<"last_seq">> := LastSeq,
+ <<"pending">> := Pending,
+ <<"results">> := Results
+ } = Resp,
+ Results1 = lists:map(fun parse_row/1, Results),
+ {seq(LastSeq), Pending, Results1};
+parse_response([#{} | _] = Lines) ->
+ #{<<"pending">> := Pending, <<"last_seq">> := LastSeq} = lists:last(Lines),
+ Results1 = lists:map(fun parse_row/1, lists:droplast(Lines)),
+ {seq(LastSeq), Pending, Results1}.
+
+parse_row(#{} = Row) ->
+ #{
+ <<"changes">> := Revs,
+ <<"id">> := Id,
+ <<"seq">> := Seq
+ } = Row,
+ Revs1 = lists:map(fun(#{<<"rev">> := Rev}) -> Rev end, Revs),
+ Revs2 =
+ case Revs1 of
+ [Rev] -> Rev;
+ [_ | _] -> Revs1
+ end,
+ case maps:get(<<"deleted">>, Row, false) of
+ true -> {seq(Seq), {Id, Revs2}, ?DELETED};
+ false -> {seq(Seq), {Id, Revs2}, ?LEAFREV}
+ end.
+
+% This will be reliable for q=1 dbs only.
+%
+seq(<<_/binary>> = Seq) ->
+ [NumStr, _] = binary:split(Seq, <<"-">>),
+ binary_to_integer(NumStr);
+seq(null) ->
+ null.
diff --git a/src/couch/src/couch_changes.erl b/src/couch/src/couch_changes.erl
index 089cda975..0aa0c277f 100644
--- a/src/couch/src/couch_changes.erl
+++ b/src/couch/src/couch_changes.erl
@@ -12,7 +12,6 @@
-module(couch_changes).
-include_lib("couch/include/couch_db.hrl").
--include_lib("couch_mrview/include/couch_mrview.hrl").
-export([
handle_db_changes/3,
@@ -24,7 +23,8 @@
handle_db_event/3,
handle_view_event/3,
send_changes_doc_ids/6,
- send_changes_design_docs/6
+ send_changes_design_docs/6,
+ doc_ids_limit/0
]).
-export([changes_enumerator/2]).
@@ -34,6 +34,9 @@
keep_sending_changes/3
]).
+% Default max doc ids optimization limit.
+-define(MAX_DOC_IDS, 1000).
+
-record(changes_acc, {
db,
seq,
@@ -457,15 +460,10 @@ send_changes(Acc, Dir, FirstRound) ->
end.
can_optimize(true, {doc_ids, _Style, DocIds}) ->
- MaxDocIds = config:get_integer(
- "couchdb",
- "changes_doc_ids_optimization_threshold",
- 100
- ),
- if
- length(DocIds) =< MaxDocIds ->
- {true, fun send_changes_doc_ids/6};
+ case length(DocIds) =< doc_ids_limit() of
true ->
+ {true, fun send_changes_doc_ids/6};
+ false ->
false
end;
can_optimize(true, {design_docs, _Style}) ->
@@ -473,6 +471,9 @@ can_optimize(true, {design_docs, _Style}) ->
can_optimize(_, _) ->
false.
+doc_ids_limit() ->
+ config:get_integer("couchdb", "changes_doc_ids_optimization_threshold", ?MAX_DOC_IDS).
+
send_changes_doc_ids(Db, StartSeq, Dir, Fun, Acc0, {doc_ids, _Style, DocIds}) ->
Results = couch_db:get_full_doc_infos(Db, DocIds),
FullInfos = lists:foldl(
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 8b9c94c87..b781eea99 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -116,36 +116,21 @@ changes(DbName, Options, StartVector, DbOptions) ->
rexi:stream_last(Error)
end.
-do_changes(Db, StartSeq, Enum, Acc0, Opts) ->
- #fabric_changes_acc{
- args = Args
- } = Acc0,
- #changes_args{
- filter = Filter
- } = Args,
+do_changes(Db, Seq, Enum, #fabric_changes_acc{args = Args} = Acc0, Opts) ->
+ #changes_args{filter_fun = Filter, dir = Dir} = Args,
+
case Filter of
- "_doc_ids" ->
- % optimised code path, we’re looking up all doc_ids in the by-id instead of filtering
- % the entire by-seq tree to find the doc_ids one by one
- #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}
- );
- "_design_docs" ->
- % optimised code path, we’re looking up all design_docs in the by-id instead of
- % filtering the entire by-seq tree to find the design_docs one by one
- #changes_args{
- filter_fun = {design_docs, Style},
- dir = Dir
- } = Args,
- couch_changes:send_changes_design_docs(
- Db, StartSeq, Dir, Enum, Acc0, {design_docs, Style}
- );
+ {doc_ids, _Style, DocIds} ->
+ case length(DocIds) =< couch_changes:doc_ids_limit() of
+ true ->
+ couch_changes:send_changes_doc_ids(Db, Seq, Dir, Enum, Acc0, Filter);
+ false ->
+ couch_db:fold_changes(Db, Seq, Enum, Acc0, Opts)
+ end;
+ {design_docs, _Style} ->
+ couch_changes:send_changes_design_docs(Db, Seq, Dir, Enum, Acc0, Filter);
_ ->
- couch_db:fold_changes(Db, StartSeq, Enum, Acc0, Opts)
+ couch_db:fold_changes(Db, Seq, Enum, Acc0, Opts)
end.
all_docs(DbName, Options, Args0) ->