Pluggable datatypes prototype. Fixes #109

include/epgsql.hrl

@@ -1,8 +1,11 @@
--type epgsql_type() :: atom() | {array, atom()} | {unknown_oid, integer()}.
+-type epgsql_type() :: epgsql:type_name()
+                      | {array, epgsql:type_name()}
+                      | {unknown_oid, integer()}.
 
 -record(column, {
     name :: binary(),
     type :: epgsql_type(),
+    oid :: integer(),
     size :: -1 | pos_integer(),
     modifier :: -1 | pos_integer(),
     format :: integer()
@@ -11,7 +14,8 @@
 -record(statement, {
     name :: string(),
     columns :: [#column{}],
-    types :: [epgsql_type()]
+    types :: [epgsql_type()],
+    parameter_info :: [epgsql_oid_db:oid_entry()]
 }).
 
 -record(error, {

src/commands/epgsql_cmd_connect.erl

@@ -150,7 +150,8 @@ handle_message(?READY_FOR_QUERY, _, Sock, _State) ->
         <<"on">>  -> put(datetime_mod, epgsql_idatetime);
         <<"off">> -> put(datetime_mod, epgsql_fdatetime)
     end,
-    Sock1 = epgsql_sock:set_attr(codec, epgsql_binary:new_codec([]), Sock),
+    Codec = epgsql_binary:new_codec(epgsql_oid_db, Sock),
+    Sock1 = epgsql_sock:set_attr(codec, Codec, Sock),
     {finish, connected, connected, Sock1};
 
 

src/commands/epgsql_cmd_describe_statement.erl

@@ -14,7 +14,8 @@
 
 -record(desc_stmt,
         {name :: iodata(),
-         parameter_descr}).
+         parameter_typenames = [],
+         parameter_descr = []}).
 
 init(Name) ->
     #desc_stmt{name = Name}.
@@ -30,16 +31,23 @@ execute(Sock, #desc_stmt{name = Name} = St) ->
 
 handle_message(?PARAMETER_DESCRIPTION, Bin, Sock, State) ->
     Codec = epgsql_sock:get_codec(Sock),
-    Types = epgsql_wire:decode_parameters(Bin, Codec),
-    Sock2 = epgsql_sock:notify(Sock, {types, Types}),
-    {noaction, Sock2, State#desc_stmt{parameter_descr = Types}};
+    TypeInfos = epgsql_wire:decode_parameters(Bin, Codec),
+    OidInfos = [epgsql_binary:typeinfo_to_oid_info(Type, Codec) || Type <- TypeInfos],
+    TypeNames = [epgsql_binary:typeinfo_to_name_array(Type, Codec) || Type <- TypeInfos],
+    Sock2 = epgsql_sock:notify(Sock, {types, TypeNames}),
+    {noaction, Sock2, State#desc_stmt{parameter_descr = OidInfos,
+                                      parameter_typenames = TypeNames}};
 handle_message(?ROW_DESCRIPTION, <<Count:?int16, Bin/binary>>, Sock,
-               #desc_stmt{name = Name, parameter_descr = Params}) ->
+               #desc_stmt{name = Name, parameter_descr = Params,
+                          parameter_typenames = TypeNames}) ->
     Codec = epgsql_sock:get_codec(Sock),
     Columns = epgsql_wire:decode_columns(Count, Bin, Codec),
-    Columns2 = [Col#column{format = epgsql_wire:format(Col#column.type, Codec)}
+    Columns2 = [Col#column{format = epgsql_wire:format(Col, Codec)}
                 || Col <- Columns],
-    Result = {ok, #statement{name = Name, types = Params, columns = Columns2}},
+    Result = {ok, #statement{name = Name,
+                             types = TypeNames,
+                             parameter_info = Params,
+                             columns = Columns2}},
     {finish, Result, {columns, Columns2}, Sock};
 handle_message(?NO_DATA, <<>>, Sock, #desc_stmt{name = Name, parameter_descr = Params}) ->
     Result = {ok, #statement{name = Name, types = Params, columns = []}},

src/commands/epgsql_cmd_parse.erl

@@ -17,8 +17,10 @@
         {name :: iodata(),
          sql :: iodata(),
          types :: [atom()],
-         parameter_descr = []}).
+         parameter_typenames = [] :: [epgsql:type_name() | {array, epgsql:type_name()}],
+         parameter_descr = [] :: [epgsql_oid_db:oid_info()]}).
 
+%% FIXME: make it use oids instead of type names!
 init({Name, Sql, Types}) ->
     #parse{name = Name, sql = Sql, types = Types}.
 
@@ -38,19 +40,30 @@ handle_message(?PARSE_COMPLETE, <<>>, Sock, _State) ->
     {noaction, Sock};
 handle_message(?PARAMETER_DESCRIPTION, Bin, Sock, State) ->
     Codec = epgsql_sock:get_codec(Sock),
-    Types = epgsql_wire:decode_parameters(Bin, Codec),
-    Sock2 = epgsql_sock:notify(Sock, {types, Types}),
-    {noaction, Sock2, State#parse{parameter_descr = Types}};
+    TypeInfos = epgsql_wire:decode_parameters(Bin, Codec),
+    OidInfos = [epgsql_binary:typeinfo_to_oid_info(Type, Codec) || Type <- TypeInfos],
+    TypeNames = [epgsql_binary:typeinfo_to_name_array(Type, Codec) || Type <- TypeInfos],
+    Sock2 = epgsql_sock:notify(Sock, {types, TypeNames}),
+    {noaction, Sock2, State#parse{parameter_descr = OidInfos,
+                                  parameter_typenames = TypeNames}};
 handle_message(?ROW_DESCRIPTION, <<Count:?int16, Bin/binary>>, Sock,
-               #parse{name = Name, parameter_descr = Params}) ->
+               #parse{name = Name, parameter_descr = Params,
+                      parameter_typenames = TypeNames}) ->
     Codec = epgsql_sock:get_codec(Sock),
     Columns = epgsql_wire:decode_columns(Count, Bin, Codec),
-    Columns2 = [Col#column{format = epgsql_wire:format(Col#column.type, Codec)}
+    Columns2 = [Col#column{format = epgsql_wire:format(Col, Codec)}
                 || Col <- Columns],
-    Result = {ok, #statement{name = Name, types = Params, columns = Columns2}},
+    Result = {ok, #statement{name = Name,
+                             types = TypeNames,
+                             columns = Columns2,
+                             parameter_info = Params}},
     {finish, Result, {columns, Columns2}, Sock};
-handle_message(?NO_DATA, <<>>, Sock, #parse{name = Name, parameter_descr = Params}) ->
-    Result = {ok, #statement{name = Name, types = Params, columns = []}},
+handle_message(?NO_DATA, <<>>, Sock, #parse{name = Name, parameter_descr = Params,
+                                            parameter_typenames = TypeNames}) ->
+    Result = {ok, #statement{name = Name,
+                             types = TypeNames,
+                             parameter_info = Params,
+                             columns = []}},
     {finish, Result, no_data, Sock};
 handle_message(?ERROR, Error, _Sock, _State) ->
     Result = {error, Error},

src/commands/epgsql_cmd_update_type_cache.erl

@@ -0,0 +1,61 @@
+%% Special command. Executes Squery over pg_type table and updates codecs.
+-module(epgsql_cmd_update_type_cache).
+-behaviour(epgsql_command).
+-export([init/1, execute/2, handle_message/4]).
+-export_type([response/0]).
+
+-type response() ::
+        {ok, [epgsql:type_name()]}
+      | {error, epgsql:query_error()}.
+
+-include("protocol.hrl").
+
+-record(upd,
+        {codecs :: [{epgsql_codec:codec_mod(), Opts :: any()}],
+         codec_entries :: [epgsql_codec:codec_entry()],
+         decoder}).
+
+init(Codecs) ->
+    #upd{codecs = Codecs}.
+
+execute(Sock, #upd{codecs = Codecs} = State) ->
+    CodecEntries = epgsql_codec:init_mods(Codecs, Sock),
+    TypeNames = [element(1, Entry) || Entry <- CodecEntries],
+    Query = epgsql_oid_db:build_query(TypeNames),
+    epgsql_sock:send(Sock, ?SIMPLEQUERY, [Query, 0]),
+    {ok, Sock, State#upd{codec_entries = CodecEntries}}.
+
+handle_message(?ROW_DESCRIPTION, <<Count:?int16, Bin/binary>>, Sock, State) ->
+    Codec = epgsql_sock:get_codec(Sock),
+    Columns = epgsql_wire:decode_columns(Count, Bin, Codec),
+    Decoder = epgsql_wire:build_decoder(Columns, Codec),
+    {noaction, Sock, State#upd{decoder = Decoder}};
+handle_message(?DATA_ROW, <<_Count:?int16, Bin/binary>>,
+               Sock, #upd{decoder = Decoder} = St) ->
+    Row = epgsql_wire:decode_data(Bin, Decoder),
+    {add_row, Row, Sock, St};
+handle_message(?COMMAND_COMPLETE, Bin, Sock, St) ->
+    Complete = epgsql_wire:decode_complete(Bin),
+    Rows = epgsql_sock:get_rows(Sock),
+    {add_result, Rows, {complete, Complete}, Sock, St};
+handle_message(?READY_FOR_QUERY, _Status, Sock, State) ->
+    [Result] = epgsql_sock:get_results(Sock),
+    handle_result(Result, Sock, State);
+handle_message(?ERROR, Error, Sock, St) ->
+    Result = {error, Error},
+    {add_result, Result, Result, Sock, St};
+handle_message(_, _, _, _) ->
+    unknown.
+
+handle_result({error, _} = Err, Sock, _State) ->
+    {finish, Err, done, Sock};
+handle_result(Rows, Sock, #upd{codec_entries = CodecEntries} = _State) ->
+    OidEntries = epgsql_oid_db:parse_rows(Rows),
+    Types = epgsql_oid_db:join_codecs_oids(OidEntries, CodecEntries),
+
+    Codec = epgsql_sock:get_codec(Sock),
+    Codec1 = epgsql_binary:update_codec(Types, Codec),
+    Sock1 = epgsql_sock:set_attr(codec, Codec1, Sock),
+
+    TypeNames = [element(1, Entry) || Entry <- CodecEntries],
+    {finish, {ok, TypeNames}, done, Sock1}.

src/datatypes/epgsql_codec_boolean.erl

@@ -0,0 +1,29 @@
+%%% @doc
+%%% Codec for `bool'.
+%%% `unknown' is represented by `null'.
+%%% https://www.postgresql.org/docs/current/static/datatype-boolean.html
+%%% $PG$/src/backend/utils/adt/bool.c
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_boolean).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: boolean().
+
+init(_, _) -> [].
+
+names() ->
+    [bool].
+
+encode(true, bool, _) ->
+    <<1:1/big-signed-unit:8>>;
+encode(false, bool, _) ->
+    <<0:1/big-signed-unit:8>>.
+
+decode(<<1:1/big-signed-unit:8>>, bool, _) -> true;
+decode(<<0:1/big-signed-unit:8>>, bool, _) -> false.

src/datatypes/epgsql_codec_bpchar.erl

@@ -0,0 +1,30 @@
+%%% @doc
+%%% Codec for `bpchar', `char' (CHAR(N), char).
+%%% ```SELECT 1::char''' ```SELECT 'abc'::char(10)'''
+%%% For 'text', 'varchar' see epgsql_codec_text.erl.
+%%% https://www.postgresql.org/docs/10/static/datatype-character.html
+%%% $PG$/src/backend/utils/adt/varchar.c
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_bpchar).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: binary() | byte().
+
+init(_, _) -> [].
+
+names() ->
+    [bpchar, char].
+
+encode(C, _, _) when is_integer(C), C =< 255 ->
+    <<C:1/big-unsigned-unit:8>>;
+encode(Bin, bpchar, _) when is_binary(Bin) ->
+    Bin.
+
+decode(<<C:1/big-unsigned-unit:8>>, _, _) -> C;
+decode(Bin, bpchar, _) -> Bin.

src/datatypes/epgsql_codec_datetime.erl

@@ -0,0 +1,48 @@
+%%% @doc
+%%% Codec for `time', `timetz', `date', `timestamp', `timestamptz', `interval'
+%%% https://www.postgresql.org/docs/current/static/datatype-datetime.html
+%%% $PG$/src/backend/utils/adt/timestamp.c // `timestamp', `timestamptz', `interval'
+%%% $PG$/src/backend/utils/adt/datetime.c // helpers
+%%% $PG$/src/backend/utils/adt/date.c // `time', `timetz', `date'
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_datetime).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: pg_date() | pg_time() | pg_datetime() | pg_interval() | pg_timetz().
+
+%% Ranges are from https://www.postgresql.org/docs/current/static/datatype-datetime.html
+-type pg_date() ::
+        {Year :: -4712..294276,
+         Month :: 1..12,
+         Day :: 1..31}.
+-type pg_time() ::
+        {Hour :: 0..24,  % Max value is 24:00:00
+         Minute :: 0..59,
+         Second :: 0..59 | float()}.
+-type pg_timetz() :: {pg_time(), UtcOffset :: integer()}.
+-type pg_datetime() :: {pg_date(), pg_time()}.
+-type pg_interval() :: {pg_time(), Days :: integer(), Months :: integer()}.
+
+
+init(_, Sock) ->
+    case epgsql_sock:get_parameter_internal(<<"integer_datetimes">>, Sock) of
+        <<"on">>  -> epgsql_idatetime;
+        <<"off">> -> epgsql_fdatetime
+    end.
+
+names() ->
+    [time, timetz, date, timestamp, timestamptz, interval].
+
+%% FIXME: move common logick out from fdatetime/idatetime; make them more
+%% low-level
+encode(Val, Type, Mod) ->
+    Mod:encode(Type, Val).
+
+decode(Bin, Type, Mod) ->
+    Mod:decode(Type, Bin).

src/datatypes/epgsql_codec_float.erl

@@ -0,0 +1,55 @@
+%%% @doc
+%%% Codec for `float4', `float8' (real, double precision).
+%%% https://www.postgresql.org/docs/current/static/datatype-numeric.html#datatype-float
+%%% $PG$/src/backend/utils/adt/float.c
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_float).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: in_data() | out_data().
+-type in_data() :: integer() | float() | nan | plus_infinity | minus_infinity.
+-type out_data() :: float() | nan | plus_infinity | minus_infinity.
+
+-define(POS_INF,  <<0:1, 255:8, 0:23>>).
+-define(NEG_INF,  <<1:1, 255:8, 0:23>>).
+-define(NAN_PATTERN, <<_:1, 255:8, _:23>>).
+-define(NAN, <<0:1, 255:8, 1:1, 0:22>>).
+
+-define(POS_INF8, <<0:1, 2047:11, 0:52>>).
+-define(NEG_INF8, <<1:1, 2047:11, 0:52>>).
+-define(NAN_PATTERN8, <<_:1, 2047:11, _:52>>).
+-define(NAN8, <<0:1, 2047:11, 1:1, 0:51>>).
+
+init(_, _) -> [].
+
+names() ->
+    [float4, float8].
+
+encode(Int, Type, State) when is_integer(Int) ->
+    encode(Int * 1.0, Type, State);
+encode(N, float4, _) when is_float(N) ->
+    <<N:1/big-float-unit:32>>;
+encode(N, float8, _) when is_float(N) ->
+    <<N:1/big-float-unit:64>>;
+encode(nan, float4, _) -> ?NAN;
+encode(nan, float8, _) -> ?NAN8;
+encode(plus_infinity, float4, _) -> ?POS_INF;
+encode(plus_infinity, float8, _) -> ?POS_INF8;
+encode(minus_infinity, float4, _) -> ?NEG_INF;
+encode(minus_infinity, float8, _) -> ?NEG_INF8.
+
+
+decode(<<N:1/big-float-unit:32>>, float4, _) -> N;
+decode(<<N:1/big-float-unit:64>>, float8, _) -> N;
+decode(?NAN_PATTERN, float4, _) -> nan;
+decode(?NAN_PATTERN8, float8, _) -> nan;
+decode(?POS_INF, float4, _) -> plus_infinity;
+decode(?POS_INF8, float8, _) -> plus_infinity;
+decode(?NEG_INF, float4, _) -> minus_infinity;
+decode(?NEG_INF8, float8, _) -> minus_infinity.

src/datatypes/epgsql_codec_geometric.erl

@@ -0,0 +1,30 @@
+%%% @doc
+%%% Codec for `point'.
+%%% https://www.postgresql.org/docs/current/static/datatype-geometric.html
+%%% $PG$/src/backend/utils/adt/geo_ops.c
+%%% XXX: it's not PostGIS!
+%%% @end
+%%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+%%% TODO: line, lseg, box, path, polygon, circle
+
+-module(epgsql_codec_geometric).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: point().
+-type point() :: {float(), float()}.
+
+init(_, _) -> [].
+
+names() ->
+    [point].
+
+encode({X, Y}, point, _) when is_number(X), is_number(Y) ->
+    %% XXX: looks like it doesn't have size prefix?!
+    <<X:1/big-float-unit:64, Y:1/big-float-unit:64>>.
+
+decode(<<X:1/big-float-unit:64, Y:1/big-float-unit:64>>, point, _) ->
+    {X, Y}.

src/datatypes/epgsql_codec_hstore.erl

@@ -0,0 +1,71 @@
+%%% @doc
+%%% Codec for `hstore' type.
+%%% https://www.postgresql.org/docs/current/static/hstore.html
+%%% XXX: hstore not a part of postgresql builtin datatypes, it's in contrib.
+%%% It should be enabled in postgresql by command
+%%% `CREATE EXTENSION hstore`
+%%% $PG$/contrib/hstore/
+%%% @end
+%%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_hstore).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-include("protocol.hrl").
+
+-export_type([data/0]).
+
+-type data() :: data_in() | data_out().
+
+-type key_in() :: list() | binary() | atom() | integer() | float().
+%% jiffy-style maps
+-type data_in() :: { [{key_in(), binary()}] }.
+-type data_out() :: { [{Key :: binary(), Value :: binary()}] }.
+
+%% TODO: option for output format: proplist | jiffy-object | map
+init(_, _) -> [].
+
+names() ->
+    [hstore].
+
+encode({Hstore}, hstore, _) when is_list(Hstore) ->
+    Size = length(Hstore),
+    Body = [[encode_key(K) | encode_value(V)]
+           || {K, V} <- Hstore],
+    [<<Size:?int32>> | Body].
+
+decode(<<Size:?int32, Elements/binary>>, hstore, _) ->
+    {do_decode(Size, Elements)}.
+
+
+encode_key(K) ->
+    encode_string(K).
+
+encode_value(null) ->
+    <<-1:?int32>>;
+encode_value(undefined) ->
+    <<-1:?int32>>;
+encode_value(V) ->
+    encode_string(V).
+
+encode_string(Str) when is_binary(Str) ->
+    <<(byte_size(Str)):?int32, Str/binary>>;
+encode_string(Str) when is_list(Str) ->
+    encode_string(list_to_binary(Str));
+encode_string(Str) when is_atom(Str) ->
+    encode_string(atom_to_binary(Str, utf8));
+encode_string(Str) when is_integer(Str) ->
+    encode_string(integer_to_binary(Str));
+encode_string(Str) when is_float(Str) ->
+    encode_string(io_lib:format("~w", [Str])).
+    %% encode_string(erlang:float_to_binary(Str)).
+
+
+do_decode(0, _) -> [];
+do_decode(N, <<KeyLen:?int32, Key:KeyLen/binary, -1:?int32, Rest/binary>>) ->
+    [{Key, null} | do_decode(N - 1, Rest)];
+do_decode(N, <<KeyLen:?int32, Key:KeyLen/binary,
+               ValLen:?int32, Value:ValLen/binary, Rest/binary>>) ->
+    [{Key, Value} | do_decode(N - 1, Rest)].

src/datatypes/epgsql_codec_integer.erl

@@ -0,0 +1,38 @@
+%%% @doc
+%%% Codec for `int2', `int4', `int8' (smallint, integer, bigint).
+%%% https://www.postgresql.org/docs/current/static/datatype-numeric.html#datatype-int
+%%% $PG$/src/backend/utils/adt/int.c
+%%% $PG$/src/backend/utils/adt/int8.c
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_integer).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+%% See table 8.2
+%% https://www.postgresql.org/docs/current/static/datatype-numeric.html
+-define(BIGINT_MAX, 16#7fffffffffffffff).  % 9223372036854775807, (2^63 - 1)
+-define(BIGINT_MIN, -16#7fffffffffffffff). % -9223372036854775807
+
+-type data() :: ?BIGINT_MIN..?BIGINT_MAX.
+
+
+init(_, _) -> [].
+
+names() ->
+    [int2, int4, int8].
+
+encode(N, int2, _) ->
+    <<N:1/big-signed-unit:16>>;
+encode(N, int4, _) ->
+    <<N:1/big-signed-unit:32>>;
+encode(N, int8, _) ->
+    <<N:1/big-signed-unit:64>>.
+
+decode(<<N:1/big-signed-unit:16>>, int2, _)    -> N;
+decode(<<N:1/big-signed-unit:32>>, int4, _)    -> N;
+decode(<<N:1/big-signed-unit:64>>, int8, _)    -> N.

src/datatypes/epgsql_codec_intrange.erl

@@ -0,0 +1,88 @@
+%%% @doc
+%%% Codec for `int4range', `int8range' types.
+%%% https://www.postgresql.org/docs/current/static/rangetypes.html#rangetypes-builtin
+%%% $PG$/src/backend/utils/adt/rangetypes.c
+%%% @end
+%%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+%%% TODO: universal range, based on pg_range table
+%%% TODO: inclusive/exclusive ranges `[]' `[)' `(]' `()'
+
+-module(epgsql_codec_intrange).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-include("protocol.hrl").
+
+-export_type([data/0]).
+
+-type data() :: {left(), right()}.
+
+-type left() :: minus_infinity | integer().
+-type right() :: plus_infinity | integer().
+
+
+init(_, _) -> [].
+
+names() ->
+    [int4range, int8range].
+
+encode(Range, int4range, _) ->
+    encode_int4range(Range);
+encode(Range, int8range, _) ->
+    encode_int8range(Range).
+
+decode(Bin, int4range, _) ->
+    decode_int4range(Bin);
+decode(Bin, int8range, _) ->
+    decode_int8range(Bin).
+
+
+encode_int4range({minus_infinity, plus_infinity}) ->
+    <<24:1/big-signed-unit:8>>;
+encode_int4range({From, plus_infinity}) ->
+    FromInt = to_int(From),
+    <<18:1/big-signed-unit:8, 4:?int32, FromInt:?int32>>;
+encode_int4range({minus_infinity, To}) ->
+    ToInt = to_int(To),
+    <<8:1/big-signed-unit:8, 4:?int32, ToInt:?int32>>;
+encode_int4range({From, To}) ->
+    FromInt = to_int(From),
+    ToInt = to_int(To),
+    <<2:1/big-signed-unit:8, 4:?int32, FromInt:?int32, 4:?int32, ToInt:?int32>>.
+
+encode_int8range({minus_infinity, plus_infinity}) ->
+    <<24:1/big-signed-unit:8>>;
+encode_int8range({From, plus_infinity}) ->
+    FromInt = to_int(From),
+    <<18:1/big-signed-unit:8, 8:?int32, FromInt:?int64>>;
+encode_int8range({minus_infinity, To}) ->
+    ToInt = to_int(To),
+    <<8:1/big-signed-unit:8, 8:?int32, ToInt:?int64>>;
+encode_int8range({From, To}) ->
+    FromInt = to_int(From),
+    ToInt = to_int(To),
+    <<2:1/big-signed-unit:8, 8:?int32, FromInt:?int64, 8:?int32, ToInt:?int64>>.
+
+to_int(N) when is_integer(N) -> N;
+to_int(S) when is_list(S) -> erlang:list_to_integer(S);
+to_int(B) when is_binary(B) -> erlang:binary_to_integer(B).
+
+
+decode_int4range(<<2:1/big-signed-unit:8, 4:?int32, From:?int32, 4:?int32, To:?int32>>) ->
+    {From, To};
+decode_int4range(<<8:1/big-signed-unit:8, 4:?int32, To:?int32>>) ->
+    {minus_infinity, To};
+decode_int4range(<<18:1/big-signed-unit:8, 4:?int32, From:?int32>>) ->
+    {From, plus_infinity};
+decode_int4range(<<24:1/big-signed-unit:8>>) ->
+    {minus_infinity, plus_infinity}.
+
+decode_int8range(<<2:1/big-signed-unit:8, 8:?int32, From:?int64, 8:?int32, To:?int64>>) ->
+    {From, To};
+decode_int8range(<<8:1/big-signed-unit:8, 8:?int32, To:?int64>>) ->
+    {minus_infinity, To};
+decode_int8range(<<18:1/big-signed-unit:8, 8:?int32, From:?int64>>) ->
+    {From, plus_infinity};
+decode_int8range(<<24:1/big-signed-unit:8>>) ->
+    {minus_infinity, plus_infinity}.

src/datatypes/epgsql_codec_json.erl

@@ -0,0 +1,35 @@
+%%% @doc
+%%% Codec for `json', `jsonb'
+%%% https://www.postgresql.org/docs/current/static/datatype-json.html
+%%% $PG$/src/backend/utils/adt/json.c // `json'
+%%% $PG$/src/backend/utils/adt/jsonb.c // `jsonb'
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_json).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: binary().
+
+-define(JSONB_VERSION_1, 1).
+
+%% TODO: JSON encode/decode `fun Mod:Name/1` / `{Mod, Name}` as option.
+%% Shall not pass `fun(_) -> .. end`, because of hot code upgrade problems.
+init(_, _) -> [].
+
+names() ->
+    [json, jsonb].
+
+encode(Bin, json, _) ->
+    Bin;
+encode(Bin, jsonb, _) ->
+    [<<?JSONB_VERSION_1:8>> | Bin].
+
+decode(Bin, json, _) ->
+    Bin;
+decode(<<?JSONB_VERSION_1:8, Bin/binary>>, jsonb, _) ->
+    Bin.

src/datatypes/epgsql_codec_net.erl

@@ -0,0 +1,63 @@
+%%% @doc
+%%% Codec for `inet', `cidr'
+%%% https://www.postgresql.org/docs/10/static/datatype-net-types.html
+%%% $PG$/src/backend/utils/adt/network.c
+%%%
+%%% TIP: use `inet:ntoa/1' to convert `ip()' to string.
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+%% TODO: `macaddr', `macaddr8' `mac.c`
+-module(epgsql_codec_net).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: ip() | ip_mask().
+
+-type ip() :: inet:ip_address().
+-type mask() :: 0..32.
+-type ip_mask() :: {ip(), mask()}.
+
+-define(INET, 2).
+-define(INET6, 3).
+-define(IP_SIZE, 4).
+-define(IP6_SIZE, 16).
+-define(MAX_IP_MASK, 32).
+-define(MAX_IP6_MASK, 128).
+
+init(_, _) -> [].
+
+names() ->
+    [inet, cidr].
+
+encode(IpMask, _, _) ->
+    encode_net(IpMask).
+
+decode(Bin, _, _) ->
+    decode_net(Bin).
+
+-spec encode_net(data()) -> binary().
+encode_net({{_, _, _, _} = IP, Mask}) ->
+    Bin = list_to_binary(tuple_to_list(IP)),
+    <<?INET, Mask:8, 1, ?IP_SIZE, Bin/binary>>;
+encode_net({{_, _, _, _, _, _, _, _} = IP, Mask}) ->
+    Bin = << <<X:16>> || X <- tuple_to_list(IP) >>,
+    <<?INET6, Mask:8, 1, ?IP6_SIZE, Bin/binary>>;
+encode_net({_, _, _, _} = IP) ->
+    Bin = list_to_binary(tuple_to_list(IP)),
+    <<?INET, ?MAX_IP_MASK, 0, ?IP_SIZE, Bin/binary>>;
+encode_net({_, _, _, _, _, _, _, _} = IP) ->
+    Bin = << <<X:16>> || X <- tuple_to_list(IP) >>,
+    <<?INET6, ?MAX_IP6_MASK, 0, ?IP6_SIZE, Bin/binary>>.
+
+-spec decode_net(binary()) -> data().
+decode_net(<<?INET, Mask:8, 1, ?IP_SIZE, Bin/binary>>) ->
+    {list_to_tuple(binary_to_list(Bin)), Mask};
+decode_net(<<?INET6, Mask:8, 1, ?IP6_SIZE, Bin/binary>>) ->
+    {list_to_tuple([X || <<X:16>> <= Bin]), Mask};
+decode_net(<<?INET, ?MAX_IP_MASK, 0, ?IP_SIZE, Bin/binary>>) ->
+    list_to_tuple(binary_to_list(Bin));
+decode_net(<<?INET6, ?MAX_IP6_MASK, 0, ?IP6_SIZE, Bin/binary>>) ->
+    list_to_tuple([X || <<X:16>> <= Bin]).

src/datatypes/epgsql_codec_noop.erl

@@ -0,0 +1,23 @@
+%%% @doc
+%%% Dummy codec. Used internally
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_noop).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3, decode_text/3]).
+
+-export_type([data/0]).
+
+-type data() :: binary().
+
+init(_, _) -> [].
+
+names() -> [].
+
+encode(Bin, _, _) when is_binary(Bin) -> Bin.
+
+decode(Bin, _, _) -> Bin.
+
+decode_text(Bin, _, _) -> Bin.

src/datatypes/epgsql_codec_postgis.erl

@@ -0,0 +1,27 @@
+%%% @doc
+%%% Codec for `geometry' PostGIS umbrella datatype.
+%%% http://postgis.net/docs/manual-2.4/geometry.html
+%%% $POSTGIS$/postgis/lwgeom_inout.c
+%%% @end
+%%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_postgis).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: point().
+-type point() :: {}.
+
+init(_, _) -> [].
+
+names() ->
+    [geometry].
+
+encode(Geo, geometry, _) ->
+    ewkb:encode_geometry(Geo).
+
+decode(Bin, geometry, _) ->
+    ewkb:decode_geometry(Bin).

src/datatypes/epgsql_codec_text.erl

@@ -0,0 +1,30 @@
+%%% @doc
+%%% Codec for `text', `varchar', `bytea'.
+%%% For 'char' see epgsql_codec_bpchar.erl.
+%%% https://www.postgresql.org/docs/10/static/datatype-character.html
+%%% $PG$/src/backend/utils/adt/varchar.c
+%%% $PG$/src/backend/utils/adt/varlena.c
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_text).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: in_data() | out_data().
+-type in_data() :: binary() | string().
+-type out_data() :: binary().
+
+init(_, _) -> [].
+
+names() ->
+    [text, varchar, bytea].
+
+encode(String, Name, State) when is_list(String) ->
+    encode(list_to_binary(String), Name, State);
+encode(Bin, _, _) when is_binary(Bin) -> Bin.
+
+decode(Bin, _, _) -> Bin.

src/datatypes/epgsql_codec_uuid.erl

@@ -0,0 +1,35 @@
+%%% @doc
+%%% Codec for `uuid' type.
+%%% Input expected to be in hex string, eg
+%%% `<<"550e8400-e29b-41d4-a716-446655440000">>'.
+%%% https://www.postgresql.org/docs/current/static/datatype-uuid.html
+%%% $PG$/src/backend/utils/adt/uuid.c
+%%% @end
+%%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec_uuid).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+-export_type([data/0]).
+
+-type data() :: in_data() | out_data().
+-type in_data() :: string() | binary().
+-type out_data() :: binary().
+
+init(_, _) -> [].
+
+names() ->
+    [uuid].
+
+encode(Uuid, uuid, St) when is_list(Uuid) ->
+    encode(list_to_binary(Uuid), uuid, St);
+encode(Uuid, uuid, _) when is_binary(Uuid) ->
+    Hex = binary:replace(Uuid, <<"-">>, <<>>, [global]),
+    Int = erlang:binary_to_integer(Hex, 16),
+    <<Int:128/big-unsigned-integer>>.
+
+decode(<<U0:32, U1:16, U2:16, U3:16, U4:48>>, uuid, _) ->
+    Format = "~8.16.0b-~4.16.0b-~4.16.0b-~4.16.0b-~12.16.0b",
+    iolist_to_binary(io_lib:format(Format, [U0, U1, U2, U3, U4])).

src/epgsql.erl

@@ -30,8 +30,11 @@
          to_proplist/1]).
 
 -export_type([connection/0, connect_option/0, connect_opts/0,
-              connect_error/0, query_error/0,
-              sql_query/0, column/0, bind_param/0, typed_param/0,
+              connect_error/0, query_error/0, sql_query/0, column/0,
+              type_name/0]).
+
+%% Deprecated types
+-export_type([bind_param/0, typed_param/0,
               squery_row/0, equery_row/0, reply/1,
               pg_time/0, pg_date/0, pg_datetime/0, pg_interval/0]).
 
@@ -76,6 +79,13 @@
       | invalid_password.
 -type query_error() :: #error{}.
 
+
+-type type_name() :: atom().
+
+
+
+
+
 %% Ranges are from https://www.postgresql.org/docs/current/static/datatype-datetime.html
 -type pg_date() ::
         {Year :: -4712..294276,
@@ -173,22 +183,20 @@ connect(C, Host, Username, Password, Opts0) ->
 
 -spec update_type_cache(connection()) -> ok.
 update_type_cache(C) ->
-    update_type_cache(C, [<<"hstore">>,<<"geometry">>]).
-
--spec update_type_cache(connection(), [binary()]) -> ok.
-update_type_cache(C, DynamicTypes) ->
-    Query = "SELECT typname, oid::int4, typarray::int4"
-            " FROM pg_type"
-            " WHERE typname = ANY($1::varchar[])",
-    case equery(C, Query, [DynamicTypes]) of
-        {ok, _, TypeInfos} ->
-            ok = gen_server:call(C, {update_type_cache, TypeInfos});
-        {error, {error, error, _, _,
-                 <<"column \"typarray\" does not exist in pg_type">>, _}} ->
-            %% Do not fail connect if pg_type table in not in the expected
-            %% format. Known to happen for Redshift which is based on PG v8.0.2
-            ok
-    end.
+    update_type_cache(C, [{epgsql_codec_hstore, []},
+                          {epgsql_codec_postgis, []}]).
+
+-spec update_type_cache(connection(), [{epgsql_codec:codec_mod(), Opts :: any()}]) ->
+                               {ok, [type_name()]} | {error, empty} | {error, query_error()}.
+update_type_cache(_C, []) ->
+    {error, empty};
+update_type_cache(C, Codecs) ->
+    %% {error, #error{severity = error,
+    %%                message = <<"column \"typarray\" does not exist in pg_type">>, _}}
+    %% Do not fail connect if pg_type table in not in the expected
+    %% format. Known to happen for Redshift which is based on PG v8.0.2
+    epgsql_sock:sync_command(C, epgsql_cmd_update_type_cache, Codecs).
+
 
 -spec close(connection()) -> ok.
 close(C) ->

src/epgsql_binary.erl

@@ -2,346 +2,296 @@
 
 -module(epgsql_binary).
 
--export([new_codec/1,
-         update_type_cache/2,
-         type2oid/2, oid2type/2,
-         encode/3, decode/3, supports/1]).
-
--export_type([codec/0]).
-
--record(codec, {
-    type2oid = [],
-    oid2type = []
-}).
+-export([new_codec/2,
+         update_codec/2,
+         type_to_oid/2,
+         typeinfo_to_name_array/2,
+         typeinfo_to_oid_info/2,
+         oid_to_name/2,
+         oid_to_info/2,
+         oid_to_decoder/3,
+         decode/2, encode/3, supports/2]).
+%% Composite type decoders
+-export([decode_record/3, decode_array/3]).
+
+-export_type([codec/0, decoder/0]).
 
 -include("protocol.hrl").
 
--opaque codec() :: #codec{}.
+-opaque codec() :: {module(), epgsql_oid_db:db()}.
+-opaque decoder() :: {fun((binary(), epgsql:type_name(), epgsql_codec:codec_state()) -> any()),
+                      epgsql:type_name(),
+                      epgsql_codec:state()}.
+
+-define(RECORD_OID, 2249).
+-define(RECORD_ARRAY_OID, 2287).
+
+-spec new_codec(module(), epgsql_sock:pg_sock()) -> codec().
+new_codec(OidDb, PgSock) ->
+    Codecs = default_codecs(),
+    Oids = default_oids(),
+    new_codec(OidDb, PgSock, Codecs, Oids).
+
+new_codec(OidDb, PgSock, Codecs, Oids) ->
+    CodecEntries = epgsql_codec:init_mods(Codecs, PgSock),
+    Types = OidDb:join_codecs_oids(Oids, CodecEntries),
+    {OidDb, OidDb:from_list(Types)}.
+
+-spec update_codec([epgsql_oid_db:type_info()], codec()) -> codec().
+update_codec(TypeInfos, {OidDb, Db}) ->
+    {OidDb, OidDb:update(TypeInfos, Db)}.
+
+-spec oid_to_name(epgsql_oid_db:oid(), codec()) -> Type | {unknown_oid, epgsql_oid_db:oid()} when
+      Type :: epgsql:type_name() | {array, epgsql:type_name()}.
+oid_to_name(Oid, Codec) ->
+    case oid_to_info(Oid, Codec) of
+        undefined ->
+            {unknown_oid, Oid};
+        Type ->
+            case epgsql_oid_db:type_to_oid_info(Type) of
+                {_, Name, true} -> {array, Name};
+                {_, Name, false} -> Name
+            end
+    end.
+
+type_to_oid({array, Name}, Codec) ->
+    type_to_oid(Name, true, Codec);
+type_to_oid(Name, Codec) ->
+    type_to_oid(Name, false, Codec).
+
+-spec type_to_oid(epgsql:type_name(), boolean(), codec()) -> epgsql_oid_db:oid().
+type_to_oid(TypeName, IsArray, {OidDb, Db}) ->
+    OidDb:oid_by_name(TypeName, IsArray, Db).
+
+type_to_oid_info({array, Name}, Codec) ->
+    type_to_info(Name, true, Codec);
+type_to_oid_info(Name, Codec) ->
+    type_to_info(Name, false, Codec).
+
+-spec oid_to_info(epgsql_oid_db:oid(), codec()) -> epgsql_oid_db:type_info().
+oid_to_info(Oid, {OidDb, Db}) ->
+    OidDb:find_by_oid(Oid, Db).
+
+-spec type_to_info(epgsql:type_name(), boolean(), codec()) -> epgsql_oid_db:type_info().
+type_to_info(TypeName, IsArray, {OidDb, Db}) ->
+    OidDb:find_by_name(TypeName, IsArray, Db).
+
+typeinfo_to_name_array({unknown_oid, _} = Unknown, _) -> Unknown;
+typeinfo_to_name_array(TypeInfo, {OidDb, _}) ->
+    case OidDb:type_to_oid_info(TypeInfo) of
+        {_, Name, false} -> Name;
+        {_, Name, true} -> {array, Name}
+    end.
 
--define(datetime, (get(datetime_mod))).
+typeinfo_to_oid_info({unknown_oid, _} = Unknown, _) -> Unknown;
+typeinfo_to_oid_info(TypeInfo, {OidDb, _}) ->
+    OidDb:type_to_oid_info(TypeInfo).
+
+%%
+%% Decode
+%%
+
+-spec decode(binary(), decoder()) -> any().
+decode(Bin, {Fun, TypeName, State}) ->
+    Fun(Bin, TypeName, State).
+
+-spec oid_to_decoder(epgsql_oid_db:oid(), binary | text, codec()) -> decoder().
+oid_to_decoder(?RECORD_OID, binary, Codec) ->
+    {fun ?MODULE:decode_record/3, record, Codec};
+oid_to_decoder(?RECORD_ARRAY_OID, binary, Codec) ->
+    %% See `make_array_decoder/3'
+    {fun ?MODULE:decode_array/3, [], oid_to_decoder(?RECORD_OID, binary, Codec)};
+oid_to_decoder(Oid, Format, {OidDb, Db}) ->
+    case OidDb:find_by_oid(Oid, Db) of
+        undefined when Format == binary ->
+            {fun epgsql_codec_noop:decode/3, undefined, []};
+        undefined when Format == text ->
+            {fun epgsql_codec_noop:decode_text/3, undefined, []};
+        Type ->
+            make_decoder(Type, Format, OidDb)
+    end.
 
--define(INET, 2).
--define(INET6, 3).
--define(IP_SIZE, 4).
--define(IP6_SIZE, 16).
--define(MAX_IP_MASK, 32).
--define(MAX_IP6_MASK, 128).
--define(JSONB_VERSION_1, 1).
+-spec make_decoder(epgsql_oid_db:type_info(), binary | text, module()) -> decoder().
+make_decoder(Type, Format, OidDb) ->
+    {Name, Mod, State} = OidDb:type_to_codec_entry(Type),
+    {_Oid, Name, IsArray} = OidDb:type_to_oid_info(Type),
+    make_decoder(Name, Mod, State, Format, IsArray).
+
+make_decoder(_Name, _Mod, _State, text, true) ->
+    %% Don't try to decode text arrays
+    {fun epgsql_codec_noop:decode_text/3, undefined, []};
+make_decoder(Name, Mod, State, text, false) ->
+    %% decode_text/3 is optional callback. If it's not defined, do NOOP.
+    case erlang:function_exported(Mod, decode_text, 3) of
+        true ->
+            {fun Mod:decode_text/3, Name, State};
+        false ->
+            {fun epgsql_codec_noop:decode_text/3, undefined, []}
+    end;
+make_decoder(Name, Mod, State, binary, true) ->
+    make_array_decoder(Name, Mod, State);
+make_decoder(Name, Mod, State, binary, false) ->
+    {fun Mod:decode/3, Name, State}.
+
+
+%% Array decoding
+%%% $PG$/src/backend/utils/adt/arrayfuncs.c
+make_array_decoder(Name, Mod, State) ->
+    {fun ?MODULE:decode_array/3, [], {fun Mod:decode/3, Name, State}}.
+
+decode_array(<<NDims:?int32, _HasNull:?int32, _Oid:?int32, Rest/binary>>, _, ElemDecoder) ->
+    %% 4b: n_dimensions;
+    %% 4b: flags;
+    %% 4b: Oid // should be the same as in column spec;
+    %%   (4b: n_elements;
+    %%    4b: lower_bound) * n_dimensions
+    %% (dynamic-size data)
+    %% Lower bound - eg, zero-bound or 1-bound or N-bound array. We ignore it, see
+    %% https://www.postgresql.org/docs/current/static/arrays.html#arrays-io
+    {Dims, Data} = erlang:split_binary(Rest, NDims * 2 * 4),
+    Lengths = [Len || <<Len:?int32, _LBound:?int32>> <= Dims],
+    {Array, <<>>} = decode_array1(Data, Lengths, ElemDecoder),
+    Array.
 
--spec new_codec(list()) -> codec().
-new_codec([]) -> #codec{}.
+decode_array1(Data, [], _)  ->
+    %% zero-dimensional array
+    {[], Data};
+decode_array1(Data, [Len], ElemDecoder) ->
+    %% 1-dimensional array
+    decode_elements(Data, [], Len, ElemDecoder);
+decode_array1(Data, [Len | T], ElemDecoder) ->
+    %% multidimensional array
+    F = fun(_N, Rest) -> decode_array1(Rest, T, ElemDecoder) end,
+    lists:mapfoldl(F, Data, lists:seq(1, Len)).
 
--spec update_type_cache(list(), codec()) -> codec().
-update_type_cache(TypeInfos, Codec) ->
-    Type2Oid = lists:flatmap(
-        fun({NameBin, ElementOid, ArrayOid}) ->
-            Name = erlang:binary_to_atom(NameBin, utf8),
-            [{Name, ElementOid}, {{array, Name}, ArrayOid}]
+decode_elements(Rest, Acc, 0, _ElDec) ->
+    {lists:reverse(Acc), Rest};
+decode_elements(<<-1:?int32, Rest/binary>>, Acc, N, ElDec) ->
+    decode_elements(Rest, [null | Acc], N - 1, ElDec);
+decode_elements(<<Len:?int32, Value:Len/binary, Rest/binary>>, Acc, N, ElemDecoder) ->
+    Value2 = decode(Value, ElemDecoder),
+    decode_elements(Rest, [Value2 | Acc], N - 1, ElemDecoder).
+
+
+
+%% Record decoding
+%% $PG$/src/backend/utils/adt/rowtypes.c
+decode_record(<<Size:?int32, Bin/binary>>, record, Codec) ->
+    list_to_tuple(decode_record1(Bin, Size, Codec)).
+
+decode_record1(<<>>, 0, _Codec) -> [];
+decode_record1(<<_Type:?int32, -1:?int32, Rest/binary>>, Size, Codec) ->
+    [null | decode_record1(Rest, Size - 1, Codec)];
+decode_record1(<<Oid:?int32, Len:?int32, ValueBin:Len/binary, Rest/binary>>, Size, {OidDb, Db} = Codec) ->
+    Value =
+        case OidDb:find_by_oid(Oid, Db) of
+            undefined -> ValueBin;
+            Type ->
+                {Name, Mod, State} = OidDb:type_to_codec_entry(Type),
+                Mod:decode(ValueBin, Name, State)
         end,
-        TypeInfos),
-    Oid2Type = [{Oid, Type} || {Type, Oid} <- Type2Oid],
-    Codec#codec{type2oid = Type2Oid, oid2type = Oid2Type}.
-
--spec oid2type(integer(), codec()) -> Type | {unknown_oid, integer()} when
-      Type :: atom() | {array, atom()}.
-oid2type(Oid, #codec{oid2type = Oid2Type}) ->
-    case epgsql_types:oid2type(Oid) of
-        {unknown_oid, _} ->
-            proplists:get_value(Oid, Oid2Type, {unknown_oid, Oid});
-        Type -> Type
+    [Value | decode_record1(Rest, Size - 1, Codec)].
+
+
+%%
+%% Encode
+%%
+-spec encode(epgsql:type_name() | {array, epgsql:type_name()}, any(), codec()) ->
+                    {error, unsupported} | iolist().
+encode(TypeName, Value, {OidDb, _Db} = Codec) ->
+    case type_to_oid_info(TypeName, Codec) of
+        undefined -> {error, unsupported};
+        Type ->
+            encode_with_type(Type, Value, OidDb)
     end.
 
--spec type2oid(Type, codec()) -> integer() | {unknown_type, Type} when
-      Type :: atom() | {array, atom()}.
-type2oid(Type, #codec{type2oid = Type2Oid}) ->
-    case epgsql_types:type2oid(Type) of
-        {unknown_type, _} ->
-            proplists:get_value(Type, Type2Oid, {unknown_type, Type});
-        Oid -> Oid
+encode_with_type(Type, Value, OidDb) ->
+    {Name, Mod, State} = OidDb:type_to_codec_entry(Type),
+    case OidDb:type_to_oid_info(Type) of
+        {_ArrayOid, _, true} ->
+            %FIXME: check if this OID is the same as was returned by 'Describe'
+            ElementOid = OidDb:type_to_element_oid(Type),
+            encode_array(Value, ElementOid, {Mod, Name, State});
+        {_Oid, _, false} ->
+            encode_value(Value, {Mod, Name, State})
     end.
 
-encode(_Any, null, _)                       -> <<-1:?int32>>;
-encode(_Any, undefined, _)                  -> <<-1:?int32>>;
-encode(bool, true, _)                       -> <<1:?int32, 1:1/big-signed-unit:8>>;
-encode(bool, false, _)                      -> <<1:?int32, 0:1/big-signed-unit:8>>;
-encode(int2, N, _)                          -> <<2:?int32, N:1/big-signed-unit:16>>;
-encode(int4, N, _)                          -> <<4:?int32, N:1/big-signed-unit:32>>;
-encode(int8, N, _)                          -> <<8:?int32, N:1/big-signed-unit:64>>;
-encode(float4, N, _)                        -> <<4:?int32, N:1/big-float-unit:32>>;
-encode(float8, N, _)                        -> <<8:?int32, N:1/big-float-unit:64>>;
-encode(bpchar, C, _) when is_integer(C)     -> <<1:?int32, C:1/big-unsigned-unit:8>>;
-encode(bpchar, B, _) when is_binary(B)      -> <<(byte_size(B)):?int32, B/binary>>;
-encode(time = Type, B, _)                   -> ?datetime:encode(Type, B);
-encode(timetz = Type, B, _)                 -> ?datetime:encode(Type, B);
-encode(date = Type, B, _)                   -> ?datetime:encode(Type, B);
-encode(timestamp = Type, B, _)              -> ?datetime:encode(Type, B);
-encode(timestamptz = Type, B, _)            -> ?datetime:encode(Type, B);
-encode(interval = Type, B, _)               -> ?datetime:encode(Type, B);
-encode(bytea, B, _) when is_binary(B)       -> <<(byte_size(B)):?int32, B/binary>>;
-encode(text, B, _) when is_binary(B)        -> <<(byte_size(B)):?int32, B/binary>>;
-encode(varchar, B, _) when is_binary(B)     -> <<(byte_size(B)):?int32, B/binary>>;
-encode(json, B, _) when is_binary(B)        -> <<(byte_size(B)):?int32, B/binary>>;
-encode(jsonb, B, _) when is_binary(B)       -> <<(byte_size(B) + 1):?int32, ?JSONB_VERSION_1:8, B/binary>>;
-encode(uuid, B, _) when is_binary(B)        -> encode_uuid(B);
-encode({array, char}, L, Codec) when is_list(L) -> encode_array(bpchar, type2oid(bpchar, Codec), L, Codec);
-encode({array, Type}, L, Codec) when is_list(L) -> encode_array(Type, type2oid(Type, Codec), L, Codec);
-encode(hstore, {L}, _) when is_list(L)      -> encode_hstore(L);
-encode(point, {X,Y}, _)                     -> encode_point({X,Y});
-encode(geometry, Data, _)                   -> encode_geometry(Data);
-encode(cidr, B, Codec)                      -> encode(bytea, encode_net(B), Codec);
-encode(inet, B, Codec)                      -> encode(bytea, encode_net(B), Codec);
-encode(int4range, R, _) when is_tuple(R)    -> encode_int4range(R);
-encode(int8range, R, _) when is_tuple(R)    -> encode_int8range(R);
-encode(Type, L, Codec) when is_list(L)      -> encode(Type, list_to_binary(L), Codec);
-encode(_Type, _Value, _)                    -> {error, unsupported}.
-
-decode(bool, <<1:1/big-signed-unit:8>>, _)     -> true;
-decode(bool, <<0:1/big-signed-unit:8>>, _)     -> false;
-decode(bpchar, <<C:1/big-unsigned-unit:8>>, _) -> C;
-decode(int2, <<N:1/big-signed-unit:16>>, _)    -> N;
-decode(int4, <<N:1/big-signed-unit:32>>, _)    -> N;
-decode(int8, <<N:1/big-signed-unit:64>>, _)    -> N;
-decode(float4, <<N:1/big-float-unit:32>>, _)   -> N;
-decode(float8, <<N:1/big-float-unit:64>>, _)   -> N;
-decode(record, <<_:?int32, Rest/binary>>, Codec) -> list_to_tuple(decode_record(Rest, [], Codec));
-decode(jsonb, <<?JSONB_VERSION_1:8, Value/binary>>, _) -> Value;
-decode(time = Type, B, _)                      -> ?datetime:decode(Type, B);
-decode(timetz = Type, B, _)                    -> ?datetime:decode(Type, B);
-decode(date = Type, B, _)                      -> ?datetime:decode(Type, B);
-decode(timestamp = Type, B, _)                 -> ?datetime:decode(Type, B);
-decode(timestamptz = Type, B, _)               -> ?datetime:decode(Type, B);
-decode(interval = Type, B, _)                  -> ?datetime:decode(Type, B);
-decode(uuid, B, _)                             -> decode_uuid(B);
-decode(hstore, Hstore, _)                      -> decode_hstore(Hstore);
-decode(inet, B, _)                             -> decode_net(B);
-decode(cidr, B, _)                             -> decode_net(B);
-decode({array, _Type}, B, Codec)               -> decode_array(B, Codec);
-decode(point, B, _)                            -> decode_point(B);
-decode(geometry, B, _)                         -> ewkb:decode_geometry(B);
-decode(int4range, B, _)                        -> decode_int4range(B);
-decode(int8range, B, _)                        -> decode_int8range(B);
-decode(_Other, Bin, _)                         -> Bin.
-
-encode_array(Type, Oid, A, Codec) ->
-    {Data, {NDims, Lengths}} = encode_array(Type, A, 0, [], Codec),
+encode_value(Value, {Mod, Name, State}) ->
+    Payload = Mod:encode(Value, Name, State),
+    [<<(iolist_size(Payload)):?int32>> | Payload].
+
+
+%% Number of dimensions determined at encode-time by introspection of data, so,
+%% we can't encode array of lists (eg. strings).
+encode_array(Array, Oid, ValueEncoder) ->
+    {Data, {NDims, Lengths}} = encode_array(Array, 0, [], ValueEncoder),
     Lens = [<<N:?int32, 1:?int32>> || N <- lists:reverse(Lengths)],
     Hdr  = <<NDims:?int32, 0:?int32, Oid:?int32>>,
-    Bin  = iolist_to_binary([Hdr, Lens, Data]),
-    <<(byte_size(Bin)):?int32, Bin/binary>>.
+    Payload  = [Hdr, Lens, Data],
+    [<<(iolist_size(Payload)):?int32>> | Payload].
 
-encode_array(_Type, [], NDims, Lengths, _Codec) ->
-    {<<>>, {NDims, Lengths}};
-encode_array(Type, [H | _] = Array, NDims, Lengths, Codec) when not is_list(H) ->
-    F = fun(E, Len) -> {encode(Type, E, Codec), Len + 1} end,
-    {Data, Len} = lists:mapfoldl(F, 0, Array),
-    {Data, {NDims + 1, [Len | Lengths]}};
-encode_array(uuid, [_H | _] = Array, NDims, Lengths, Codec) ->
-    F = fun(E, Len) -> {encode(uuid, E, Codec), Len + 1} end,
+encode_array([], NDims, Lengths, _Codec) ->
+    {[], {NDims, Lengths}};
+encode_array([H | _] = Array, NDims, Lengths, ValueEncoder) when not is_list(H) ->
+    F = fun(E, Len) -> {encode_value(E, ValueEncoder), Len + 1} end,
     {Data, Len} = lists:mapfoldl(F, 0, Array),
     {Data, {NDims + 1, [Len | Lengths]}};
-encode_array(Type, Array, NDims, Lengths, Codec) ->
+encode_array(Array, NDims, Lengths, Codec) ->
     Lengths2 = [length(Array) | Lengths],
-    F = fun(A2, {_NDims, _Lengths}) -> encode_array(Type, A2, NDims, Lengths2, Codec) end,
+    F = fun(A2, {_NDims, _Lengths}) -> encode_array(A2, NDims, Lengths2, Codec) end,
     {Data, {NDims2, Lengths3}} = lists:mapfoldl(F, {NDims, Lengths2}, Array),
     {Data, {NDims2 + 1, Lengths3}}.
 
-encode_uuid(U) when is_binary(U) ->
-    encode_uuid(binary_to_list(U));
-encode_uuid(U) ->
-    Hex = [H || H <- U, H =/= $-],
-    {ok, [Int], _} = io_lib:fread("~16u", Hex),
-    <<16:?int32,Int:128>>.
-
-encode_hstore(HstoreEntries) ->
-    Body = << <<(encode_hstore_entry(Entry))/binary>> || Entry <- HstoreEntries >>,
-    <<(byte_size(Body) + 4):?int32, (length(HstoreEntries)):?int32, Body/binary>>.
-
-encode_hstore_entry({Key, Value}) ->
-    <<(encode_hstore_key(Key))/binary, (encode_hstore_value(Value))/binary>>.
-
-encode_hstore_key(Key) -> encode_hstore_string(Key).
-
-encode_hstore_value(null)      -> <<-1:?int32>>;
-encode_hstore_value(undefined) -> <<-1:?int32>>;
-encode_hstore_value(Val)       -> encode_hstore_string(Val).
-
-encode_hstore_string(Str) when is_list(Str) -> encode_hstore_string(list_to_binary(Str));
-encode_hstore_string(Str) when is_atom(Str) -> encode_hstore_string(atom_to_binary(Str, utf8));
-encode_hstore_string(Str) when is_integer(Str) ->
-    %% FIXME - we can use integer_to_binary when we deprecate R15
-    encode_hstore_string(list_to_binary(integer_to_list(Str)));
-encode_hstore_string(Str) when is_float(Str) ->
-    encode_hstore_string(iolist_to_binary(io_lib:format("~w", [Str])));
-encode_hstore_string(Str) when is_binary(Str) -> <<(byte_size(Str)):?int32, Str/binary>>.
-
-encode_net({{_, _, _, _} = IP, Mask}) ->
-    Bin = list_to_binary(tuple_to_list(IP)),
-    <<?INET, Mask, 1, ?IP_SIZE, Bin/binary>>;
-encode_net({{_, _, _, _, _, _, _, _} = IP, Mask}) ->
-    Bin = << <<X:16>> || X <- tuple_to_list(IP) >>,
-    <<?INET6, Mask, 1, ?IP6_SIZE, Bin/binary>>;
-encode_net({_, _, _, _} = IP) ->
-    Bin = list_to_binary(tuple_to_list(IP)),
-    <<?INET, ?MAX_IP_MASK, 0, ?IP_SIZE, Bin/binary>>;
-encode_net({_, _, _, _, _, _, _, _} = IP) ->
-    Bin = << <<X:16>> || X <- tuple_to_list(IP) >>,
-    <<?INET6, ?MAX_IP6_MASK, 0, ?IP6_SIZE, Bin/binary>>.
-
-decode_array(<<NDims:?int32, _HasNull:?int32, Oid:?int32, Rest/binary>>, Codec) ->
-    {Dims, Data} = erlang:split_binary(Rest, NDims * 2 * 4),
-    Lengths = [Len || <<Len:?int32, _LBound:?int32>> <= Dims],
-    Type = oid2type(Oid, Codec),
-    {Array, <<>>} = decode_array(Data, Type, Lengths, Codec),
-    Array.
 
-decode_array(Data, _Type, [], _Codec)  ->
-    {[], Data};
-decode_array(Data, Type, [Len], Codec) ->
-    decode_elements(Data, Type, [], Len, Codec);
-decode_array(Data, Type, [Len | T], Codec) ->
-    F = fun(_N, Rest) -> decode_array(Rest, Type, T, Codec) end,
-    lists:mapfoldl(F, Data, lists:seq(1, Len)).
-
-decode_elements(Rest, _Type, Acc, 0, _Codec) ->
-    {lists:reverse(Acc), Rest};
-decode_elements(<<-1:?int32, Rest/binary>>, Type, Acc, N, Codec) ->
-    decode_elements(Rest, Type, [null | Acc], N - 1, Codec);
-decode_elements(<<Len:?int32, Value:Len/binary, Rest/binary>>, Type, Acc, N, Codec) ->
-    Value2 = decode(Type, Value, Codec),
-    decode_elements(Rest, Type, [Value2 | Acc], N - 1, Codec).
-
-decode_record(<<>>, Acc, _Codec) ->
-    lists:reverse(Acc);
-decode_record(<<_Type:?int32, -1:?int32, Rest/binary>>, Acc, Codec) ->
-    decode_record(Rest, [null | Acc], Codec);
-decode_record(<<Type:?int32, Len:?int32, Value:Len/binary, Rest/binary>>, Acc, Codec) ->
-    Value2 = decode(oid2type(Type, Codec), Value, Codec),
-    decode_record(Rest, [Value2 | Acc], Codec).
-
-decode_uuid(<<U0:32, U1:16, U2:16, U3:16, U4:48>>) ->
-    Format = "~8.16.0b-~4.16.0b-~4.16.0b-~4.16.0b-~12.16.0b",
-    iolist_to_binary(io_lib:format(Format, [U0, U1, U2, U3, U4])).
-
-decode_hstore(<<NumElements:?int32, Elements/binary>>) ->
-    {decode_hstore1(NumElements, Elements, [])}.
-
-decode_hstore1(0, _Elements, Acc) -> Acc;
-decode_hstore1(N, <<KeyLen:?int32, Key:KeyLen/binary, -1:?int32, Rest/binary>>, Acc) ->
-    decode_hstore1(N - 1, Rest, [{Key, null} | Acc]);
-decode_hstore1(N, <<KeyLen:?int32, Key:KeyLen/binary, ValLen:?int32, Value:ValLen/binary, Rest/binary>>, Acc) ->
-    decode_hstore1(N - 1, Rest, [{Key, Value} | Acc]).
-
-encode_point({X, Y}) when is_number(X), is_number(Y) ->
-    <<X:1/big-float-unit:64, Y:1/big-float-unit:64>>.
-
-decode_point(<<X:1/big-float-unit:64, Y:1/big-float-unit:64>>) ->
-    {X, Y}.
-
-encode_geometry(Data) ->
-    Bin = ewkb:encode_geometry(Data),
-    Size = byte_size(Bin),
-    <<Size:?int32, Bin/binary>>.
-
-decode_net(<<?INET, Mask, 1, ?IP_SIZE, Bin/binary>>) ->
-    {list_to_tuple(binary_to_list(Bin)), Mask};
-decode_net(<<?INET6, Mask, 1, ?IP6_SIZE, Bin/binary>>) ->
-    {list_to_tuple([X || <<X:16>> <= Bin]), Mask};
-decode_net(<<?INET, ?MAX_IP_MASK, 0, ?IP_SIZE, Bin/binary>>) ->
-    list_to_tuple(binary_to_list(Bin));
-decode_net(<<?INET6, ?MAX_IP6_MASK, 0, ?IP6_SIZE, Bin/binary>>) ->
-    list_to_tuple([X || <<X:16>> <= Bin]).
-
-%% @doc encode an int4range
-encode_int4range({minus_infinity, plus_infinity}) ->
-    <<1:?int32, 24:1/big-signed-unit:8>>;
-encode_int4range({From, plus_infinity}) ->
-    FromInt = to_int(From),
-    <<9:?int32, 18:1/big-signed-unit:8, 4:?int32, FromInt:?int32>>;
-encode_int4range({minus_infinity, To}) ->
-    ToInt = to_int(To),
-    <<9:?int32, 8:1/big-signed-unit:8, 4:?int32, ToInt:?int32>>;
-encode_int4range({From, To}) ->
-    FromInt = to_int(From),
-    ToInt = to_int(To),
-    <<17:?int32, 2:1/big-signed-unit:8, 4:?int32, FromInt:?int32, 4:?int32, ToInt:?int32>>.
-
-%% @doc encode an int8range
-encode_int8range({minus_infinity, plus_infinity}) ->
-    <<1:?int32, 24:1/big-signed-unit:8>>;
-encode_int8range({From, plus_infinity}) ->
-    FromInt = to_int(From),
-    <<13:?int32, 18:1/big-signed-unit:8, 8:?int32, FromInt:?int64>>;
-encode_int8range({minus_infinity, To}) ->
-    ToInt = to_int(To),
-    <<13:?int32, 8:1/big-signed-unit:8, 8:?int32, ToInt:?int64>>;
-encode_int8range({From, To}) ->
-    FromInt = to_int(From),
-    ToInt = to_int(To),
-    <<25:?int32, 2:1/big-signed-unit:8, 8:?int32, FromInt:?int64, 8:?int32, ToInt:?int64>>.
-
-to_int(N) when is_integer(N) -> N;
-to_int(S) when is_list(S) -> erlang:list_to_integer(S);
-to_int(B) when is_binary(B) -> erlang:binary_to_integer(B).
-
-%% @doc decode an int4range
-decode_int4range(<<2:1/big-signed-unit:8, 4:?int32, From:?int32, 4:?int32, To:?int32>>) -> {From, To};
-decode_int4range(<<8:1/big-signed-unit:8, 4:?int32, To:?int32>>) -> {minus_infinity, To};
-decode_int4range(<<18:1/big-signed-unit:8, 4:?int32, From:?int32>>) -> {From, plus_infinity};
-decode_int4range(<<24:1/big-signed-unit:8>>) -> {minus_infinity, plus_infinity}.
-
-%% @doc decode an int8range
-decode_int8range(<<2:1/big-signed-unit:8, 8:?int32, From:?int64, 8:?int32, To:?int64>>) -> {From, To};
-decode_int8range(<<8:1/big-signed-unit:8, 8:?int32, To:?int64>>) -> {minus_infinity, To};
-decode_int8range(<<18:1/big-signed-unit:8, 8:?int32, From:?int64>>) -> {From, plus_infinity};
-decode_int8range(<<24:1/big-signed-unit:8>>) -> {minus_infinity, plus_infinity}.
-
-supports(bool)    -> true;
-supports(bpchar)  -> true;
-supports(int2)    -> true;
-supports(int4)    -> true;
-supports(int8)    -> true;
-supports(float4)  -> true;
-supports(float8)  -> true;
-supports(bytea)   -> true;
-supports(text)    -> true;
-supports(varchar) -> true;
-supports(record)  -> true;
-supports(date)    -> true;
-supports(time)    -> true;
-supports(timetz)  -> true;
-supports(timestamp)   -> true;
-supports(timestamptz) -> true;
-supports(interval)    -> true;
-supports(uuid)        -> true;
-supports(hstore)      -> true;
-supports(cidr)        -> true;
-supports(inet)        -> true;
-supports(geometry)    -> true;
-supports(point)       -> true;
-supports(json)        -> true;
-supports(jsonb)        -> true;
-supports({array, bool})   -> true;
-supports({array, int2})   -> true;
-supports({array, int4})   -> true;
-supports({array, int8})   -> true;
-supports({array, float4}) -> true;
-supports({array, float8}) -> true;
-supports({array, char})   -> true;
-supports({array, text})   -> true;
-supports({array, date})   -> true;
-supports({array, time})   -> true;
-supports({array, timetz}) -> true;
-supports({array, timestamp})     -> true;
-supports({array, timestamptz})   -> true;
-supports({array, interval})      -> true;
-supports({array, hstore})        -> true;
-supports({array, varchar}) -> true;
-supports({array, uuid})   -> true;
-supports({array, cidr})   -> true;
-supports({array, inet})   -> true;
-supports({array, record}) -> true;
-supports({array, json})   -> true;
-supports({array, jsonb})   -> true;
-supports(int4range)       -> true;
-supports(int8range)       -> true;
-supports(_Type)       -> false.
+%% Supports
+supports(RecOid, _) when RecOid == ?RECORD_OID; RecOid == ?RECORD_ARRAY_OID ->
+    true;
+supports(Oid, {OidDb, Db}) ->
+    OidDb:find_by_oid(Oid, Db) =/= undefined.
+
+%% Default codec set
+-spec default_codecs() -> [epgsql_codec:codec_entry()].
+default_codecs() ->
+    [{epgsql_codec_boolean,[]},
+     {epgsql_codec_bpchar,[]},
+     {epgsql_codec_datetime,[]},
+     {epgsql_codec_float,[]},
+     {epgsql_codec_geometric, []},
+     %% {epgsql_codec_hstore, []},
+     {epgsql_codec_integer,[]},
+     {epgsql_codec_intrange,[]},
+     {epgsql_codec_json,[]},
+     {epgsql_codec_net,[]},
+     %% {epgsql_codec_postgis,[]},
+     {epgsql_codec_text,[]},
+     {epgsql_codec_uuid,[]}].
+
+-spec default_oids() -> [epgsql_oid_db:oid_entry()].
+default_oids() ->
+    [{bool, 16, 1000},
+     {bpchar, 1042, 1014},
+     {bytea, 17, 1001},
+     {char, 18, 1002},
+     {cidr, 650, 651},
+     {date, 1082, 1182},
+     {float4, 700, 1021},
+     {float8, 701, 1022},
+     %% {geometry, 17063, 17071},
+     %% {hstore, 16935, 16940},
+     {inet, 869, 1041},
+     {int2, 21, 1005},
+     {int4, 23, 1007},
+     {int4range, 3904, 3905},
+     {int8, 20, 1016},
+     {int8range, 3926, 3927},
+     {interval, 1186, 1187},
+     {json, 114, 199},
+     {jsonb, 3802, 3807},
+     {point, 600, 1017},
+     {text, 25, 1009},
+     {time, 1083, 1183},
+     {timestamp, 1114, 1115},
+     {timestamptz, 1184, 1185},
+     {timetz, 1266, 1270},
+     {uuid, 2950, 2951},
+     {varchar, 1043, 1015}].

src/epgsql_codec.erl

@@ -0,0 +1,69 @@
+%%% @doc
+%%% Behaviour for postgresql datatype codecs.
+%%% XXX: this module and callbacks "know nothing" about OIDs.
+%%% XXX: state of codec shouldn't leave epgsql_sock process. If you need to
+%%% return "pointer" to data type/codec, it's better to return OID or type name.
+%%% @end
+%%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
+
+-module(epgsql_codec).
+-export([init_mods/2]).
+
+-export_type([codec_state/0, codec_mod/0]).
+
+%%
+%% Behaviour
+%%
+-type codec_state() :: any().
+-type codec_mod() :: module().
+
+-optional_callbacks([decode_text/3]).
+
+%% Called on connection start-up
+-callback init(any(), epgsql_sock:pg_sock()) -> codec_state().
+
+%% List of supported type names
+-callback names() -> [epgsql:type_name()].
+
+%% Encode Erlang representation to PG binary
+%% Called for each parameter, binary protocol (equery)
+-callback encode(Cell :: any(), epgsql:type_name(), codec_state()) -> iodata().
+
+%% Decode PG binary to erlang representation
+%% Called for each cell in each row, binary protocol (equery)
+-callback decode(Cell :: binary(), epgsql:type_name(), codec_state()) -> any().
+
+%% Decode PG string representation (text protocol) to erlang term.
+%% Called for each cell in each row, text protocol (squery)
+-callback decode_text(Cell :: binary(), epgsql:type_name(), codec_state()) ->
+    any().
+
+%% ==========
+-type codec_entry() :: {epgsql:type_name(),
+                        Mod :: codec_mod(),
+                        CallbackState :: any()}.
+
+-spec init_mods([{codec_mod(), any()}], epgsql_sock:pg_sock()) ->
+                       ordsets:ordset(codec_entry()).
+init_mods(Codecs, PgSock) ->
+    ModState = [{Mod, Mod:init(Opts, PgSock)} || {Mod, Opts} <- Codecs],
+    build_mapping(ModState, sets:new(), []).
+
+build_mapping([{Mod, _State} = MS | ModStates], Set, Acc) ->
+    Names = Mod:names(),
+    {Set1, Acc1} = add_names(Names, MS, Set, Acc),
+    build_mapping(ModStates, Set1, Acc1);
+build_mapping([], _, Acc) ->
+    ordsets:from_list(Acc).
+
+add_names([Name | Names], {Mod, State} = MS, Set, Acc) ->
+    case sets:is_element(Name, Set) of
+        true ->
+            add_names(Names, MS, Set, Acc);
+        false ->
+            Set1 = sets:add_element(Name, Set),
+            Acc1 = [{Name, Mod, State} | Acc],
+            add_names(Names, MS, Set1, Acc1)
+    end;
+add_names([], _, Set, Acc) ->
+    {Set, Acc}.

src/epgsql_fdatetime.erl

@@ -14,57 +14,21 @@
 -define(secs_per_hour, 3600.0).
 -define(secs_per_minute, 60.0).
 
-decode(date, <<J:1/big-signed-unit:32>>)             -> j2date(?postgres_epoc_jdate + J);
+decode(date, <<J:1/big-signed-unit:32>>)             -> epgsql_idatetime:j2date(?postgres_epoc_jdate + J);
 decode(time, <<N:1/big-float-unit:64>>)              -> f2time(N);
 decode(timetz, <<N:1/big-float-unit:64, TZ:?int32>>) -> {f2time(N), TZ};
 decode(timestamp, <<N:1/big-float-unit:64>>)         -> f2timestamp(N);
 decode(timestamptz, <<N:1/big-float-unit:64>>)       -> f2timestamp(N);
 decode(interval, <<N:1/big-float-unit:64, D:?int32, M:?int32>>) -> {f2time(N), D, M}.
 
-encode(date, D)         -> <<4:?int32, (date2j(D) - ?postgres_epoc_jdate):1/big-signed-unit:32>>;
-encode(time, T)         -> <<8:?int32, (time2f(T)):1/big-float-unit:64>>;
-encode(timetz, {T, TZ}) -> <<12:?int32, (time2f(T)):1/big-float-unit:64, TZ:?int32>>;
-encode(timestamp, TS = {_, _, _})   -> <<8:?int32, (now2f(TS)):1/big-float-unit:64>>;
-encode(timestamp, TS)   -> <<8:?int32, (timestamp2f(TS)):1/big-float-unit:64>>;
-encode(timestamptz, TS = {_, _, _})   -> <<8:?int32, (now2f(TS)):1/big-float-unit:64>>;
-encode(timestamptz, TS) -> <<8:?int32, (timestamp2f(TS)):1/big-float-unit:64>>;
-encode(interval, {T, D, M}) -> <<16:?int32, (time2f(T)):1/big-float-unit:64, D:?int32, M:?int32>>.
-
-j2date(N) ->
-    J = N + 32044,
-    Q1 = J div 146097,
-    Extra = (J - Q1 * 146097) * 4 + 3,
-    J2 = J + 60 + Q1 * 3 + Extra div 146097,
-    Q2 = J2 div 1461,
-    J3 = J2 - Q2 * 1461,
-    Y = J3 * 4 div 1461,
-    J4 = case Y of
-        0 -> ((J3 + 306) rem 366) + 123;
-        _ -> ((J3 + 305) rem 365) + 123
-    end,
-    Year = (Y + Q2 * 4) - 4800,
-    Q3 = J4 * 2141 div 65536,
-    Day = J4 - 7834 * Q3 div 256,
-    Month = (Q3 + 10) rem 12 + 1,
-    {Year, Month, Day}.
-
-date2j({Y, M, D}) ->
-    M2 = case M > 2 of
-        true ->
-            M + 1;
-        false ->
-            M + 13
-    end,
-    Y2 = case M > 2 of
-        true ->
-            Y + 4800;
-        false ->
-            Y + 4799
-    end,
-    C = Y2 div 100,
-    J1 = Y2 * 365 - 32167,
-    J2 = J1 + (Y2 div 4 - C + C div 4),
-    J2 + 7834 * M2 div 256 + D.
+encode(date, D)         -> <<(epgsql_idatetime:date2j(D) - ?postgres_epoc_jdate):1/big-signed-unit:32>>;
+encode(time, T)         -> <<(time2f(T)):1/big-float-unit:64>>;
+encode(timetz, {T, TZ}) -> <<(time2f(T)):1/big-float-unit:64, TZ:?int32>>;
+encode(timestamp, TS = {_, _, _})   -> <<(now2f(TS)):1/big-float-unit:64>>;
+encode(timestamp, TS)   -> <<(timestamp2f(TS)):1/big-float-unit:64>>;
+encode(timestamptz, TS = {_, _, _})   -> <<(now2f(TS)):1/big-float-unit:64>>;
+encode(timestamptz, TS) -> <<(timestamp2f(TS)):1/big-float-unit:64>>;
+encode(interval, {T, D, M}) -> <<(time2f(T)):1/big-float-unit:64, D:?int32, M:?int32>>.
 
 f2time(N) ->
     {R1, Hour} = tmodulo(N, ?secs_per_hour),
@@ -86,7 +50,7 @@ f2timestamp(N) ->
 
 f2timestamp2(D, T) ->
     {_H, _M, S} = Time = f2time(T),
-    Date = j2date(D),
+    Date = epgsql_idatetime:j2date(D),
     case tsround(S - trunc(S)) of
         N when N >= 1.0 ->
             case ceiling(T) of
@@ -98,7 +62,7 @@ f2timestamp2(D, T) ->
     {Date, Time}.
 
 timestamp2f({Date, Time}) ->
-    D = date2j(Date) - ?postgres_epoc_jdate,
+    D = epgsql_idatetime:date2j(Date) - ?postgres_epoc_jdate,
     D * ?secs_per_day + time2f(Time).
 
 now2f({MegaSecs, Secs, MicroSecs}) ->

src/epgsql_idatetime.erl

@@ -3,6 +3,7 @@
 -module(epgsql_idatetime).
 
 -export([decode/2, encode/2]).
+-export([j2date/1, date2j/1]).
 
 -include("protocol.hrl").
 
@@ -24,15 +25,17 @@ decode(timestamp, <<N:?int64>>)                    -> i2timestamp(N);
 decode(timestamptz, <<N:?int64>>)                  -> i2timestamp(N);
 decode(interval, <<N:?int64, D:?int32, M:?int32>>) -> {i2time(N), D, M}.
 
-encode(date, D)         -> <<4:?int32, (date2j(D) - ?postgres_epoc_jdate):?int32>>;
-encode(time, T)         -> <<8:?int32, (time2i(T)):?int64>>;
-encode(timetz, {T, TZ}) -> <<12:?int32, (time2i(T)):?int64, TZ:?int32>>;
-encode(timestamp, TS = {_, _, _})   -> <<8:?int32, (now2i(TS)):?int64>>;
-encode(timestamp, TS)   -> <<8:?int32, (timestamp2i(TS)):?int64>>;
-encode(timestamptz, TS = {_, _, _})   -> <<8:?int32, (now2i(TS)):?int64>>;
-encode(timestamptz, TS) -> <<8:?int32, (timestamp2i(TS)):?int64>>;
-encode(interval, {T, D, M}) -> <<16:?int32, (time2i(T)):?int64, D:?int32, M:?int32>>.
-
+encode(date, D)         -> <<(date2j(D) - ?postgres_epoc_jdate):?int32>>;
+encode(time, T)         -> <<(time2i(T)):?int64>>;
+encode(timetz, {T, TZ}) -> <<(time2i(T)):?int64, TZ:?int32>>;
+encode(timestamp, TS = {_, _, _})   -> <<(now2i(TS)):?int64>>;
+encode(timestamp, TS)   -> <<(timestamp2i(TS)):?int64>>;
+encode(timestamptz, TS = {_, _, _})   -> <<(now2i(TS)):?int64>>;
+encode(timestamptz, TS) -> <<(timestamp2i(TS)):?int64>>;
+encode(interval, {T, D, M}) -> <<(time2i(T)):?int64, D:?int32, M:?int32>>.
+
+%% Julian calendar
+%% See $PG$/src/backend/utils/adt/datetime.c
 j2date(N) ->
     J = N + 32044,
     Q1 = J div 146097,

src/epgsql_oid_db.erl

@@ -0,0 +1,147 @@
+%%% @author Sergey Prokhorov <me@seriyps.ru>
+%%% @doc
+%%% Holds Oid <-> Type mappings (forward and reverse).
+%%% See https://www.postgresql.org/docs/current/static/catalog-pg-type.html
+%%% @end
+
+-module(epgsql_oid_db).
+
+-export([build_query/1, parse_rows/1, join_codecs_oids/2]).
+-export([from_list/1, to_list/1, update/2,
+         find_by_oid/2, find_by_name/3, oid_by_name/3,
+         type_to_codec_entry/1, type_to_oid_info/1, type_to_element_oid/1]).
+-export_type([oid/0, oid_info/0, oid_entry/0, type_info/0, db/0]).
+
+-record(type,
+        {oid :: oid(),
+         name :: epgsql:type_name(),
+         is_array :: boolean(),
+         array_element_oid :: oid(),
+         codec :: module(),
+         codec_state :: any()}).
+-record(oid_db,
+        {by_oid :: dict:dict(oid(), #type{}),
+         by_name :: dict:dict(epgsql:type_name(), #type{})}).
+
+-type oid() :: non_neg_integer().
+-type oid_entry() :: {epgsql:type_name(), Oid :: oid(), ArrayOid :: oid()}.
+-type oid_info() :: {Oid :: oid(), epgsql:type_name(), IsArray :: boolean()}.
+-opaque db() :: #oid_db{}.
+-opaque type_info() :: #type{}.
+
+-define(DICT, dict).
+-define(RECORD_OID, 2249).
+
+
+%%
+%% pg_type Data preparation
+%%
+
+%% @doc build query to fetch OID<->type_name information from PG server
+-spec build_query([epgsql:type_name() | binary()]) -> iolist().
+build_query(TypeNames) ->
+    %% TODO: lists:join/2, ERL 19+
+    %% XXX: we don't escape type names!
+    ToBin = fun(B) when is_binary(B) -> B;
+               (A) when is_atom(A) -> atom_to_binary(A, utf8)
+            end,
+    Types = join(",",
+                 [["'", ToBin(TypeName) | "'"]
+                  || TypeName <- TypeNames]),
+    [<<"SELECT typname, oid::int4, typarray::int4 "
+       "FROM pg_type "
+       "WHERE typname IN (">>, Types, <<") ORDER BY typname">>].
+
+%% Parse result of `squery(build_query(...))'
+-spec parse_rows(ordsets:ordset({binary(), binary(), binary()})) ->
+                        ordsets:ordset(oid_entry()).
+parse_rows(Rows) ->
+    [{binary_to_existing_atom(TypeName, utf8),
+      binary_to_integer(Oid),
+      binary_to_integer(ArrayOid)}
+     || {TypeName, Oid, ArrayOid} <- Rows].
+
+%% Build list of #type{}'s by merging oid and codec lists by type name.
+-spec join_codecs_oids(ordsets:ordset(oid_entry()),
+                       ordsets:ordset(epgsql_codec:codec_entry())) -> [#type{}].
+join_codecs_oids(Oids, Codecs) ->
+    do_join(lists:sort(Oids), lists:sort(Codecs)).
+
+do_join([{TypeName, Oid, ArrayOid} | Oids],
+        [{TypeName, CallbackMod, CallbackState} | Codecs]) ->
+    [#type{oid = Oid, name = TypeName, is_array = false,
+           codec = CallbackMod, codec_state = CallbackState},
+     #type{oid = ArrayOid, name = TypeName, is_array = true,
+           codec = CallbackMod, codec_state = CallbackState,
+           array_element_oid = Oid}
+     | do_join(Oids, Codecs)];
+do_join([OidEntry | _Oids] = Oids, [CodecEntry | Codecs])
+  when element(1, OidEntry) > element(1, CodecEntry) ->
+    %% This type isn't supported by PG server. That's ok, but not vice-versa.
+    do_join(Oids, Codecs);
+do_join([], _) ->
+    %% Codecs list may be not empty. See prev clause.
+    [].
+
+
+%%
+%% Storage API
+%%
+
+-spec from_list([type_info()]) -> db().
+from_list(Types) ->
+    #oid_db{by_oid = ?DICT:from_list(
+                       [{Oid, Type} || #type{oid = Oid} = Type <- Types]),
+            by_name = ?DICT:from_list(
+                        [{{Name, IsArray}, Oid}
+                         || #type{name = Name, is_array = IsArray, oid = Oid}
+                                <- Types])}.
+
+to_list(#oid_db{by_oid = Dict}) ->
+    [Type || {_Oid, Type} <- ?DICT:to_list(Dict)].
+
+-spec update([type_info()], db()) -> db().
+update(Types, #oid_db{by_oid = OldByOid, by_name = OldByName} = Store) ->
+    #oid_db{by_oid = NewByOid, by_name = NewByName} = from_list(Types),
+    ByOid = ?DICT:merge(fun(_, _, V2) -> V2 end, OldByOid, NewByOid),
+    ByName = ?DICT:merge(fun(_, _, V2) -> V2 end, OldByName, NewByName),
+    Store#oid_db{by_oid = ByOid,
+                 by_name = ByName}.
+
+-spec find_by_oid(oid(), db()) -> type_info() | undefined.
+%% find_by_oid(?RECORD_OID, _) ->
+%%     '$record';
+find_by_oid(Oid, #oid_db{by_oid = Dict}) ->
+    case ?DICT:find(Oid, Dict) of
+        {ok, Type} -> Type;
+        error -> undefined
+    end.
+
+-spec find_by_name(epgsql:type_name(), boolean(), db()) -> type_info().
+find_by_name(Name, IsArray, #oid_db{by_oid = ByOid} = Db) ->
+    Oid = oid_by_name(Name, IsArray, Db),
+    ?DICT:fetch(Oid, ByOid).                  % or maybe find_by_oid(Oid, Store)
+
+-spec oid_by_name(epgsql:type_name(), boolean(), db()) -> oid().
+oid_by_name(Name, IsArray, #oid_db{by_name = ByName}) ->
+    ?DICT:fetch({Name, IsArray}, ByName).
+
+-spec type_to_codec_entry(type_info()) -> epgsql_codec:codec_entry().
+type_to_codec_entry(#type{name = Name, codec = Codec, codec_state = State}) ->
+    {Name, Codec, State}.
+
+-spec type_to_oid_info(type_info()) -> oid_info().
+type_to_oid_info(#type{name = Name, is_array = IsArray, oid = Oid}) ->
+    {Oid, Name, IsArray}.
+
+-spec type_to_element_oid(type_info()) -> oid() | undefined.
+type_to_element_oid(#type{array_element_oid = ElementOid}) ->
+    ElementOid.
+
+%% Internal
+
+join(_Sep, []) -> [];
+join(Sep, [H | T]) -> [H | join_prepend(Sep, T)].
+
+join_prepend(_Sep, []) -> [];
+join_prepend(Sep, [H | T]) -> [Sep, H | join_prepend(Sep, T)].

src/epgsql_sock.erl

@@ -176,10 +176,6 @@ get_parameter_internal(Name, #state{parameters = Parameters}) ->
 init([]) ->
     {ok, #state{}}.
 
-handle_call({update_type_cache, TypeInfos}, _From, #state{codec = Codec} = State) ->
-    Codec2 = epgsql_binary:update_type_cache(TypeInfos, Codec),
-    {reply, ok, State#state{codec = Codec2}};
-
 handle_call({get_parameter, Name}, _From, State) ->
     {reply, {ok, get_parameter_internal(Name, State)}, State};
 

src/epgsql_wire.erl

@@ -18,11 +18,14 @@
          format/2,
          encode_parameters/2,
          encode_standby_status_update/3]).
+-export_type([row_decoder/0]).
 
 -include("epgsql.hrl").
 -include("protocol.hrl").
 
+-opaque row_decoder() :: {[epgsql_binary:decoder()], [epgsql:column()], epgsql_binary:codec()}.
 
+-spec decode_message(binary()) -> {byte(), binary(), binary()} | binary().
 decode_message(<<Type:8, Len:?int32, Rest/binary>> = Bin) ->
     Len2 = Len - 4,
     case Rest of
@@ -31,20 +34,22 @@ decode_message(<<Type:8, Len:?int32, Rest/binary>> = Bin) ->
         _Other ->
             Bin
     end;
-
 decode_message(Bin) ->
     Bin.
 
 %% decode a single null-terminated string
+-spec decode_string(binary()) -> [binary(), ...].
 decode_string(Bin) ->
     binary:split(Bin, <<0>>).
 
 %% decode multiple null-terminated string
+-spec decode_strings(binary()) -> [binary(), ...].
 decode_strings(Bin) ->
     [<<>> | T] = lists:reverse(binary:split(Bin, <<0>>, [global])),
     lists:reverse(T).
 
 %% decode field
+-spec decode_fields(binary()) -> [{byte(), binary()}].
 decode_fields(Bin) ->
     decode_fields(Bin, []).
 
@@ -56,6 +61,7 @@ decode_fields(<<Type:8, Rest/binary>>, Acc) ->
 
 %% decode ErrorResponse
 %% See http://www.postgresql.org/docs/current/interactive/protocol-error-fields.html
+-spec decode_error(binary()) -> #error{}.
 decode_error(Bin) ->
     Fields = decode_fields(Bin),
     ErrCode = proplists:get_value($C, Fields),
@@ -111,59 +117,58 @@ lower_atom(Str) when is_binary(Str) ->
 lower_atom(Str) when is_list(Str) ->
     list_to_atom(string:to_lower(Str)).
 
-%% FIXME: return iolist
-encode(Data) ->
-    Bin = iolist_to_binary(Data),
-    <<(byte_size(Bin) + 4):?int32, Bin/binary>>.
-
-encode(Type, Data) ->
-    Bin = iolist_to_binary(Data),
-    <<Type:8, (byte_size(Bin) + 4):?int32, Bin/binary>>.
 
 %% Build decoder for DataRow
+-spec build_decoder([epgsql:column()], epgsql_binary:codec()) -> row_decoder().
 build_decoder(Columns, Codec) ->
-    {Columns, Codec}.
+    Decoders = lists:map(
+                 fun(#column{oid = Oid, format = Format}) ->
+                         Fmt = case Format of
+                                   1 -> binary;
+                                   0 -> text
+                               end,
+                         epgsql_binary:oid_to_decoder(Oid, Fmt, Codec)
+                 end, Columns),
+    {Decoders, Columns, Codec}.
 
 %% decode row data
-%% FIXME: use body recursion
-decode_data(Bin, {Columns, Codec}) ->
-    decode_data(Columns, Bin, [], Codec).
-
-decode_data([], _Bin, Acc, _Codec) ->
-    list_to_tuple(lists:reverse(Acc));
-decode_data([_C | T], <<-1:?int32, Rest/binary>>, Acc, Codec) ->
-    decode_data(T, Rest, [null | Acc], Codec);
-decode_data([C | T], <<Len:?int32, Value:Len/binary, Rest/binary>>, Acc, Codec) ->
-    Value2 = case C of
-        #column{type = Type, format = 1} ->
-            epgsql_binary:decode(Type, Value, Codec);
-        #column{} ->
-            Value
-    end,
-    decode_data(T, Rest, [Value2 | Acc], Codec).
+-spec decode_data(binary(), row_decoder()) -> tuple().
+decode_data(Bin, {Decoders, Columns, Codec}) ->
+    list_to_tuple(decode_data(Bin, Decoders, Columns, Codec)).
+
+decode_data(_, [], [], _) -> [];
+decode_data(<<-1:?int32, Rest/binary>>, [_Dec | Decs], [_Col | Cols], Codec) ->
+    [null | decode_data(Rest, Decs, Cols, Codec)];
+decode_data(<<Len:?int32, Value:Len/binary, Rest/binary>>, [Decoder | Decs], [_Col | Cols], Codec) ->
+    [epgsql_binary:decode(Value, Decoder)
+     | decode_data(Rest, Decs, Cols, Codec)].
 
 %% decode column information
-%% TODO: use body-recursion
+-spec decode_columns(non_neg_integer(), binary(), epgsql_binary:codec()) -> [#column{}].
+decode_columns(0, _Bin, _Codec) -> [];
 decode_columns(Count, Bin, Codec) ->
-    decode_columns(Count, Bin, [], Codec).
-
-decode_columns(0, _Bin, Acc, _Codec) ->
-    lists:reverse(Acc);
-decode_columns(N, Bin, Acc, Codec) ->
     [Name, Rest] = decode_string(Bin),
-    <<_Table_Oid:?int32, _Attrib_Num:?int16, Type_Oid:?int32,
-     Size:?int16, Modifier:?int32, Format:?int16, Rest2/binary>> = Rest,
+    <<_TableOid:?int32, _AttribNum:?int16, TypeOid:?int32,
+      Size:?int16, Modifier:?int32, Format:?int16, Rest2/binary>> = Rest,
+    %% TODO: get rid of this 'type' (extra oid_db lookup)
+    Type = epgsql_binary:oid_to_name(TypeOid, Codec),
     Desc = #column{
       name     = Name,
-      type     = epgsql_binary:oid2type(Type_Oid, Codec),
+      type     = Type,
+      oid      = TypeOid,
       size     = Size,
       modifier = Modifier,
       format   = Format},
-    decode_columns(N - 1, Rest2, [Desc | Acc], Codec).
+    [Desc | decode_columns(Count - 1, Rest2, Codec)].
 
 %% decode ParameterDescription
+-spec decode_parameters(binary(), epgsql_binary:codec()) ->
+                               [epgsql_oid_db:type_info() | {unknown_oid, epgsql_oid_db:oid()}].
 decode_parameters(<<_Count:?int16, Bin/binary>>, Codec) ->
-    [epgsql_binary:oid2type(Oid, Codec) || <<Oid:?int32>> <= Bin].
+    [case epgsql_binary:oid_to_info(Oid, Codec)  of
+         undefined -> {unknown_oid, Oid};
+         TypeInfo -> TypeInfo
+     end || <<Oid:?int32>> <= Bin].
 
 %% decode command complete msg
 decode_complete(<<"SELECT", 0>>)        -> select;
@@ -181,6 +186,7 @@ decode_complete(Bin) ->
         [Type | _Rest]         -> lower_atom(Type)
     end.
 
+
 %% encode types
 encode_types(Types, Codec) ->
     encode_types(Types, 0, <<>>, Codec).
@@ -190,12 +196,13 @@ encode_types([], Count, Acc, _Codec) ->
 
 encode_types([Type | T], Count, Acc, Codec) ->
     Oid = case Type of
-        undefined -> 0;
-        _Any      -> epgsql_binary:type2oid(Type, Codec)
+              undefined -> 0;
+              _Any -> epgsql_binary:type_to_oid(Type, Codec)
     end,
     encode_types(T, Count + 1, <<Acc/binary, Oid:?int32>>, Codec).
 
-%% encode column formats
+%% encode expected column formats
+-spec encode_formats([#column{}]) -> binary().
 encode_formats(Columns) ->
     encode_formats(Columns, 0, <<>>).
 
@@ -205,42 +212,64 @@ encode_formats([], Count, Acc) ->
 encode_formats([#column{format = Format} | T], Count, Acc) ->
     encode_formats(T, Count + 1, <<Acc/binary, Format:?int16>>).
 
-format(Type, _Codec) ->
-    case epgsql_binary:supports(Type) of
-        true  -> 1;
-        false -> 0
+format({unknown_oid, _}, _) -> 0;
+format(#column{oid = Oid}, Codec) ->
+    case epgsql_binary:supports(Oid, Codec) of
+        true  -> 1;                             %binary
+        false -> 0                              %text
     end.
 
-%% encode parameters
+%% encode parameters for 'Bind'
+-spec encode_parameters([], epgsql_binary:codec()) -> iolist().
 encode_parameters(Parameters, Codec) ->
-    encode_parameters(Parameters, 0, <<>>, <<>>, Codec).
+    encode_parameters(Parameters, 0, <<>>, [], Codec).
 
 encode_parameters([], Count, Formats, Values, _Codec) ->
-    <<Count:?int16, Formats/binary, Count:?int16, Values/binary>>;
+    [<<Count:?int16>>, Formats, <<Count:?int16>> | lists:reverse(Values)];
 
 encode_parameters([P | T], Count, Formats, Values, Codec) ->
     {Format, Value} = encode_parameter(P, Codec),
     Formats2 = <<Formats/binary, Format:?int16>>,
-    Values2 = <<Values/binary, Value/binary>>,
+    Values2 = [Value | Values],
     encode_parameters(T, Count + 1, Formats2, Values2, Codec).
 
 %% encode parameter
 
+-spec encode_parameter({Type, Val :: any()},
+                       epgsql_binary:codec()) -> {0..1, iolist()} when
+      Type :: epgsql:type_name()
+            | {array, epgsql:type_name()}
+            | {unknown_oid, epgsql_oid_db:oid()}.
+encode_parameter({T, undefined}, Codec) ->
+    encode_parameter({T, null}, Codec);
+encode_parameter({_, null}, _Codec) ->
+    {1, <<-1:?int32>>};
+encode_parameter({{unknown_oid, _Oid}, Value}, _Codec) ->
+    encode_text(Value);
 encode_parameter({Type, Value}, Codec) ->
     case epgsql_binary:encode(Type, Value, Codec) of
-        Bin when is_binary(Bin) -> {1, Bin};
-        {error, unsupported}    -> encode_parameter(Value)
+        {error, unsupported} -> encode_text(Value);
+        Encoded -> {1, Encoded}
     end;
-encode_parameter(Value, _Codec) -> encode_parameter(Value).
+encode_parameter(Value, _Codec) -> encode_text(Value).
+
+encode_text(B) when is_binary(B)  -> {0, encode_bin(B)};
+encode_text(A) when is_atom(A)    -> {0, encode_bin(atom_to_binary(A, utf8))};
+encode_text(I) when is_integer(I) -> {0, encode_bin(integer_to_binary(I))};
+encode_text(F) when is_float(F)   -> {0, encode_bin(float_to_binary(F))};
+encode_text(L) when is_list(L)    -> {0, encode_bin(list_to_binary(L))}.
+
+
+encode(Data) ->
+    Size = iolist_size(Data),
+    [<<(Size + 4):?int32>> | Data].
+
+encode(Type, Data) ->
+    Size = iolist_size(Data),
+    [<<Type:8, (Size + 4):?int32>> | Data].
 
-encode_parameter(A) when is_atom(A)    -> {0, encode_list(atom_to_list(A))};
-encode_parameter(B) when is_binary(B)  -> {0, <<(byte_size(B)):?int32, B/binary>>};
-encode_parameter(I) when is_integer(I) -> {0, encode_list(integer_to_list(I))};
-encode_parameter(F) when is_float(F)   -> {0, encode_list(float_to_list(F))};
-encode_parameter(L) when is_list(L)    -> {0, encode_list(L)}.
 
-encode_list(L) ->
-    Bin = list_to_binary(L),
+encode_bin(Bin) ->
     <<(byte_size(Bin)):?int32, Bin/binary>>.
 
 encode_standby_status_update(ReceivedLSN, FlushedLSN, AppliedLSN) ->

src/epgsqla.erl

@@ -152,7 +152,7 @@ complete_connect(C, Ref) ->
             Retval =
                 case Msg of
                     connected ->
-                        ok = epgsql:update_type_cache(C),
+                        {ok, _} = epgsql:update_type_cache(C),
                         {C, Ref, connected};
                     {error, Error} ->
                         {C, Ref, {error, Error}}

src/ewkb.erl

@@ -1,3 +1,5 @@
+%% https://en.wikipedia.org/wiki/Well-known_text
+%% http://postgis.net/docs/manual-2.4/using_postgis_dbmanagement.html#EWKB_EWKT
 -module(ewkb).
 -include("epgsql_geometry.hrl").
 -export([decode_geometry/1, encode_geometry/1]).

test/epgsql_SUITE.erl

@@ -167,10 +167,10 @@ end_per_group(_GroupName, _Config) ->
         }}).
 
 %% From uuid.erl in http://gitorious.org/avtobiff/erlang-uuid
-uuid_to_string(<<U0:32, U1:16, U2:16, U3:16, U4:48>>) ->
-    lists:flatten(io_lib:format(
-                    "~8.16.0b-~4.16.0b-~4.16.0b-~4.16.0b-~12.16.0b",
-                    [U0, U1, U2, U3, U4])).
+uuid_to_bin_string(<<U0:32, U1:16, U2:16, U3:16, U4:48>>) ->
+    iolist_to_binary(io_lib:format(
+                       "~8.16.0b-~4.16.0b-~4.16.0b-~4.16.0b-~12.16.0b",
+                       [U0, U1, U2, U3, U4])).
 
 connect(Config) ->
     epgsql_ct:connect_only(Config, []).
@@ -554,8 +554,8 @@ describe_with_param(Config) ->
     Module = ?config(module, Config),
     epgsql_ct:with_connection(Config, fun(C) ->
         {ok, S} = Module:parse(C, "select id from test_table1 where id = $1"),
-        [int4] = S#statement.types,
-        [#column{name = <<"id">>}] = S#statement.columns,
+        ?assertEqual([int4], S#statement.types),
+        ?assertMatch([#column{name = <<"id">>}], S#statement.columns),
         {ok, S} = Module:describe(C, S),
         ok = Module:close(C, S),
         ok = Module:sync(C)
@@ -671,8 +671,8 @@ character_type(Config) ->
 
 uuid_type(Config) ->
     check_type(Config, uuid,
-        io_lib:format("'~s'", [uuid_to_string(?UUID1)]),
-        list_to_binary(uuid_to_string(?UUID1)), []).
+               io_lib:format("'~s'", [uuid_to_bin_string(?UUID1)]),
+               uuid_to_bin_string(?UUID1), []).
 
 point_type(Config) ->
     check_type(Config, point, "'(23.15, 100)'", {23.15, 100.0}, []).
@@ -694,9 +694,9 @@ geometry_type(Config) ->
 uuid_select(Config) ->
     Module = ?config(module, Config),
     epgsql_ct:with_rollback(Config, fun(C) ->
-        U1 = uuid_to_string(?UUID1),
-        U2 = uuid_to_string(?UUID2),
-        U3 = uuid_to_string(?UUID3),
+        U1 = uuid_to_bin_string(?UUID1),
+        U2 = uuid_to_bin_string(?UUID2),
+        U3 = uuid_to_bin_string(?UUID3),
         {ok, 1} =
             Module:equery(C, "insert into test_table2 (c_varchar, c_uuid) values ('UUID1', $1)",
                    [U1]),
@@ -708,12 +708,11 @@ uuid_select(Config) ->
                    [U3]),
         Res = Module:equery(C, "select c_varchar, c_uuid from test_table2 where c_uuid = any($1)",
                     [[U1, U2]]),
-        U1Bin = list_to_binary(U1),
-        U2Bin = list_to_binary(U2),
-        {ok,[{column,<<"c_varchar">>,varchar,_,_,_},
-             {column,<<"c_uuid">>,uuid,_,_,_}],
-         [{<<"UUID1">>, U1Bin},
-          {<<"UUID2">>, U2Bin}]} = Res
+        ?assertMatch(
+           {ok,[#column{name = <<"c_varchar">>, type = varchar},
+                #column{name = <<"c_uuid">>, type = uuid}],
+            [{<<"UUID1">>, U1},
+             {<<"UUID2">>, U2}]}, Res)
     end).
 
 date_time_type(Config) ->
@@ -761,7 +760,7 @@ hstore_type(Config) ->
     check_type(Config, hstore, "''", {[]}, []),
     check_type(Config, hstore,
                "'a => 1, b => 2.0, c => null'",
-               {[{<<"c">>, null}, {<<"b">>, <<"2.0">>}, {<<"a">>, <<"1">>}]}, Values).
+               {[{<<"a">>, <<"1">>}, {<<"b">>, <<"2.0">>}, {<<"c">>, null}]}, Values).
 
 net_type(Config) ->
     check_type(Config, cidr, "'127.0.0.1/32'", {{127,0,0,1}, 32}, [{{127,0,0,1}, 32}, {{0,0,0,0,0,0,0,1}, 128}]),
@@ -811,13 +810,14 @@ custom_types(Config) ->
     Module = ?config(module, Config),
     epgsql_ct:with_connection(Config, fun(C) ->
         Module:squery(C, "drop table if exists t_foo;"),
-        Module:squery(C, "drop type foo;"),
-        {ok, [], []} = Module:squery(C, "create type foo as enum('foo', 'bar');"),
-        ok = epgsql:update_type_cache(C, [<<"foo">>]),
-        {ok, [], []} = Module:squery(C, "create table t_foo (col foo);"),
-        {ok, S} = Module:parse(C, "insert_foo", "insert into t_foo values ($1)", [foo]),
-        ok = Module:bind(C, S, ["bar"]),
-        {ok, 1} = Module:execute(C, S)
+        Module:squery(C, "drop type if exists my_type;"),
+        {ok, [], []} = Module:squery(C, "create type my_type as enum('foo', 'bar');"),
+        {ok, [my_type]} = epgsql:update_type_cache(C, [{epgsql_codec_test_enum, [foo, bar]}]),
+        {ok, [], []} = Module:squery(C, "create table t_foo (col my_type);"),
+        {ok, S} = Module:parse(C, "insert_foo", "insert into t_foo values ($1)", [my_type]),
+        ok = Module:bind(C, S, [bar]),
+        {ok, 1} = Module:execute(C, S),
+        ?assertMatch({ok, _, [{bar}]}, Module:equery(C, "SELECT col FROM t_foo"))
     end).
 
 text_format(Config) ->
@@ -826,8 +826,8 @@ text_format(Config) ->
         Select = fun(Type, V) ->
             V2 = list_to_binary(V),
             Query = "select $1::" ++ Type,
-            {ok, _Cols, [{V2}]} = Module:equery(C, Query, [V]),
-            {ok, _Cols, [{V2}]} = Module:equery(C, Query, [V2])
+            ?assertMatch({ok, _Cols, [{V2}]}, Module:equery(C, Query, [V])),
+            ?assertMatch({ok, _Cols, [{V2}]}, Module:equery(C, Query, [V2]))
         end,
         Select("numeric", "123456")
     end).
@@ -1121,7 +1121,7 @@ check_type(Config, Type, In, Out, Values, Column) ->
     epgsql_ct:with_connection(Config, fun(C) ->
         Select = io_lib:format("select ~s::~w", [In, Type]),
         Res = Module:equery(C, Select),
-        {ok, [#column{type = Type}], [{Out}]} = Res,
+        ?assertMatch({ok, [#column{type = Type}], [{Out}]}, Res),
         Sql = io_lib:format("insert into test_table2 (~s) values ($1) returning ~s", [Column, Column]),
         {ok, #statement{columns = [#column{type = Type}]} = S} = Module:parse(C, Sql),
         Insert = fun(V) ->
@@ -1129,7 +1129,10 @@ check_type(Config, Type, In, Out, Values, Column) ->
             {ok, 1, [{V2}]} = Module:execute(C, S),
             case compare(Type, V, V2) of
                 true  -> ok;
-                false -> ?debugFmt("~p =/= ~p~n", [V, V2]), ?assert(false)
+                false ->
+                    error({write_read_compare_failed,
+                           iolist_to_binary(
+                             io_lib:format("~p =/= ~p~n", [V, V2]))})
             end,
             ok = Module:sync(C)
         end,

test/epgsql_codec_test_enum.erl

@@ -0,0 +1,19 @@
+-module(epgsql_codec_test_enum).
+-behaviour(epgsql_codec).
+
+-export([init/2, names/0, encode/3, decode/3]).
+
+
+init(Choices, _) -> Choices.
+
+names() ->
+    [my_type].
+
+encode(Atom, my_type, Choices) ->
+    true = lists:member(Atom, Choices),
+    atom_to_binary(Atom, utf8).
+
+decode(Bin, my_type, Choices) ->
+    Atom = binary_to_existing_atom(Bin, utf8),
+    true = lists:member(Atom, Choices),
+    Atom.