【6日目】シンプルデータベース実装④

まえがき

本日は、今まで説明してきたシンプルデータベースのErlangソースを公開し、ソース自体の説明を少ししようと思います。Erlangを触るのは今回が初めてですので、変なところ、こうしたほうが良いよーというところはコメントにて指摘ください。随時修正していく予定です。

↓↓今回RDBMSを実装していくgithubリポジトリ
https://github.com/sh0yu/erlang-rdbms
ディレクトリ構成とか何もないですが、それは追々直していきます。メインのソースはsimple_db_server.erlになります。simple_db_server.erlは下記のソース群からできています。それぞれソースを見せながら、簡単に説明します。

• [クライアント-DB]インターフェース
• gen_serverコールバック関数
• [DB-キーバリューストア]インターフェース
• [DB-カラムインデックス]インターフェース
• システムテーブル操作関数
• ユーティリティ関数

[クライアント-DB]インターフェース

DBを操作するためのインターフェースを紹介。今まで説明してきた、下記6操作にそれぞれ対応しています。

• create_table/3 [テーブルを作成する]
• insert_data/3 [データを挿入する]
• select_data/4 [データを取得する]
• update_data/5 [データを更新する]
• delete_data/4 [データを削除する]
• drop_data/2 [テーブルを削除する]

※Erlangの世界では、関数をドキュメントに記載する際、関数名/(スラッシュ)引数の数(アリティ)で表記するようです。

simple_db_server.erl

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% DB Server APIs.
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
create_table(Pid, TableName, ColumnList) ->
    gen_server:call(Pid, {create_table, {TableName, ColumnList}}).

drop_table(Pid, TableName) ->
    gen_server:call(Pid, {drop_table, {TableName}}).

insert_data(Pid, TableName, Val) ->
    gen_server:call(Pid, {insert, {TableName, Val}}).

select_data(Pid, TableName, ColName, Val) ->
    gen_server:call(Pid, {select, {TableName, ColName, Val}}).

update_data(Pid, TableName, SetQuery, ColName, Val) ->
    gen_server:call(Pid, {update, {TableName, SetQuery, ColName, Val}}).

delete_data(Pid, TableName, ColName, Val) ->
    gen_server:call(Pid, {delete, {TableName, ColName, Val}}).

Erlangには汎用的なプログラムを実装するためにビヘイビアと呼ばれるものがいくつか用意されています。今回は gen_serverビヘイビア を利用しDBサーバを実装しています。上のソース中に出てくるPidはシンプルデータベースサーバのプロセスIDです。サーバを起動したときの返り値から取得できます。
クライアントからの呼び出しはこんな感じです。名前、価格という列を持つフルーツというテーブルを作り、データを挿入、更新、取得、削除し、最後にテーブルを削除します。

{ok, Pid} = simple_db_server:start_link().
simple_db_server:create_table(Pid, fruit, [name, price]).
simple_db_server:insert_data(Pid, fruit, [apple, 100]).
simple_db_server:update_data(Pid, fruit, [{price, 120}], name, apple).
simple_db_server:select_data(Pid, fruit, name, apple).
simple_db_server:delete_data(Pid, fruit, name, apple).
simple_db_server:drop_table(Pid, fruit).
simple_db_server:stop(Pid).

gen_serverコールバック関数

次にコールバック関数です。gen_serverビヘイビアでは、init/1, handle_call/3, handle_cast/2 などのコールバック関数を実装しなければいけません。

simple_db_server.erl

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Callback functions of gen_server.
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
init([]) ->
    %% システムテーブル作成
    create_system_tables(),
    {ok, []}.

handle_call({create_table, {TableName, ColumnList}}, _From, _State) ->
    register_table(TableName, ColumnList),
    register_column(TableName, ColumnList),
    {reply, ok, []};
handle_call({drop_table, {TableName}}, _From, _State) ->
    ColumnList = get_column_list(TableName),
    unregister_table(TableName),
    unregister_column(TableName, ColumnList),
    {reply, ok, []};
handle_call({insert, {TableName, Val}}, _From, _State) ->
    Oid = generate_object_id(),
    insert_kvstore(TableName, Oid, Val),
    insert_all_column_index(TableName, Val, Oid),
    {reply, Oid, []};
handle_call({select, {TableName, ColumnName, Val}}, _From, _State) ->
    OidList = select_object_id_list(TableName, ColumnName, Val),
    case select_kvstore(TableName, OidList) of
            table_not_found -> table_not_found;
            [] -> {reply, not_found, []};
            RetVal -> {reply, RetVal, []}
    end;
handle_call({update, {TableName, SetQuery, ColumnName, Val}}, _From, _State) ->
    OidList = select_object_id_list(TableName, ColumnName, Val),
    ColumnList = get_column_list(TableName),
    SetQueryConverted = convert_set_query(SetQuery, ColumnList),
    F = fun(Oid) ->
        %% OldVal -> [banana, 100]
        %% NewVal -> [apple, 100]
        OldVal = select_kvstore(TableName, Oid),
        OldValWithCol = select_kvstore_with_column(TableName, Oid),
        NewVal = build_new_val(OldVal, SetQueryConverted),
        %% kvstoreを更新する
        update_kvstore(TableName, Oid, NewVal),
        %% 更新対象のカラムごとにカラムインデックスを更新する
        %% ex. {name, apple}
        FF = fun({ColumnN, NewColVal}) ->
            ColumnIndexId = get_column_index_id(TableName, ColumnN),
            {_Key, OldColVal} = lists:keyfind(ColumnN, 1, OldValWithCol),
            update_column_index(ColumnIndexId, OldColVal, NewColVal, Oid)
        end,
        lists:map(FF, SetQuery)
    end,
    lists:map(F, OidList),
    {reply, ok, []};
handle_call({delete, {TableName, ColumnName, Val}}, _From, _State) ->
    OidList = select_object_id_list(TableName, ColumnName, Val),
    ColumnList = get_column_list(TableName),
    F = fun(Oid) ->
        KvVal = select_kvstore(TableName, Oid),
        %% kvstoreを削除する
        delete_kvstore(TableName, Oid),
        %% 各カラムごとにカラムインデックスを削除更新する
        %% ex. {name, apple}
        FF = fun({ColumnN, ColVal}) ->
            %% 古いインデックス情報を削除する
            ColumnIndexId = get_column_index_id(TableName, ColumnN),
            delete_column_index(ColumnIndexId, ColVal, Oid)
        end,
        lists:map(FF, lists:zip(ColumnList, KvVal))
    end,
    lists:map(F, OidList),
    {reply, ok, []};
handle_call(terminate, _From, _State) ->
    {stop, normal, ok, []}.

handle_cast({get_config}, []) ->
    {noreply, []}.

handle_info(Msg, _State) ->
    io:format("Unexpected message: ~p~n", [Msg]),
    {noreply, _State}.

terminate(normal, _State) ->
    io:format("Server teminated.~n"),
    ok.

gen_server:call/2が呼ばれた時、gen_serverによってコールバック関数が呼ばれます。
例）
1. simple_db_server:create_table(Pid, test, [col1, col2])を実行する(DBのインターフェース)
2. コールバック関数であるhandle_call/3にメッセージが送られる
3. メッセージのパターンマッチが行われ、handle_call({create_table, {TableName, ColumnList}}, _From, _State)が実行される

[DB-キーバリューストア]インターフェース

続いて、オブジェクトIDをキー、テーブルレコードをバリューとして格納するキーバリューストアを操作するソースです。キーバリューストアは、ErlangのETSという組み込みのインメモリデータベースを使います。テーブルをCREATEするときに、ETSを一つ作成します。ETS名はテーブル名と同じです。テーブルをDROPするときはETSも削除します。データをSELECTするときは、オブジェクトIDを指定します。

simple_db_server.erl

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Datastore mng functions.
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
create_kvstore(TableName) ->
    ets:new(TableName, [set, named_table, public]).

drop_kvstore(TableName) ->
    ets:delete(TableName).

insert_kvstore(TableName, Oid, Val) ->
    case get_table_id(TableName) of
        not_found -> table_not_found;
        TableId -> ets:insert(TableId, {Oid, Val})
    end.

select_kvstore(TableName, OidList) when is_list(OidList) ->
    lists:map(fun(Oid) ->
        select_kvstore(TableName, Oid) end,
        OidList);
select_kvstore(TableName, Oid) ->
    case get_table_id(TableName) of
        not_found -> table_not_found;
        TableId ->
            case ets:lookup(TableId, Oid) of
            [] -> [];
            [{_Oid, Val}] -> Val
            end        
    end.

select_kvstore_with_column(TableName, Oid) ->
        case get_table_id(TableName) of
                not_found -> table_not_found;
                TableId ->
                    case ets:lookup(TableId, Oid) of
                    [] -> [];
                    [{_Oid, Val}] -> Val,
                    ColumnList = get_column_list(TableName),
                    lists:zip(ColumnList, Val)
                    end        
            end.

update_kvstore(TableName, Oid, Val) ->
    case get_table_id(TableName) of
        not_found -> table_not_found;
        TableId -> ets:insert(TableId, {Oid, Val})
    end.

delete_kvstore(TableName, Oid) ->
    case get_table_id(TableName) of
        not_found -> table_not_found;
        TableId -> ets:delete(TableId, Oid)
    end.

[DB-カラムインデックス]インターフェース

そして、カラムインデックスを操作するソースです。カラムインデックスもETSで実装します。テーブルをCREATEするときに、カラムインデックスのETSを作成します。カラムインデックスを保持するETS名はテーブル名+"_"+カラム名としています。
テーブルレコードをSELECTするときに、カラムインデックスから対象レコードのオブジェクトIDを取得するために、カラムインデックスをSELECTします。
例）価格が150円のフルーツは？

1>select_column_index(fruit_price, 150).
[0001, 0002]

simple_db_server.erl

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% ColumnIndex mng functions.
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% カラムインデックスを挿入する
insert_column_index(ColumnIndexId, Val, Oid) ->
    case ets:lookup(ColumnIndexId, Val) of
        %% カラムの値が新規登録されるパターン
        [] -> ets:insert(ColumnIndexId, {Val, [Oid]});
        %% すでにカラムの値が登録されているパターン
        %% 既存のオブジェクトIDのリストに新しいオブジェクトIDを追加する
        [{_Val, OidList}] ->
            case lists:member(Oid, OidList) of
                false ->
                    ets:insert(ColumnIndexId, {Val, [Oid | OidList]});
                %% ただし、すでに同じオブジェクトIDが登録されている場合は追加しない
                true -> nop
            end
    end.

%% カラムインデックスを更新する
update_column_index(ColumnIndexId, OldVal, NewVal, Oid) ->
    %% 更新前の値に紐づくオブジェクトIDを削除する
    delete_column_index(ColumnIndexId, OldVal, Oid),
    insert_column_index(ColumnIndexId, NewVal, Oid).

%% カラムインデックスからオブジェクトIDを削除する
delete_column_index(ColumnIndexId, Val, Oid) ->
    case ets:lookup(ColumnIndexId, Val) of
        [] -> nop;
        [{_Val, OidList}] ->
            NewOidList = lists:filter(fun(X) -> X =/= Oid end, OidList),
            case NewOidList of
                [] -> ets:delete(ColumnIndexId, Val);
                _ -> ets:insert(ColumnIndexId, {Val, NewOidList})
            end
    end.

%% カラムインデックスからオブジェクトIDを取得する
select_column_index(ColumnIndexId, Val) ->
    ets:lookup(ColumnIndexId, Val).

%% カラムインデックスを作成する
create_column_index(ColumnIndexName) ->
    ets:new(ColumnIndexName, [set, named_table, public]).

%% カラムインデックスを削除する
drop_column_index(ColumnIndexId) ->
    ets:delete(ColumnIndexId).

%% 複数のカラムインデックスに値を挿入する
insert_all_column_index(TableName, ValList, Oid) ->
    ColumnList = get_column_list(TableName),
    InsertColumnIndex = fun({ColumnName, ColumnVal}) ->
        ColumnIndexId = get_column_index_id(TableName, ColumnName),
        insert_column_index(ColumnIndexId, ColumnVal, Oid)
    end,
    lists:map(InsertColumnIndex, lists:zip(ColumnList, ValList)).

%% 指定されたテーブルで条件に当てはまるオブジェクトIDのリストを取得する
select_object_id_list(TableName, ColumnName, Val) ->
    ColumnIndexId = get_column_index_id(TableName, ColumnName),
    case ets:lookup(ColumnIndexId, Val) of
        [] -> [];
        [{_Val, OidList}] -> OidList
    end.

システムテーブル操作関数

システムテーブルは、テーブルの全量とそのテーブルが持つカラムを保持するms_tablesと、カラムの全量を保持するms_tab_columnsがあります。システムテーブルはサーバ起動時に作成されます。

simple_db_server.erl

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% System Table mng functions.
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% システムテーブルを作成する
create_system_tables() ->
    ets:new(ms_tables, [set, named_table, public]),
    ets:new(ms_tab_columns, [set, named_table, public]).

%% テーブルを作成して、ms_tablesにテーブルを登録する
register_table(TableName, ColumnList) ->
    TableId = create_kvstore(TableName),
    ets:insert(ms_tables, {TableName, TableId, ColumnList}).

%% ms_tablesからテーブル情報を削除し、テーブルを削除する
unregister_table(TableName) ->
    TableId = get_table_id(TableName),
    ets:delete(ms_tables, TableId),
    drop_kvstore(TableName).

%% テーブル名からテーブルIDを取得する
get_table_id(TableName)->
    ets:lookup_element(ms_tables, TableName, 2).

%% ms_tab_columnsにカラムを登録する
register_column(TableName, ColumnNameList) when is_list(ColumnNameList) ->
    lists:map(fun(ColumnName) -> 
        register_column(TableName, ColumnName) end,
        ColumnNameList);
register_column(TableName, ColumnName) ->
    ColumnIndexName = get_tab_column_key(TableName, ColumnName),
    ColumnIndexId = create_column_index(ColumnIndexName),
    ets:insert(ms_tab_columns, {ColumnIndexName, TableName, ColumnName, ColumnIndexId}).

%% ms_tab_columnsからカラムを削除する
unregister_column(TableName, ColumnNameList) when is_list(ColumnNameList) ->
    lists:map(fun(ColumnName) ->
        unregister_column(TableName, ColumnName) end,
        ColumnNameList);
unregister_column(TableName, ColumnName) ->
    ColumnIndexName = get_tab_column_key(TableName, ColumnName),
    ets:delete(ms_tab_columns, ColumnIndexName),
    drop_column_index(ColumnIndexName).

get_column_list(TableName) ->
    case ets:lookup_element(ms_tables, TableName, 3) of
        [] -> not_found;
        ColumnList -> ColumnList
end.

get_column_index_id(TableName, ColumnName) ->
    case ets:lookup_element(ms_tab_columns, get_tab_column_key(TableName, ColumnName), 4) of
        [] -> not_found;
        ColumnIndexId -> ColumnIndexId
end.

ユーティリティ関数

最後は雑多なユーティリティの関数です。
オブジェクトIDは呼び出し時点のナノ秒値です。
データをUPDATEする際に渡すSetQueryは、カラム名と更新後値のタプルのリストです。[{price, 150}]。それを使って、データを更新するための関数がbuild_new_val/2です。SetQueryはそのままだと使いにくいので、convert_set_query/2によって列番号と更新後値のタプルのリストに変換します。[{2, 150}]。

simple_db_server.erl

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% util functions.
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% オブジェクトIDを取得する
generate_object_id() ->
    erlang:system_time(nanosecond).

%% tab_column_keyを取得する
get_tab_column_key(TableName, ColumnNameList) when is_list(ColumnNameList) ->
    lists:map(fun(ColumnName) ->
        get_tab_column_key(TableName, ColumnName) end,
        ColumnNameList);
get_tab_column_key(TableName, ColumnName) ->
    TableNameStr = atom_to_list(TableName),
    ColumnNameStr = atom_to_list(ColumnName),
    list_to_atom(TableNameStr ++ "_" ++ ColumnNameStr).

%% OldVal -> [apple, 100]
%% SetQueryConverted -> [{1, banana}, {2, 200}]
build_new_val(OldVal, SetQueryConverted) ->
    Sub = fun Sub([], _SetQuery, Res) ->
                lists:reverse(Res);
            Sub(OldV, [{1, NewVal} | Rest], Res) ->
                Sub(tl(OldV), decrement_set_query(Rest), [NewVal | Res]);
            Sub(OldV, SetQ, Res) ->
                Sub(tl(OldV), decrement_set_query(SetQ), [hd(OldV) | Res])
            end,
    Sub(OldVal, SetQueryConverted, []).

%% decrement_set_query([{2, 200}, {3, apple}]) -> [{1, 200}, {2, apple}]
decrement_set_query([]) -> [];
decrement_set_query([{Num, Val} | T]) -> [{Num - 1, Val} | decrement_set_query(T)].

%% ([{name, banana}, {price, 200}], [name, price]) -> [{1, banana}, {2, 200}]
convert_set_query(SetQuery, ColumnList) ->
    %% ColumnListの先頭がSetQueryにあれば、タプルの左の値をカラムリストの番号に書き換える
    Sub = fun   Sub(_SetQ, [], _N, Res) ->
                    lists:reverse(Res);
                Sub(SetQ, ColumnL, N, Res) ->
                Key = hd(ColumnL),
                case lists:keymember(hd(ColumnL), 1, SetQ) of
                    true -> 
                        {_Key, Val} = lists:keyfind(Key, 1, SetQ),
                        Filter = fun (X) ->
                                case X of 
                                    ({Key, _Val}) -> false;
                                    (_) -> true
                                end
                            end,
                        Sub(lists:filter(Filter, SetQ), tl(ColumnL), N + 1, [{N, Val} | Res]);
                    false ->
                        Sub(SetQ, tl(ColumnL), N + 1, Res)
                end
            end,
    Sub(SetQuery, ColumnList, 1, []).

あとがき

シンプルデータベースサーバの実装をざっと紹介しました。クライアントソースと合わせて見たほうがわかりやすいかもしれませんね。各ETSのテーブルたちはpublicで定義しているため、サーバ実行中にETSの中身を見てみると良いと思います。

ソースの説明ってどうやるのがベストなんでしょうか。あんまり一つ一つ説明するつもりはないんですが、説明しなさすぎても意味がないし。。