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/07/09 21:40:56 UTC
[couchdb] 02/02: WIP plumbing
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch user-partitioned-dbs-wip
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 28e0dfa1f97c3b3dc9c2585d2a032868e49a11c0
Author: Robert Newson <rn...@apache.org>
AuthorDate: Mon Jul 9 22:39:16 2018 +0100
WIP plumbing
---
src/couch_mrview/include/couch_mrview.hrl | 3 ++-
src/couch_mrview/src/couch_mrview_util.erl | 29 +++++++++++++++--------------
src/fabric/src/fabric.erl | 14 +++++++++-----
src/fabric/src/fabric_db_create.erl | 6 +++++-
src/mem3/src/mem3.erl | 10 +++++++++-
src/mem3/src/mem3_util.erl | 15 +++++++++++++--
6 files changed, 53 insertions(+), 24 deletions(-)
diff --git a/src/couch_mrview/include/couch_mrview.hrl b/src/couch_mrview/include/couch_mrview.hrl
index 67b3cd9..09faf51 100644
--- a/src/couch_mrview/include/couch_mrview.hrl
+++ b/src/couch_mrview/include/couch_mrview.hrl
@@ -32,7 +32,7 @@
doc_queue,
write_queue,
qserver=nil,
- partitioned=false
+ partitioned
}).
@@ -88,6 +88,7 @@
conflicts,
callback,
sorted = true,
+ partitioned,
partition_key,
extra = []
}).
diff --git a/src/couch_mrview/src/couch_mrview_util.erl b/src/couch_mrview/src/couch_mrview_util.erl
index 35554c4..aa67c88 100644
--- a/src/couch_mrview/src/couch_mrview_util.erl
+++ b/src/couch_mrview/src/couch_mrview_util.erl
@@ -24,7 +24,7 @@
-export([temp_view_to_ddoc/1]).
-export([calculate_external_size/1]).
-export([calculate_active_size/1]).
--export([validate_and_update_args/1, validate_and_update_args/2]).
+-export([validate_args/1, validate_and_update_args/1]).
-export([maybe_load_doc/3, maybe_load_doc/4]).
-export([maybe_update_index_file/1]).
-export([extract_view/4, extract_view_reduce/1]).
@@ -33,6 +33,7 @@
-export([changes_key_opts/2]).
-export([fold_changes/4]).
-export([to_key_seq/1]).
+-export([set_view_options/3]).
-export([partition_key/2, unpartition_key/1]).
-define(MOD, couch_mrview_index).
@@ -60,7 +61,7 @@ get_view(Db, DDoc, ViewName, Args0) ->
get_view_index_pid(Db, DDoc, ViewName, Args0) ->
ArgCheck = fun(InitState) ->
Args1 = set_view_type(Args0, ViewName, InitState#mrst.views),
- {ok, validate_and_update_args(Args1)}
+ {ok, validate_args(Args1)}
end,
couch_index_server:get_index(?MOD, Db, DDoc, ArgCheck).
@@ -449,7 +450,7 @@ fold_reduce({NthRed, Lang, View}, Fun, Acc, Options) ->
couch_btree:fold_reduce(Bt, WrapperFun, Acc, Options).
-validate_args(Args, Options) ->
+validate_args(Args) ->
GroupLevel = determine_group_level(Args),
Reduce = Args#mrargs.reduce,
case Reduce == undefined orelse is_boolean(Reduce) of
@@ -559,21 +560,20 @@ validate_args(Args, Options) ->
_ -> mrverror(<<"Invalid value for `sorted`.">>)
end,
- case {lists:member(partitioned, Options), Args#mrargs.partition_key} of
+ case {Args#mrargs.partitioned, Args#mrargs.partition_key} of
{true, undefined} ->
mrverror(<<"`partition_key` parameter is mandatory for queries to this database.">>);
{true, _PartitionKey} ->
ok;
- {false, undefined} ->
+ {undefined, undefined} ->
ok;
- {false, _PartitionKey} ->
+ {undefined, _PartitionKey} ->
mrverror(<<"`partition_key` parameter is not supported in this database.">>)
end,
+ Args.
- true.
-
-update_args(#mrargs{} = Args, _Options) ->
+update_args(#mrargs{} = Args) ->
GroupLevel = determine_group_level(Args),
SKDocId = case {Args#mrargs.direction, Args#mrargs.start_key_docid} of
@@ -621,11 +621,8 @@ update_args(#mrargs{} = Args, _Options) ->
validate_and_update_args(#mrargs{} = Args) ->
- validate_and_update_args(Args, []).
-
-validate_and_update_args(#mrargs{} = Args, Options) ->
- true = validate_args(Args, Options),
- update_args(Args, Options).
+ Args = validate_args(Args),
+ update_args(Args).
determine_group_level(#mrargs{group=undefined, group_level=undefined}) ->
@@ -1259,6 +1256,10 @@ kv_external_size(KVList, Reduction) ->
?term_size(Key) + ?term_size(Value) + Acc
end, ?term_size(Reduction), KVList).
+set_view_options(#mrargs{} = Args, partitioned, true) ->
+ Args#mrargs{partitioned=true};
+set_view_options(#mrargs{} = Args, partitioned, false) ->
+ Args#mrargs{partitioned=false}.
partition_key(Key, DocId) ->
[hd(binary:split(DocId, <<":">>)), Key].
diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl
index 1b761ab..59fbb2b 100644
--- a/src/fabric/src/fabric.erl
+++ b/src/fabric/src/fabric.erl
@@ -354,16 +354,20 @@ query_view(DbName, Options, DDoc, ViewName, Callback, Acc0, QueryArgs0) ->
end,
{ok, #mrst{views=Views, language=Lang}} =
couch_mrview_util:ddoc_to_mrst(Db, DDoc),
+
+ Partitioned = mem3:is_partitioned(hd(mem3:shards(Db))), % hideous
+
QueryArgs1 = couch_mrview_util:set_view_type(QueryArgs0, View, Views),
- QueryArgs2 = couch_mrview_util:validate_and_update_args(QueryArgs1),
- VInfo = couch_mrview_util:extract_view(Lang, QueryArgs2, View, Views),
- case is_reduce_view(QueryArgs2) of
+ QueryArgs2 = couch_mrview_util:set_view_options(QueryArgs1, partitioned, Partitioned),
+ QueryArgs3 = couch_mrview_util:validate_and_update_args(QueryArgs2),
+ VInfo = couch_mrview_util:extract_view(Lang, QueryArgs3, View, Views),
+ case is_reduce_view(QueryArgs3) of
true ->
fabric_view_reduce:go(
Db,
DDoc,
View,
- QueryArgs2,
+ QueryArgs3,
Callback,
Acc0,
VInfo
@@ -374,7 +378,7 @@ query_view(DbName, Options, DDoc, ViewName, Callback, Acc0, QueryArgs0) ->
Options,
DDoc,
View,
- QueryArgs2,
+ QueryArgs3,
Callback,
Acc0,
VInfo
diff --git a/src/fabric/src/fabric_db_create.erl b/src/fabric/src/fabric_db_create.erl
index 94ffd56..35e38cb 100644
--- a/src/fabric/src/fabric_db_create.erl
+++ b/src/fabric/src/fabric_db_create.erl
@@ -168,6 +168,10 @@ make_document([#shard{dbname=DbName}|_] = Shards, Suffix, Options) ->
E when is_binary(E) -> [{<<"engine">>, E}];
_ -> []
end,
+ PartitionedProp = case lists:member(partitioned, Options) of
+ true -> [{<<"partitioned">>, true}];
+ false -> []
+ end,
#doc{
id = DbName,
body = {[
@@ -175,7 +179,7 @@ make_document([#shard{dbname=DbName}|_] = Shards, Suffix, Options) ->
{<<"changelog">>, lists:sort(RawOut)},
{<<"by_node">>, {[{K,lists:sort(V)} || {K,V} <- ByNodeOut]}},
{<<"by_range">>, {[{K,lists:sort(V)} || {K,V} <- ByRangeOut]}}
- ] ++ EngineProp}
+ ] ++ EngineProp ++ PartitionedProp}
}.
db_exists(DbName) -> is_list(catch mem3:shards(DbName)).
diff --git a/src/mem3/src/mem3.erl b/src/mem3/src/mem3.erl
index ca083c4..f4bed76 100644
--- a/src/mem3/src/mem3.erl
+++ b/src/mem3/src/mem3.erl
@@ -23,7 +23,7 @@
-export([get_placement/1]).
%% For mem3 use only.
--export([name/1, node/1, range/1, engine/1]).
+-export([name/1, node/1, range/1, engine/1, is_partitioned/1]).
-include_lib("mem3/include/mem3.hrl").
-include_lib("couch/include/couch_db.hrl").
@@ -331,6 +331,14 @@ engine(Opts) when is_list(Opts) ->
[]
end.
+is_partitioned(#shard{opts=Opts}) ->
+ is_partitioned(Opts);
+is_partitioned(#ordered_shard{opts=Opts}) ->
+ is_partitioned(Opts);
+is_partitioned(Opts) when is_list(Opts) ->
+ lists:member(partitioned, Opts).
+
+
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
diff --git a/src/mem3/src/mem3_util.erl b/src/mem3/src/mem3_util.erl
index fc6123d..5af8c95 100644
--- a/src/mem3/src/mem3_util.erl
+++ b/src/mem3/src/mem3_util.erl
@@ -177,7 +177,7 @@ build_shards_by_node(DbName, DocProps) ->
dbname = DbName,
node = to_atom(Node),
range = [Beg, End],
- opts = get_engine_opt(DocProps)
+ opts = get_opts(DocProps)
}, Suffix)
end, Ranges)
end, ByNode).
@@ -195,7 +195,7 @@ build_shards_by_range(DbName, DocProps) ->
node = to_atom(Node),
range = [Beg, End],
order = Order,
- opts = get_engine_opt(DocProps)
+ opts = get_opts(DocProps)
}, Suffix)
end, lists:zip(Nodes, lists:seq(1, length(Nodes))))
end, ByRange).
@@ -212,6 +212,9 @@ to_integer(N) when is_binary(N) ->
to_integer(N) when is_list(N) ->
list_to_integer(N).
+get_opts(DocProps) ->
+ get_engine_opt(DocProps) ++ get_partitioned_opt(DocProps).
+
get_engine_opt(DocProps) ->
case couch_util:get_value(<<"engine">>, DocProps) of
Engine when is_binary(Engine) ->
@@ -220,6 +223,14 @@ get_engine_opt(DocProps) ->
[]
end.
+get_partitioned_opt(DocProps) ->
+ case couch_util:get_value(<<"partitioned">>, DocProps) of
+ true ->
+ [partitioned];
+ _ ->
+ []
+ end.
+
n_val(undefined, NodeCount) ->
n_val(config:get("cluster", "n", "3"), NodeCount);
n_val(N, NodeCount) when is_list(N) ->