ecql

%%==============================================================================
%% Copyright (c) Exosite LLC
%%
%% ecql_mnesia.erl - Mnesia drop-in replacement
%%==============================================================================
-module(ecql_mnesia).

-include("ecql.hrl").

%% Mnesia APIs
-export([
   add_table_copy/3
  ,all_keys/1
  ,change_config/2
  ,change_table_copy_type/3
  ,clear_table/1
  ,create_schema/1
  ,create_table/2
  ,delete/1, delete/3
  ,delete_object/1
  ,delete_table/1
  ,dirty_all_keys/1
  ,dirty_delete/1, dirty_delete/3
  ,dirty_delete_object/1
  ,dirty_index_match_object/3, dirty_index_match_object/2
  ,dirty_index_read/3
  ,dirty_match_object/1, dirty_match_object/2, dirty_match_object/3
  ,dirty_read/2, dirty_read/3
  ,dirty_select/1, dirty_select/2, dirty_select/4
  ,dirty_update_counter/3
  ,dirty_write/1
  ,first/1
  ,foldr/3
  ,info/0
  ,index_match_object/3, index_match_object/2
  ,index_read/3
  ,load_textfile/1
  ,match_object/1, match_object/2, match_object/3
  ,next/2
  ,read/2, read/3
  ,select/1
  ,select/2
  ,select/4
  ,start/0
  ,stop/0
  ,system_info/1
  ,table_info/2
  ,transaction/1
  ,wait_for_tables/2
  ,write/1
]).

%% Extensions Updating (from ets)
-export([
   update_element/3
]).


%% Extensions Counting
-export([
   count/2
  ,dirty_count/2
  ,dirty_index_count/3
  ,dirty_index_match_count/3, dirty_index_match_count/2
  ,dirty_match_count/1, dirty_match_count/2
  ,index_count/3
  ,index_match_count/3, index_match_count/2
  ,match_count/1, match_count/2
]).

%% Extensions Finding (only prim key)
-export([
   dirty_index_key/3
  ,dirty_index_match_key/3, dirty_index_match_key/2
  ,dirty_key/2
  ,dirty_match_key/1, dirty_match_key/2
  ,index_key/3
  ,index_match_key/3, index_match_key/2
  ,key/2
  ,match_key/1, match_key/2
]).

%% Extensions Traversing (inspired by gb_trees)
-export([
   iterator/1
  ,next/1
]).

%% Other Extensions
-export([
   select_records/3
]).

%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
%% Mnesia API
%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

%%------------------------------------------------------------------------------
start()  ->
  {ok, _Apps} = application:ensure_all_started(ecql),
  ok
.

%%------------------------------------------------------------------------------
stop()  ->
  application:stop(ecql)
.

%%------------------------------------------------------------------------------
transaction(Fun)  ->
  try Fun() of
    Value -> {atomic ,Value}
  catch
    Throw -> {aborted , Throw}
  end
.

%%------------------------------------------------------------------------------
% Returns list of records
dirty_index_read(RecordName, KeyValue, KeyIndex) ->
  index_read(RecordName, KeyValue, KeyIndex)
.
index_read(RecordName, KeyValue, KeyIndex) ->
  case table_type(RecordName) of
    set ->
      ecql_cache:get({RecordName, KeyValue, KeyIndex}, fun() ->
        do_index_read(RecordName, KeyValue, KeyIndex)
      end)
    ;
    bag ->
      do_index_read(RecordName, KeyValue, KeyIndex)
    %~
  end
.
do_index_read(RecordName, KeyValue, KeyIndex) when is_atom(RecordName) ->
  % KeyIndex is actually the list position (starting at 1) behind the RecordName
  % so we need to subtract 2 (1 for starting at one + 1 for skipping the record)
   Index = KeyIndex - 2
  ,select_records(RecordName, [
     "SELECT * FROM ", map_recordname(RecordName)
    ," WHERE ", map_fieldindex(Index), " = ?"
   ], [ecql:term_to_bin(KeyValue)])
.

%%------------------------------------------------------------------------------
dirty_read(RecordName, KeyValue) ->
  read(RecordName, KeyValue)
.
read(RecordName, KeyValue) ->
  index_read(RecordName, KeyValue, 2)
.

%%------------------------------------------------------------------------------
dirty_read(RecordName, KeyValue, _Lock) ->
  read(RecordName, KeyValue)
.
read(RecordName, KeyValue, _Lock) ->
  read(RecordName, KeyValue)
.

%%------------------------------------------------------------------------------
dirty_write(Record)  ->
  write(Record)
.
write(Record) when is_tuple(Record) ->
   [RecordName, Key | RecordValues] = tuple_to_list(Record)
  ,List = lists:zip(lists:seq(3, tuple_size(Record)), RecordValues)
  ,update_element(RecordName, Key, List)
.

%%------------------------------------------------------------------------------
dirty_index_match_object(_RecordName, RecordPattern, _KeyIndex) ->
  match_object(RecordPattern)
.
index_match_object(_RecordName, RecordPattern, _KeyIndex) ->
  match_object(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_index_match_object(RecordPattern, _KeyIndex) ->
  match_object(RecordPattern)
.
index_match_object(RecordPattern, _KeyIndex) ->
  match_object(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_match_object(_RecordName, RecordPattern, _LockKind) ->
  match_object(_RecordName, RecordPattern, _LockKind)
.
match_object(_RecordName, RecordPattern, _LockKind) ->
  match_object(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_match_object(_RecordName, RecordPattern) ->
  match_object(RecordPattern)
.
match_object(_RecordName, RecordPattern) ->
  match_object(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_match_object(RecordPattern) ->
  match_object(RecordPattern)
.
match_object(RecordPattern) when is_tuple(RecordPattern) ->
  match_object(tuple_to_list(RecordPattern))
;
match_object([RecordName | RecordValues]) when is_atom(RecordName) ->
   IndexedValues = lists:zip(lists:seq(1, length(RecordValues)), RecordValues)
  ,[{KeyIndex, KeyValue} | Rest] = [
    {Index+1, Value} || {Index, Value} <- IndexedValues, not is_ref(Value)
  ]
 ,do_match_object(index_read(RecordName, KeyValue, KeyIndex), Rest)
.
do_match_object(Results, []) ->
  Results
;
do_match_object(Results, Other) ->
  [Obj || Obj <- Results, do_match_object_test(Obj, Other)]
.
do_match_object_test(_Obj, []) ->
  true
;
do_match_object_test(Obj, [{Index, Value} | Other]) ->
  element(Index, Obj) == Value andalso do_match_object_test(Obj, Other)
.

%%------------------------------------------------------------------------------
dirty_delete(RecordName, KeyValue, Lock) ->
  delete(RecordName ,KeyValue ,Lock)
.
delete(RecordName, KeyValue, _Lock) ->
  delete({RecordName, KeyValue})
.

%%------------------------------------------------------------------------------
dirty_delete({RecordName, KeyValue}) ->
  delete({RecordName, KeyValue})
.
delete({RecordName, KeyValue}) when is_atom(RecordName) ->
  delete_object(read(RecordName, KeyValue))
.

%%------------------------------------------------------------------------------
dirty_delete_object(Records) ->
  delete_object(Records)
.
delete_object(Records) when is_list(Records) ->
   lists:foreach(fun delete_object/1, Records)
  ,ok
;
delete_object(Record) when is_tuple(Record) ->
  % Correct Impl?
  % Should delete all object according to the pattern or only based on prim key?
   [RecordName | RecordValues] = tuple_to_list(Record)
  ,do_delete_object(table_type(RecordName), RecordName, RecordValues)
.
do_delete_object(set, RecordName, RecordValues) ->
  Ret = ecql:execute(
     ["DELETE FROM ", map_recordname(RecordName), " WHERE a = ?"]
    ,[ecql:term_to_bin(hd(RecordValues))]
  )
  ,dirty_cache([RecordName | RecordValues])
  ,Ret
;
do_delete_object(bag, RecordName, RecordValues) ->
   {Keys, Values} = and_pairs(RecordValues)
  ,ecql:execute([
     "DELETE FROM ", map_recordname(RecordName)
    ," WHERE ", Keys
  ], Values)
.

%%------------------------------------------------------------------------------
% REALLY DONT!
dirty_all_keys(RecordName) ->
  all_keys(RecordName)
.
all_keys(RecordName) when is_atom(RecordName) ->
  ecql:eval_all(ecql:select_column([
    "SELECT a FROM ", map_recordname(RecordName)
  ]))
.

%%------------------------------------------------------------------------------
first(RecordName) when is_atom(RecordName) ->
  key_or_end(firstn(RecordName, 1))
.

%%------------------------------------------------------------------------------
next(RecordName, KeyValue) when is_atom(RecordName) ->
  key_or_end(nextn(RecordName, KeyValue, 1))
.

%%------------------------------------------------------------------------------
wait_for_tables([], _Timeout) ->
  ok
;
wait_for_tables(Tables, infinity) ->
   NewTables = lists:subtract(Tables, system_info(tables))
  ,do_wait_for_tables(NewTables, infinity)
;
wait_for_tables(Tables, Number) ->
   NewTables = lists:subtract(Tables, system_info(tables))
  ,do_wait_for_tables(NewTables, Number - 100)
.

do_wait_for_tables([], _Number) ->
   ok
;
do_wait_for_tables(Tables, Number) when Number < 0 ->
   {timeout, Tables}
;
do_wait_for_tables(Tables, Number) ->
   timer:sleep(100)
  ,wait_for_tables(Tables, Number)
.

%%------------------------------------------------------------------------------
system_info(tables) ->
  [TableName || {TableName, _} <- get_tables()]
;
system_info(use_dir) ->
  false
.

%%------------------------------------------------------------------------------
% NOP
change_table_copy_type(_Schema,_This,_DiscCopies) ->
  {atomic ,ok}
.

%%------------------------------------------------------------------------------
% NOP
create_schema(_DiscNodes) ->
  ok
.

%%------------------------------------------------------------------------------
% NOP
info() ->
  ok
.

%%------------------------------------------------------------------------------
clear_table(RecordName) when is_atom(RecordName) ->
  case ecql:execute(["TRUNCATE ", map_recordname(RecordName)]) of
     ok ->
       ecql_cache:match_clear({{RecordName, '_', '_'}, '_'})
      ,{atomic ,ok}
    ;
    Error ->
      {error, Error}
    %~
  end
.

%%------------------------------------------------------------------------------
delete_table(RecordName) when is_atom(RecordName) ->
  case ecql:execute(["DROP TABLE ", map_recordname(RecordName)]) of
     ok -> {atomic ,ok}
    ;Error -> {error, Error}
  end
.

%%------------------------------------------------------------------------------
% NOP
table_info(_Table ,where_to_commit) ->
  [{nop ,nop}]
.

%%------------------------------------------------------------------------------
create_table(Name ,Def)  ->
  case is_tablename(Name) of
    false -> {aborted, "Tablename is invalid. Can't contain underscore"};
    true ->
      Type = case lists:keyfind(type, 1, Def) of
        false -> set;
        {type, T} -> T
      end
      ,{attributes, Fields} = lists:keyfind(attributes, 1, Def)
      ,ok = do_create_table(Type, Name, Fields)
      ,case lists:keyfind(index, 1, Def) of
        {index, Indexes} ->
           lists:foreach(
             fun(IndexName) ->
               Index = ecql:indexof(IndexName, Fields)
              ,ok = ecql:create_index(
                 map_recordname(Name) ++ "_" ++ map_recordname(IndexName)
                ,map_recordname(Name)
                ,map_fieldindex(Index)
              )
             end
            ,Indexes
           )
        ;
        false ->
           ok
        %~
       end
      ,ok = ecql_cache:dirty({ecql_mnesia, tables})
      ,{atomic ,ok}
    %~
  end
.
do_create_table(set, Name, Fields) ->
  ecql:create_table(
     map_recordname(Name)
    ,[
       "a blob PRIMARY KEY, "
      ,implode(" blob, ", [map_fieldindex(I) || I <- lists:seq(1, length(Fields)-1)])
      ," blob"
     ]
    ,"ecql_mnesia_set"
  )
;
do_create_table(bag, Name, Fields) ->
  ecql:create_table(
     map_recordname(Name)
    ,[
       "a blob, "
      ,implode(" blob, ", [map_fieldindex(I) || I <- lists:seq(1, length(Fields)-1)])
      ," blob, "
      ,"PRIMARY KEY(a, "
      ,implode(", ", [map_fieldindex(I) || I <- lists:seq(1, length(Fields)-1)])
      ,")"
     ]
    ,"ecql_mnesia_bag"
  )
.

%%------------------------------------------------------------------------------
add_table_copy(_Table ,_This ,_Type) ->
  {atomic ,ok}
.


%%------------------------------------------------------------------------------
dirty_update_counter(RecordName, KeyValue, Increment) when is_atom(RecordName) ->
   case read(RecordName, KeyValue) of
    [{RecordName, KeyValue, Value}] ->
      write({RecordName, KeyValue, Value+Increment})
    ;
    [] ->
      write({RecordName, KeyValue, Increment})
    %~
   end
.

%%------------------------------------------------------------------------------
dirty_select(Tuple) ->
  select(Tuple)
.
select({continuation, [], '$end_of_table', _NObjects, _Filter}) ->
  '$end_of_table'
;
select({continuation, ResultList, Cont, NObjects, Filter}) ->
  case ((length(ResultList) >= NObjects) orelse (Cont == '$end_of_table')) of
    true ->
       {Frame ,Rest} = split(NObjects ,ResultList)
      ,Records = apply_filter(Frame, Filter)
      ,{Records, {continuation, Rest, Cont, NObjects, Filter}}
    ;
    false ->
       {{_, Ret}, Cont2} = ecql:select_nextpage(Cont)
      ,select({continuation, ResultList ++ Ret, Cont2, NObjects, Filter})
    %~
  end
;
select({continuation ,[] ,_NObjects}) ->
  '$end_of_table'
;
select({continuation ,ResultList ,NObjects}) ->
  {Frame ,Rest} = split(NObjects ,ResultList)
 ,{Frame ,{continuation ,Rest ,NObjects}}
;
select(_Cont) ->
  '$end_of_table'
.
create_filter(RecordValues, ResultTuple) ->
   IndexedValues = lists:zip(lists:seq(1, length(RecordValues)), RecordValues)
  ,Symbols = dict:from_list([
    {Value, Index} || {Index, Value} <- IndexedValues, is_ref(Value)
  ])
  ,Matches = [{I, ecql:term_to_bin(Value)} || {I, Value} <- IndexedValues, not is_ref(Value)]
  ,ResultList = tuple_to_list(ResultTuple)
  ,fun(Row) ->
    case lists:all(
       fun({Index, Value}) -> lists:nth(Index, Row) == Value end
      ,Matches
    ) of
      true ->
        {true, list_to_tuple(lists:map(
          fun(Specifier) ->
            case is_ref(Specifier) of
              true ->
                ecql:eval(lists:nth(dict:fetch(Specifier, Symbols), Row))
              ;
              false ->
                Specifier
              %~
            end
          end
          ,ResultList
        ))}
      ;
      false ->
        false
      %~
    end
  end
.
apply_filter(Ret, Filter) ->
  lists:filtermap(Filter, Ret)
.


%%------------------------------------------------------------------------------
dirty_select(RecordName ,MatchSpec) ->
  select(RecordName ,MatchSpec)
.
select(RecordName ,MatchSpec) when is_atom(RecordName) ->
  foldr(
     fun(Record, Acc) ->
      case ets:test_ms(Record, MatchSpec) of
        {ok, false} ->
          Acc
        ;
        {ok, Result} ->
          [Result | Acc]
        ;
        _ ->
          Acc
        %~
      end
     end
    ,[]
    ,RecordName
  )
.

%%------------------------------------------------------------------------------
dirty_select(RecordName ,MatchSpec ,NObjects ,_Lock) ->
  select(RecordName ,MatchSpec ,NObjects ,_Lock)
.
select(RecordName ,[{RecordPattern, [], [{ResultTuple}]}] ,NObjects ,_Lock) when is_atom(RecordName) ->
   [RecordName | RecordValues] = tuple_to_list(RecordPattern)
  ,{Keys, Values} = and_pairs(RecordValues, table_type(RecordName))
  ,{{_, Ret}, Cont} = ecql:select_firstpage(
    [
       "SELECT * FROM ", map_recordname(RecordName)
      ," WHERE ", Keys
      ," ALLOW FILTERING;"
    ]
    ,Values
  )
  ,Filter = create_filter(RecordValues, ResultTuple)
  ,select({continuation, Ret, Cont, NObjects, Filter})
;
select(RecordName ,MatchSpec ,NObjects ,_Lock) when is_atom(RecordName) ->
  select({continuation, select(RecordName, MatchSpec), NObjects})
.

%%------------------------------------------------------------------------------
load_textfile(Filename) ->
   {ok ,[{tables, Tables} | Records]} = file:consult(Filename)
  ,lists:foreach(
     fun({Name ,Def}) ->
       {atomic ,ok} = create_table(Name ,Def)
     end
    ,Tables
   )
  ,lists:foreach(
     fun(Record) ->
       ok = dirty_write(Record)
     end
    ,Records
   )
.

%%------------------------------------------------------------------------------
change_config(_What, NewValue) ->
  {ok, NewValue}
.

%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
%% Extended API
%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

%%------------------------------------------------------------------------------
dirty_index_match_count(_RecordName, RecordPattern, _KeyIndex) ->
  match_count(RecordPattern)
.
index_match_count(_RecordName, RecordPattern, _KeyIndex) ->
  match_count(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_index_match_count(RecordPattern, _KeyIndex) ->
  match_count(RecordPattern)
.
index_match_count(RecordPattern, _KeyIndex) ->
  match_count(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_match_count(_RecordName, RecordPattern) ->
  match_count(RecordPattern)
.
match_count(_RecordName, RecordPattern) ->
  match_count(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_match_count(RecordPattern) ->
  match_count(RecordPattern)
.
match_count(RecordPattern) when is_tuple(RecordPattern) ->
  match_count(tuple_to_list(RecordPattern))
;
match_count([RecordName | RecordValues]) when is_atom(RecordName) ->
   {Keys, Values} = and_pairs(RecordValues)
  ,ecql:select_value([
     "SELECT COUNT(*) FROM ", map_recordname(RecordName)
    ," WHERE ", Keys
    ," ALLOW FILTERING;"
   ], Values)
.

%%------------------------------------------------------------------------------
% Returns list of records
dirty_index_count(RecordName, KeyValue, KeyIndex) ->
  index_count(RecordName, KeyValue, KeyIndex)
.
index_count(RecordName, KeyValue, KeyIndex) when is_atom(RecordName) ->
  % KeyIndex is actually the list position (starting at 1) behind the RecordName
  % so we need to subtract 2 (1 for starting at one + 1 for skipping the record)
   Index = KeyIndex - 2
  ,ecql:select_value([
     "SELECT COUNT(*) FROM ", map_recordname(RecordName)
    ," WHERE ", map_fieldindex(Index), " = ?"
   ], [ecql:term_to_bin(KeyValue)])
.

%%------------------------------------------------------------------------------
dirty_count(RecordName, KeyValue) ->
  count(RecordName, KeyValue)
.
count(RecordName, KeyValue) ->
  index_count(RecordName, KeyValue, 2)
.

%%------------------------------------------------------------------------------
dirty_index_match_key(_RecordName, RecordPattern, _KeyIndex) ->
  match_key(RecordPattern)
.
index_match_key(_RecordName, RecordPattern, _KeyIndex) ->
  match_key(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_index_match_key(RecordPattern, _KeyIndex) ->
  match_key(RecordPattern)
.
index_match_key(RecordPattern, _KeyIndex) ->
  match_key(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_match_key(_RecordName, RecordPattern) ->
  match_key(RecordPattern)
.
match_key(_RecordName, RecordPattern) ->
  match_key(RecordPattern)
.

%%------------------------------------------------------------------------------
dirty_match_key(RecordPattern) ->
  match_key(RecordPattern)
.
match_key(RecordPattern) when is_tuple(RecordPattern) ->
  match_key(tuple_to_list(RecordPattern))
;
match_key([RecordName | RecordValues]) when is_atom(RecordName) ->
   {Keys, Values} = and_pairs(RecordValues)
  ,ecql:eval_all(ecql:select_column([
     "SELECT a FROM ", map_recordname(RecordName)
    ," WHERE ", Keys
    ," ALLOW FILTERING;"
   ], 1, Values))
.

%%------------------------------------------------------------------------------
% Returns list of records
dirty_index_key(RecordName, KeyValue, KeyIndex) ->
  index_key(RecordName, KeyValue, KeyIndex)
.
index_key(RecordName, KeyValue, KeyIndex) when is_atom(RecordName) ->
  % KeyIndex is actually the list position (starting at 1) behind the RecordName
  % so we need to subtract 2 (1 for starting at one + 1 for skipping the record)
   Index = KeyIndex - 2
  ,ecql:eval_all(ecql:select_column([
     "SELECT a FROM ", map_recordname(RecordName)
    ," WHERE ", map_fieldindex(Index), " = ?"
   ], 1, [ecql:term_to_bin(KeyValue)]))
.

%%------------------------------------------------------------------------------
dirty_key(RecordName, KeyValue) ->
  key(RecordName, KeyValue)
.
key(RecordName, KeyValue) ->
  index_key(RecordName, KeyValue, 2)
.

%%------------------------------------------------------------------------------
foldr(Fun, Acc, RecordName) ->
  do_foldr(Fun, next(iterator(RecordName)), Acc)
.
do_foldr(_Fun, none, Acc) ->
  Acc
;
do_foldr(Fun, {Value, Iter}, Acc) ->
  do_foldr(Fun, next(Iter), Fun(Value, Acc))
.

%%------------------------------------------------------------------------------
select_records(RecordName, Cql, Args) when is_atom(RecordName) ->
   {_Keys, Rows} = ecql:select(Cql, Args)
  ,Context = {Cql, Args}
  ,[
     list_to_tuple([RecordName | ecql:eval_all(RecordValues)])
     ||
     RecordValues <- Rows
    ,is_valid_record(RecordValues, Context)
  ]
.

%%------------------------------------------------------------------------------
update_element(RecordName, Key, Tuple) when is_tuple(Tuple) ->
  update_element(RecordName, Key, [Tuple])
;
update_element(RecordName, Key, List) ->
  do_update_element(table_type(RecordName), RecordName, Key, List)
.
do_update_element(set, RecordName, Key, List) ->
   Changes = [{2, Key} | List]
  ,{RecordIndexes, RecordValues} = lists:unzip(Changes)
  ,OldRec = read(RecordName, Key)
  ,Ret = do_update_element(bag, RecordName, Key, List)
  ,case OldRec of
    [Record] ->
      dirty_cache(Record)
      ,NewRec = lists:foldl(
        fun({Index, Value}, R0) ->
          case catch setelement(Index, R0, Value) of
            {'EXIT', {badarg, Badarg}} ->
              case erlang:size(R0) + 1 of
                Index ->
                  erlang:append_element(R0, Value)
                ;
                _ ->
                  throw({badarg, Badarg})
                %~
              end
            ;
            Tuple ->
              Tuple
            %~
          end
        end
        ,Record
        ,Changes
      )
      ,ecql_cache:set({RecordName, Key}, [NewRec])
    ;
    [] ->
      ok
    %~
  end
  ,dirty_cache(RecordName, RecordIndexes, RecordValues)
  ,Ret
;
do_update_element(bag, RecordName, Key, List) ->
   {RecordIndexes, RecordValues} = lists:unzip([{2, Key} | List])
  ,FieldNames = [map_fieldindex(RecordIndex - 2) || RecordIndex <- RecordIndexes]
  ,ecql:execute(
     [
       "INSERT INTO ", map_recordname(RecordName) ," ("
      ,implode($,, FieldNames)
      ,") VALUES (?"
      ,string:copies(",?", length(RecordValues) - 1) ,");"
    ]
    ,[ecql:term_to_bin(Value) || Value <- RecordValues]
    ,?CL_LOCAL_QUORUM
  )
.

%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
%% Private API
%%-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

dirty_cache(Record) when is_tuple(Record) ->
  dirty_cache(tuple_to_list(Record))
;
dirty_cache([RecordName | RecordValues]) when is_atom(RecordName) ->
   dirty_cache(RecordName, lists:seq(2, length(RecordValues) + 1), RecordValues)
.

%%------------------------------------------------------------------------------
dirty_cache(_RecordName, [], []) ->
   ok
;
dirty_cache(RecordName, [N | RecordIndexes], [Value | RecordValues]) ->
   ecql_cache:dirty({RecordName, Value, N})
  ,dirty_cache(RecordName, RecordIndexes, RecordValues)
.

%%------------------------------------------------------------------------------
iterator(RecordName) when is_atom(RecordName) ->
  do_next({iterator, RecordName, [], undefined})
.

%%------------------------------------------------------------------------------
next({iterator, _RecordName, [], '$end_of_table'}) ->
  none
;
next({iterator, _RecordName, [], _LastID} = Iter) ->
  next(do_next(Iter))
;
next({iterator, RecordName, [Head | Values], LastID}) ->
  {Head, {iterator, RecordName, Values, LastID}}
.
do_next({iterator, RecordName, [], LastID}) ->
   N = 100
  ,Values = case LastID of
     undefined -> firstn(RecordName, N)
    ;_ -> nextn(RecordName, LastID, N)
   end
  ,case length(Values) of
    N ->
       LastID2 = element(2, lists:last(Values))
      ,{iterator, RecordName, Values, LastID2}
    ;
    _ ->
       {iterator, RecordName, Values, '$end_of_table'}
   end
.

%%------------------------------------------------------------------------------
firstn(RecordName, N) when is_atom(RecordName) ->
  select_records(
     RecordName
    ,["SELECT * FROM ", map_recordname(RecordName), " LIMIT ?"]
    ,[N]
  )
.

%%------------------------------------------------------------------------------
nextn(RecordName, KeyValue, N) when is_atom(RecordName) ->
  select_records(
     RecordName
    ,[
       "SELECT * FROM ", map_recordname(RecordName)
      ," WHERE token(a) > token(?) LIMIT ?"
     ]
    ,[ecql:term_to_bin(KeyValue), N]
  )
.

%%------------------------------------------------------------------------------
key_or_end([]) ->
  '$end_of_table'
;
key_or_end([Other]) ->
  element(2, Other)
.

%%------------------------------------------------------------------------------
map_fieldindex(FieldIndex) when is_integer(FieldIndex) ->
  $a + FieldIndex
.

%%------------------------------------------------------------------------------
map_recordname(RecordName) when is_atom(RecordName) ->
  [case C of $. -> $_; _ -> C end || C <- atom_to_list(RecordName)]
.

%%------------------------------------------------------------------------------
table_type(RecordName) when is_atom(RecordName) ->
   {RecordName, Table} = lists:keyfind(RecordName, 1, get_tables())
  ,{type, Type} = lists:keyfind(type, 1, Table)
  ,Type
.

%%------------------------------------------------------------------------------
unmap_recordname(TableName) when is_binary(TableName) ->
  unmap_recordname(binary_to_list(TableName))
;
unmap_recordname(TableName) when is_list(TableName) ->
  list_to_atom([case C of $_ -> $.; _ -> C end || C <- TableName])
.

%%------------------------------------------------------------------------------
is_ref(Reference) when is_atom(Reference) ->
  case Reference of
    '$1' -> true;
    '$2' -> true;
    '$3' -> true;
    '$4' -> true;
    '$5' -> true;
    '$6' -> true;
    '$7' -> true;
    '$8' -> true;
    '$9' -> true;
    '_' -> true;
    _ -> false
  end
;
is_ref(_Reference) ->
  false
.

%%------------------------------------------------------------------------------
fields(RecordValues) ->
  [map_fieldindex(I) || I <- lists:seq(0, length(RecordValues)-1)]
.

%%------------------------------------------------------------------------------
and_pairs(List) ->
   KeyValueList = [
    {[Key, " = ?"], ecql:term_to_bin(Value)}
    || {Key, Value} <- lists:zip(fields(List), List), not is_ref(Value)
   ]
  ,{Keys, Values} = lists:unzip(KeyValueList)
  ,{implode(" AND ", Keys), Values}
.

%%------------------------------------------------------------------------------
and_pairs(List, set) ->
  and_pairs(List)
;
and_pairs(List, bag) ->
  % bags are all primary key
   Candiate = lists:takewhile(fun(Value) -> not is_ref(Value) end, List)
  ,case Candiate of
    [] ->
      and_pairs(List)
    ;
    _Other ->
      and_pairs(Candiate)
    %~
  end
.

%%------------------------------------------------------------------------------
implode(Sep, List) ->
   tl(implode_concat(Sep, List))
.

%%------------------------------------------------------------------------------
implode_concat(_Sep, []) ->
  []
;
implode_concat(Sep, [Head | Tail]) ->
  [Sep, Head] ++ implode_concat(Sep, Tail)
.

%%------------------------------------------------------------------------------
is_valid_record([], Context) ->
   error_logger:error_msg("ecql_mnesia: Empty record ~p~n", [Context])
  ,false
;
is_valid_record([undefined | _RecordList], Context) ->
  is_valid_tail(_RecordList, Context)
;
is_valid_record([_Element], _Context) ->
  true
;
is_valid_record(RecordList, Context) ->
  is_valid_record(tl(RecordList), Context)
.

%%------------------------------------------------------------------------------
is_valid_tail([], Context) ->
   error_logger:error_msg("ecql_mnesia: Non-migrated record ~p~n", [Context])
  ,true
;
is_valid_tail([undefined | RecordList], Context) ->
   is_valid_tail(RecordList, Context)
;
is_valid_tail(_RecordList, Context) ->
   error_logger:error_msg("ecql_mnesia: Partial record ~p~n", [Context])
  ,false
.

%%------------------------------------------------------------------------------
split(NObjects ,ResultList) when length(ResultList) > NObjects ->
  lists:split(NObjects ,ResultList)
;
split(_NObjects ,ResultList) ->
  {ResultList ,[]}
.

%%------------------------------------------------------------------------------
get_tables() ->
  ecql_cache:get({ecql_mnesia, tables}, fun() ->
     {_, CQLTables} = ecql:select([
       "SELECT table_name, comment FROM system_schema.tables"
     ])
    ,Tables = [
      {
         unmap_recordname(TableName)
        ,[{
           type
          ,case Comment of
             <<"ecql_mnesia_set">> -> set
            ;<<"ecql_mnesia_bag">> -> bag
           end
         }]
      }
      || [TableName, Comment] <- CQLTables
        ,string:left(binary_to_list(Comment), 11) =:= "ecql_mnesia"
     ]
    ,Tables
  end)
.

%%------------------------------------------------------------------------------
is_tablename(Table) when is_atom(Table) ->
  not lists:member($_ ,atom_to_list(Table))
;
is_tablename(_) ->
  false
.

%%==============================================================================
%% END OF FILE