Current section
Files
Jump to
Current section
Files
src/erlcloud_ddb_util.erl
%% -*- mode: erlang;erlang-indent-level: 4;indent-tabs-mode: nil -*-
%% @author Ransom Richardson <ransom@ransomr.net>
%% @doc
%% Helpers for using DynamoDB from Erlang.
%%
%% This is a higher layer API that augments the operations supported
%% by erlcloud_ddb2. The functions in this file do not map directly to
%% DynamoDB operations. Instead they will perform multiple operations
%% in order to implement functionality that isn't available directly
%% using the DynamoDB API.
%%
%% @end
-module(erlcloud_ddb_util).
-include("erlcloud.hrl").
-include("erlcloud_ddb2.hrl").
-include("erlcloud_aws.hrl").
%%% DynamoDB Higher Layer API
-export([delete_all/2, delete_all/3, delete_all/4,
delete_hash_key/3, delete_hash_key/4, delete_hash_key/5,
get_all/2, get_all/3, get_all/4,
put_all/2, put_all/3, put_all/4,
list_tables_all/0, list_tables_all/1,
q_all/2, q_all/3, q_all/4,
scan_all/1, scan_all/2, scan_all/3,
wait_for_table_active/1, wait_for_table_active/2, wait_for_table_active/3, wait_for_table_active/4,
write_all/2, write_all/3, write_all/4
]).
-ifdef(TEST).
-export([set_out_opt/1]).
-endif.
-define(BATCH_WRITE_LIMIT, 25).
-define(BATCH_GET_LIMIT, 100).
-type typed_out() :: {typed_out, boolean()}.
-type batch_read_ddb_opt() :: typed_out() | erlcloud_ddb2:out_opt().
-type batch_read_ddb_opts() :: [batch_read_ddb_opt()].
-type conditions() :: erlcloud_ddb2:conditions().
-type ddb_opts() :: erlcloud_ddb2:ddb_opts().
-type expression() :: erlcloud_ddb2:expression().
-type hash_key() :: erlcloud_ddb2:in_attr().
-type in_item() :: erlcloud_ddb2:in_item().
-type key() :: erlcloud_ddb2:key().
-type out_item() :: erlcloud_ddb2:out_item().
-type range_key_name() :: erlcloud_ddb2:range_key_name().
-type table_name() :: erlcloud_ddb2:table_name().
-type items_return() :: {ok, [out_item()]}
| {ok, non_neg_integer()}
| {error, term()}.
-export_type(
[batch_read_ddb_opt/0,
batch_read_ddb_opts/0,
typed_out/0]).
default_config() -> erlcloud_aws:default_config().
%%%------------------------------------------------------------------------------
%%% delete_all
%%%------------------------------------------------------------------------------
-spec delete_all(table_name(), [key()]) -> ok | {error, term()}.
delete_all(Table, Keys) ->
delete_all(Table, Keys, [], default_config()).
-spec delete_all(table_name(), [key()], ddb_opts()) -> ok | {error, term()}.
delete_all(Table, Keys, Opts) ->
delete_all(Table, Keys, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more BatchWriteItem operations to delete all items.
%% Operations are performed in parallel. Writing to only one table is supported.
%%
%% ===Example===
%%
%% `
%% ok =
%% erlcloud_ddb_util:delete_all(
%% [{<<"Forum">>,
%% [{<<"Name">>, {s, <<"Amazon DynamoDB">>}},
%% {<<"Name">>, {s, <<"Amazon RDS">>}},
%% {<<"Name">>, {s, <<"Amazon Redshift">>}},
%% {<<"Name">>, {s, <<"Amazon ElastiCache">>}}
%% ]}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec delete_all(table_name(), [key()], ddb_opts(), aws_config()) -> ok | {error, term()}.
delete_all(Table, Keys, Opts, Config) ->
write_all(Table, [{delete, Key} || Key <- Keys], Opts, Config).
%%%------------------------------------------------------------------------------
%%% delete_hash_key
%%%------------------------------------------------------------------------------
-spec delete_hash_key(table_name(), hash_key(), range_key_name()) -> ok | {error, term()}.
delete_hash_key(Table, HashKey, RangeKeyName) ->
delete_hash_key(Table, HashKey, RangeKeyName, [], default_config()).
-spec delete_hash_key(table_name(), hash_key(), range_key_name(), ddb_opts()) -> ok | {error, term()}.
delete_hash_key(Table, HashKey, RangeKeyName, Opts) ->
delete_hash_key(Table, HashKey, RangeKeyName, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Delete all items with the specified table. Table must be a
%% hash-and-range primary key table. Opts is currently ignored and is
%% provided for future enhancements. This method is not transacted.
%%
%% ===Example===
%%
%% `
%% ok = erlcloud_ddb_util:delete_hash_key(<<"tn">>, {<<"hash-key-name">>, <<"hash-key-value">>}, <<"range-key-name">>, [])),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec delete_hash_key(table_name(), hash_key(), range_key_name(), ddb_opts(), aws_config()) -> ok | {error, term()}.
delete_hash_key(Table, HashKey, RangeKeyName, Opts, Config) ->
case erlcloud_ddb2:q(Table, HashKey,
[{consistent_read, true},
{limit, ?BATCH_WRITE_LIMIT},
{attributes_to_get, [RangeKeyName]},
{out, typed_record}],
Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_q{count = 0}} ->
ok;
{ok, QResult} ->
case erlcloud_ddb2:batch_write_item(
[{Table, [{delete, [HashKey, RangeKey]} || [RangeKey] <- QResult#ddb2_q.items]}],
[{out, record}], Config) of
{error, Reason} ->
{error, Reason};
{ok, BatchResult} ->
if QResult#ddb2_q.last_evaluated_key == undefined andalso
BatchResult#ddb2_batch_write_item.unprocessed_items == [] ->
%% No more work to do
ok;
true ->
%% Some stuff was unprocessed - keep going
delete_hash_key(Table, HashKey, RangeKeyName, Opts, Config)
end
end
end.
%%%------------------------------------------------------------------------------
%%% get_all
%%%------------------------------------------------------------------------------
-type get_all_opts() :: erlcloud_ddb2:batch_get_item_request_item_opts().
-spec get_all(table_name(), [key()]) -> items_return().
get_all(Table, Keys) ->
get_all(Table, Keys, [], default_config()).
-spec get_all(table_name(), [key()], get_all_opts()) -> items_return().
get_all(Table, Keys, Opts) ->
get_all(Table, Keys, Opts, default_config()).
-spec get_all(table_name(), [key()], get_all_opts(), aws_config() | batch_read_ddb_opts()) -> items_return().
get_all(Table, Keys, Opts, Config) when is_record(Config, aws_config) ->
get_all(Table, Keys, Opts, [], Config);
get_all(Table, Keys, Opts, DdbOpts) ->
get_all(Table, Keys, Opts, DdbOpts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more BatchGetItem operations to get all matching
%% items. Operations are performed in parallel. Order may not be preserved.
%% Getting from only one table is supported.
%%
%% ===Example===
%%
%% `
%% {ok, Items} =
%% erlcloud_ddb_util:get_all(
%% <<"Forum">>,
%% [{<<"Name">>, {s, <<"Amazon DynamoDB">>}},
%% {<<"Name">>, {s, <<"Amazon RDS">>}},
%% {<<"Name">>, {s, <<"Amazon Redshift">>}}],
%% [{projection_expression, <<"Name, Threads, Messages, Views">>}],
%% [{typed_out, false}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec get_all(table_name(), [key()], get_all_opts(), batch_read_ddb_opts(), aws_config()) -> items_return().
get_all(Table, Keys, Opts, DdbOpts, Config) when length(Keys) =< ?BATCH_GET_LIMIT ->
batch_get_retry([{Table, Keys, Opts}], DdbOpts, Config, []);
get_all(Table, Keys, Opts, DdbOpts, Config) ->
BatchList = chop(?BATCH_GET_LIMIT, Keys),
Results = pmap_unordered(
fun(Batch) ->
%% try/catch to prevent hang forever if there is an exception
try
batch_get_retry([{Table, Batch, Opts}], DdbOpts, Config, [])
catch
Type:Ex ->
{error, {Type, Ex}}
end
end,
BatchList),
lists:foldl(fun parfold/2, {ok, []}, Results).
-spec batch_get_retry([erlcloud_ddb2:batch_get_item_request_item()], ddb_opts(), aws_config(), [out_item()]) -> items_return().
batch_get_retry(RequestItems, DdbOpts, Config, Acc) ->
case erlcloud_ddb2:batch_get_item(RequestItems, set_out_opt(DdbOpts), Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_batch_get_item{unprocessed_keys = [],
responses = [#ddb2_batch_get_item_response{items = Items}]}} ->
{ok, Items ++ Acc};
{ok, #ddb2_batch_get_item{unprocessed_keys = Unprocessed,
responses = [#ddb2_batch_get_item_response{items = Items}]}} ->
batch_get_retry(Unprocessed, DdbOpts, Config, Items ++ Acc)
end.
%%%------------------------------------------------------------------------------
%%% list_tables_all
%%%------------------------------------------------------------------------------
list_tables_all() ->
list_tables_all(default_config()).
-spec list_tables_all(aws_config()) -> {ok, [table_name()]} | {error, any()}.
list_tables_all(Config) ->
do_list_tables_all(undefined, Config, []).
do_list_tables_all(LastTable, Config, Result) ->
Options = [{exclusive_start_table_name, LastTable}, {out, record}],
case erlcloud_ddb2:list_tables(Options, Config) of
{ok, #ddb2_list_tables{table_names = TableNames, last_evaluated_table_name = undefined}} ->
{ok, flatreverse([TableNames, Result])};
{ok, #ddb2_list_tables{table_names = TableNames, last_evaluated_table_name = LastTableName}} ->
do_list_tables_all(LastTableName, Config, flatreverse([TableNames, Result]));
{error, _} = Error ->
Error
end.
%%%------------------------------------------------------------------------------
%%% put_all
%%%------------------------------------------------------------------------------
-spec put_all(table_name(), [in_item()]) -> ok | {error, term()}.
put_all(Table, Items) ->
put_all(Table, Items, [], default_config()).
-spec put_all(table_name(), [in_item()], ddb_opts()) -> ok | {error, term()}.
put_all(Table, Items, Opts) ->
put_all(Table, Items, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more BatchWriteItem operations to put all items.
%% Operations are performed in parallel. Writing to only one table is supported.
%%
%% ===Example===
%%
%% `
%% ok =
%% erlcloud_ddb_util:put_all(
%% [{<<"Forum">>,
%% [[{<<"Name">>, {s, <<"Amazon DynamoDB">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}],
%% [{<<"Name">>, {s, <<"Amazon RDS">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}],
%% [{<<"Name">>, {s, <<"Amazon Redshift">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}],
%% [{<<"Name">>, {s, <<"Amazon ElastiCache">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]
%% ]}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec put_all(table_name(), [in_item()], ddb_opts(), aws_config()) -> ok | {error, term()}.
put_all(Table, Items, Opts, Config) ->
write_all(Table, [{put, Item} || Item <- Items], Opts, Config).
%%%------------------------------------------------------------------------------
%%% q_all
%%%------------------------------------------------------------------------------
-type q_all_opts() :: [erlcloud_ddb2:q_opt() | batch_read_ddb_opt()].
-spec q_all(table_name(), conditions() | expression()) -> items_return().
q_all(Table, KeyConditionsOrExpression) ->
q_all(Table, KeyConditionsOrExpression, [], default_config()).
-spec q_all(table_name(), conditions() | expression(), q_all_opts()) -> items_return().
q_all(Table, KeyConditionsOrExpression, Opts) ->
q_all(Table, KeyConditionsOrExpression, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more Query operations to get all matching items.
%%
%% ===Example===
%%
%% `
%% {ok, Items} =
%% erlcloud_ddb_util:q_all(
%% <<"Thread">>,
%% <<"ForumName = :n AND LastPostDateTime BETWEEN :t1 AND :t2">>,
%% [{expression_attribute_values,
%% [{<<":n">>, <<"Amazon DynamoDB">>},
%% {<<":t1">>, <<"20130101">>},
%% {<<":t2">>, <<"20130115">>}]},
%% {index_name, <<"LastPostIndex">>},
%% {select, all_attributes},
%% {consistent_read, true},
%% {typed_out, true}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec q_all(table_name(),
conditions() | expression(),
q_all_opts(),
aws_config()) -> items_return().
q_all(Table, KeyConditionsOrExpression, Opts, Config) ->
q_all(Table, KeyConditionsOrExpression, Opts, Config, [], undefined).
-spec q_all(table_name(),
conditions() | expression(),
q_all_opts(),
aws_config(),
[[out_item()]],
key() | undefined) -> items_return().
q_all(Table, KeyCondOrExpr, Opts0, Config, Acc, StartKey) ->
Opts = [{exclusive_start_key, StartKey}|set_out_opt(Opts0)],
case erlcloud_ddb2:q(Table, KeyCondOrExpr, Opts, Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_q{last_evaluated_key = undefined,
items = undefined,
count = Count}} ->
{ok, lists:sum([Count|Acc])};
{ok, #ddb2_q{last_evaluated_key = undefined,
items = Items}} ->
{ok, flatreverse([Items|Acc])};
{ok, #ddb2_q{last_evaluated_key = LastKey,
items = undefined,
count = Count}} ->
q_all(Table, KeyCondOrExpr, Opts0, Config, [Count|Acc], LastKey);
{ok, #ddb2_q{last_evaluated_key = LastKey,
items = Items}} ->
q_all(Table, KeyCondOrExpr, Opts0, Config, [Items|Acc], LastKey)
end.
%%%------------------------------------------------------------------------------
%%% scan_all
%%%------------------------------------------------------------------------------
-type scan_all_opts() :: [erlcloud_ddb2:scan_opt() | batch_read_ddb_opt()].
-spec scan_all(table_name()) -> items_return().
scan_all(Table) ->
scan_all(Table, [], default_config()).
-spec scan_all(table_name(), scan_all_opts()) -> items_return().
scan_all(Table, Opts) ->
scan_all(Table, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more Scan operations to get all matching items.
%%
%% ===Example===
%%
%% `
%% {ok, Items} =
%% erlcloud_ddb_util:scan_all(
%% <<"Thread">>,
%% [{segment, 0},
%% {total_segments, 4},
%% {typed_out, true}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec scan_all(table_name(), scan_all_opts(), aws_config()) -> items_return().
scan_all(Table, Opts, Config) ->
scan_all(Table, Opts, Config, [], undefined).
-spec scan_all(table_name(),
scan_all_opts(),
aws_config(),
[[out_item()]],
key() | undefined) -> items_return().
scan_all(Table, Opts0, Config, Acc, StartKey) ->
Opts = [{exclusive_start_key, StartKey}|set_out_opt(Opts0)],
case erlcloud_ddb2:scan(Table, Opts, Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_scan{last_evaluated_key = undefined,
items = undefined,
count = Count}} ->
{ok, lists:sum([Count|Acc])};
{ok, #ddb2_scan{last_evaluated_key = undefined,
items = Items}} ->
{ok, flatreverse([Items|Acc])};
{ok, #ddb2_scan{last_evaluated_key = LastKey,
items = undefined,
count = Count}} ->
scan_all(Table, Opts0, Config, [Count|Acc], LastKey);
{ok, #ddb2_scan{last_evaluated_key = LastKey,
items = Items}} ->
scan_all(Table, Opts0, Config, [Items|Acc], LastKey)
end.
%%%------------------------------------------------------------------------------
%%% write_all
%%%------------------------------------------------------------------------------
-type write_all_item() :: erlcloud_ddb2:batch_write_item_request().
-spec write_all(table_name(), [write_all_item()]) -> ok | {error, term()}.
write_all(Table, Items) ->
write_all(Table, Items, [], default_config()).
-spec write_all(table_name(), [write_all_item()], ddb_opts()) -> ok | {error, term()}.
write_all(Table, Items, Opts) ->
write_all(Table, Items, Opts, default_config()).
%%------------------------------------------------------------------------------
%% @doc
%%
%% Perform one or more BatchWriteItem operations to put or delete all items.
%% Operations are performed in parallel. Writing to only one table is supported.
%%
%% ===Example===
%%
%% `
%% ok =
%% erlcloud_ddb_util:write_all(
%% [{<<"Forum">>,
%% [{put, [{<<"Name">>, {s, <<"Amazon DynamoDB">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]},
%% {put, [{<<"Name">>, {s, <<"Amazon RDS">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]},
%% {put, [{<<"Name">>, {s, <<"Amazon Redshift">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]},
%% {put, [{<<"Name">>, {s, <<"Amazon ElastiCache">>}},
%% {<<"Category">>, {s, <<"Amazon Web Services">>}}]}
%% ]}]),
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec write_all(table_name(), [write_all_item()], ddb_opts(), aws_config()) -> ok | {error, term()}.
write_all(Table, Items, _Opts, Config) when length(Items) =< ?BATCH_WRITE_LIMIT ->
batch_write_retry([{Table, Items}], Config);
write_all(Table, Items, _Opts, Config) ->
BatchList = chop(?BATCH_WRITE_LIMIT, Items),
Results = pmap_unordered(
fun(Batch) ->
%% try/catch to prevent hang forever if there is an exception
try
batch_write_retry([{Table, Batch}], Config)
catch
Type:Ex ->
{error, {Type, Ex}}
end
end,
BatchList),
write_all_result(Results).
-spec batch_write_retry([erlcloud_ddb2:batch_write_item_request_item()], aws_config()) -> ok | {error, term()}.
batch_write_retry(RequestItems, Config) ->
case erlcloud_ddb2:batch_write_item(RequestItems, [{out, record}], Config) of
{error, Reason} ->
{error, Reason};
{ok, #ddb2_batch_write_item{unprocessed_items = []}} ->
ok;
{ok, #ddb2_batch_write_item{unprocessed_items = Unprocessed}} ->
batch_write_retry(Unprocessed, Config)
end.
%%------------------------------------------------------------------------------
%% @doc
%% wait until table_status==active.
%%
%% ===Example===
%%
%% `
%% erlcloud_ddb2:wait_for_table_active(<<"TableName">>, 3000, 40, Config)
%% '
%% @end
%%------------------------------------------------------------------------------
-spec wait_for_table_active(table_name(), pos_integer() | infinity, non_neg_integer() | infinity, aws_config()) ->
ok | {error, deleting | retry_threshold_exceeded | any()}.
wait_for_table_active(Table, Interval, RetryTimes, Config) when is_binary(Table), Interval > 0, RetryTimes >= 0 ->
case erlcloud_ddb2:describe_table(Table, [{out, record}], Config) of
{ok, #ddb2_describe_table{table = #ddb2_table_description{table_status = active}}} ->
ok;
{ok, #ddb2_describe_table{table = #ddb2_table_description{table_status = deleting}}} ->
{error, deleting};
{ok, _} ->
case RetryTimes of
infinity ->
timer:sleep(Interval),
wait_for_table_active(Table, infinity, RetryTimes, Config);
1 ->
{error, retry_threshold_exceeded};
_ ->
timer:sleep(Interval),
wait_for_table_active(Table, Interval, RetryTimes - 1, Config)
end;
{error, Reason} ->
{error, Reason}
end.
wait_for_table_active(Table, Interval, RetryTimes) ->
wait_for_table_active(Table, Interval, RetryTimes, default_config()).
wait_for_table_active(Table, AWSCfg) ->
wait_for_table_active(Table, 3000, 100, AWSCfg).
wait_for_table_active(Table) ->
wait_for_table_active(Table, default_config()).
write_all_result([ok | T]) ->
write_all_result(T);
write_all_result([{error, Reason} | _]) ->
{error, Reason};
write_all_result([]) ->
ok.
%%%------------------------------------------------------------------------------
%%% Internal Functions
%%%------------------------------------------------------------------------------
%% Set `out' option to record/typed_record output formats based on `typed_out'
%% boolean setting for get_all, scan_all, q_all. Other output formats are not
%% supported for multi_call reads. Validation is bypassed for backwards
%% compatibility.
-spec set_out_opt(batch_read_ddb_opts()) -> ddb_opts().
set_out_opt(Opts) ->
{OutOpt, NewOpts} = case lists:keytake(typed_out, 1, Opts) of
{value, {typed_out, true}, Opts1} -> {{out, typed_record}, Opts1};
{value, {typed_out, _}, Opts2} -> {{out, record}, Opts2};
false -> {{out, record}, Opts}
end,
lists:keystore(out, 1, NewOpts, OutOpt).
%% Reverses a list of lists and flattens one level
flatreverse(List) ->
lists:foldl(fun(I, A) -> I ++ A end, [], List).
%% fold a set of results from parallel operations producing lists
parfold(_, {error, Reason}) ->
{error, Reason};
parfold({error, Reason}, _) ->
{error, Reason};
parfold({ok, I}, {ok, A}) ->
{ok, I ++ A}.
%% parallel map implementation. See Armstrong's Programming Erlang and
%% http://bc.tech.coop/blog/070601.html
pmap_unordered(F, L) ->
Parent = self(),
Ref = make_ref(),
Pids = [spawn(fun() -> Parent ! {Ref, F(X)} end) || X <- L],
[receive {Ref, Result} -> Result end || _ <- Pids].
%% creates a list of list each of which has N or fewer elements
chop(N, List) ->
chop(N, List, []).
chop(_, [], Acc) ->
lists:reverse(Acc);
chop(N, List, Acc) ->
{H, T} = safe_split(N, List),
chop(N, T, [H | Acc]).
%% lists:split throws if N is larger than the list, safe_split doesn't
safe_split(N, List) ->
safe_split(N, List, []).
safe_split(0, List, Acc) ->
{lists:reverse(Acc), List};
safe_split(_, [], Acc) ->
{lists:reverse(Acc), []};
safe_split(N, [H|T], Acc) ->
safe_split(N - 1, T, [H | Acc]).