Permalink
Browse files

MB-6126 Correctly split database name into components

If the bucket name had forward slashes in it (/), then all the
code that splitted a database name into bucket name (set name)
and vbucket Id (partition Id) didn't work correctly.

Change-Id: I6b266f8d16b30a97c9ddafa2ce1bb70ae5408875
Reviewed-on: http://review.couchbase.org/19301
Reviewed-by: Volker Mische <volker.mische@gmail.com>
Tested-by: Filipe David Borba Manana <fdmanana@gmail.com>
  • Loading branch information...
1 parent 04e3dab commit e024aaf9e3a4a2057046046868fa272de1580788 @fdmanana fdmanana committed with fdmanana Aug 6, 2012
@@ -792,10 +792,10 @@ handle_call({get_group_server, SetName, Group}, From, Server) ->
handle_call({before_database_delete, DbName}, _From, Server) ->
#server{root_dir = RootDir} = Server,
- case is_set_db(DbName) of
- false ->
+ case couch_set_view_util:split_set_db_name(DbName) of
+ error ->
ok;
- {true, SetName, PartId} ->
+ {ok, SetName, PartId} ->
lists:foreach(
fun({_SetName, {_DDocId, Sig}}) ->
[{_, Pid}] = ets:lookup(couch_sig_to_setview_pid, {SetName, Sig}),
@@ -978,9 +978,8 @@ modify_bitmasks(#set_view_group{replica_group = RepGroup} = Group, Partitions) -
handle_db_event({before_delete, DbName}) ->
ok = gen_server:call(?MODULE, {before_database_delete, DbName}, infinity);
handle_db_event({Event, {DbName, DDocId}}) when Event == ddoc_updated; Event == ddoc_deleted ->
- case string:tokens(?b2l(DbName), "/") of
- [SetNameStr, "master"] ->
- SetName = ?l2b(SetNameStr),
+ case couch_set_view_util:split_set_db_name(DbName) of
+ {ok, SetName, master} ->
lists:foreach(
fun({_SetName, {_DDocId, Sig}}) ->
case ets:lookup(couch_sig_to_setview_pid, {SetName, Sig}) of
@@ -1047,23 +1046,6 @@ reduce_view_merge_callback({debug_info, _From, _Info}, Acc) ->
{ok, Acc}.
-is_set_db(DbName) ->
- case string:tokens(?b2l(DbName), "/") of
- [SetName, "master"] ->
- {true, ?l2b(SetName), master};
- [SetName, Rest] ->
- PartId = (catch list_to_integer(Rest)),
- case is_integer(PartId) of
- true ->
- {true, ?l2b(SetName), PartId};
- false ->
- false
- end;
- _ ->
- false
- end.
-
-
% Returns whether the results should be filtered based on a bitmask or not
-spec filter(#set_view_group{}) -> false | {true, bitmask(), bitmask()}.
filter(#set_view_group{type = main} = Group) ->
@@ -190,16 +190,15 @@ handle_config_change("set_views", "ddoc_cache_size", NewSizeList) ->
handle_db_event({deleted, DbName}) ->
- case string:tokens(?b2l(DbName), "/") of
- [SetNameList, "master"] ->
- ok = gen_server:call(?MODULE, {set_deleted, ?l2b(SetNameList)}, infinity);
+ case couch_set_view_util:split_set_db_name(DbName) of
+ {ok, SetName, master} ->
+ ok = gen_server:call(?MODULE, {set_deleted, SetName}, infinity);
_ ->
ok
end;
handle_db_event({ddoc_updated, {DbName, Id}}) ->
- case string:tokens(?b2l(DbName), "/") of
- [SetNameList, "master"] ->
- SetName = ?l2b(SetNameList),
+ case couch_set_view_util:split_set_db_name(DbName) of
+ {ok, SetName, master} ->
case couch_db:open_int(DbName, []) of
{ok, Db} ->
try
@@ -225,9 +224,9 @@ handle_db_event({ddoc_updated, {DbName, Id}}) ->
ok
end;
handle_db_event({ddoc_deleted, {DbName, Id}}) ->
- case string:tokens(?b2l(DbName), "/") of
- [SetNameList, "master"] ->
- ok = gen_server:call(?MODULE, {delete_ddoc, ?l2b(SetNameList), Id}, infinity);
+ case couch_set_view_util:split_set_db_name(DbName) of
+ {ok, SetName, master} ->
+ ok = gen_server:call(?MODULE, {delete_ddoc, SetName, Id}, infinity);
_ ->
ok
end;
@@ -25,6 +25,7 @@
-export([new_sort_file_path/1, delete_sort_files/1]).
-export([encode_key_docid/2, decode_key_docid/1, split_key_docid/1]).
-export([parse_values/1, parse_reductions/1, parse_view_id_keys/1]).
+-export([split_set_db_name/1]).
-include("couch_db.hrl").
@@ -392,3 +393,30 @@ split_key_docid(<<KeyLen:16, KeyJson:KeyLen/binary, DocId/binary>>) ->
-spec encode_key_docid(binary(), binary()) -> binary().
encode_key_docid(JsonKey, DocId) ->
<<(byte_size(JsonKey)):16, JsonKey/binary, DocId/binary>>.
+
+
+-spec split_set_db_name(string() | binary()) ->
+ {'ok', SetName::binary(), Partition::master} |
+ {'ok', SetName::binary(), Partition::non_neg_integer()} |
+ 'error'.
+split_set_db_name(DbName) when is_binary(DbName) ->
+ split_set_db_name(?b2l(DbName));
+split_set_db_name(DbName) ->
+ Len = length(DbName),
+ case string:rchr(DbName, $/) of
+ Pos when (Pos > 0), (Pos < Len) ->
+ {SetName, [$/ | Partition]} = lists:split(Pos - 1, DbName),
+ case Partition of
+ "master" ->
+ {ok, ?l2b(SetName), master};
+ _ ->
+ case (catch list_to_integer(Partition)) of
+ Id when is_integer(Id), Id >= 0 ->
+ {ok, ?l2b(SetName), Id};
+ _ ->
+ error
+ end
+ end;
+ _ ->
+ error
+ end.

0 comments on commit e024aaf

Please sign in to comment.