%%%-------------------------------------------------------------------
%%% @doc
%%% Partitioned/Sharded ETS tables.
%%%
%%% <h2>Features</h2>
%%%
%%% <ul>
%%% <li>
%%% `shards' implements the ETS API to keep compatibility and make the usage
%%% straightforward; exactly the same if you were using `ets'.
%%% </li>
%%% <li>
%%% Sharded or partitioned tables under-the-hood. This feature is managed
%%% entirely by `shards' and it is 100% transparent for the client. Provides
%%% a logical view of a single ETS table, but internally, it is composed by
%%% `N' number of `partitions' and each partition has associated an ETS table.
%%% </li>
%%% <li>
%%% High performance and scalability. `shards' keeps the lock contention under
%%% control enabling ETS tables to scale out and support higher levels of
%%% concurrency without lock issues; specially write-locks, which most of the
%%% cases might cause significant performance degradation.
%%% </li>
%%% </ul>
%%%
%%% <h2>Partitioned Table</h2>
%%%
%%% When a table is created with `shards:new/2', a new supervision tree is
%%% created to represent the partitioned table. There is a main supervisor
%%% `shards_partition_sup' that create an ETS table to store the metadata
%%% and also starts the children which are the partitions to create. Each
%%% partition is owned by `shards_partition' (it is a `gen_server') and it
%%% creates an ETS table for storing data mapped to that partition. The
%%% supervision tree looks like:
%%%
%%% <br></br>
%%% ```
%%%                           [shards_partition]--><ETS-Table>
%%%                           /
%%% [shards_partition_sup]--<-[shards_partition]--><ETS-Table>
%%%           |               \
%%%  <Metadata-ETS-Table>     [shards_partition]--><ETS-Table>
%%% '''
%%% <br></br>
%%%
%%% The returned value by `shards:new/2' may be an atom if it is a named
%%% table or a reference otherwise, and in the second case the returned
%%% reference is the one of the metadata table, which is the main entry
%%% point and it is owned by the main supervisor. See `shards:new/2' for
%%% more information.
%%%
%%% <h2>Usage</h2>
%%%
%%% ```
%%% > Tab = shards:new(tab, []).
%%% #Ref<0.1541908042.2337144842.31535>
%%% > shards:insert(Tab, [{a, 1}, {b, 2}, {c, 3}]).
%%% true
%%% > shards:lookup(Tab, a).
%%% [{a,1}]
%%% > shards:lookup(Tab, b).
%%% [{b,2}]
%%% > shards:lookup(Tab, c).
%%% [{c,3}]
%%% > shards:lookup_element(Tab, c, 2).
%%% 3
%%% > shards:lookup(Tab, d).
%%% []
%%% > shards:delete(Tab, c).
%%% true
%%% > shards:lookup(Tab, c).
%%% []
%%% '''
%%%
%%% As you can see, the usage is exactly the same if you were using `ets',
%%% you can try the rest of the ETS API but with `shards' module.
%%%
%%% <h2>Important</h2>
%%%
%%% Despite `shards' aims to keep 100% compatibility with current ETS API,
%%% the semantic for some of the functions may be a bit different due to
%%% the nature of sharding; it is not the same having all the entries in
%%% a single table than distributed across multiple ones. For example,
%%% for query-based functions like `select', `match', etc., the returned
%%% entries are the same but not necessary the same order than `ets'.
%%% For `first', `next', and `last' they behave the similar in the sense
%%% by means of them a partitioned table is traversed, so the final result
%%% is the same, but the order in which the entries are traversed may be
%%% different. Therefore, it is highly recommended to read the documentation
%%% of the functions.
%%%
%%% @end
%%%-------------------------------------------------------------------
-module(shards).

%% Helpers
-export([
  meta/1,
  partition_owners/1
]).

%% ETS API
-export([
  all/0,
  delete/1, delete/2, delete/3,
  delete_all_objects/1, delete_all_objects/2,
  delete_object/2, delete_object/3,
  file2tab/1, file2tab/2,
  first/1, first/2,
  foldl/3, foldl/4,
  foldr/3, foldr/4,
  i/0,
  info/1, info/2,
  insert/2, insert/3,
  insert_new/2, insert_new/3,
  is_compiled_ms/1,
  last/1,
  lookup/2, lookup/3,
  lookup_element/3, lookup_element/4,
  match/1, match/2, match/3, match/4,
  match_delete/2, match_delete/3,
  match_object/1, match_object/2, match_object/3, match_object/4,
  match_spec_compile/1,
  match_spec_run/2,
  member/2, member/3,
  new/2,
  next/2, next/3,
  prev/2,
  rename/2, rename/3,
  safe_fixtable/2,
  select/1, select/2, select/3, select/4,
  select_count/2, select_count/3,
  select_delete/2, select_delete/3,
  select_replace/2, select_replace/3,
  select_reverse/1, select_reverse/2, select_reverse/3, select_reverse/4,
  setopts/2, setopts/3,
  tab2file/2, tab2file/3,
  tab2list/1, tab2list/2,
  tabfile_info/1,
  table/1, table/2, table/3,
  test_ms/2,
  take/2, take/3,
  update_counter/3, update_counter/4, update_counter/5,
  update_element/3, update_element/4
]).

%%%===================================================================
%%% Types & Macros
%%%===================================================================

%% @type tab() = atom() | ets:tid().
%%
%% Table parameter
-type tab() :: atom() | ets:tid().

%% ETS Types
-type access() :: public | protected | private.
-type type() :: set | ordered_set | bag | duplicate_bag.
-type ets_continuation() ::
        '$end_of_table'
        | {tab(), integer(), integer(), ets:comp_match_spec(), list(), integer()}
        | {tab(), _, _, integer(), ets:comp_match_spec(), list(), integer(), integer()}.

%% @type tweaks() =
%%         {write_concurrency, boolean()}
%%         | {read_concurrency, boolean()}
%%         | compressed.
%%
%% ETS tweaks option
-type tweaks() ::
        {write_concurrency, boolean()}
        | {read_concurrency, boolean()}
        | compressed.

%% @type shards_opt() =
%%         {partitions, pos_integer()}
%%         | {keyslot_fun, shards_meta:keyslot_fun()}
%%         | {restore, term(), term()}.
%%
%% Shards extended options.
-type shards_opt() ::
        {partitions, pos_integer()}
        | {keyslot_fun, shards_meta:keyslot_fun()}
        | {restore, term(), term()}.

%% @type option() =
%%         type() | access() | named_table
%%         | {keypos, pos_integer()}
%%         | {heir, pid(), HeirData :: term()}
%%         | {heir, none} | tweaks()
%%         | shards_opt().
%%
%% Create table options – used by `new/2'.
-type option() ::
        type()
        | access()
        | named_table
        | {keypos, pos_integer()}
        | {heir, pid(), HeirData :: term()}
        | {heir, none}
        | tweaks()
        | shards_opt().

%% ETS Info Tuple
-type info_tuple() ::
        {compressed, boolean()}
        | {heir, pid() | none}
        | {keypos, pos_integer()}
        | {memory, non_neg_integer()}
        | {name, atom()}
        | {named_table, boolean()}
        | {node, node()}
        | {owner, pid()}
        | {protection, access()}
        | {size, non_neg_integer()}
        | {type, type()}
        | {write_concurrency, boolean()}
        | {read_concurrency, boolean()}
        | {shards, [atom()]}.

%% ETS Info Item
-type info_item() ::
        compressed | fixed | heir | keypos | memory
        | name | named_table | node | owner | protection
        | safe_fixed | size | stats | type
        | write_concurrency | read_concurrency
        | shards.

%% ETS TabInfo Item
-type tabinfo_item() ::
        {name, atom()}
        | {type, type()}
        | {protection, access()}
        | {named_table, boolean()}
        | {keypos, non_neg_integer()}
        | {size, non_neg_integer()}
        | {extended_info, [md5sum | object_count]}
        | {version, {Major :: non_neg_integer(), Minor :: non_neg_integer()}}
        | {shards, [atom()]}.

%% @type continuation() = {
%%         Tab          :: tab(),
%%         MatchSpec    :: ets:match_spec(),
%%         Limit        :: pos_integer(),
%%         Partition    :: non_neg_integer(),
%%         Continuation :: ets_continuation()
%%       }.
%%
%% Defines the convention for the query functions:
%% <ul>
%% <li>`Tab': Table reference.</li>
%% <li>`MatchSpec': The `ets:match_spec()'.</li>
%% <li>`Limit': Results limit.</li>
%% <li>`Partition': Partition index.</li>
%% <li>`Continuation': The `ets:continuation()'.</li>
%% </ul>
-type continuation() :: {
        Tab          :: tab(),
        MatchSpec    :: ets:match_spec(),
        Limit        :: pos_integer(),
        Partition    :: non_neg_integer(),
        Continuation :: ets_continuation()
      }.

%% @type filename() = string() | binary() | atom().
-type filename() :: string() | binary() | atom().

% Exported Types
-export_type([
  tab/0,
  option/0,
  info_tuple/0,
  info_item/0,
  tabinfo_item/0,
  continuation/0,
  filename/0
]).

%% Macro to check if the given Filename has the right type
-define(is_filename(FN_), is_list(FN_); is_binary(FN_); is_atom(FN_)).

%%%===================================================================
%%% Helpers
%%%===================================================================

%% @doc Returns the metadata associated with the given table `Tab'.
-spec meta(Tab :: tab()) -> shards_meta:t().
meta(Tab) ->
  shards_meta:get(Tab).

%% @doc Returns the partition PIDs associated with the given table `TabOrPid'.
-spec partition_owners(TabOrPid) -> [OwnerPid] when
      TabOrPid :: pid() | tab(),
      OwnerPid :: pid().
partition_owners(TabOrPid) when is_pid(TabOrPid) ->
  [Child || {_, Child, _, _} <- supervisor:which_children(TabOrPid)];
partition_owners(TabOrPid) when is_atom(TabOrPid); is_reference(TabOrPid) ->
  partition_owners(shards_meta:tab_pid(TabOrPid)).

%%%===================================================================
%%% ETS API
%%%===================================================================

%% @equiv ets:all()
all() ->
  ets:all().

%% @doc
%% Equivalent to `ets:delete/1'.
%%
%% @see ets:delete/1.
%% @end
-spec delete(Tab :: tab()) -> true.
delete(Tab) ->
  Meta = shards_meta:get(Tab),
  TabPid = shards_meta:tab_pid(Meta),
  ok = shards_partition_sup:stop(TabPid),
  true.

%% @equiv delete(Tab, Key, shards_meta:get(Tab))
delete(Tab, Key) ->
  delete(Tab, Key, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:delete/2'.
%%
%% @see ets:delete/2.
%% @end
-spec delete(Tab, Key, Meta) -> true when
      Tab  :: tab(),
      Key  :: term(),
      Meta :: shards_meta:t().
delete(Tab, Key, Meta) ->
  PartTid = shards_partition:tid(Tab, Key, Meta),
  ets:delete(PartTid, Key).

%% @equiv delete_all_objects(Tab, shards_meta:get(Tab))
delete_all_objects(Tab) ->
  delete_all_objects(Tab, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:delete_all_objects/1'.
%%
%% @see ets:delete_all_objects/1.
%% @end
-spec delete_all_objects(Tab, Meta) -> true when
      Tab  :: tab(),
      Meta :: shards_meta:t().
delete_all_objects(Tab, Meta) ->
  _ = mapred(Tab, fun ets:delete_all_objects/1, Meta),
  true.

%% @equiv delete_object(Tab, Object, shards_meta:get(Tab))
delete_object(Tab, Object) ->
  delete_object(Tab, Object, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:delete_object/2'.
%%
%% @see ets:delete_object/2.
%% @end
-spec delete_object(Tab, Object, Meta) -> true when
      Tab    :: tab(),
      Object :: tuple(),
      Meta   :: shards_meta:t().
delete_object(Tab, Object, Meta) when is_tuple(Object) ->
  Key = shards_lib:object_key(Object, Meta),
  PartTid = shards_partition:tid(Tab, Key, Meta),
  ets:delete_object(PartTid, Object).

%% @equiv file2tab(Filename, [])
file2tab(Filename) ->
  file2tab(Filename, []).

%% @doc
%% Equivalent to `shards:file2tab/2'. Moreover, it restores the
%% supervision tree for the `shards' corresponding to the given
%% file, such as if they had been created using `shards:new/2'.
%%
%% @see ets:file2tab/2.
%% @end
-spec file2tab(Filename, Options) -> {ok, Tab} | {error, Reason} when
      Filename :: filename(),
      Tab      :: tab(),
      Options  :: [Option],
      Option   :: {verify, boolean()},
      Reason   :: term().
file2tab(Filename, Options) when ?is_filename(Filename) ->
  try
    StrFilename = shards_lib:to_string(Filename),
    Header = shards_lib:read_tabfile(StrFilename),

    TabName = maps:get(name, Header),
    Meta = maps:get(metadata, Header),
    Partitions = maps:get(partitions, Header),

    TabOpts = [
      {partitions, shards_meta:partitions(Meta)},
      {keyslot_fun, shards_meta:keyslot_fun(Meta)}
      | shards_meta:ets_opts(Meta)
    ],

    Return = new(TabName, [{restore, Partitions, Options} | TabOpts]),
    {ok, Return}
  catch
    throw:Error ->
      Error;

    error:{error, _} = Error ->
      Error;

    error:{badarg, Arg} ->
      {error, {read_error, {file_error, Arg, enoent}}}
  end.

%% @equiv first(Tab, shards_meta:get(Tab))
first(Tab) ->
  first(Tab, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:first/1'.
%%
%% However, the order in which results are returned might be not the same as
%% the original ETS function, since it is a sharded table.
%%
%% @see ets:first/1.
%% @end
-spec first(Tab, Meta) -> Key | '$end_of_table' when
      Tab  :: tab(),
      Key  :: term(),
      Meta :: shards_meta:t().
first(Tab, Meta) ->
  N = shards_meta:partitions(Meta),
  Partition = N - 1,
  first(Tab, ets:first(shards_partition:tid(Tab, Partition)), Partition).

%% @private
first(Tab, '$end_of_table', Partition) when Partition > 0 ->
  NextPartition = Partition - 1,
  first(Tab, ets:first(shards_partition:tid(Tab, NextPartition)), NextPartition);
first(_, '$end_of_table', _) ->
  '$end_of_table';
first(_, Key, _) ->
  Key.

%% @equiv foldl(Fun, Acc, Tab, shards_meta:get(Tab))
foldl(Fun, Acc, Tab) ->
  foldl(Fun, Acc, Tab, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:foldl/3'.
%%
%% However, the order in which the entries are traversed may be different
%% since they are distributed across multiple partitions.
%%
%% @see ets:foldl/3.
%% @end
-spec foldl(Fun, Acc0, Tab, Meta) -> Acc1 when
      Fun    :: fun((Element :: term(), AccIn) -> AccOut),
      Tab    :: tab(),
      Meta   :: shards_meta:t(),
      Acc0   :: term(),
      Acc1   :: term(),
      AccIn  :: term(),
      AccOut :: term().
foldl(Fun, Acc, Tab, Meta) ->
  fold(foldl, Fun, Acc, Tab, shards_meta:partitions(Meta) - 1).

%% @equiv foldr(Fun, Acc, Tab, shards_meta:get(Tab))
foldr(Fun, Acc, Tab) ->
  foldr(Fun, Acc, Tab, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:foldr/3'.
%%
%% However, the order in which the entries are traversed may be different
%% since they are distributed across multiple partitions.
%%
%% @see ets:foldr/3.
%% @end
-spec foldr(Fun, Acc0, Tab, Meta) -> Acc1 when
      Fun    :: fun((Element :: term(), AccIn) -> AccOut),
      Tab    :: tab(),
      Meta   :: shards_meta:t(),
      Acc0   :: term(),
      Acc1   :: term(),
      AccIn  :: term(),
      AccOut :: term().
foldr(Fun, Acc, Tab, Meta) ->
  fold(foldr, Fun, Acc, Tab, shards_meta:partitions(Meta) - 1).

%% @private
fold(Fold, Fun, Acc, Tab, Partition) when Partition >= 0 ->
  NewAcc = ets:Fold(Fun, Acc, shards_partition:tid(Tab, Partition)),
  fold(Fold, Fun, NewAcc, Tab, Partition - 1);
fold(_Fold, _Fun, Acc, _Tab, _Partition) ->
  Acc.

%% @equiv ets:i()
i() -> ets:i().

%% @doc
%% Similar to ets:info/1' but extra information about the partitioned
%% table is added.
%%
%% <h2>Extra Info:</h2>
%%
%% <ul>
%% <li>
%% `{partitions, post_integer()}' - Number of partitions.
%% </li>
%% <li>
%% `{keyslot_fun, shards_meta:keyslot_fun()}' - Functions used for compute
%% the keyslot.
%% </li>
%% <li>
%% `{parallel, boolean()}' - Whether the parallel mode is enabled or not.
%% </li>
%% </ul>
%%
%% @see ets:info/1.
%% @end
-spec info(Tab) -> InfoList | undefined when
      Tab       :: tab(),
      InfoList  :: [InfoTuple],
      InfoTuple :: info_tuple().
info(Tab) ->
  with_meta(Tab, fun(Meta) ->
    do_info(Tab, Meta)
  end).

%% @doc
%% Equivalent to `ets:info/2'.
%%
%% See the added items by `shards:info/1'.
%%
%% @see ets:info/2.
%% @end
-spec info(Tab, Item) -> Value | undefined when
      Tab   :: tab(),
      Item  :: info_item(),
      Value :: term().
info(Tab, Item) when is_atom(Item) ->
  with_meta(Tab, fun(Meta) ->
    shards_lib:keyfind(Item, do_info(Tab, Meta))
  end).

%% @private
do_info(Tab, Meta) ->
  InfoLists = mapred(Tab, fun ets:info/1, Meta),

  [
    {partitions, shards_meta:partitions(Meta)},
    {keyslot_fun, shards_meta:keyslot_fun(Meta)},
    {parallel, shards_meta:parallel(Meta)}
    | parts_info(Tab, InfoLists, [memory])
  ].

%% @equiv insert(Tab, ObjOrObjs, shards_meta:get(Tab))
insert(Tab, ObjOrObjs) ->
  insert(Tab, ObjOrObjs, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:insert/2'.
%%
%% Despite this functions behaves exactly the same as `ets:insert/2'
%% and produces the same result, there is a big difference due to the
%% nature of the sharding distribution model, <b>IT IS NOT ATOMIC</b>.
%% Therefore, if it fails by inserting an object at some partition,
%% previous inserts execution on other partitions are not rolled back,
%% but an error is raised instead.
%%
%% @see ets:insert/2.
%% @end
-spec insert(Tab, ObjOrObjs, Meta) -> true | no_return() when
      Tab       :: tab(),
      ObjOrObjs :: tuple() | [tuple()],
      Meta      :: shards_meta:t().
insert(Tab, ObjOrObjs, Meta) when is_list(ObjOrObjs) ->
  maps:fold(fun(Partition, Group, Acc) ->
    Acc = ets:insert(Partition, Group)
  end, true, group_keys_by_partition(Tab, ObjOrObjs, Meta));
insert(Tab, ObjOrObjs, Meta) when is_tuple(ObjOrObjs) ->
  ets:insert(get_part_tid(Tab, ObjOrObjs, Meta), ObjOrObjs).

%% @equiv insert_new(Tab, ObjOrObjs, shards_meta:get(Tab))
insert_new(Tab, ObjOrObjs) ->
  insert_new(Tab, ObjOrObjs, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:insert_new/2'.
%%
%% Despite this functions behaves exactly the same as `ets:insert_new/2'
%% and produces the same result, there is a big difference due to the
%% nature of the sharding distribution model, <b>IT IS NOT ATOMIC</b>.
%% Opposite to `shards:insert/2', this function tries to roll-back
%% previous inserts execution on other partitions if it fails by
%% inserting an object at some partition, but there might be race
%% conditions during roll-back execution.
%%
%% <b>Example:</b>
%%
%% ```
%% > shards:insert_new(mytab, {k1, 1}).
%% true
%%
%% > shards:insert_new(mytab, {k1, 1}).
%% false
%%
%% > shards:insert_new(mytab, [{k1, 1}, {k2, 2}]).
%% false
%% '''
%%
%% @see ets:insert_new/2.
%% @end
-spec insert_new(Tab, ObjOrObjs, Meta) -> boolean() when
      Tab       :: tab(),
      ObjOrObjs :: tuple() | [tuple()],
      Meta      :: shards_meta:t().
insert_new(Tab, ObjOrObjs, Meta) when is_list(ObjOrObjs) ->
  Result =
    shards_enum:reduce_while(fun({Partition, Group}, Acc) ->
      case ets:insert_new(Partition, Group) of
        true ->
          {cont, Group ++ Acc};

        false ->
          ok = rollback_insert(Tab, Acc, Meta),
          {halt, false}
      end
    end, [], group_keys_by_partition(Tab, ObjOrObjs, Meta)),

  case Result of
    false -> false;
    _     -> true
  end;
insert_new(Tab, ObjOrObjs, Meta) when is_tuple(ObjOrObjs) ->
  ets:insert_new(get_part_tid(Tab, ObjOrObjs, Meta), ObjOrObjs).

%% @private
rollback_insert(Tab, Entries, Meta) ->
  lists:foreach(fun(Entry) ->
    Key = element(shards_meta:keypos(Meta), Entry),
    ?MODULE:delete(Tab, Key)
  end, Entries).

%% @equiv ets:is_compiled_ms(Term)
is_compiled_ms(Term) ->
  ets:is_compiled_ms(Term).

%% @doc
%% Equivalent to `ets:last/1'.
%%
%% However, the order in which results are returned might be not the same as
%% the original ETS function, since it is a sharded table.
%%
%% @see ets:last/1.
%% @end
-spec last(Tab) -> Key | '$end_of_table' when
      Tab :: tab(),
      Key :: term().
last(Tab) ->
  Partition0 = shards_partition:tid(Tab, 0),

  case ets:info(Partition0, type) of
    ordered_set -> ets:last(Partition0);
    _TableType  -> first(Tab)
  end.

%% @equiv lookup(Tab, Key, shards_meta:get(Tab))
lookup(Tab, Key) ->
  lookup(Tab, Key, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:lookup/2'.
%%
%% @see ets:lookup/2.
%% @end
-spec lookup(Tab, Key, Meta) -> [Object] when
      Tab    :: tab(),
      Key    :: term(),
      Meta   :: shards_meta:t(),
      Object :: tuple().
lookup(Tab, Key, Meta) ->
  PartTid = shards_partition:tid(Tab, Key, Meta),
  ets:lookup(PartTid, Key).

%% @equiv lookup_element(Tab, Key, Pos, shards_meta:get(Tab))
lookup_element(Tab, Key, Pos) ->
  lookup_element(Tab, Key, Pos, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:lookup_element/3'.
%%
%% @see ets:lookup_element/3.
%% @end
-spec lookup_element(Tab, Key, Pos, Meta) -> Elem when
      Tab  :: tab(),
      Key  :: term(),
      Pos  :: pos_integer(),
      Meta :: shards_meta:t(),
      Elem :: term() | [term()].
lookup_element(Tab, Key, Pos, Meta) ->
  PartTid = shards_partition:tid(Tab, Key, Meta),
  ets:lookup_element(PartTid, Key, Pos).

%% @equiv match(Tab, Pattern, shards_meta:get(Tab))
match(Tab, Pattern) ->
  match(Tab, Pattern, shards_meta:get(Tab)).

%% @doc
%% If 3rd argument is `pos_integer()' this function behaves
%% like `ets:match/3', otherwise, the 3rd argument is
%% assumed as `shards_meta:t()` and it behaves like
%% `ets:match/2'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:match/2.
%% @see ets:match/3.
%% @end
-spec match(Tab, Pattern, LimitOrMeta) -> {[Match], Cont} | '$end_of_table' | [Match] when
      Tab         :: tab(),
      Pattern     :: ets:match_pattern(),
      LimitOrMeta :: pos_integer() | shards_meta:t(),
      Match       :: [term()],
      Cont        :: continuation().
match(Tab, Pattern, Limit) when is_integer(Limit), Limit > 0 ->
  match(Tab, Pattern, Limit, shards_meta:get(Tab));
match(Tab, Pattern, Meta) ->
  Map = {fun ets:match/2, [Pattern]},
  Reduce = fun erlang:'++'/2,
  mapred(Tab, Map, Reduce, Meta).

%% @doc
%% Equivalent to `ets:match/3'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:match/3.
%% @end
-spec match(Tab, Pattern, Limit, Meta) -> {[Match], Cont} | '$end_of_table' when
      Tab     :: tab(),
      Pattern :: ets:match_pattern(),
      Limit   :: pos_integer(),
      Meta    :: shards_meta:t(),
      Match   :: [term()],
      Cont    :: continuation().
match(Tab, Pattern, Limit, Meta) when is_integer(Limit), Limit > 0 ->
  N = shards_meta:partitions(Meta),
  q(match, Tab, Pattern, Limit, q_fun(), Limit, N - 1, {[], nil}).

%% @doc
%% Equivalent to `ets:match/1'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:match/1.
%% @end
-spec match(Continuation) -> {[Match], continuation()} | '$end_of_table' when
      Match        :: [term()],
      Continuation :: continuation().
match({_, _, Limit, _, _} = Continuation) ->
  q(match, Continuation, q_fun(), Limit, []).

%% @equiv match_delete(Tab, Pattern, shards_meta:get(Tab))
match_delete(Tab, Pattern) ->
  match_delete(Tab, Pattern, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:match_delete/2'.
%%
%% @see ets:match_delete/2.
%% @end
-spec match_delete(Tab, Pattern, Meta) -> true when
      Tab     :: tab(),
      Pattern :: ets:match_pattern(),
      Meta    :: shards_meta:t().
match_delete(Tab, Pattern, Meta) ->
  Map = {fun ets:match_delete/2, [Pattern]},
  Reduce = {fun erlang:'and'/2, true},
  mapred(Tab, Map, Reduce, Meta).

%% @equiv match_object(Tab, Pattern, shards_meta:get(Tab))
match_object(Tab, Pattern) ->
  match_object(Tab, Pattern, shards_meta:get(Tab)).

%% @doc
%% If 3rd argument is `pos_integer()' this function behaves
%% like `ets:match_object/3', otherwise, the 3rd argument is
%% assumed as `shards_meta:t()` and it behaves like
%% `ets:match_object/2'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:match_object/3.
%% @end
-spec match_object(Tab, Pattern, LimitOrMeta) -> {[Object], Cont} | '$end_of_table' | [Object] when
      Tab         :: tab(),
      Pattern     :: ets:match_pattern(),
      LimitOrMeta :: pos_integer() | shards_meta:t(),
      Object      :: tuple(),
      Cont        :: continuation().
match_object(Tab, Pattern, Limit) when is_integer(Limit), Limit > 0 ->
  match_object(Tab, Pattern, Limit, shards_meta:get(Tab));
match_object(Tab, Pattern, Meta) ->
  Map = {fun ets:match_object/2, [Pattern]},
  Reduce = fun erlang:'++'/2,
  mapred(Tab, Map, Reduce, Meta).

%% @doc
%% Equivalent to `ets:match_object/3'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:match_object/3.
%% @end
-spec match_object(Tab, Pattern, Limit, Meta) -> {[Object], Cont} | '$end_of_table' when
      Tab     :: tab(),
      Pattern :: ets:match_pattern(),
      Limit   :: pos_integer(),
      Meta    :: shards_meta:t(),
      Object  :: tuple(),
      Cont    :: continuation().
match_object(Tab, Pattern, Limit, Meta) when is_integer(Limit), Limit > 0 ->
  N = shards_meta:partitions(Meta),
  q(match_object, Tab, Pattern, Limit, q_fun(), Limit, N - 1, {[], nil}).

%% @doc
%% Equivalent to `ets:match_object/1'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:match_object/1.
%% @end
-spec match_object(Cont) -> {[Object], Cont} | '$end_of_table' when
      Object :: tuple(),
      Cont   :: continuation().
match_object({_, _, Limit, _, _} = Continuation) ->
  q(match_object, Continuation, q_fun(), Limit, []).

%% @equiv ets:match_spec_compile(MatchSpec)
match_spec_compile(MatchSpec) ->
  ets:match_spec_compile(MatchSpec).

%% @equiv ets:match_spec_run(List, CompiledMatchSpec)
match_spec_run(List, CompiledMatchSpec) ->
  ets:match_spec_run(List, CompiledMatchSpec).

%% @equiv member(Tab, Key, shards_meta:get(Tab))
member(Tab, Key) ->
  member(Tab, Key, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:member/2'.
%%
%% @see ets:member/2.
%% @end
-spec member(Tab, Key, Meta) -> boolean() when
      Tab  :: tab(),
      Key  :: term(),
      Meta :: shards_meta:t().
member(Tab, Key, Meta) ->
  PartTid = shards_partition:tid(Tab, Key, Meta),
  ets:member(PartTid, Key).

%% @doc
%% This operation is equivalent to `ets:new/2', but when is called,
%% instead of create a single ETS table, it creates a new supervision
%% tree for the partitioned table.
%%
%% The supervision tree is composed by a main supervisor `shards_partition_sup'
%%  and `N' number of workers or partitions handled by `shards_partition'
%% (partition owner). Each worker creates an ETS table to handle the partition.
%% Also, the main supervisor `shards_partition_sup' creates an ETS table to
%% keep the metadata for the partitioned table.
%%
%% Returns an atom if the created table is a named table, otherwise,
%% a reference is returned. In the last case, the returned reference
%% is the one of the metadata table, which is the main entry-point
%% and it is owned by the main supervisor `shards_partition_sup'.
%%
%% <h3>Options:</h3>
%% In addition to the options given by `ets:new/2', this functions provides
%% the next options:
%% <ul>
%% <li>
%% `{partitions, N}' - Specifies the number of partitions for the sharded
%% table. By default, `N = erlang:system_info(schedulers_online)'.
%% </li>
%% <li>
%% `{keyslot_fun, F}' - Specifies the function used to compute the partition
%% where the action will be evaluated. Defaults to `erlang:phash2/2'.
%% </li>
%% <li>
%% `{parallel, P}' - Specifies whether `shards' should work in parallel mode
%% or not, for the applicable functions, e.g.: `select', `match', etc. By
%% default is set to `false'.
%% </li>
%% <li>
%% `{parallel_timeout, T}' - When `parallel' is set to `true', it specifies
%% the max timeout for a parallel execution. Defaults to `infinity'.
%% </li>
%% </ul>
%%
%% <h3>Access:</h3>
%% Currently, only `public' access is supported by `shards:new/2'. Since a
%% partitioned table is started with its own supervision tree when created,
%% it is very tricky to provide `private' or `protected' access since there
%% are multiple partitions (or ETS tables) and they are owned by the
%% supervisor's children, and the supervisor along with their children
%% (or partitions) are managed by `shards' under-the-hood; it is completely
%% transparent for the client.
%%
%% <h3>Examples:</h3>
%%
%% ```
%% > Tab = shards:new(tab1, []).
%% #Ref<0.1541908042.2337144842.31535>
%%
%% > shards:new(tab2, [named_table]).
%% tab2
%% '''
%%
%% See also the <b>"Partitioned Table"</b> section at the module documentation
%% for more information.
%%
%% @see ets:new/2.
%% @end
-spec new(Name, Options) -> Tab when
      Name    :: atom(),
      Options :: [option()],
      Tab     :: tab().
new(Name, Options) ->
  with_trap_exit(fun() ->
    ParsedOpts = shards_opts:parse(Options),
    StartResult = shards_partition_sup:start_link(Name, ParsedOpts),
    do_new(StartResult, Name, ParsedOpts)
  end).

%% @private
do_new({ok, Pid}, Name, Options) ->
  ok = maybe_register(Name, Pid, maps:get(ets_opts, Options)),
  shards_partition:retrieve_tab(shards_lib:get_sup_child(Pid, 0));
do_new({error, {shutdown, {_, _, {restore_error, Error}}}}, _Name, _Options) ->
  ok = wrap_exit(),
  error(Error);
do_new({error, {Reason, _}}, _Name, _Options) ->
  ok = wrap_exit(),
  error(Reason).

%% @private
maybe_register(Name, Pid, Options) ->
  case lists:member(named_table, Options) of
    true ->
      true = register(Name, Pid),
      ok;

    false ->
      ok
  end.

%% @private
wrap_exit() ->
  % We wait for the 'EXIT' signal from the partition supervisor knowing
  % that it will reply at some point, either after roughly timeout or
  % when an 'EXIT' signal response is ready.
  receive
    {'EXIT', _Pid, _Reason} -> ok
  end.

%% @equiv next(Tab, Key1, shards_meta:get(Tab))
next(Tab, Key1) ->
  next(Tab, Key1, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:next/2'.
%%
%% However, the order in which results are returned might be not the same as
%% the original ETS function, since it is a sharded table.
%%
%% @see ets:next/2.
%% @end
-spec next(Tab, Key1, Meta) -> Key2 | '$end_of_table' when
      Tab  :: tab(),
      Key1 :: term(),
      Key2 :: term(),
      Meta :: shards_meta:t().
next(Tab, Key1, Meta) ->
  KeyslotFun = shards_meta:keyslot_fun(Meta),
  Partitions = shards_meta:partitions(Meta),
  Idx = KeyslotFun(Key1, Partitions),
  PartTid = shards_partition:tid(Tab, Idx),
  next_(Tab, ets:next(PartTid, Key1), Idx).

%% @private
next_(Tab, '$end_of_table', Partition) when Partition > 0 ->
  NextPartition = Partition - 1,
  next_(Tab, ets:first(shards_partition:tid(Tab, NextPartition)), NextPartition);
next_(_, '$end_of_table', _) ->
  '$end_of_table';
next_(_, Key2, _) ->
  Key2.

%% @doc
%% Equivalent to `ets:next/2'.
%%
%% However, the order in which results are returned might be not the same as
%% the original ETS function, since it is a sharded table.
%%
%% @see ets:prev/2.
%% @end
-spec prev(Tab, Key1) -> Key2 | '$end_of_table' when
      Tab  :: tab(),
      Key1 :: term(),
      Key2 :: term().
prev(Tab, Key1) ->
  Partition0 = shards_partition:tid(Tab, 0),

  case ets:info(Partition0, type) of
    ordered_set -> ets:prev(Partition0, Key1);
    _TableType  -> next(Tab, Key1)
  end.

%% @equiv rename(Tab, Name, shards_meta:get(Tab))
rename(Tab, Name) ->
  rename(Tab, Name, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:rename/2'.
%%
%% Renames the table name and all its associated shard tables.
%% If something unexpected occurs during the process, an exception
%% will be raised.
%%
%% @see ets:rename/2.
%% @end
-spec rename(Tab, Name, Meta) -> Name when
      Tab  :: tab(),
      Name :: atom(),
      Meta :: shards_meta:t().
rename(Tab, Name, Meta) ->
  Pid = shards_meta:tab_pid(Meta),
  true = unregister(Tab),
  true = register(Name, Pid),
  Name = shards_meta:rename(Tab, Name).

%% @equiv safe_fixtable(Tab, Fix, shards_meta:get(Tab))
safe_fixtable(Tab, Fix) ->
  safe_fixtable(Tab, Fix, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:safe_fixtable/2'.
%%
%% Returns `true' if the function was applied successfully
%% on each partition, otherwise `false' is returned.
%%
%% @see ets:safe_fixtable/2.
%% @end
-spec safe_fixtable(Tab, Fix, Meta) -> true when
      Tab  :: tab(),
      Fix  :: boolean(),
      Meta :: shards_meta:t().
safe_fixtable(Tab, Fix, Meta) ->
  Map = {fun ets:safe_fixtable/2, [Fix]},
  Reduce = {fun erlang:'and'/2, true},
  mapred(Tab, Map, Reduce, Meta).

%% @equiv select(Tab, MatchSpec, shards_meta:get(Tab))
select(Tab, MatchSpec) ->
  select(Tab, MatchSpec, shards_meta:get(Tab)).

%% @doc
%% If 3rd argument is `pos_integer()' this function behaves
%% like `ets:select/3', otherwise, the 3rd argument is
%% assumed as `shards_meta:t()` and it behaves like
%% `ets:select/2'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:select/3.
%% @end
-spec select(Tab, MatchSpec, LimitOrMeta) -> {[Match], Cont} | '$end_of_table' | [Match] when
      Tab         :: tab(),
      MatchSpec   :: ets:match_spec(),
      LimitOrMeta :: pos_integer() | shards_meta:t(),
      Match       :: term(),
      Cont        :: continuation().
select(Tab, MatchSpec, Limit) when is_integer(Limit) ->
  select(Tab, MatchSpec, Limit, shards_meta:get(Tab));
select(Tab, MatchSpec, Meta) ->
  Map = {fun ets:select/2, [MatchSpec]},
  Reduce = fun erlang:'++'/2,
  mapred(Tab, Map, Reduce, Meta).

%% @doc
%% Equivalent to `ets:select/3'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:select/3.
%% @end
-spec select(Tab, MatchSpec, Limit, Meta) -> {[Match], Cont} | '$end_of_table' when
      Tab       :: tab(),
      MatchSpec :: ets:match_spec(),
      Limit     :: pos_integer(),
      Meta      :: shards_meta:t(),
      Match     :: term(),
      Cont      :: continuation().
select(Tab, MatchSpec, Limit, Meta) ->
  N = shards_meta:partitions(Meta),
  q(select, Tab, MatchSpec, Limit, q_fun(), Limit, N - 1, {[], nil}).

%% @doc
%% Equivalent to `ets:select/1'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:select/1.
%% @end
-spec select(Cont) -> {[Match], Cont} | '$end_of_table' when
      Match :: term(),
      Cont  :: continuation().
select({_, _, Limit, _, _} = Continuation) ->
  q(select, Continuation, q_fun(), Limit, []).

%% @equiv select_count(Tab, MatchSpec, shards_meta:get(Tab))
select_count(Tab, MatchSpec) ->
  select_count(Tab, MatchSpec, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:select_count/2'.
%%
%% @see ets:select_count/2.
%% @end
-spec select_count(Tab, MatchSpec, Meta) -> NumMatched when
      Tab        :: tab(),
      MatchSpec  :: ets:match_spec(),
      Meta       :: shards_meta:t(),
      NumMatched :: non_neg_integer().
select_count(Tab, MatchSpec, Meta) ->
  Map = {fun ets:select_count/2, [MatchSpec]},
  Reduce = {fun erlang:'+'/2, 0},
  mapred(Tab, Map, Reduce, Meta).

%% @equiv select_delete(Tab, MatchSpec, shards_meta:get(Tab))
select_delete(Tab, MatchSpec) ->
  select_delete(Tab, MatchSpec, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:select_delete/2'.
%%
%% @see ets:select_delete/2.
%% @end
-spec select_delete(Tab, MatchSpec, Meta) -> NumDeleted when
      Tab        :: tab(),
      MatchSpec  :: ets:match_spec(),
      Meta       :: shards_meta:t(),
      NumDeleted :: non_neg_integer().
select_delete(Tab, MatchSpec, Meta) ->
  Map = {fun ets:select_delete/2, [MatchSpec]},
  Reduce = {fun erlang:'+'/2, 0},
  mapred(Tab, Map, Reduce, Meta).

%% @equiv select_replace(Tab, MatchSpec, shards_meta:get(Tab))
select_replace(Tab, MatchSpec) ->
  select_replace(Tab, MatchSpec, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:select_replace/2'.
%%
%% @see ets:select_replace/2.
%% @end
-spec select_replace(Tab, MatchSpec, Meta) -> NumReplaced when
      Tab         :: tab(),
      MatchSpec   :: ets:match_spec(),
      Meta        :: shards_meta:t(),
      NumReplaced :: non_neg_integer().
select_replace(Tab, MatchSpec, Meta) ->
  Map = {fun ets:select_replace/2, [MatchSpec]},
  Reduce = {fun erlang:'+'/2, 0},
  mapred(Tab, Map, Reduce, Meta).

%% @equiv select_reverse(Tab, MatchSpec, shards_meta:get(Tab))
select_reverse(Tab, MatchSpec) ->
  select_reverse(Tab, MatchSpec, shards_meta:get(Tab)).

%% @doc
%% If 3rd argument is `pos_integer()' this function behaves
%% like `ets:select_reverse/3', otherwise, the 3rd argument is
%% assumed as `shards_meta:t()` and it behaves like
%% `ets:select_reverse/2'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:select_reverse/3.
%% @end
-spec select_reverse(Tab, MatchSpec, LimitOrMeta) -> {[Match], Cont} | '$end_of_table' | [Match] when
      Tab         :: tab(),
      MatchSpec   :: ets:match_spec(),
      LimitOrMeta :: pos_integer() | shards_meta:t(),
      Match       :: term(),
      Cont        :: continuation().
select_reverse(Tab, MatchSpec, Limit) when is_integer(Limit) ->
  select_reverse(Tab, MatchSpec, Limit, shards_meta:get(Tab));
select_reverse(Tab, MatchSpec, Meta) ->
  Map = {fun ets:select_reverse/2, [MatchSpec]},
  Reduce = fun erlang:'++'/2,
  mapred(Tab, Map, Reduce, Meta).

%% @doc
%% Equivalent to `ets:select_reverse/3'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:select_reverse/3.
%% @end
-spec select_reverse(Tab, MatchSpec, Limit, Meta) -> {[Match], Cont} | '$end_of_table' when
      Tab       :: tab(),
      MatchSpec :: ets:match_spec(),
      Limit     :: pos_integer(),
      Meta      :: shards_meta:t(),
      Match     :: term(),
      Cont      :: continuation().
select_reverse(Tab, MatchSpec, Limit, Meta) ->
  N = shards_meta:partitions(Meta),
  q(select_reverse, Tab, MatchSpec, Limit, q_fun(), Limit, N - 1, {[], nil}).

%% @doc
%% Equivalent to `ets:select_reverse/1'.
%%
%% The order in which results are returned might be not the same
%% as the original ETS function.
%%
%% @see ets:select_reverse/1.
%% @end
-spec select_reverse(Cont) -> {[Match], Cont} | '$end_of_table' when
      Cont  :: continuation(),
      Match :: term().
select_reverse({_, _, Limit, _, _} = Continuation) ->
  q(select_reverse, Continuation, q_fun(), Limit, []).

%% @equiv setopts(Tab, Opts, shards_meta:get(Tab))
setopts(Tab, Opts) ->
  setopts(Tab, Opts, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:setopts/2'.
%%
%% Returns `true' if the function was applied successfully on each partition,
%% otherwise, `false' is returned.
%%
%% @see ets:setopts/2.
%% @end
-spec setopts(Tab, Opts, Meta) -> true when
      Tab      :: tab(),
      Opts     :: Opt | [Opt],
      Opt      :: {heir, pid(), HeirData} | {heir, none},
      Meta     :: shards_meta:t(),
      HeirData :: term().
setopts(Tab, Opts, Meta) ->
  Map = {fun shards_partition:apply_ets_fun/3, [setopts, [Opts]]},
  Reduce = {fun erlang:'and'/2, true},
  mapred(Tab, Map, Reduce, {pid, Meta}).

%% @equiv tab2file(Tab, Filename, [])
tab2file(Tab, Filename) ->
  tab2file(Tab, Filename, []).

%% @doc
%% Equivalent to `ets:tab2file/3'.
%%
%% This function generates one file per partition using `ets:tab2file/3',
%% and also generates a master file with the given `Filename' that holds
%% the information of the created partition files so that they can be
%% recovered by calling `ets:file2tab/1,2'.
%%
%% @see ets:tab2file/3.
%% @end
-spec tab2file(Tab, Filename, Options) -> ok | {error, Reason} when
      Tab      :: tab(),
      Filename :: filename(),
      Options  :: [Option],
      Option   :: {extended_info, [md5sum | object_count]} | {sync, boolean()},
      Reason   :: term().
tab2file(Tab, Filename, Options) when ?is_filename(Filename) ->
  StrFilename = shards_lib:to_string(Filename),
  Metadata = shards_meta:get(Tab),

  PartitionFilenamePairs =
    shards_enum:reduce_while(fun({Idx, Partition}, Acc) ->
      PartitionFilename = StrFilename ++ "." ++ integer_to_list(Idx),

      case ets:tab2file(Partition, PartitionFilename, Options) of
        ok ->
          {cont, Acc#{Idx => PartitionFilename}};

        {error, _} = Error ->
          {halt, Error}
      end
    end, #{}, shards_meta:get_partition_tids(Tab)),

  case PartitionFilenamePairs of
    {error, _} = Error ->
      Error;

    _ ->
      TabInfo = maps:from_list(shards:info(Tab)),

      Header = #{
        name          => maps:get(name, TabInfo),
        type          => maps:get(type, TabInfo),
        protection    => maps:get(protection, TabInfo),
        keypos        => maps:get(keypos, TabInfo),
        size          => maps:get(size, TabInfo),
        named_table   => maps:get(named_table, TabInfo),
        extended_info => maps:get(extended_info, TabInfo, []),
        metadata      => Metadata,
        partitions    => PartitionFilenamePairs
      },

      shards_lib:write_tabfile(StrFilename, Header)
  end.

%% @equiv tab2list(Tab, shards_meta:get(Tab))
tab2list(Tab) ->
  tab2list(Tab, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:tab2list/1'.
%%
%% @see ets:tab2list/1.
%% @end
-spec tab2list(Tab, Meta) -> [Object] when
      Tab    :: tab(),
      Meta   :: shards_meta:t(),
      Object :: tuple().
tab2list(Tab, Meta) ->
  mapred(Tab, fun ets:tab2list/1, fun erlang:'++'/2, Meta).

%% @doc
%% Equivalent to `ets:tabfile_info/1'.
%%
%% Adds extra information about the partitions.
%%
%% @see ets:tabfile_info/1.
%% @end
-spec tabfile_info(Filename) -> {ok, TableInfo} | {error, Reason} when
      Filename  :: filename(),
      TableInfo :: [tabinfo_item()],
      Reason    :: term().
tabfile_info(Filename) when ?is_filename(Filename) ->
  try
    StrFilename = shards_lib:to_string(Filename),
    Header = shards_lib:read_tabfile(StrFilename),

    TabName = maps:get(name, Header),
    Metadata = maps:get(metadata, Header),
    NamedTable = lists:member(named_table, shards_meta:ets_opts(Metadata)),
    Partitions = maps:get(partitions, Header),

    ShardsTabInfo =
      maps:fold(fun(_, PartitionFN, Acc) ->
        case ets:tabfile_info(PartitionFN) of
          {ok, TabInfo} -> [TabInfo | Acc];
          Error         -> throw(Error)
        end
      end, [], Partitions),

    PartitionedTabInfo =
      lists:keystore(
        named_table,
        1,
        [
          {partitions, map_size(Partitions)}
          | parts_info(TabName, ShardsTabInfo)
        ],
        {named_table, NamedTable}
      ),

    {ok, PartitionedTabInfo}
  catch
    throw:Error -> Error
  end.

%% @equiv table(Tab, [])
table(Tab) ->
  table(Tab, []).

%% @equiv table(Tab, Options, shards_meta:get(Tab))
table(Tab, Options) ->
  table(Tab, Options, shards_meta:get(Tab)).

%% @doc
%% Similar to `ets:table/2', but it returns a list of `qlc:query_handle()';
%% one per partition.
%%
%% @see ets:table/2.
%% @end
-spec table(Tab, Options, Meta) -> QueryHandle when
      Tab            :: tab(),
      QueryHandle    :: qlc:query_handle(),
      Options        :: [Option] | Option,
      Meta           :: shards_meta:t(),
      Option         :: {n_objects, NObjects} | {traverse, TraverseMethod},
      NObjects       :: 'default' | pos_integer(),
      MatchSpec      :: ets:match_spec(),
      TraverseMethod :: first_next | last_prev | select | {select, MatchSpec}.
table(Tab, Options, Meta) ->
  mapred(Tab, {fun ets:table/2, [Options]}, Meta).

%% @equiv ets:test_ms(Tuple, MatchSpec)
test_ms(Tuple, MatchSpec) ->
  ets:test_ms(Tuple, MatchSpec).

%% @equiv take(Tab, Key, shards_meta:get(Tab))
take(Tab, Key) ->
  take(Tab, Key, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:take/2'.
%%
%% @see ets:take/2.
%% @end
-spec take(Tab, Key, Meta) -> [Object] when
      Tab    :: tab(),
      Key    :: term(),
      Meta   :: shards_meta:t(),
      Object :: tuple().
take(Tab, Key, Meta) ->
  PartTid = shards_partition:tid(Tab, Key, Meta),
  ets:take(PartTid, Key).

%% @equiv update_counter(Tab, Key, UpdateOp, shards_meta:get(Tab))
update_counter(Tab, Key, UpdateOp) ->
  update_counter(Tab, Key, UpdateOp, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:update_counter/4'.
%%
%% If the 4th argument is `shards_meta:t()', it behaves like
%% `ets:update_counter/3'.
%%
%% @see ets:update_counter/4.
%% @end
-spec update_counter(Tab, Key, UpdateOp, DefaultOrMeta) -> Result | [Result] when
      Tab           :: tab(),
      Key           :: term(),
      UpdateOp      :: term(),
      DefaultOrMeta :: tuple() | shards_meta:t(),
      Result        :: integer().
update_counter(Tab, Key, UpdateOp, DefaultOrMeta) ->
  case shards_meta:is_metadata(DefaultOrMeta) of
    true ->
      PartTid = shards_partition:tid(Tab, Key, DefaultOrMeta),
      ets:update_counter(PartTid, Key, UpdateOp);

    false ->
      update_counter(Tab, Key, UpdateOp, DefaultOrMeta, shards_meta:get(Tab))
  end.

%% @doc
%% Equivalent to `ets:update_counter/4'.
%%
%% @see ets:update_counter/4.
%% @end
-spec update_counter(Tab, Key, UpdateOp, Default, Meta) -> Result | [Result] when
      Tab      :: tab(),
      Key      :: term(),
      UpdateOp :: term(),
      Default  :: tuple(),
      Meta     :: shards_meta:t(),
      Result   :: integer().
update_counter(Tab, Key, UpdateOp, Default, Meta) ->
  PartTid = shards_partition:tid(Tab, Key, Meta),
  ets:update_counter(PartTid, Key, UpdateOp, Default).

%% @equiv update_element(Tab, Key, ElementSpec, shards_meta:get(Tab))
update_element(Tab, Key, ElementSpec) ->
  update_element(Tab, Key, ElementSpec, shards_meta:get(Tab)).

%% @doc
%% Equivalent to `ets:update_element/3'.
%%
%% @see ets:update_element/3.
%% @end
-spec update_element(Tab, Key, ElementSpec, Meta) -> boolean() when
      Tab         :: tab(),
      Key         :: term(),
      ElementSpec :: {Pos, Value} | [{Pos, Value}],
      Meta        :: shards_meta:t().
update_element(Tab, Key, ElementSpec, Meta) ->
  PartTid = shards_partition:tid(Tab, Key, Meta),
  ets:update_element(PartTid, Key, ElementSpec).

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% @private
with_trap_exit(Fun) ->
  Flag = process_flag(trap_exit, true),

  try
    Fun()
  after
    process_flag(trap_exit, Flag)
  end.

%% @private
with_meta(Tab, Fun) ->
  try shards_meta:get(Tab) of
    Meta -> Fun(Meta)
  catch
    error:{unknown_table, Tab} -> undefined
  end.

%% @private
get_part_tid(Tab, Object, Meta) ->
  Key = shards_lib:object_key(Object, Meta),
  shards_partition:tid(Tab, Key, Meta).

%% @private
group_keys_by_partition(Tab, Objects, Meta) ->
  lists:foldr(fun(Object, Acc) ->
    PartTid = get_part_tid(Tab, Object, Meta),
    Acc#{PartTid => [Object | maps:get(PartTid, Acc, [])]}
  end, #{}, Objects).

%% @private
parts_info(Tab, InfoLists) ->
  parts_info(Tab, InfoLists, []).

%% @private
parts_info(Tab, [FirstInfo | RestInfoLists], ExtraAttrs) ->
  FirstInfo1 =
    shards_lib:keyupdate(fun(K, _V) ->
      ets:info(Tab, K)
    end, [id, name, named_table, owner], FirstInfo),

  Keys = [size | ExtraAttrs],

  lists:foldl(fun(InfoList, InfoListAcc) ->
    shards_lib:keyupdate(fun(K, V) ->
      {K, V1} = lists:keyfind(K, 1, InfoList),
      V + V1
    end, Keys, InfoListAcc)
  end, FirstInfo1, RestInfoLists).

%% @private
mapred(Tab, Map, Meta) ->
  mapred(Tab, Map, nil, Meta).

%% @private
mapred(Tab, Map, nil, Meta) ->
  mapred(Tab, Map, fun(E, Acc) -> [E | Acc] end, Meta);
mapred(Tab, Map, Reduce, {PartitionFun, Meta}) ->
  do_mapred(Tab, Map, Reduce, PartitionFun, Meta);
mapred(Tab, Map, Reduce, Meta) ->
  do_mapred(Tab, Map, Reduce, tid, Meta).

%% @private
do_mapred(Tab, Map, Reduce, PartFun, Meta) ->
  case {shards_meta:partitions(Meta), shards_meta:parallel(Meta)} of
    {Partitions, true} when Partitions > 1 ->
      ParallelTimeout = shards_meta:parallel_timeout(Meta),
      p_mapred(Tab, Map, Reduce, PartFun, Partitions, ParallelTimeout);

    {Partitions, false} ->
      s_mapred(Tab, Map, Reduce, PartFun, Partitions)
  end.

%% @private
s_mapred(Tab, {MapFun, Args}, {ReduceFun, AccIn}, PartFun, Partitions) ->
  shards_enum:reduce(fun(Part, Acc) ->
    PartitionId = shards_partition:PartFun(Tab, Part),
    MapRes = apply(MapFun, [PartitionId | Args]),
    ReduceFun(MapRes, Acc)
  end, AccIn, Partitions);
s_mapred(Tab, MapFun, ReduceFun, PartFun, Partitions) ->
  {Map, Reduce} = mapred_funs(MapFun, ReduceFun),
  s_mapred(Tab, Map, Reduce, PartFun, Partitions).

%% @private
p_mapred(Tab, {MapFun, Args}, {ReduceFun, AccIn}, PartFun, Partitions, ParallelTimeout) ->
  MapResults =
    shards_enum:pmap(fun(Idx) ->
      PartitionId = shards_partition:PartFun(Tab, Idx),
      apply(MapFun, [PartitionId | Args])
    end, ParallelTimeout, lists:seq(0, Partitions - 1)),

  lists:foldl(ReduceFun, AccIn, MapResults);
p_mapred(Tab, MapFun, ReduceFun, PartFun, Partitions, ParallelTimeout) ->
  {Map, Reduce} = mapred_funs(MapFun, ReduceFun),
  p_mapred(Tab, Map, Reduce, PartFun, Partitions, ParallelTimeout).

%% @private
mapred_funs(MapFun, ReduceFun) ->
  Map =
    case is_function(MapFun) of
      true -> {MapFun, []};
      _    -> MapFun
    end,

  {Map, {ReduceFun, []}}.

%% @private
q(_, Tab, MatchSpec, Limit, _, 0, Shard, {Acc, Continuation}) ->
  {Acc, {Tab, MatchSpec, Limit, Shard, Continuation}};
q(_, _, _, _, _, _, Shard, {[], _}) when Shard < 0 ->
  '$end_of_table';
q(_, Tab, MatchSpec, Limit, _, _, Shard, {Acc, _}) when Shard < 0 ->
  {Acc, {Tab, MatchSpec, Limit, Shard, '$end_of_table'}};
q(F, Tab, MatchSpec, Limit, QFun, I, Shard, {Acc, '$end_of_table'}) ->
  q(F, Tab, MatchSpec, Limit, QFun, I, Shard - 1, {Acc, nil});
q(F, Tab, MatchSpec, Limit, QFun, I, Shard, {Acc, _}) ->
  case ets:F(shards_partition:tid(Tab, Shard), MatchSpec, I) of
    {L, Cont} ->
      NewAcc = {QFun(L, Acc), Cont},
      q(F, Tab, MatchSpec, Limit, QFun, I - length(L), Shard, NewAcc);

    '$end_of_table' ->
      q(F, Tab, MatchSpec, Limit, QFun, I, Shard, {Acc, '$end_of_table'})
  end.

%% @private
q(_, {Tab, MatchSpec, Limit, Shard, Continuation}, _, 0, Acc) ->
  {Acc, {Tab, MatchSpec, Limit, Shard, Continuation}};
q(F, {Tab, MatchSpec, Limit, Shard, '$end_of_table'}, QFun, I, Acc) ->
  q(F, Tab, MatchSpec, Limit, QFun, I, Shard - 1, {Acc, nil});
q(F, {Tab, MatchSpec, Limit, Shard, Continuation}, QFun, I, Acc) ->
  case ets:F(Continuation) of
    {L, Cont} ->
      NewAcc = QFun(L, Acc),
      q(F, {Tab, MatchSpec, Limit, Shard, Cont}, QFun, I - length(L), NewAcc);

    '$end_of_table' ->
      q(F, {Tab, MatchSpec, Limit, Shard, '$end_of_table'}, QFun, I, Acc)
  end.

%% @private
q_fun() ->
  fun(L1, L0) -> L1 ++ L0 end.