You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by rn...@apache.org on 2018/08/20 15:19:29 UTC
[couchdb] branch user-partitioned-dbs-6 updated (fcbe16c -> 6f38cb7)
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a change to branch user-partitioned-dbs-6
in repository https://gitbox.apache.org/repos/asf/couchdb.git.
discard fcbe16c optimize _all_docs requests that are bounded within a single partition
discard e99839f Implement _all_docs and _find support
discard b715be5 mem3 tests for partitioned databases
discard 6cf8819 implement partitioned views
new 775098c implement partitioned views
new d650f30 mem3 tests for partitioned databases
new 1d6bf96 Implement _all_docs and _find support
new 6f38cb7 optimize _all_docs requests that are bounded within a single partition
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (fcbe16c)
\
N -- N -- N refs/heads/user-partitioned-dbs-6 (6f38cb7)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 4 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.
Summary of changes:
src/couch_mrview/src/couch_mrview_util.erl | 3 +++
1 file changed, 3 insertions(+)
[couchdb] 02/04: mem3 tests for partitioned databases
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-6
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit d650f30fc2b650a1bd075b66c450b3608028b481
Author: Garren Smith <ga...@gmail.com>
AuthorDate: Tue Aug 7 12:15:14 2018 +0200
mem3 tests for partitioned databases
---
src/mem3/src/mem3.erl | 22 ++++++++++++++
src/mem3/src/mem3_shards.erl | 68 +++++++++++++++++++++++++++++++++++++++++++-
src/mem3/src/mem3_util.erl | 59 +++++++++++++++++++++++++++++++++++++-
3 files changed, 147 insertions(+), 2 deletions(-)
diff --git a/src/mem3/src/mem3.erl b/src/mem3/src/mem3.erl
index 99c0863..aecca2f 100644
--- a/src/mem3/src/mem3.erl
+++ b/src/mem3/src/mem3.erl
@@ -389,4 +389,26 @@ allowed_nodes_test_() ->
]
}]}.
+is_partitioned_false_shards_db_test() ->
+ meck:expect(config, get, fun (_, _, Default) -> Default end),
+ ?assertEqual(is_partitioned(<<"_dbs">>), false),
+ meck:unload().
+
+is_partitioned_false_nodes_db_test() ->
+ meck:expect(config, get, fun (_, _, Default) -> Default end),
+ ?assertEqual(is_partitioned(<<"_nodes">>), false),
+ meck:unload().
+
+is_partitioned_true_partitioned_db_test() ->
+ Shard = #shard{
+ opts = [{partitioned, true}]
+ },
+ ?assertEqual(is_partitioned([Shard]), true).
+
+is_partitioned_false_partitioned_db_test() ->
+ Shard = #shard{
+ opts = []
+ },
+ ?assertEqual(is_partitioned([Shard]), false).
+
-endif.
diff --git a/src/mem3/src/mem3_shards.erl b/src/mem3/src/mem3_shards.erl
index 1cd7be6..cfbc32e 100644
--- a/src/mem3/src/mem3_shards.erl
+++ b/src/mem3/src/mem3_shards.erl
@@ -559,7 +559,9 @@ mem3_shards_test_() ->
t_writer_does_not_delete_other_writers_for_same_shard(),
t_spawn_writer_in_load_shards_from_db(),
t_cache_insert_takes_new_update(),
- t_cache_insert_ignores_stale_update_and_kills_worker()
+ t_cache_insert_ignores_stale_update_and_kills_worker(),
+ t_load_shards_from_disk_returns_correct_shard_for_partition(),
+ t_for_docid_returns_correct_shard_for_partition()
]
}.
@@ -703,6 +705,70 @@ t_cache_insert_ignores_stale_update_and_kills_worker() ->
?assertEqual([], ets:tab2list(?OPENERS))
end).
+t_load_shards_from_disk_returns_correct_shard_for_partition() ->
+ ?_test(begin
+ Shards = [
+ #ordered_shard{
+ name = <<"shards/80000000-9fffffff/db1.1533630706">>,
+ node = 'node1@127.0.0.1',
+ dbname = <<"db1">>,
+ range = [2147483648,2684354559],
+ ref = undefined,
+ order = 1,
+ opts = [{partitioned,true}]
+ }
+ ],
+ DbName = <<"db1">>,
+ DocId = <<"foo:123">>,
+ Doc = #doc{body = {[]}},
+ meck:expect(couch_db, open_doc, 3, {ok, Doc}),
+ meck:expect(couch_db, get_update_seq, 1, 1),
+ meck:expect(mem3_util, build_ordered_shards, 2, Shards),
+ meck:expect(mem3_util, ensure_exists, 1, {ok, <<"shard-name">>}),
+ meck:expect(couch_db, close, 1, ok),
+
+ [Shard] = load_shards_from_disk(DbName, DocId),
+
+ meck:validate(couch_db),
+ meck:validate(mem3_util),
+
+ ShardName = Shard#ordered_shard.name,
+ ?assertEqual(ShardName, <<"shards/80000000-9fffffff/db1.1533630706">>)
+ end).
+
+t_for_docid_returns_correct_shard_for_partition() ->
+ ?_test(begin
+ Shards = [
+ #ordered_shard{
+ name = <<"shards/60000000-7fffffff/db1.1533630706">>,
+ node = 'node1@127.0.0.1',
+ dbname = <<"db1">>,
+ range = [1610612736,2147483647],
+ ref = undefined,
+ order = 1,
+ opts = [{partitioned,true}]
+ },
+ #ordered_shard{
+ name = <<"shards/80000000-9fffffff/db1.1533630706">>,
+ node = 'node1@127.0.0.1',
+ dbname = <<"db1">>,
+ range = [2147483648,2684354559],
+ ref = undefined,
+ order = 1,
+ opts = [{partitioned,true}]
+ }
+ ],
+ DbName = <<"db1">>,
+ DocId = <<"foo:123">>,
+
+ true = ets:insert(?SHARDS, Shards),
+
+ [Shard] = for_docid(DbName, DocId, [ordered]),
+
+ ShardName = Shard#ordered_shard.name,
+ ?assertEqual(ShardName, <<"shards/80000000-9fffffff/db1.1533630706">>)
+ end).
+
mock_state(UpdateSeq) ->
#st{
diff --git a/src/mem3/src/mem3_util.erl b/src/mem3/src/mem3_util.erl
index 9620e98..7b8dd16 100644
--- a/src/mem3/src/mem3_util.erl
+++ b/src/mem3/src/mem3_util.erl
@@ -35,7 +35,6 @@ hash(Item) when is_binary(Item) ->
hash(Item) ->
erlang:crc32(term_to_binary(Item)).
-
docid_hash(DocId) when is_binary(DocId) ->
docid_hash(DocId, []).
@@ -294,3 +293,61 @@ downcast(#ordered_shard{}=S) ->
};
downcast(Shards) when is_list(Shards) ->
[downcast(Shard) || Shard <- Shards].
+
+-ifdef(TEST).
+
+-include_lib("eunit/include/eunit.hrl").
+
+docid_hash_design_doc_test() ->
+ Id = <<"_design/ddoc">>,
+ Hash = docid_hash(Id),
+ ?assertEqual(Hash, erlang:crc32(Id)).
+
+docid_hash_doc_partition_false_test() ->
+ Id = <<"partitionkey:docid">>,
+ IdHash = erlang:crc32(Id),
+ Hash = docid_hash(Id),
+ ?assertEqual(Hash, IdHash),
+ Hash = docid_hash(Id, []),
+ ?assertEqual(Hash, IdHash).
+
+docid_hash_doc_partition_true_test() ->
+ Id = <<"partitionkey:doc:id">>,
+ Hash = docid_hash(Id, [{partitioned, true}]),
+ ?assertEqual(Hash, erlang:crc32(<<"partitionkey">>)).
+
+
+add_shards_by_node_adds_partition_prop_test() ->
+ DocProp = [
+ {<<"_id">>, <<"database-name">>},
+ {<<"_rev">>,<<"1-fb8e28457a6e0c49de1848b5e4a28238">>},
+ {<<"shard_suffix">>,".1533550200"},
+ {<<"changelog">>, [[<<"add">>,<<"00000000-1fffffff">>,<<"node1@127.0.0.1">>]]},
+ {<<"by_node">>, {[{<<"node1@127.0.0.1">>, [<<"00000000-1fffffff">>,<<"20000000-3fffffff">>]}]}},
+ {<<"by_range">>, {[{<<"00000000-1fffffff">>,[<<"node1@127.0.0.1">>]}]}},
+ {<<"options">>,{[{partitioned,true}]}}
+ ],
+
+ [ShardRange | _] = build_shards_by_node(<<"database-name">>, DocProp),
+ Opts = ShardRange#shard.opts,
+ Partitioned = lists:keyfind(partitioned, 1, Opts),
+ ?assertEqual(Partitioned, {partitioned, true}).
+
+
+add_shards_by_range_adds_partition_prop_test() ->
+ DocProp = [
+ {<<"_id">>, <<"database-name">>},
+ {<<"_rev">>,<<"1-fb8e28457a6e0c49de1848b5e4a28238">>},
+ {<<"shard_suffix">>,".1533550200"},
+ {<<"changelog">>, [[<<"add">>,<<"00000000-1fffffff">>,<<"node1@127.0.0.1">>]]},
+ {<<"by_node">>, {[{<<"node1@127.0.0.1">>, [<<"00000000-1fffffff">>,<<"20000000-3fffffff">>]}]}},
+ {<<"by_range">>, {[{<<"00000000-1fffffff">>,[<<"node1@127.0.0.1">>]}]}},
+ {<<"options">>,{[{partitioned,true}]}}
+ ],
+
+ [ShardRange | _] = build_shards_by_range(<<"database-name">>, DocProp),
+ Opts = ShardRange#ordered_shard.opts,
+ Partitioned = lists:keyfind(partitioned, 1, Opts),
+ ?assertEqual(Partitioned, {partitioned, true}).
+
+-endif.
[couchdb] 04/04: optimize _all_docs requests that are bounded
within a single partition
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-6
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 6f38cb763cb393823023da41e9b42ed7cef527af
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Aug 13 22:34:59 2018 +0100
optimize _all_docs requests that are bounded within a single partition
---
src/fabric/src/fabric_view_all_docs.erl | 24 +++++++++++++++++++++++-
1 file changed, 23 insertions(+), 1 deletion(-)
diff --git a/src/fabric/src/fabric_view_all_docs.erl b/src/fabric/src/fabric_view_all_docs.erl
index d515ab8..83c3790 100644
--- a/src/fabric/src/fabric_view_all_docs.erl
+++ b/src/fabric/src/fabric_view_all_docs.erl
@@ -21,7 +21,7 @@
-include_lib("couch_mrview/include/couch_mrview.hrl").
go(DbName, Options, #mrargs{keys=undefined} = QueryArgs, Callback, Acc) ->
- Shards = mem3:shards(DbName),
+ Shards = shards(DbName, QueryArgs),
Workers0 = fabric_util:submit_jobs(
Shards, fabric_rpc, all_docs, [Options, QueryArgs]),
RexiMon = fabric_util:create_monitors(Workers0),
@@ -136,6 +136,28 @@ go(DbName, _Options, Workers, QueryArgs, Callback, Acc0) ->
{ok, Resp}
end.
+shards(DbName, Args) ->
+ case couch_mrview_util:get_extra(Args, partitioned) of
+ true ->
+ StartKey = partition(Args#mrargs.start_key),
+ EndKey = partition(Args#mrargs.end_key),
+ case {StartKey, EndKey} of
+ {Same, Same} when Same =/= undefined ->
+ mem3:shards(DbName, <<Same/binary, ":foo">>);
+ {_, _} ->
+ mem3:shards(DbName)
+ end;
+ _ ->
+ mem3:shards(DbName)
+ end.
+
+partition(undefined) ->
+ undefined;
+partition(null) ->
+ null;
+partition(Key) when is_binary(Key) ->
+ hd(binary:split(Key, <<":">>)).
+
handle_message({rexi_DOWN, _, {_, NodeRef}, _}, _, State) ->
fabric_view:check_down_shards(State, NodeRef);
[couchdb] 03/04: Implement _all_docs and _find support
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-6
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 1d6bf963d6b7959e725bd7898bcc3772c149a735
Author: Robert Newson <rn...@apache.org>
AuthorDate: Thu Aug 9 13:43:17 2018 +0100
Implement _all_docs and _find support
Co-authored-by: Garren Smith <ga...@gmail.com>
Co-authored-by: Robert Newson <rn...@apache.org>
---
src/chttpd/src/chttpd_db.erl | 9 ++--
src/couch_mrview/src/couch_mrview.erl | 7 +--
src/couch_mrview/src/couch_mrview_updater.erl | 4 +-
src/couch_mrview/src/couch_mrview_util.erl | 77 ++++++++++++++++++++++-----
src/fabric/src/fabric.erl | 7 ++-
src/fabric/src/fabric_view.erl | 2 +-
src/fabric/src/fabric_view_all_docs.erl | 2 +-
src/mango/src/mango_cursor_view.erl | 23 ++++++--
src/mango/src/mango_opts.erl | 6 +++
src/mem3/src/mem3.erl | 5 +-
10 files changed, 113 insertions(+), 29 deletions(-)
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index 57d85e1..83d0489 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -684,12 +684,15 @@ multi_all_docs_view(Req, Db, OP, Queries) ->
all_docs_view(Req, Db, Keys, OP) ->
Args0 = couch_mrview_http:parse_params(Req, Keys),
Args1 = Args0#mrargs{view_type=map},
- Args2 = couch_mrview_util:validate_args(Args1),
- Args3 = set_namespace(OP, Args2),
+ DbPartitioned = mem3:is_partitioned(couch_db:name(Db)),
+ Args2 = couch_mrview_util:set_extra(Args1, partitioned, DbPartitioned),
+ Args3 = couch_mrview_util:set_extra(Args2, style, all_docs),
+ Args4 = couch_mrview_util:validate_args(Args3),
+ Args5 = set_namespace(OP, Args4),
Options = [{user_ctx, Req#httpd.user_ctx}],
Max = chttpd:chunked_response_buffer_size(),
VAcc = #vacc{db=Db, req=Req, threshold=Max},
- {ok, Resp} = fabric:all_docs(Db, Options, fun couch_mrview_http:view_cb/2, VAcc, Args3),
+ {ok, Resp} = fabric:all_docs(Db, Options, fun couch_mrview_http:view_cb/2, VAcc, Args5),
{ok, Resp#vacc.resp}.
db_doc_req(#httpd{method='DELETE'}=Req, Db, DocId) ->
diff --git a/src/couch_mrview/src/couch_mrview.erl b/src/couch_mrview/src/couch_mrview.erl
index 09945f5..f5963e7 100644
--- a/src/couch_mrview/src/couch_mrview.erl
+++ b/src/couch_mrview/src/couch_mrview.erl
@@ -228,12 +228,13 @@ query_all_docs(Db, Args0, Callback, Acc) ->
couch_index_util:hexsig(couch_hash:md5_hash(term_to_binary(Info)))
end),
Args1 = Args0#mrargs{view_type=map},
- Args2 = couch_mrview_util:validate_args(Args1),
- {ok, Acc1} = case Args2#mrargs.preflight_fun of
+ Args2 = couch_mrview_util:set_extra(Args1, style, all_docs),
+ Args3 = couch_mrview_util:validate_args(Args2),
+ {ok, Acc1} = case Args3#mrargs.preflight_fun of
PFFun when is_function(PFFun, 2) -> PFFun(Sig, Acc);
_ -> {ok, Acc}
end,
- all_docs_fold(Db, Args2, Callback, Acc1).
+ all_docs_fold(Db, Args3, Callback, Acc1).
query_view(Db, DDoc, VName) ->
diff --git a/src/couch_mrview/src/couch_mrview_updater.erl b/src/couch_mrview/src/couch_mrview_updater.erl
index 2b69eee..0c1a17c 100644
--- a/src/couch_mrview/src/couch_mrview_updater.erl
+++ b/src/couch_mrview/src/couch_mrview_updater.erl
@@ -315,7 +315,9 @@ write_kvs(State, UpdateSeq, ViewKVs, DocIdKeys, Seqs, Log0) ->
design_opts=DesignOpts
} = State,
- Partitioned = couch_util:get_value(<<"partitioned">>, DesignOpts, false),
+ DbPartitioned = mem3:is_partitioned(State#mrst.db_name),
+ Partitioned = couch_util:get_value(<<"partitioned">>, DesignOpts, DbPartitioned),
+
Revs = dict:from_list(dict:fetch_keys(Log0)),
Log = dict:fold(fun({Id, _Rev}, DIKeys, Acc) ->
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index bb1e365..4cfea46 100644
--- a/src/couch_mrview/src/couch_mrview_util.erl
+++ b/src/couch_mrview/src/couch_mrview_util.erl
@@ -39,8 +39,11 @@
-define(GET_VIEW_RETRY_COUNT, 1).
-define(GET_VIEW_RETRY_DELAY, 50).
-define(LOWEST_KEY, null).
--define(HIGHEST_KEY, {[{<<239, 191, 176>>, null}]}). % is {"\ufff0": null}
-
+-define(HIGHEST_KEY, {<<255, 255, 255, 255>>}).
+-define(PARTITION_START(P), <<P/binary, $:>>).
+-define(PARTITION_END(P), <<P/binary, $;>>).
+-define(LOWEST(A, B), (if A < B -> A; true -> B end)).
+-define(HIGHEST(A, B), (if A > B -> A; true -> B end)).
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
@@ -583,20 +586,38 @@ validate_args(Args) ->
_ -> mrverror(<<"Invalid value for `sorted`.">>)
end,
- case {get_extra(Args, partitioned, false), get_extra(Args, partition)} of
- {true, undefined} ->
+ Style = get_extra(Args, style, normal),
+ Partitioned = get_extra(Args, partitioned, false),
+ Partition = get_extra(Args, partition),
+
+ case {Style, Partitioned, Partition} of
+ {all_docs, true, _} ->
+ ok; % _all_docs can be called with or without partition parameter.
+ {all_docs, false, undefined} ->
+ ok;
+ {all_docs, false, _Partition} ->
+ mrverror(<<"`partition` parameter is not supported in this db.">>);
+ {normal, true, undefined} ->
mrverror(<<"`partition` parameter is mandatory for queries to this view.">>);
- {true, _Partition} ->
+ {normal, true, _Partition} ->
ok;
- {false, undefined} ->
+ {normal, false, undefined} ->
ok;
- {false, _Partition} ->
+ {normal, false, _Partition} ->
mrverror(<<"`partition` parameter is not supported in this view.">>)
end,
- Args1 = case get_extra(Args, partitioned, false) of
- true -> apply_partition(Args);
- false -> Args
+ Args1 = case {Style, Partitioned, Partition} of
+ {all_docs, true, undefined} ->
+ Args;
+ {all_docs, true, Partition} ->
+ apply_partition(Args, all_docs);
+ {all_docs, false, _} ->
+ Args;
+ {normal, true, _} ->
+ apply_partition(Args, normal);
+ {normal, false, _} ->
+ Args
end,
Args1#mrargs{
@@ -617,15 +638,18 @@ determine_group_level(#mrargs{group=true, group_level=undefined}) ->
determine_group_level(#mrargs{group_level=GroupLevel}) ->
GroupLevel.
-apply_partition(#mrargs{} = Args0) ->
+apply_partition(#mrargs{} = Args0, Style) ->
case get_extra(Args0, partition_applied, false) of
true ->
Args0;
false ->
Partition = get_extra(Args0, partition),
- Args1 = apply_partition(Partition, Args0),
+ Args1 = case Style of
+ normal -> apply_partition(Partition, Args0);
+ all_docs -> apply_all_docs_partition(Partition, Args0)
+ end,
set_extra(Args1, partition_applied, true)
- end.
+ end;
apply_partition(Partition, #mrargs{keys=Keys} = Args) when Keys /= undefined ->
Args#mrargs{keys=[{p, Partition, K} || K <- Keys]};
@@ -652,6 +676,33 @@ apply_partition(Partition, #mrargs{start_key=SK0, end_key=EK0} = Args) ->
Args#mrargs{start_key={p, Partition, SK0}, end_key={p, Partition, EK0}}.
+%% all_docs is special as it's not really a view and is already
+%% effectively partitioned as the partition is a prefix of all keys.
+apply_all_docs_partition(Partition, #mrargs{direction=fwd, start_key=undefined, end_key=undefined} = Args) ->
+ Args#mrargs{start_key = ?PARTITION_START(Partition), end_key = ?PARTITION_END(Partition)};
+
+apply_all_docs_partition(Partition, #mrargs{direction=rev, start_key=undefined, end_key=undefined} = Args) ->
+ Args#mrargs{start_key = ?PARTITION_END(Partition), end_key = ?PARTITION_START(Partition)};
+
+apply_all_docs_partition(Partition, #mrargs{direction=fwd, start_key=SK, end_key=undefined} = Args) ->
+ Args#mrargs{start_key = ?HIGHEST(?PARTITION_START(Partition), SK), end_key = ?PARTITION_END(Partition)};
+
+apply_all_docs_partition(Partition, #mrargs{direction=rev, start_key=SK, end_key=undefined} = Args) ->
+ Args#mrargs{start_key = ?LOWEST(?PARTITION_END(Partition), SK), end_key = ?PARTITION_START(Partition)};
+
+apply_all_docs_partition(Partition, #mrargs{direction=fwd, start_key=undefined, end_key=EK} = Args) ->
+ Args#mrargs{start_key = ?PARTITION_START(Partition), end_key = ?LOWEST(?PARTITION_END(Partition), EK)};
+
+apply_all_docs_partition(Partition, #mrargs{direction=rev, start_key=undefined, end_key=EK} = Args) ->
+ Args#mrargs{start_key = ?PARTITION_END(Partition), end_key = ?HIGHEST(?PARTITION_START(Partition), EK)};
+
+apply_all_docs_partition(Partition, #mrargs{direction=fwd, start_key=SK, end_key=EK} = Args) ->
+ Args#mrargs{start_key = ?HIGHEST(?PARTITION_START(Partition), SK), end_key = ?LOWEST(?PARTITION_END(Partition), EK)};
+
+apply_all_docs_partition(Partition, #mrargs{direction=rev, start_key=SK, end_key=EK} = Args) ->
+ Args#mrargs{start_key = ?LOWEST(?PARTITION_END(Partition), SK), end_key = ?HIGHEST(?PARTITION_START(Partition), EK)}.
+
+
check_range(#mrargs{start_key=undefined}, _Cmp) ->
ok;
check_range(#mrargs{end_key=undefined}, _Cmp) ->
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 97e323e..4de365f 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -294,9 +294,12 @@ all_docs(DbName, Callback, Acc, QueryArgs) ->
#mrargs{} | [option()]) ->
{ok, any()} | {error, Reason :: term()}.
-all_docs(DbName, Options, Callback, Acc0, #mrargs{} = QueryArgs) when
+
+all_docs(DbName, Options, Callback, Acc0, #mrargs{} = QueryArgs0) when
is_function(Callback, 2) ->
- fabric_view_all_docs:go(dbname(DbName), opts(Options), QueryArgs, Callback, Acc0);
+ DbPartitioned = mem3:is_partitioned(dbname(DbName)),
+ QueryArgs1 = couch_mrview_util:set_extra(QueryArgs0, partitioned, DbPartitioned),
+ fabric_view_all_docs:go(dbname(DbName), opts(Options), QueryArgs1, Callback, Acc0);
%% @doc convenience function that takes a keylist rather than a record
%% @equiv all_docs(DbName, Callback, Acc0, kl_to_query_args(QueryArgs))
diff --git a/src/fabric/src/fabric_view.erl b/src/fabric/src/fabric_view.erl
index 844b44d..c0e2974 100644
--- a/src/fabric/src/fabric_view.erl
+++ b/src/fabric/src/fabric_view.erl
@@ -122,7 +122,7 @@ maybe_send_row(State) ->
user_acc = AccIn,
query_args = QueryArgs
} = State,
- Partitioned = couch_mrview_util:get_extra(QueryArgs, partitioned, false),
+ Partitioned = couch_mrview_util:get_extra(QueryArgs, partitioned),
case fabric_dict:any(0, Counters) of
true ->
{ok, State};
diff --git a/src/fabric/src/fabric_view_all_docs.erl b/src/fabric/src/fabric_view_all_docs.erl
index ac16dac..d515ab8 100644
--- a/src/fabric/src/fabric_view_all_docs.erl
+++ b/src/fabric/src/fabric_view_all_docs.erl
@@ -118,7 +118,7 @@ go(DbName, _Options, Workers, QueryArgs, Callback, Acc0) ->
#mrargs{limit = Limit, skip = Skip, update_seq = UpdateSeq} = QueryArgs,
State = #collector{
db_name = DbName,
- query_args = QueryArgs,
+ query_args = couch_mrview_util:set_extra(QueryArgs, style, all_docs),
callback = Callback,
counters = fabric_dict:init(Workers, 0),
skip = Skip,
diff --git a/src/mango/src/mango_cursor_view.erl b/src/mango/src/mango_cursor_view.erl
index dbea36e..63c010f 100644
--- a/src/mango/src/mango_cursor_view.erl
+++ b/src/mango/src/mango_cursor_view.erl
@@ -29,6 +29,7 @@
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
-include("mango_cursor.hrl").
+-include("mango_idx.hrl").
-include("mango_idx_view.hrl").
create(Db, Indexes, Selector, Opts) ->
@@ -71,7 +72,8 @@ explain(Cursor) ->
{direction, Args#mrargs.direction},
{stable, Args#mrargs.stable},
{update, Args#mrargs.update},
- {conflicts, Args#mrargs.conflicts}
+ {conflicts, Args#mrargs.conflicts},
+ {extra, {Args#mrargs.extra}}
]}}].
@@ -93,14 +95,27 @@ maybe_replace_max_json([H | T] = EndKey) when is_list(EndKey) ->
maybe_replace_max_json(EndKey) ->
EndKey.
-base_args(#cursor{index = Idx} = Cursor) ->
- #mrargs{
+base_args(#cursor{index = Idx, opts = Opts} = Cursor) ->
+ Args1 = #mrargs{
view_type = map,
reduce = false,
start_key = mango_idx:start_key(Idx, Cursor#cursor.ranges),
end_key = mango_idx:end_key(Idx, Cursor#cursor.ranges),
include_docs = true
- }.
+ },
+ Args2 = add_partition_opts(Args1, couch_util:get_value(partition, Opts)),
+ add_style(Idx, Args2).
+
+add_partition_opts(#mrargs{} = Args, <<>>) ->
+ Args;
+add_partition_opts(#mrargs{} = Args, Partition) ->
+ Args1 = couch_mrview_util:set_extra(Args, partitioned, true),
+ couch_mrview_util:set_extra(Args1, partition, Partition).
+
+add_style(#idx{def = all_docs}, Args) ->
+ couch_mrview_util:set_extra(Args, style, all_docs);
+add_style(_, Args) ->
+ Args.
execute(#cursor{db = Db, index = Idx, execution_stats = Stats} = Cursor0, UserFun, UserAcc) ->
diff --git a/src/mango/src/mango_opts.erl b/src/mango/src/mango_opts.erl
index 7bae9c9..87d876a 100644
--- a/src/mango/src/mango_opts.erl
+++ b/src/mango/src/mango_opts.erl
@@ -81,6 +81,12 @@ validate_find({Props}) ->
{tag, selector},
{validator, fun validate_selector/1}
]},
+ {<<"partition">>, [
+ {tag, partition},
+ {optional, true},
+ {default, <<>>},
+ {validator, fun is_string/1}
+ ]},
{<<"use_index">>, [
{tag, use_index},
{optional, true},
diff --git a/src/mem3/src/mem3.erl b/src/mem3/src/mem3.erl
index aecca2f..f113357 100644
--- a/src/mem3/src/mem3.erl
+++ b/src/mem3/src/mem3.erl
@@ -352,7 +352,10 @@ is_partitioned(DbName0) when is_binary(DbName0) ->
is_partitioned(mem3:shards(DbName))
end;
-is_partitioned(Shards) when is_list(Shards) ->
+is_partitioned([#shard{} | _] = Shards) ->
+ lists:all(fun is_partitioned/1, Shards);
+
+is_partitioned([#ordered_shard{} | _] = Shards) ->
lists:all(fun is_partitioned/1, Shards);
is_partitioned(#shard{opts=Opts}) ->
[couchdb] 01/04: implement partitioned views
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-6
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 775098cb1f3e4077c3bab5d3981d6682177d790a
Author: Robert Newson <rn...@apache.org>
AuthorDate: Tue Aug 7 15:44:33 2018 +0100
implement partitioned views
---
src/couch/src/couch_btree.erl | 14 +++++++
src/couch/src/couch_ejson_compare.erl | 4 ++
src/couch_mrview/src/couch_mrview.erl | 2 +
src/couch_mrview/src/couch_mrview_updater.erl | 14 ++++++-
src/couch_mrview/src/couch_mrview_util.erl | 55 ++++++++++++++++++++++++---
src/fabric/src/fabric_view.erl | 18 +++++++--
6 files changed, 97 insertions(+), 10 deletions(-)
diff --git a/src/couch/src/couch_btree.erl b/src/couch/src/couch_btree.erl
index ea224b1..d11d7e6 100644
--- a/src/couch/src/couch_btree.erl
+++ b/src/couch/src/couch_btree.erl
@@ -133,6 +133,20 @@ make_group_fun(Bt, exact) ->
end;
make_group_fun(Bt, GroupLevel) when is_integer(GroupLevel), GroupLevel > 0 ->
fun
+ ({{p, _Partition, Key1}, _}, {{p, _Partition, Key2}, _}) ->
+ SL1 = lists:sublist(Key1, GroupLevel),
+ SL2 = lists:sublist(Key2, GroupLevel),
+ case less(Bt, {SL1, nil}, {SL2, nil}) of
+ false ->
+ case less(Bt, {SL2, nil}, {SL1, nil}) of
+ false ->
+ true;
+ _ ->
+ false
+ end;
+ _ ->
+ false
+ end;
({[_|_] = Key1, _}, {[_|_] = Key2, _}) ->
SL1 = lists:sublist(Key1, GroupLevel),
SL2 = lists:sublist(Key2, GroupLevel),
diff --git a/src/couch/src/couch_ejson_compare.erl b/src/couch/src/couch_ejson_compare.erl
index 81adbb8..ca36c86 100644
--- a/src/couch/src/couch_ejson_compare.erl
+++ b/src/couch/src/couch_ejson_compare.erl
@@ -22,6 +22,10 @@ init() ->
Dir = code:priv_dir(couch),
ok = erlang:load_nif(filename:join(Dir, ?MODULE), NumScheds).
+% partitioned row comparison
+less({p, PA, A}, {p, PB, B}) ->
+ less([PA, A], [PB, B]);
+
less(A, B) ->
try
less_nif(A, B)
diff --git a/src/couch_mrview/src/couch_mrview.erl b/src/couch_mrview/src/couch_mrview.erl
index db467f0..09945f5 100644
--- a/src/couch_mrview/src/couch_mrview.erl
+++ b/src/couch_mrview/src/couch_mrview.erl
@@ -614,6 +614,8 @@ red_fold(Db, {NthRed, _Lang, View}=RedView, Args, Callback, UAcc) ->
end, Acc, OptList),
finish_fold(Acc2, []).
+red_fold({p, _Partition, Key}, Red, Acc) ->
+ red_fold(Key, Red, Acc);
red_fold(_Key, _Red, #mracc{skip=N}=Acc) when N > 0 ->
{ok, Acc#mracc{skip=N-1, last_go=ok}};
red_fold(Key, Red, #mracc{meta_sent=false}=Acc) ->
diff --git a/src/couch_mrview/src/couch_mrview_updater.erl b/src/couch_mrview/src/couch_mrview_updater.erl
index 214f487..2b69eee 100644
--- a/src/couch_mrview/src/couch_mrview_updater.erl
+++ b/src/couch_mrview/src/couch_mrview_updater.erl
@@ -311,9 +311,11 @@ write_kvs(State, UpdateSeq, ViewKVs, DocIdKeys, Seqs, Log0) ->
#mrst{
id_btree=IdBtree,
log_btree=LogBtree,
- first_build=FirstBuild
+ first_build=FirstBuild,
+ design_opts=DesignOpts
} = State,
+ Partitioned = couch_util:get_value(<<"partitioned">>, DesignOpts, false),
Revs = dict:from_list(dict:fetch_keys(Log0)),
Log = dict:fold(fun({Id, _Rev}, DIKeys, Acc) ->
@@ -328,8 +330,9 @@ write_kvs(State, UpdateSeq, ViewKVs, DocIdKeys, Seqs, Log0) ->
_ -> update_log(LogBtree, Log, Revs, Seqs, FirstBuild)
end,
- UpdateView = fun(#mrview{id_num=ViewId}=View, {ViewId, {KVs, SKVs}}) ->
+ UpdateView = fun(#mrview{id_num=ViewId}=View, {ViewId, {KVs0, SKVs}}) ->
#mrview{seq_indexed=SIndexed, keyseq_indexed=KSIndexed} = View,
+ KVs = if Partitioned -> inject_partition(KVs0); true -> KVs0 end,
ToRem = couch_util:dict_find(ViewId, ToRemByView, []),
{ok, VBtree2} = couch_btree:add_remove(View#mrview.btree, KVs, ToRem),
NewUpdateSeq = case VBtree2 =/= View#mrview.btree of
@@ -378,6 +381,13 @@ write_kvs(State, UpdateSeq, ViewKVs, DocIdKeys, Seqs, Log0) ->
log_btree=LogBtree2
}.
+inject_partition(KVs) ->
+ [{{{p, partition(DocId), Key}, DocId}, Value} || {{Key, DocId}, Value} <- KVs].
+
+partition(DocId) ->
+ [Partition, _Rest] = binary:split(DocId, <<":">>),
+ Partition.
+
update_id_btree(Btree, DocIdKeys, true) ->
ToAdd = [{Id, DIKeys} || {Id, DIKeys} <- DocIdKeys, DIKeys /= []],
couch_btree:query_modify(Btree, [], ToAdd, []);
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index 592bfb5..bb1e365 100644
--- a/src/couch_mrview/src/couch_mrview_util.erl
+++ b/src/couch_mrview/src/couch_mrview_util.erl
@@ -38,6 +38,9 @@
-define(MOD, couch_mrview_index).
-define(GET_VIEW_RETRY_COUNT, 1).
-define(GET_VIEW_RETRY_DELAY, 50).
+-define(LOWEST_KEY, null).
+-define(HIGHEST_KEY, {[{<<239, 191, 176>>, null}]}). % is {"\ufff0": null}
+
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
@@ -226,11 +229,12 @@ view_sig(_Db, State, View, Args0) ->
PurgeSeq = View#mrview.purge_seq,
SeqIndexed = View#mrview.seq_indexed,
KeySeqIndexed = View#mrview.keyseq_indexed,
+ Partitioned = get_extra(Args0, partitioned, false),
Args = Args0#mrargs{
preflight_fun=undefined,
extra=[]
},
- Term = view_sig_term(Sig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed, Args),
+ Term = view_sig_term(Sig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed, Partitioned, Args),
couch_index_util:hexsig(couch_hash:md5_hash(term_to_binary(Term))).
view_sig_term(BaseSig, UpdateSeq, PurgeSeq, false, false) ->
@@ -238,10 +242,12 @@ view_sig_term(BaseSig, UpdateSeq, PurgeSeq, false, false) ->
view_sig_term(BaseSig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed) ->
{BaseSig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed}.
-view_sig_term(BaseSig, UpdateSeq, PurgeSeq, false, false, Args) ->
+view_sig_term(BaseSig, UpdateSeq, PurgeSeq, false, false, false, Args) ->
{BaseSig, UpdateSeq, PurgeSeq, Args};
-view_sig_term(BaseSig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed, Args) ->
- {BaseSig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed, Args}.
+view_sig_term(BaseSig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed, false, Args) ->
+ {BaseSig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed, Args};
+view_sig_term(BaseSig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed, Partitioned, Args) ->
+ {BaseSig, UpdateSeq, PurgeSeq, KeySeqIndexed, SeqIndexed, Partitioned, Args}.
init_state(Db, Fd, #mrst{views=Views}=State, nil) ->
@@ -588,7 +594,12 @@ validate_args(Args) ->
mrverror(<<"`partition` parameter is not supported in this view.">>)
end,
- Args#mrargs{
+ Args1 = case get_extra(Args, partitioned, false) of
+ true -> apply_partition(Args);
+ false -> Args
+ end,
+
+ Args1#mrargs{
start_key_docid=SKDocId,
end_key_docid=EKDocId,
group_level=GroupLevel
@@ -606,6 +617,40 @@ determine_group_level(#mrargs{group=true, group_level=undefined}) ->
determine_group_level(#mrargs{group_level=GroupLevel}) ->
GroupLevel.
+apply_partition(#mrargs{} = Args0) ->
+ case get_extra(Args0, partition_applied, false) of
+ true ->
+ Args0;
+ false ->
+ Partition = get_extra(Args0, partition),
+ Args1 = apply_partition(Partition, Args0),
+ set_extra(Args1, partition_applied, true)
+ end.
+
+apply_partition(Partition, #mrargs{keys=Keys} = Args) when Keys /= undefined ->
+ Args#mrargs{keys=[{p, Partition, K} || K <- Keys]};
+
+apply_partition(Partition, #mrargs{direction=fwd, start_key=undefined, end_key=undefined} = Args) ->
+ Args#mrargs{start_key={p, Partition, ?LOWEST_KEY}, end_key={p, Partition, ?HIGHEST_KEY}};
+
+apply_partition(Partition, #mrargs{direction=rev, start_key=undefined, end_key=undefined} = Args) ->
+ Args#mrargs{start_key={p, Partition, ?HIGHEST_KEY}, end_key={p, Partition, ?LOWEST_KEY}};
+
+apply_partition(Partition, #mrargs{direction=fwd, start_key=SK0, end_key=undefined} = Args) ->
+ Args#mrargs{start_key={p, Partition, SK0}, end_key={p, Partition, ?HIGHEST_KEY}};
+
+apply_partition(Partition, #mrargs{direction=rev, start_key=SK0, end_key=undefined} = Args) ->
+ Args#mrargs{start_key={p, Partition, SK0}, end_key={p, Partition, ?LOWEST_KEY}};
+
+apply_partition(Partition, #mrargs{direction=fwd, start_key=undefined, end_key=EK0} = Args) ->
+ Args#mrargs{start_key={p, Partition, ?LOWEST_KEY}, end_key={p, Partition, EK0}};
+
+apply_partition(Partition, #mrargs{direction=rev, start_key=undefined, end_key=EK0} = Args) ->
+ Args#mrargs{start_key={p, Partition, ?HIGHEST_KEY}, end_key={p, Partition, EK0}};
+
+apply_partition(Partition, #mrargs{start_key=SK0, end_key=EK0} = Args) ->
+ Args#mrargs{start_key={p, Partition, SK0}, end_key={p, Partition, EK0}}.
+
check_range(#mrargs{start_key=undefined}, _Cmp) ->
ok;
diff --git a/src/fabric/src/fabric_view.erl b/src/fabric/src/fabric_view.erl
index b4b8a8c..844b44d 100644
--- a/src/fabric/src/fabric_view.erl
+++ b/src/fabric/src/fabric_view.erl
@@ -119,8 +119,10 @@ maybe_send_row(State) ->
counters = Counters,
skip = Skip,
limit = Limit,
- user_acc = AccIn
+ user_acc = AccIn,
+ query_args = QueryArgs
} = State,
+ Partitioned = couch_mrview_util:get_extra(QueryArgs, partitioned, false),
case fabric_dict:any(0, Counters) of
true ->
{ok, State};
@@ -128,8 +130,14 @@ maybe_send_row(State) ->
try get_next_row(State) of
{_, NewState} when Skip > 0 ->
maybe_send_row(NewState#collector{skip=Skip-1});
- {Row, NewState} ->
- case Callback(transform_row(possibly_embed_doc(NewState,Row)), AccIn) of
+ {Row0, NewState} ->
+ Row1 = possibly_embed_doc(NewState, Row0),
+ Row2 = if
+ Partitioned -> detach_partition(Row1);
+ true -> Row1
+ end,
+ Row3 = transform_row(Row2),
+ case Callback(Row3, AccIn) of
{stop, Acc} ->
{stop, NewState#collector{user_acc=Acc, limit=Limit-1}};
{ok, Acc} ->
@@ -194,6 +202,10 @@ possibly_embed_doc(#collector{db_name=DbName, query_args=Args},
_ -> Row
end.
+detach_partition(#view_row{key={p, _Partition, Key}} = Row) ->
+ Row#view_row{key = Key};
+detach_partition(#view_row{} = Row) ->
+ Row.
keydict(undefined) ->
undefined;