% Licensed under the Apache License, Version 2.0 (the "License"); you may not % use this file except in compliance with the License. You may obtain a copy of % the License at % % http://www.apache.org/licenses/LICENSE-2.0 % % Unless required by applicable law or agreed to in writing, software % distributed under the License is distributed on an "AS IS" BASIS, WITHOUT % WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the % License for the specific language governing permissions and limitations under % the License. -module(erlfdb). -compile({no_auto_import, [get/1]}). -export([ open/0, open/1, open_tenant/2, create_transaction/1, tenant_create_transaction/1, transactional/2, snapshot/1, % Db/Tx configuration set_option/2, set_option/3, % Lifecycle Management commit/1, reset/1, cancel/1, cancel/2, % Future Specific functions is_ready/1, get/1, get_error/1, block_until_ready/1, wait/1, wait/2, wait_for_any/1, wait_for_any/2, wait_for_all/1, wait_for_all/2, % Data retrieval get/2, get_ss/2, get_key/2, get_key_ss/2, get_range/3, get_range/4, get_range_startswith/2, get_range_startswith/3, get_mapped_range/4, get_mapped_range/5, fold_range/5, fold_range/6, fold_range_future/4, fold_range_wait/4, % Data modifications set/3, clear/2, clear_range/3, clear_range_startswith/2, % Atomic operations add/3, bit_and/3, bit_or/3, bit_xor/3, min/3, max/3, byte_min/3, byte_max/3, set_versionstamped_key/3, set_versionstamped_value/3, atomic_op/4, % Watches watch/2, get_and_watch/2, set_and_watch/3, clear_and_watch/2, % Conflict ranges add_read_conflict_key/2, add_read_conflict_range/3, add_write_conflict_key/2, add_write_conflict_range/3, add_conflict_range/4, % Transaction versioning set_read_version/2, get_read_version/1, get_committed_version/1, get_versionstamp/1, % Transaction size info get_approximate_size/1, % Transaction status get_next_tx_id/1, is_read_only/1, has_watches/1, get_writes_allowed/1, % Locality and Statistics get_addresses_for_key/2, get_estimated_range_size/3, % Get conflict information get_conflicting_keys/1, % Misc on_error/2, error_predicate/2, get_last_error/0, get_error_string/1 ]). -export_type([ atomic_mode/0, atomic_operand/0, cluster_filename/0, database/0, database_option/0, error/0, error_predicate/0, fold_future/0, fold_option/0, future/0, key/0, key_selector/0, mapper/0, result/0, snapshot/0, tenant/0, tenant_name/0, transaction/0, transaction_option/0, value/0, version/0, wait_option/0 ]). -define(IS_FUTURE, {erlfdb_future, _, _}). -define(IS_FOLD_FUTURE, {fold_info, _, _}). -define(IS_DB, {erlfdb_database, _}). -define(IS_TENANT, {erlfdb_tenant, _}). -define(IS_TX, {erlfdb_transaction, _}). -define(IS_SS, {erlfdb_snapshot, _}). -define(GET_TX(SS), element(2, SS)). -define(ERLFDB_ERROR, '$erlfdb_error'). -record(fold_st, { start_key, end_key, mapper, limit, target_bytes, streaming_mode, iteration, snapshot, reverse }). -type atomic_mode() :: erlfdb_nif:atomic_mode(). -type atomic_operand() :: erlfdb_nif:atomic_operand(). -type cluster_filename() :: binary(). -type database() :: erlfdb_nif:database(). -type database_option() :: erlfdb_nif:database_option(). -type error() :: erlfdb_nif:error(). -type error_predicate() :: erlfdb_nif:error_predicate(). -type fold_future() :: {fold_info, #fold_st{}, future()}. -type fold_option() :: {reverse, boolean() | integer()} | {limit, non_neg_integer()} | {target_bytes, non_neg_integer()} | {streaming_mode, atom()} | {iteration, pos_integer()} | {snapshot, boolean()} | {mapper, binary()}. -type future() :: erlfdb_nif:future(). -type key() :: erlfdb_nif:key(). -type key_selector() :: erlfdb_nif:key_selector(). -type result() :: erlfdb_nif:future_result(). -type mapper() :: tuple(). -type snapshot() :: {erlfdb_snapshot, transaction()}. -type tenant() :: erlfdb_nif:tenant(). -type tenant_name() :: binary(). -type transaction() :: erlfdb_nif:transaction(). -type transaction_option() :: erlfdb_nif:transaction_option(). -type value() :: erlfdb_nif:value(). -type version() :: erlfdb_nif:version(). -type wait_option() :: {timeout, non_neg_integer() | infinity}. -spec open() -> database(). open() -> open(<<>>). -spec open(cluster_filename()) -> database(). open(ClusterFile) -> erlfdb_nif:create_database(ClusterFile). -spec open_tenant(database(), tenant_name()) -> tenant(). open_tenant(?IS_DB = Db, TenantName) -> erlfdb_nif:database_open_tenant(Db, TenantName). -spec create_transaction(database()) -> transaction(). create_transaction(?IS_DB = Db) -> erlfdb_nif:database_create_transaction(Db). -spec tenant_create_transaction(tenant()) -> transaction(). tenant_create_transaction(?IS_TENANT = Tenant) -> erlfdb_nif:tenant_create_transaction(Tenant). -spec transactional(database() | tenant() | transaction() | snapshot(), function()) -> any(). transactional(?IS_DB = Db, UserFun) when is_function(UserFun, 1) -> clear_erlfdb_error(), Tx = create_transaction(Db), do_transaction(Tx, UserFun); transactional(?IS_TENANT = Tenant, UserFun) when is_function(UserFun, 1) -> clear_erlfdb_error(), Tx = tenant_create_transaction(Tenant), do_transaction(Tx, UserFun); transactional(?IS_TX = Tx, UserFun) when is_function(UserFun, 1) -> UserFun(Tx); transactional(?IS_SS = SS, UserFun) when is_function(UserFun, 1) -> UserFun(SS). -spec snapshot(transaction() | snapshot()) -> snapshot(). snapshot(?IS_TX = Tx) -> {erlfdb_snapshot, Tx}; snapshot(?IS_SS = SS) -> SS. -spec set_option(database() | transaction(), database_option() | transaction_option()) -> ok. set_option(DbOrTx, Option) -> set_option(DbOrTx, Option, <<>>). -spec set_option(database() | transaction(), database_option() | transaction_option(), binary()) -> ok. set_option(?IS_DB = Db, DbOption, Value) -> erlfdb_nif:database_set_option(Db, DbOption, Value); set_option(?IS_TX = Tx, TxOption, Value) -> erlfdb_nif:transaction_set_option(Tx, TxOption, Value). -spec commit(transaction()) -> future(). commit(?IS_TX = Tx) -> erlfdb_nif:transaction_commit(Tx). -spec reset(transaction()) -> ok. reset(?IS_TX = Tx) -> ok = erlfdb_nif:transaction_reset(Tx). -spec cancel(fold_future() | future() | transaction()) -> ok. cancel(?IS_FOLD_FUTURE = FoldInfo) -> cancel(FoldInfo, []); cancel(?IS_FUTURE = Future) -> cancel(Future, []); cancel(?IS_TX = Tx) -> ok = erlfdb_nif:transaction_cancel(Tx). -spec cancel(fold_future() | future(), [{flush, boolean()}] | []) -> ok. cancel(?IS_FOLD_FUTURE = FoldInfo, Options) -> {fold_info, _St, Future} = FoldInfo, cancel(Future, Options); cancel(?IS_FUTURE = Future, Options) -> ok = erlfdb_nif:future_cancel(Future), case erlfdb_util:get(Options, flush, false) of true -> flush_future_message(Future); false -> ok end. -spec is_ready(future()) -> boolean(). is_ready(?IS_FUTURE = Future) -> erlfdb_nif:future_is_ready(Future). -spec get_error(future()) -> error(). get_error(?IS_FUTURE = Future) -> erlfdb_nif:future_get_error(Future). -spec get(future()) -> any(). get(?IS_FUTURE = Future) -> erlfdb_nif:future_get(Future). -spec block_until_ready(future()) -> ok. block_until_ready(?IS_FUTURE = Future) -> {erlfdb_future, MsgRef, _FRef} = Future, receive {MsgRef, ready} -> ok end. -spec wait(future() | result()) -> result(). wait(?IS_FUTURE = Future) -> wait(Future, []); wait(Ready) -> Ready. -spec wait(future() | any(), [wait_option()]) -> any(). wait(?IS_FUTURE = Future, Options) -> case is_ready(Future) of true -> Result = get(Future), % Flush ready message if already sent flush_future_message(Future), Result; false -> Timeout = erlfdb_util:get(Options, timeout, infinity), {erlfdb_future, MsgRef, _Res} = Future, receive {MsgRef, ready} -> get(Future) after Timeout -> erlang:error({timeout, Future}) end end; wait(Ready, _) -> Ready. -spec wait_for_any([future()]) -> any(). wait_for_any(Futures) -> wait_for_any(Futures, []). -spec wait_for_any([future()], [wait_option()]) -> any(). wait_for_any(Futures, Options) -> wait_for_any(Futures, Options, []). -spec wait_for_any([future()], [wait_option()], list()) -> any(). wait_for_any(Futures, Options, ResendQ) -> Timeout = erlfdb_util:get(Options, timeout, infinity), receive {MsgRef, ready} = Msg -> case lists:keyfind(MsgRef, 2, Futures) of ?IS_FUTURE = Future -> lists:foreach( fun(M) -> self() ! M end, ResendQ ), Future; _ -> wait_for_any(Futures, Options, [Msg | ResendQ]) end after Timeout -> lists:foreach( fun(M) -> self() ! M end, ResendQ ), erlang:error({timeout, Futures}) end. -spec wait_for_all([future()]) -> list(). wait_for_all(Futures) -> wait_for_all(Futures, []). -spec wait_for_all([future()], [wait_option()]) -> list(). wait_for_all(Futures, Options) -> % Same as wait for all. We might want to % handle timeouts here so we have a single % timeout for all future waiting. lists:map( fun(Future) -> wait(Future, Options) end, Futures ). -spec get(database() | transaction() | snapshot(), key()) -> future(). get(?IS_DB = Db, Key) -> transactional(Db, fun(Tx) -> wait(get(Tx, Key)) end); get(?IS_TX = Tx, Key) -> erlfdb_nif:transaction_get(Tx, Key, false); get(?IS_SS = SS, Key) -> get_ss(?GET_TX(SS), Key). -spec get_ss(transaction() | snapshot(), key()) -> future(). get_ss(?IS_TX = Tx, Key) -> erlfdb_nif:transaction_get(Tx, Key, true); get_ss(?IS_SS = SS, Key) -> get_ss(?GET_TX(SS), Key). -spec get_key(database() | transaction() | snapshot(), key_selector()) -> future(). get_key(?IS_DB = Db, Key) -> transactional(Db, fun(Tx) -> wait(get_key(Tx, Key)) end); get_key(?IS_TX = Tx, Key) -> erlfdb_nif:transaction_get_key(Tx, Key, false); get_key(?IS_SS = SS, Key) -> get_key_ss(?GET_TX(SS), Key). -spec get_key_ss(transaction(), key_selector()) -> future(). get_key_ss(?IS_TX = Tx, Key) -> erlfdb_nif:transaction_get_key(Tx, Key, true). -spec get_range(database() | transaction(), key(), key()) -> future(). get_range(DbOrTx, StartKey, EndKey) -> get_range(DbOrTx, StartKey, EndKey, []). -spec get_range(database() | transaction(), key(), key(), [fold_option()]) -> future(). get_range(?IS_DB = Db, StartKey, EndKey, Options) -> transactional(Db, fun(Tx) -> get_range(Tx, StartKey, EndKey, Options) end); get_range(?IS_TX = Tx, StartKey, EndKey, Options) -> Fun = fun(Rows, Acc) -> [Rows | Acc] end, Chunks = fold_range_int(Tx, StartKey, EndKey, Fun, [], Options), lists:flatten(lists:reverse(Chunks)); get_range(?IS_SS = SS, StartKey, EndKey, Options) -> get_range(?GET_TX(SS), StartKey, EndKey, [{snapshot, true} | Options]). -spec get_range_startswith(database() | transaction(), key()) -> future(). get_range_startswith(DbOrTx, Prefix) -> get_range_startswith(DbOrTx, Prefix, []). -spec get_range_startswith(database() | transaction(), key(), [fold_option()]) -> future(). get_range_startswith(DbOrTx, Prefix, Options) -> StartKey = Prefix, EndKey = erlfdb_key:strinc(Prefix), get_range(DbOrTx, StartKey, EndKey, Options). -spec get_mapped_range(database() | transaction(), key(), key(), mapper()) -> future(). get_mapped_range(DbOrTx, StartKey, EndKey, Mapper) -> get_mapped_range(DbOrTx, StartKey, EndKey, Mapper, []). -spec get_mapped_range(database() | transaction(), key(), key(), mapper(), [fold_option()]) -> future(). get_mapped_range(DbOrTx, StartKey, EndKey, Mapper, Options) -> get_range(DbOrTx, StartKey, EndKey, [{mapper, erlfdb_tuple:pack(Mapper)} | Options]). -spec fold_range(database() | transaction(), key(), key(), function(), any()) -> any(). fold_range(DbOrTx, StartKey, EndKey, Fun, Acc) -> fold_range(DbOrTx, StartKey, EndKey, Fun, Acc, []). -spec fold_range(database() | transaction(), key(), key(), function(), any(), [fold_option()]) -> any(). fold_range(?IS_DB = Db, StartKey, EndKey, Fun, Acc, Options) -> transactional(Db, fun(Tx) -> fold_range(Tx, StartKey, EndKey, Fun, Acc, Options) end); fold_range(?IS_TX = Tx, StartKey, EndKey, Fun, Acc, Options) -> fold_range_int( Tx, StartKey, EndKey, fun(Rows, InnerAcc) -> lists:foldl(Fun, InnerAcc, Rows) end, Acc, Options ); fold_range(?IS_SS = SS, StartKey, EndKey, Fun, Acc, Options) -> SSOptions = [{snapshot, true} | Options], fold_range(?GET_TX(SS), StartKey, EndKey, Fun, Acc, SSOptions). -spec fold_range_future(transaction() | snapshot(), key(), key(), [fold_option()]) -> fold_future(). fold_range_future(?IS_TX = Tx, StartKey, EndKey, Options) -> St = options_to_fold_st(StartKey, EndKey, Options), fold_range_future_int(Tx, St); fold_range_future(?IS_SS = SS, StartKey, EndKey, Options) -> SSOptions = [{snapshot, true} | Options], fold_range_future(?GET_TX(SS), StartKey, EndKey, SSOptions). -spec fold_range_wait(transaction(), fold_future(), function(), any()) -> any(). fold_range_wait(?IS_TX = Tx, ?IS_FOLD_FUTURE = FI, Fun, Acc) -> fold_range_int( Tx, FI, fun(Rows, InnerAcc) -> lists:foldl(Fun, InnerAcc, Rows) end, Acc ); fold_range_wait(?IS_SS = SS, ?IS_FOLD_FUTURE = FI, Fun, Acc) -> fold_range_wait(?GET_TX(SS), FI, Fun, Acc). -spec set(database() | transaction() | snapshot(), key(), value()) -> future(). set(?IS_DB = Db, Key, Value) -> transactional(Db, fun(Tx) -> set(Tx, Key, Value) end); set(?IS_TX = Tx, Key, Value) -> erlfdb_nif:transaction_set(Tx, Key, Value); set(?IS_SS = SS, Key, Value) -> set(?GET_TX(SS), Key, Value). -spec clear(database() | transaction() | snapshot(), key()) -> ok. clear(?IS_DB = Db, Key) -> transactional(Db, fun(Tx) -> clear(Tx, Key) end); clear(?IS_TX = Tx, Key) -> erlfdb_nif:transaction_clear(Tx, Key); clear(?IS_SS = SS, Key) -> clear(?GET_TX(SS), Key). -spec clear_range(database() | transaction() | snapshot(), key(), key()) -> ok. clear_range(?IS_DB = Db, StartKey, EndKey) -> transactional(Db, fun(Tx) -> clear_range(Tx, StartKey, EndKey) end); clear_range(?IS_TX = Tx, StartKey, EndKey) -> erlfdb_nif:transaction_clear_range(Tx, StartKey, EndKey); clear_range(?IS_SS = SS, StartKey, EndKey) -> clear_range(?GET_TX(SS), StartKey, EndKey). -spec clear_range_startswith(database() | transaction() | snapshot(), key()) -> ok. clear_range_startswith(?IS_DB = Db, Prefix) -> transactional(Db, fun(Tx) -> clear_range_startswith(Tx, Prefix) end); clear_range_startswith(?IS_TX = Tx, Prefix) -> EndKey = erlfdb_key:strinc(Prefix), erlfdb_nif:transaction_clear_range(Tx, Prefix, EndKey); clear_range_startswith(?IS_SS = SS, Prefix) -> clear_range_startswith(?GET_TX(SS), Prefix). -spec add(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. add(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, add). -spec bit_and(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. bit_and(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, bit_and). -spec bit_or(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. bit_or(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, bit_or). -spec bit_xor(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. bit_xor(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, bit_xor). -spec min(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. min(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, min). -spec max(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. max(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, max). -spec byte_min(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. byte_min(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, byte_min). -spec byte_max(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. byte_max(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, byte_max). -spec set_versionstamped_key(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. set_versionstamped_key(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, set_versionstamped_key). -spec set_versionstamped_value(database() | transaction() | snapshot(), key(), atomic_operand()) -> ok. set_versionstamped_value(DbOrTx, Key, Param) -> atomic_op(DbOrTx, Key, Param, set_versionstamped_value). -spec atomic_op(database() | transaction() | snapshot(), key(), atomic_operand(), atomic_mode()) -> ok. atomic_op(?IS_DB = Db, Key, Param, Op) -> transactional(Db, fun(Tx) -> atomic_op(Tx, Key, Param, Op) end); atomic_op(?IS_TX = Tx, Key, Param, Op) -> erlfdb_nif:transaction_atomic_op(Tx, Key, Param, Op); atomic_op(?IS_SS = SS, Key, Param, Op) -> atomic_op(?GET_TX(SS), Key, Param, Op). -spec watch(database() | transaction() | snapshot(), key()) -> future(). watch(?IS_DB = Db, Key) -> transactional(Db, fun(Tx) -> watch(Tx, Key) end); watch(?IS_TX = Tx, Key) -> erlfdb_nif:transaction_watch(Tx, Key); watch(?IS_SS = SS, Key) -> watch(?GET_TX(SS), Key). -spec get_and_watch(database(), key()) -> {result(), future()}. get_and_watch(?IS_DB = Db, Key) -> transactional(Db, fun(Tx) -> KeyFuture = get(Tx, Key), WatchFuture = watch(Tx, Key), {wait(KeyFuture), WatchFuture} end). -spec set_and_watch(database(), key(), value()) -> future(). set_and_watch(?IS_DB = Db, Key, Value) -> transactional(Db, fun(Tx) -> set(Tx, Key, Value), watch(Tx, Key) end). -spec clear_and_watch(database(), key()) -> future(). clear_and_watch(?IS_DB = Db, Key) -> transactional(Db, fun(Tx) -> clear(Tx, Key), watch(Tx, Key) end). -spec add_read_conflict_key(transaction() | snapshot(), key()) -> ok. add_read_conflict_key(TxObj, Key) -> add_read_conflict_range(TxObj, Key, <>). -spec add_read_conflict_range(transaction() | snapshot(), key(), key()) -> ok. add_read_conflict_range(TxObj, StartKey, EndKey) -> add_conflict_range(TxObj, StartKey, EndKey, read). -spec add_write_conflict_key(transaction() | snapshot(), key()) -> ok. add_write_conflict_key(TxObj, Key) -> add_write_conflict_range(TxObj, Key, <>). -spec add_write_conflict_range(transaction() | snapshot(), key(), key()) -> ok. add_write_conflict_range(TxObj, StartKey, EndKey) -> add_conflict_range(TxObj, StartKey, EndKey, write). -spec add_conflict_range(transaction() | snapshot(), key(), key(), read | write) -> ok. add_conflict_range(?IS_TX = Tx, StartKey, EndKey, Type) -> erlfdb_nif:transaction_add_conflict_range(Tx, StartKey, EndKey, Type); add_conflict_range(?IS_SS = SS, StartKey, EndKey, Type) -> add_conflict_range(?GET_TX(SS), StartKey, EndKey, Type). -spec set_read_version(transaction() | snapshot(), version()) -> ok. set_read_version(?IS_TX = Tx, Version) -> erlfdb_nif:transaction_set_read_version(Tx, Version); set_read_version(?IS_SS = SS, Version) -> set_read_version(?GET_TX(SS), Version). -spec get_read_version(transaction() | snapshot()) -> future(). get_read_version(?IS_TX = Tx) -> erlfdb_nif:transaction_get_read_version(Tx); get_read_version(?IS_SS = SS) -> get_read_version(?GET_TX(SS)). -spec get_committed_version(transaction() | snapshot()) -> version(). get_committed_version(?IS_TX = Tx) -> erlfdb_nif:transaction_get_committed_version(Tx); get_committed_version(?IS_SS = SS) -> get_committed_version(?GET_TX(SS)). -spec get_versionstamp(transaction() | snapshot()) -> future(). get_versionstamp(?IS_TX = Tx) -> erlfdb_nif:transaction_get_versionstamp(Tx); get_versionstamp(?IS_SS = SS) -> get_versionstamp(?GET_TX(SS)). -spec get_approximate_size(transaction() | snapshot()) -> non_neg_integer(). get_approximate_size(?IS_TX = Tx) -> erlfdb_nif:transaction_get_approximate_size(Tx); get_approximate_size(?IS_SS = SS) -> get_approximate_size(?GET_TX(SS)). -spec get_next_tx_id(transaction() | snapshot()) -> non_neg_integer(). get_next_tx_id(?IS_TX = Tx) -> erlfdb_nif:transaction_get_next_tx_id(Tx); get_next_tx_id(?IS_SS = SS) -> get_next_tx_id(?GET_TX(SS)). -spec is_read_only(transaction() | snapshot()) -> boolean(). is_read_only(?IS_TX = Tx) -> erlfdb_nif:transaction_is_read_only(Tx); is_read_only(?IS_SS = SS) -> is_read_only(?GET_TX(SS)). -spec has_watches(transaction() | snapshot()) -> boolean(). has_watches(?IS_TX = Tx) -> erlfdb_nif:transaction_has_watches(Tx); has_watches(?IS_SS = SS) -> has_watches(?GET_TX(SS)). -spec get_writes_allowed(transaction() | snapshot()) -> boolean(). get_writes_allowed(?IS_TX = Tx) -> erlfdb_nif:transaction_get_writes_allowed(Tx); get_writes_allowed(?IS_SS = SS) -> get_writes_allowed(?GET_TX(SS)). -spec get_addresses_for_key(database() | transaction() | snapshot(), key()) -> future() | result(). get_addresses_for_key(?IS_DB = Db, Key) -> transactional(Db, fun(Tx) -> wait(get_addresses_for_key(Tx, Key)) end); get_addresses_for_key(?IS_TX = Tx, Key) -> erlfdb_nif:transaction_get_addresses_for_key(Tx, Key); get_addresses_for_key(?IS_SS = SS, Key) -> get_addresses_for_key(?GET_TX(SS), Key). -spec get_estimated_range_size(transaction() | snapshot(), key(), key()) -> future(). get_estimated_range_size(?IS_TX = Tx, StartKey, EndKey) -> erlfdb_nif:transaction_get_estimated_range_size(Tx, StartKey, EndKey); get_estimated_range_size(?IS_SS = SS, StartKey, EndKey) -> erlfdb_nif:transaction_get_estimated_range_size(?GET_TX(SS), StartKey, EndKey). -spec get_conflicting_keys(transaction()) -> future(). get_conflicting_keys(?IS_TX = Tx) -> StartKey = <<16#FF, 16#FF, "/transaction/conflicting_keys/">>, EndKey = <<16#FF, 16#FF, "/transaction/conflicting_keys/", 16#FF>>, get_range(Tx, StartKey, EndKey). -spec on_error(transaction() | snapshot(), error() | integer()) -> future(). on_error(?IS_TX = Tx, {erlfdb_error, ErrorCode}) -> on_error(Tx, ErrorCode); on_error(?IS_TX = Tx, ErrorCode) -> erlfdb_nif:transaction_on_error(Tx, ErrorCode); on_error(?IS_SS = SS, Error) -> on_error(?GET_TX(SS), Error). -spec error_predicate(error_predicate(), error() | integer()) -> boolean(). error_predicate(Predicate, {erlfdb_error, ErrorCode}) -> error_predicate(Predicate, ErrorCode); error_predicate(Predicate, ErrorCode) -> erlfdb_nif:error_predicate(Predicate, ErrorCode). -spec get_last_error() -> error() | undefined. get_last_error() -> erlang:get(?ERLFDB_ERROR). -spec get_error_string(integer()) -> binary(). get_error_string(ErrorCode) when is_integer(ErrorCode) -> erlfdb_nif:get_error(ErrorCode). -spec clear_erlfdb_error() -> ok. clear_erlfdb_error() -> put(?ERLFDB_ERROR, undefined). -spec do_transaction(transaction(), function()) -> any(). do_transaction(?IS_TX = Tx, UserFun) -> try Ret = UserFun(Tx), case is_read_only(Tx) andalso not has_watches(Tx) of true -> ok; false -> wait(commit(Tx), [{timeout, infinity}]) end, Ret catch error:{erlfdb_error, Code} -> put(?ERLFDB_ERROR, Code), wait(on_error(Tx, Code), [{timeout, infinity}]), do_transaction(Tx, UserFun) end. -spec fold_range_int(transaction(), key(), key(), function(), any(), [fold_option()]) -> any(). fold_range_int(?IS_TX = Tx, StartKey, EndKey, Fun, Acc, Options) -> St = options_to_fold_st(StartKey, EndKey, Options), fold_range_int(Tx, St, Fun, Acc). -spec fold_range_int(transaction(), fold_future() | #fold_st{}, function(), any()) -> any(). fold_range_int(Tx, #fold_st{} = St, Fun, Acc) -> RangeFuture = fold_range_future_int(Tx, St), fold_range_int(Tx, RangeFuture, Fun, Acc); fold_range_int(Tx, ?IS_FOLD_FUTURE = FI, Fun, Acc) -> {fold_info, St, Future} = FI, #fold_st{ start_key = StartKey, end_key = EndKey, limit = Limit, iteration = Iteration, reverse = Reverse } = St, {RawRows, Count, HasMore} = wait(Future), Count = length(RawRows), % If our limit is within the current set of % rows we need to truncate the list Rows = if Limit == 0 orelse Limit > Count -> RawRows; true -> lists:sublist(RawRows, Limit) end, % Invoke our callback to update the accumulator NewAcc = if Rows == [] -> Acc; true -> Fun(Rows, Acc) end, % Determine if we have more rows to iterate Recurse = (Rows /= []) and (Limit == 0 orelse Limit > Count) and HasMore, if not Recurse -> NewAcc; true -> LastKey = fold_range_last_key_int(Rows, St), {NewStartKey, NewEndKey} = case Reverse /= 0 of true -> {StartKey, erlfdb_key:first_greater_or_equal(LastKey)}; false -> {erlfdb_key:first_greater_than(LastKey), EndKey} end, NewSt = St#fold_st{ start_key = NewStartKey, end_key = NewEndKey, limit = if Limit == 0 -> 0; true -> Limit - Count end, iteration = Iteration + 1 }, fold_range_int(Tx, NewSt, Fun, NewAcc) end. -spec fold_range_last_key_int(list(), #fold_st{}) -> key(). fold_range_last_key_int(Rows, #fold_st{mapper = undefined}) -> element(1, lists:last(Rows)); fold_range_last_key_int(Rows, #fold_st{mapper = _Mapper}) -> element(1, element(1, lists:last(Rows))). -spec fold_range_future_int(transaction(), #fold_st{}) -> fold_future(). fold_range_future_int(?IS_TX = Tx, #fold_st{mapper = undefined} = St) -> #fold_st{ start_key = StartKey, end_key = EndKey, limit = Limit, target_bytes = TargetBytes, streaming_mode = StreamingMode, iteration = Iteration, snapshot = Snapshot, reverse = Reverse } = St, Future = erlfdb_nif:transaction_get_range( Tx, StartKey, EndKey, Limit, TargetBytes, StreamingMode, Iteration, Snapshot, Reverse ), {fold_info, St, Future}; fold_range_future_int(?IS_TX = Tx, #fold_st{} = St) -> #fold_st{ start_key = StartKey, end_key = EndKey, mapper = Mapper, limit = Limit, target_bytes = TargetBytes, streaming_mode = StreamingMode, iteration = Iteration, snapshot = Snapshot, reverse = Reverse } = St, Future = erlfdb_nif:transaction_get_mapped_range( Tx, StartKey, EndKey, Mapper, Limit, TargetBytes, StreamingMode, Iteration, Snapshot, Reverse ), {fold_info, St, Future}. -spec options_to_fold_st(key(), key(), [fold_option()]) -> #fold_st{}. options_to_fold_st(StartKey, EndKey, Options) -> Reverse = case erlfdb_util:get(Options, reverse, false) of true -> 1; false -> 0; I when is_integer(I) -> I end, #fold_st{ mapper = erlfdb_util:get(Options, mapper), start_key = erlfdb_key:to_selector(StartKey), end_key = erlfdb_key:to_selector(EndKey), limit = erlfdb_util:get(Options, limit, 0), target_bytes = erlfdb_util:get(Options, target_bytes, 0), streaming_mode = erlfdb_util:get(Options, streaming_mode, want_all), iteration = erlfdb_util:get(Options, iteration, 1), snapshot = erlfdb_util:get(Options, snapshot, false), reverse = Reverse }. -spec flush_future_message(future()) -> ok. flush_future_message(?IS_FUTURE = Future) -> erlfdb_nif:future_silence(Future), {erlfdb_future, MsgRef, _Res} = Future, receive {MsgRef, ready} -> ok after 0 -> ok end.