Versioning sketch with actor model lazy iterators (#8)

1 files changed, 281 insertions, 11 deletions

diff --git a/src/operations.erl b/src/operations.erl
index 7846caf..24256d1 100644
--- a/src/operations.erl
+++ b/src/operations.erl
@@ -1,15 +1,285 @@
 -module(operations).
--export([select/2, project/2, cartesian_product/2]).
+-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]).
 
-%% @doc Selection - filter tuples based on a predicate
-%% `Predicate` is a function that takes a tuple and returns true/false
-select(Relation, Predicate) ->
-    sets:filter(Predicate, Relation).
+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}
+    ]).
 
-%% @doc Projection - select specific attributes from a relation
-%% Note that `AttributeNames` can have duplicate keys, which are ignored
-project(Relation, AttributeNames) when is_list(AttributeNames) ->
-    sets:map(fun (Tuple) -> maps:with(AttributeNames, Tuple) end, Relation).
+-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.
 
-cartesian_product(Relation1, Relation2) ->
-    sets:from_list([maps:merge(Tuple1, Tuple2) || Tuple1 <- sets:to_list(Relation1), Tuple2 <- sets:to_list(Relation2)], [{version, 2}]).