-module(operations).
-export([setup/0,
hash/1,
create_database/1,
create_relation/3,
create_tuple/3,
get_relations/1,
get_relation_hash/2,
hashes_from_tuple/1,
get_tuples_iterator/2,
next_tuple/1,
close_iterator/1,
collect_all/1]).
setup() ->
mnesia:stop(),
mnesia:delete_schema([node()]),
mnesia:create_schema([node()]),
mnesia:start(),
mnesia:create_table(attribute, [
{attributes, [hash, value]},
{disc_copies, [node()]},
{type, set}
]),
mnesia:create_table(tuple, [
{attributes, [hash, relation, attribute_map]},
{disc_copies, [node()]},
{type, set}
]),
mnesia:create_table(relation, [
{attributes, [hash, name, tree, schema]},
{disc_copies, [node()]},
{type, set}
]),
mnesia:create_table(database_state, [
{attributes, [hash, name, tree, relations, timestamp]},
{disc_copies, [node()]},
{type, set}
]).
-record(database_state, {hash, name, tree, relations, timestamp}).
-record(relation, {hash, name, tree, schema}).
-record(tuple, {hash, relation, attribute_map}).
-record(attribute, {hash, value}).
hash(Value) ->
crypto:hash(sha256, term_to_binary(Value)).
%% @doc Store a tuple into a relation and update database state
%% Database: #database_state{} record
%% Relation: relation name (atom)
%% Tuple: (map) #{name => "John", age => 18}
%% Returns: {UpdatedDatabase, UpdatedRelation}
create_tuple(Database, RelationName, Tuple) when is_map(Tuple), is_record(Database, database_state) ->
F = fun() ->
%% Step 1: Store each attribute value and build attribute_map
AttributeMap = maps:map(fun(_AttrName, Value) ->
ValueHash = hash(Value),
mnesia:write(attribute, #attribute{hash = ValueHash, value = Value}, write),
ValueHash
end, Tuple),
%% Step 2: Compute tuple hash and store tuple
TupleHash = hash(Tuple),
mnesia:write(tuple, #tuple{hash = TupleHash, relation = RelationName, attribute_map = AttributeMap}, write),
%% Step 3: Get current relation hash from database
CurrentRelationHash = maps:get(RelationName, Database#database_state.relations),
%% Step 4: Read current relation
[RelationRecord] = mnesia:read(relation, CurrentRelationHash),
%% Step 5: Insert tuple hash into relation merkle tree
%% We don't need to store the content of a tuple hash, just the key
%% merklet can spot the difference in both keys and values
%% sets:set follows the same strategy of an empty value
NewRelationTree = merklet:insert({TupleHash, <<>>}, RelationRecord#relation.tree),
%% Step 6: Compute new relation hash as the root of the new merkle tree and store
{_,NewRelationHash,_,_} = NewRelationTree,
UpdatedRelation = #relation{
hash = NewRelationHash,
name = RelationName,
tree = NewRelationTree,
schema = RelationRecord#relation.schema
},
mnesia:write(relation, UpdatedRelation, write),
%% Step 7: Update database relations map
NewRelations = maps:put(RelationName, NewRelationHash, Database#database_state.relations),
%% Step 8: Update database tree (for diffing)
RelationKey = atom_to_binary(RelationName, utf8),
NewDatabaseTree = merklet:insert({RelationKey, NewRelationHash}, Database#database_state.tree),
%% Step 9: Compute new database hash and store
{_,NewDatabaseHash,_,_} = NewDatabaseTree,
UpdatedDatabase = #database_state{
name = Database#database_state.name,
hash = NewDatabaseHash,
tree = NewDatabaseTree,
relations = NewRelations,
timestamp = erlang:timestamp()
},
mnesia:write(database_state, UpdatedDatabase, write),
{UpdatedDatabase, UpdatedRelation}
end,
{atomic, Result} = mnesia:transaction(F),
Result.
retract_tuple(Database, RelationHash, TupleHashes)
when is_record(Database, database_state) andalso is_list(TupleHashes) ->
todo;
retract_tuple(Database, RelationHash, TupleHash) when is_record(Database, database_state) ->
[Relation] = mnesia:dirty_read(relation, RelationHash),
RelationTreeWithoutTuple = merklet:delete(TupleHash, Relation#relation.tree),
{_,NewRelationHash,_,_} = RelationTreeWithoutTuple,
DatabaseTreeWithoutTuple = merklet:delete(RelationHash, Database#database_state.tree),
{_,NewDatabaseHash,_,_} = DatabaseTreeWithoutTuple,
UpdatedDatabase =
Database#database_state{timestamp = erlang:timestamp(),
hash = NewDatabaseHash,
tree = DatabaseTreeWithoutTuple},
UpdatedRelation =
Relation#relation{hash = NewRelationHash,
tree = RelationTreeWithoutTuple},
TX = fun () ->
mnesia:write(relation, UpdatedRelation, write),
mnesia:write(database_state, UpdatedDatabase, write)
end,
case mnesia:is_transaction() of
true -> TX();
false -> mnesia:transaction(TX)
end.
retract_relation(Database, Name) ->
todo.
create_relation(Database, Name, Definition) when is_record(Database, database_state) ->
F = fun() ->
%% Create relation with empty tree
%% The meaning of a relation needs to be attached to its name
%% It may sound nominalistic but that is a physical concern.
%% Two relations might share definitions but their name gives another interpretation
%% and therefore different tuples
RelationHash = hash({Name, Definition}),
NewRelation = #relation{
hash = RelationHash,
name = Name,
tree = undefined,
schema = Definition
},
mnesia:write(relation, NewRelation, write),
%% Update database relations map
%% TODO: maybe we can purge this later after including the name on the merkle trees
NewRelations = maps:put(Name, RelationHash, Database#database_state.relations),
%% Update database tree (for diffing)
%% Note that the name here is important as a key because
%% we gotta go from a name to a hash within a database state
NewTree = merklet:insert({atom_to_binary(Name), RelationHash}, Database#database_state.tree),
{_,NewHash,_,_} = NewTree,
UpdatedDatabase = #database_state{
name = Database#database_state.name,
hash = NewHash,
tree = NewTree,
relations = NewRelations,
timestamp = erlang:timestamp()
},
mnesia:write(database_state, UpdatedDatabase, write),
{UpdatedDatabase, NewRelation}
end,
{atomic, Result} = mnesia:transaction(F),
Result.
create_database(Name) ->
Entry = #database_state{
name = Name,
hash = <<>>,
tree = undefined,
relations = #{},
timestamp = erlang:timestamp()
},
{atomic, ok} = mnesia:transaction(fun () ->
mnesia:write(database_state, Entry, write)
end),
Entry.
%% @doc Get all relation names in a database
get_relations(Database) ->
maps:keys(Database#database_state.relations).
%% @doc Get relation hash by name
get_relation_hash(Database, RelationName) ->
case maps:find(RelationName, Database#database_state.relations) of
error -> {error, not_found};
X -> X
end.
%% @doc Resolve a tuple hash to actual values
%% Returns map with attribute names and actual values
resolve_tuple(TupleHash) ->
[#tuple{attribute_map = AttributeMap}] = mnesia:dirty_read(tuple, TupleHash),
maps:map(fun(_AttrName, ValueHash) ->
[#attribute{value = Value}] = mnesia:dirty_read(attribute, ValueHash),
Value
end, AttributeMap).
%% All reads can be dirty as it is all immutable
%% Writes need to be transactional
hashes_from_tuple(RelationHash) ->
[Relation] = mnesia:dirty_read(relation, RelationHash),
case Relation#relation.tree of
undefined -> [];
Tree -> merklet:keys(Tree)
end.
%% @doc Create an iterator for streaming tuples from a relation
%% Returns: Pid of iterator process
get_tuples_iterator(Database, RelationName) when is_record(Database, database_state) ->
case maps:get(RelationName, Database#database_state.relations, error) of
error -> erlang:error({error_tuple_iterator_init, RelationName});
RelationHash -> TupleHashes = hashes_from_tuple(RelationHash),
spawn(fun() -> tuple_iterator_loop(TupleHashes) end)
end.
%% X = (a, b, c, d)
%% Y = (a, x, y)
%% Z = project (a, c, y, natural_join(X, Y))
%% @doc Get next tuple from iterator1
%% Returns: {ok, Tuple} | done | {error, timeout}
next_tuple(IteratorPid) ->
IteratorPid ! {next, self()},
receive
{tuple, Tuple} -> {ok, Tuple};
done -> done
after 5000 ->
{error, timeout}
end.
%% @doc Close iterator process
close_iterator(IteratorPid) ->
IteratorPid ! stop,
ok.
%% Iterator loop: streams tuples one at a time
tuple_iterator_loop([]) ->
receive
{next, Caller} ->
Caller ! done;
stop ->
ok
end;
tuple_iterator_loop([TupleHash | Rest]) ->
receive
{next, Caller} ->
%% Resolve tuple on demand
ResolvedTuple = resolve_tuple(TupleHash),
Caller ! {tuple, ResolvedTuple},
tuple_iterator_loop(Rest);
stop ->
ok
end.
%% @doc Helper to collect all tuples from iterator (for testing)
collect_all(IteratorPid) ->
collect_all(IteratorPid, []).
collect_all(IteratorPid, Acc) ->
case next_tuple(IteratorPid) of
{ok, Tuple} -> collect_all(IteratorPid, [Tuple | Acc]);
done ->
close_iterator(IteratorPid),
Acc;
{error, Reason} ->
close_iterator(IteratorPid),
{error, Reason, Acc}
end.
