Merge branch 'devel'

.gitignore

@@ -1,3 +1,8 @@
 _build
 rebar3
 datadir/
+doc/*
+!doc/overview.edoc
+!doc/*.md
+*.idea/*
+*.iml

.travis.yml

@@ -11,13 +11,14 @@ install: "true"
 language: erlang
 matrix:
   include:
-    - otp_release: 22.0
+    - otp_release: 22.2
     - otp_release: 21.3
     - otp_release: 20.3
     - otp_release: 19.3
     - otp_release: 18.3
       dist: trusty
 script:
-  - make elvis
+  - '[ "$TRAVIS_OTP_RELEASE" = "18.3" ] || make elvis' # TODO: remove the guard when OTP18 support is dropped
   - make test
+  - make edoc
   - make dialyzer

CHANGES

@@ -1,3 +1,42 @@
+In 4.4.0
+
+* Guards are now added to avoid silent integer truncation for numeric and
+  numeric range datatype codecs. So, an attempt to encode 100000 as `int2`
+  will now crash the connection instead of silently truncating it. #218
+* `epgsql{a,i}:cancel/1` API was documented. #224
+* Version of `execute_batch` that uses the same SQL query for each request
+  in a batch. Very convenient for batch-inserts. #209
+* It's now possible to provide `#statement{}` to `prepared_query/3`. This way
+  of calling it eliminates extra `describe` round-trip thus making it more
+  efficient. #207
+* Representation of SQL `NULL` is now fully configurable. You can choose what
+  set of Erlang terms should be interpreted as `NULL` and which term to use to
+  represent `NULL`s received from database. #212
+* It's now possible to choose between 3 representations of a `hstore` datatype:
+  map(), jiffy-style objects (default) and proplist. It can also take `map()` as
+  input now. NULL value representation is also configurable. #217
+* Edocs build was fixed. Just run `rebar3 edoc` and reference documentation for
+  all modules will be generated. But it's considered to be more "internal"
+  documentation for those who want to learn more about epgsql internals or
+  to do some hacking. It complements, but not replaces README. #214
+* `epgsql:connect` `timeout` option is more strict now - it limits TCP and SSL
+  setup time as a whole. #223
+* Test coverage report was enabled in CI. We will fail the build if coverage
+  falls below 55%. We hope to improve this metric over time. #208
+* We now send `Terminate` message to the server when doing graceful connection
+  shutdown (as recommended by protocol). #219
+* We found that `describe(_, portal, _)` API was broken since release v4.0.0, but
+  was not covered by tests. So now it was fixed and tests were added. #211
+* Error code to error name conversion code was updated (see `#error.codename`).
+  Some new codes were added (mostly related to JSON datatypes) and one has changed.
+  So, if you were matching over `#error.codename` being
+  `invalid_preceding_following_size` you have to update your code. #210
+* `#column{}` record is now fully documented. It was extended to
+  include `table_oid` and `table_attr_number` fields which point to the originating
+  database table of this column (if any). #205
+* Extended timerange datatype support #204
+* Some minor typos, datatype and CI fixes #199 #201 #206 #221
+
 In 4.3.0
 
 * Erlang 22 compatibility is tested; support for Erlang 17 was dropped. Last

Makefile

@@ -1,4 +1,5 @@
 REBAR = ./rebar3
+MINIMAL_COVERAGE = 55
 
 all: compile
 
@@ -11,12 +12,25 @@ compile: src/epgsql_errcodes.erl $(REBAR)
 
 clean: $(REBAR)
 	@$(REBAR) clean
+	@rm -f doc/*.html
+	@rm -f doc/erlang.png
+	@rm -f doc/stylesheet.css
+	@rm -f doc/edoc-info
 
 src/epgsql_errcodes.erl:
 	./generate_errcodes_src.sh > src/epgsql_errcodes.erl
 
-test: compile
-	@$(REBAR) do ct -v
+common-test:
+	$(REBAR) ct -v -c
+
+eunit:
+	$(REBAR) eunit -c
+
+# Fail the build if coverage falls below 55%
+cover:
+	$(REBAR) cover -v --min_coverage $(MINIMAL_COVERAGE)
+
+test: compile eunit common-test cover
 
 dialyzer: compile
 	@$(REBAR) dialyzer
@@ -24,4 +38,7 @@ dialyzer: compile
 elvis: $(REBAR)
 	@$(REBAR) as lint lint
 
-.PHONY: all compile clean test dialyzer elvis
+edoc: $(REBAR)
+	@$(REBAR) edoc
+
+.PHONY: all compile clean common-test eunit cover test dialyzer elvis

README.md

@@ -74,6 +74,7 @@ connect(Opts) -> {ok, Connection :: epgsql:connection()} | {error, Reason :: epg
       timeout =>  timeout(),             % socket connect timeout, default: 5000 ms
       async =>    pid() | atom(),        % process to receive LISTEN/NOTIFY msgs
       codecs =>   [{epgsql_codec:codec_mod(), any()}]}
+      nulls =>    [any(), ...],          % NULL terms
       replication => Replication :: string()} % Pass "database" to connect in replication mode
     | list().
 
@@ -96,14 +97,20 @@ Only `host` and `username` are mandatory, but most likely you would need `databa
 - `password` - DB user password. It might be provided as string / binary or as a fun that returns
    string / binary. Internally, plain password is wrapped to anonymous fun before it is sent to connection
    process, so, if `connect` command crashes, plain password will not appear in crash logs.
-- `{timeout, TimeoutMs}` parameter will trigger an `{error, timeout}` result when the
-   socket fails to connect within `TimeoutMs` milliseconds.
+- `timeout` parameter will trigger an `{error, timeout}` result when the
+   socket fails to connect within provided milliseconds.
 - `ssl` if set to `true`, perform an attempt to connect in ssl mode, but continue unencrypted
   if encryption isn't supported by server. if set to `required` connection will fail if encryption
   is not available.
 - `ssl_opts` will be passed as is to `ssl:connect/3`
 - `async` see [Server notifications](#server-notifications)
 - `codecs` see [Pluggable datatype codecs](#pluggable-datatype-codecs)
+- `nulls` terms which will be used to represent SQL `NULL`. If any of those has been encountered in
+   placeholder parameters (`$1`, `$2` etc values), it will be interpreted as `NULL`.
+   1st element of the list will be used to represent NULLs received from the server. It's not recommended
+   to use `"string"`s or lists. Try to keep this list short for performance!
+   Default is `[null, undefined]`, i.e. encode `null` or `undefined` in parameters as `NULL`
+   and decode `NULL`s as atom `null`.
 - `replication` see [Streaming replication protocol](#streaming-replication-protocol)
 
 Options may be passed as proplist or as map with the same key names.
@@ -126,21 +133,15 @@ Asynchronous connect example (applies to **epgsqli** too):
 ### Simple Query
 
 ```erlang
--type query() :: string() | iodata().
--type squery_row() :: {binary()}.
+-include_lib("epgsql/include/epgsql.hrl").
 
--record(column, {
-    name :: binary(),
-    type :: epgsql_type(),
-    size :: -1 | pos_integer(),
-    modifier :: -1 | pos_integer(),
-    format :: integer()
-}).
+-type query() :: string() | iodata().
+-type squery_row() :: tuple() % tuple of binary().
 
 -type ok_reply(RowType) ::
-    {ok, ColumnsDescription :: [#column{}], RowsValues :: [RowType]} |                            % select
+    {ok, ColumnsDescription :: [epgsql:column()], RowsValues :: [RowType]} |                            % select
     {ok, Count :: non_neg_integer()} |                                                            % update/insert/delete
-    {ok, Count :: non_neg_integer(), ColumnsDescription :: [#column{}], RowsValues :: [RowType]}. % update/insert/delete + returning
+    {ok, Count :: non_neg_integer(), ColumnsDescription :: [epgsql:column()], RowsValues :: [RowType]}. % update/insert/delete + returning
 -type error_reply() :: {error, query_error()}.
 -type reply(RowType) :: ok_reply() | error_reply().
 
@@ -159,7 +160,7 @@ epgsql:squery(C, "insert into account (name) values  ('alice'), ('bob')").
 ```erlang
 epgsql:squery(C, "select * from account").
 > {ok,
-    [{column,<<"id">>,int4,4,-1,0},{column,<<"name">>,text,-1,-1,0}],
+    [#column{name = <<"id">>, type = int4, …},#column{name = <<"name">>, type = text, …}],
     [{<<"1">>,<<"alice">>},{<<"2">>,<<"bob">>}]
 }
 ```
@@ -170,13 +171,12 @@ epgsql:squery(C,
     "    values ('joe'), (null)"
     "    returning *").
 > {ok,2,
-    [{column,<<"id">>,int4,4,-1,0}, {column,<<"name">>,text,-1,-1,0}],
+    [#column{name = <<"id">>, type = int4, …}, #column{name = <<"name">>, type = text, …}],
     [{<<"3">>,<<"joe">>},{<<"4">>,null}]
 }
 ```
 
 ```erlang
--include_lib("epgsql/include/epgsql.hrl").
 epgsql:squery(C, "SELECT * FROM _nowhere_").
 > {error,
    #error{severity = error,code = <<"42P01">>,
@@ -253,7 +253,7 @@ an error occurs, all statements result in `{error, #error{}}`.
 ```erlang
 epgsql:equery(C, "select id from account where name = $1", ["alice"]),
 > {ok,
-    [{column,<<"id">>,int4,4,-1,1}],
+    [#column{name = <<"id">>, type = int4, …}],
     [{1}]
 }
 ```
@@ -283,25 +283,26 @@ squery including final `{C, Ref, done}`.
 ### Prepared Query
 
 ```erlang
-{ok, Columns, Rows}        = epgsql:prepared_query(C, StatementName, [Parameters]).
-{ok, Count}                = epgsql:prepared_query(C, StatementName, [Parameters]).
-{ok, Count, Columns, Rows} = epgsql:prepared_query(C, StatementName, [Parameters]).
+{ok, Columns, Rows}        = epgsql:prepared_query(C, Statement :: #statement{} | string(), [Parameters]).
+{ok, Count}                = epgsql:prepared_query(C, Statement, [Parameters]).
+{ok, Count, Columns, Rows} = epgsql:prepared_query(C, Statement, [Parameters]).
 {error, Error}             = epgsql:prepared_query(C, "non_existent_query", [Parameters]).
 ```
 
-`Parameters` - optional list of values to be bound to `$1`, `$2`, `$3`, etc.
-`StatementName` - name of query given with ```erlang epgsql:parse(C, StatementName, "select ...", []).```
+- `Parameters` - optional list of values to be bound to `$1`, `$2`, `$3`, etc.
+- `Statement` - name of query given with ```erlang epgsql:parse(C, StatementName, "select ...", []).```
+               (can be empty string) or `#statement{}` record returned by `epgsql:parse`.
 
 With prepared query one can parse a query giving it a name with `epgsql:parse` on start and reuse the name
 for all further queries with different parameters.
 
 ```erlang
-epgsql:parse(C, "inc", "select $1+1", []).
-epgsql:prepared_query(C, "inc", [4]).
-epgsql:prepared_query(C, "inc", [1]).
+{ok, Stmt} = epgsql:parse(C, "inc", "select $1+1", []).
+epgsql:prepared_query(C, Stmt, [4]).
+epgsql:prepared_query(C, Stmt, [1]).
 ```
 
-Asynchronous API `epgsqla:prepared_query/3` requires you to parse statement beforehand
+Asynchronous API `epgsqla:prepared_query/3` requires you to always parse statement beforehand
 
 ```erlang
 #statement{types = Types} = Statement,
@@ -384,23 +385,54 @@ Batch execution is `bind` + `execute` for several prepared statements.
 It uses unnamed portals and `MaxRows = 0`.
 
 ```erlang
-Results = epgsql:execute_batch(C, Batch).
+Results = epgsql:execute_batch(C, BatchStmt :: [{statement(), [bind_param()]}]).
+{Columns, Results} = epgsql:execute_batch(C, statement() | sql_query(), Batch :: [ [bind_param()] ]).
 ```
 
-- `Batch`   - list of {Statement, ParameterValues}
-- `Results` - list of {ok, Count} or {ok, Count, Rows}
+- `BatchStmt` - list of `{Statement, ParameterValues}`, each item has it's own `#statement{}`
+- `Batch` - list of `ParameterValues`, each item executes the same common `#statement{}` or SQL query
+- `Columns` - list of `#column{}` descriptions of `Results` columns
+- `Results` - list of `{ok, Count}` or `{ok, Count, Rows}`
+
+There are 2 versions:
+
+`execute_batch/2` - each item in a batch has it's own named statement (but it's allowed to have duplicates)
 
 example:
 
 ```erlang
-{ok, S1} = epgsql:parse(C, "one", "select $1", [int4]),
-{ok, S2} = epgsql:parse(C, "two", "select $1 + $2", [int4, int4]),
+{ok, S1} = epgsql:parse(C, "one", "select $1::integer", []),
+{ok, S2} = epgsql:parse(C, "two", "select $1::integer + $2::integer", []),
 [{ok, [{1}]}, {ok, [{3}]}] = epgsql:execute_batch(C, [{S1, [1]}, {S2, [1, 2]}]).
+ok = epgsql:close(C, "one").
+ok = epgsql:close(C, "two").
+```
+
+`execute_batch/3` - each item in a batch executed with the same common SQL query or `#statement{}`.
+It's allowed to use unnamed statement.
+
+example (the most efficient way to make batch inserts with epgsql):
+
+```erlang
+{ok, Stmt} = epgsql:parse(C, "my_insert", "INSERT INTO account (name, age) VALUES ($1, $2) RETURNING id", []).
+{[#column{name = <<"id">>}], [{ok, [{1}]}, {ok, [{2}]}, {ok, [{3}]}]} =
+    epgsql:execute_batch(C, Stmt, [ ["Joe", 35], ["Paul", 26], ["Mary", 24] ]).
+ok = epgsql:close(C, "my_insert").
+```
+
+equivalent:
+
+```erlang
+epgsql:execute_batch(C, "INSERT INTO account (name, age) VALUES ($1, $2) RETURNING id",
+                     [ ["Joe", 35], ["Paul", 26], ["Mary", 24] ]).
 ```
 
-`epgsqla:execute_batch/3` sends `{C, Ref, Results}`
+In case one of the batch items causes an error, the result returned for this particular
+item will be `{error, #error{}}` and no more results will be produced.
 
-`epgsqli:execute_batch/3` sends
+`epgsqla:execute_batch/{2,3}` sends `{C, Ref, Results}`
+
+`epgsqli:execute_batch/{2,3}` sends
 
 - `{C, Ref, {data, Row}}`
 - `{C, Ref, {error, Reason}}`
@@ -408,9 +440,45 @@ example:
 - `{C, Ref, {complete, _Type}}`
 - `{C, Ref, done}` - execution of all queries from Batch has finished
 
+### Query cancellation
+
+```erlang
+epgsql:cancel(connection()) -> ok.
+```
+
+PostgreSQL protocol supports [cancellation](https://www.postgresql.org/docs/current/protocol-flow.html#id-1.10.5.7.9)
+of currently executing command. `cancel/1` sends a cancellation request via the
+new temporary TCP connection asynchronously, it doesn't await for the command to
+be cancelled. Instead, client should expect to get
+`{error, #error{code = <<"57014">>, codename = query_canceled}}` back from
+the command that was cancelled. However, normal response can still be received as well.
+While it's not so straightforward to use with synchronous `epgsql` API, it plays
+quite nicely with asynchronous `epgsqla` API. For example, that's how a query with
+soft timeout can be implemented:
+
+```erlang
+squery(C, SQL, Timeout) ->
+    Ref = epgsqla:squery(C, SQL),
+    receive
+       {C, Ref, Result} -> Result
+    after Timeout ->
+        ok = epgsql:cancel(C),
+        % We can still receive {ok, …} as well as
+        % {error, #error{codename = query_canceled}} or any other error
+        receive
+            {C, Ref, Result} -> Result
+        end
+    end.
+```
+
+This API should be used with extreme care when pipelining is in use: it only cancels
+currently executing command, all the subsequent pipelined commands will continue
+their normal execution. And it's not always easy to see which command exactly is
+executing when we are issuing the cancellation request.
+
 ## Data Representation
 
-Data representation may be configured using [pluggable datatype codecs](pluggable_types.md),
+Data representation may be configured using [pluggable datatype codecs](doc/pluggable_types.md),
 so following is just default mapping:
 
 PG type       | Representation
@@ -433,8 +501,8 @@ PG type       | Representation
   record      | `{int2, time, text, ...}` (decode only)
   point       |  `{10.2, 100.12}`
   int4range   | `[1,5)`
-  hstore      | `{[ {binary(), binary() \| null} ]}`
-  json/jsonb  | `<<"{ \"key\": [ 1, 1.0, true, \"string\" ] }">>` (see below for codec details)
+  hstore      | `{[ {binary(), binary() \| null} ]}` (configurable)
+  json/jsonb  | `<<"{ \"key\": [ 1, 1.0, true, \"string\" ] }">>` (configurable)
   uuid        | `<<"123e4567-e89b-12d3-a456-426655440000">>`
   inet        | `inet:ip_address()`
   cidr        | `{ip_address(), Mask :: 0..32}`
@@ -444,6 +512,8 @@ PG type       | Representation
   tstzrange   | `{{Hour, Minute, Second.Microsecond}, {Hour, Minute, Second.Microsecond}}`
   daterange   | `{{Year, Month, Day}, {Year, Month, Day}}`
 
+`null` can be configured. See `nulls` `connect/1` option.
+
 `timestamp` and `timestamptz` parameters can take `erlang:now()` format: `{MegaSeconds, Seconds, MicroSeconds}`
 
 `int4range` is a range type for ints that obeys inclusive/exclusive semantics,
@@ -453,6 +523,12 @@ and `plus_infinity`
 `tsrange`, `tstzrange`, `daterange` are range types for `timestamp`, `timestamptz` and `date`
 respectively. They can return `empty` atom as the result from a database if bounds are equal
 
+`hstore` type can take map or jiffy-style objects as input. Output can be tuned by
+providing `return :: map | jiffy | proplist` option to choose the format to which
+hstore should be decoded. `nulls :: [atom(), ...]` option can be used to select the
+terms which should be interpreted as SQL `NULL` - semantics is the same as
+for `connect/1` `nulls` option.
+
 `json` and `jsonb` types can optionally use a custom JSON encoding/decoding module to accept
 and return erlang-formatted JSON. The module must implement the callbacks in `epgsql_codec_json`,
 which most popular open source JSON parsers will already, and you can specify it in the codec
@@ -584,15 +660,15 @@ Parameter's value may change during connection's lifetime.
 
 ## Streaming replication protocol
 
-See [streaming.md](streaming.md).
+See [streaming.md](doc/streaming.md).
 
 ## Pluggable commands
 
-See [pluggable_commands.md](pluggable_commands.md)
+See [pluggable_commands.md](doc/pluggable_commands.md)
 
 ## Pluggable datatype codecs
 
-See [pluggable_types.md](pluggable_types.md)
+See [pluggable_types.md](doc/pluggable_types.md)
 
 ## Mailing list
 

doc/overview.edoc

@@ -0,0 +1,57 @@
+@title epgsql - PostgreSQL driver for Erlang, internal documentation
+@doc
+This document is made mostly as internal documentation. It can be useful
+if you plan to contribute some patches to epgsql, want to implement
+custom datatypes or commands or to better understand epgsql internals.
+
+End-user interface is described in <a href="https://github.com/epgsql/epgsql#readme">README.md</a>.
+
+== Interfaces ==
+Epgsql has 3 end-user API interfaces:
+
+<ul>
+  <li>{@link epgsql} - synchronous</li>
+  <li>{@link epgsqla} - asynchronous</li>
+  <li>{@link epgsqli} - incremental</li>
+</ul>
+
+== Internals ==
+
+All 3 interfaces communicate with {@link epgsql_sock} gen_server, which holds all
+the connection state. While `epgsql_sock' holds all the state, it doesn't know
+much about Client-Server communication protocol.
+All the communication logic between epgsql and PostgreSQL server is implemented
+as a {@section Commands} and `epgsql_sock' acts as an executor for those commands.
+
+PostgreSQL binary communication protocol is represented by 2 modules:
+<ul>
+  <li>{@link epgsql_wire} - codecs for on-wire communication protocol messages</li>
+  <li>{@link epgsql_binary} - interface to PostgreSQL binary data encoding protocol(see {@section Datatypes})</li>
+</ul>
+
+`epgsql_sock' holds an internal state of `epgsql_binary' codecs as well. The
+main contents of this state is the mapping between PostgreSQL unique numeric
+datatype IDs (OIDs) and callbacks which will be used to decode this datatype.
+This mapping is handled by {@link epgsql_oid_db} module and is populated at
+connection set-up time by {@link epgsql_cmd_connect}.
+
+Most of the connection initialization (network connection, authentication, codecs)
+is performed by {@link epgsql_cmd_connect} command, wich is just a regualr command
+(but quite complex one) and can be replaced by own implementation if needed.
+
+== Commands ==
+
+Client can execute a number of built-in commands as well as define their own.
+See {@link epgsql_command} and all the `epgsql_cmd_*' pages.
+There exists a <a href="pluggable_commands.md">manual</a> that explains how to
+implement your own command.
+
+== Datatypes ==
+
+Epgsql supports both PostgreSQL <a href="https://www.postgresql.org/docs/current/protocol-overview.html#PROTOCOL-FORMAT-CODES">text and binary</a>
+data encodings to transfer the data (query placeholder parameters and result rows).
+There are a bunch of built-in codecs and it's possible to
+implement custom ones as well as fine-tune some of built-ins.
+See {@link epgsql_codec} and all the `epgsql_codec_*' pages for more details.
+There exists a <a href="pluggable_types.md">manual</a> that explains how to
+implement your own datatype codec.

generate_errcodes_src.sh

@@ -1,15 +1,15 @@
-#!/usr/bin/env bash -e
+#!/usr/bin/env bash
 #
 # Used to generate epgsql_errcodes.erl
 #
-ERRFILE="https://raw.github.com/postgres/postgres/master/src/backend/utils/errcodes.txt"
+ERRFILE="https://raw.githubusercontent.com/postgres/postgres/master/src/backend/utils/errcodes.txt"
 echo "%% DO NOT EDIT - AUTOGENERATED BY $0 ON $(date +%Y-%m-%dT%H:%M:%S%z)"
 echo "-module(epgsql_errcodes)."
 echo "-export([to_name/1])."
 echo
 wget -qO- "$ERRFILE" | awk '
 NF == 4 && \
-$1 ~ /[^\s]{5}/ && \
+$1 ~ /[0-9A_Z]+/ && \
 $2 ~ /[EWS]/ \
 {
     printf("to_name(<<\"%s\">>) -> %s;\n", $1, $4)

include/epgsql.hrl

@@ -1,10 +1,26 @@
+%% See https://www.postgresql.org/docs/current/protocol-message-formats.html
+%% Description of `RowDescription' packet
 -record(column, {
+    %% field name
     name :: binary(),
+    %% name of the field data type
     type :: epgsql:epgsql_type(),
-    oid :: integer(),
+    %% OID of the field's data type
+    oid :: non_neg_integer(),
+    %% data type size (see pg_type.typlen). negative values denote variable-width types
     size :: -1 | pos_integer(),
+    %% type modifier (see pg_attribute.atttypmod). meaning of the modifier is type-specific
     modifier :: -1 | pos_integer(),
-    format :: integer()
+    %% format code being used for the field during server->client transmission.
+    %% Currently will be zero (text) or one (binary).
+    format :: integer(),
+    %% If the field can be identified as a column of a specific table, the OID of the table; otherwise zero.
+    %% SELECT relname FROM pg_catalog.pg_class WHERE oid=<table_oid>
+    table_oid :: non_neg_integer(),
+    %% If table_oid is not zero, the attribute number of the column; otherwise zero.
+    %% SELECT attname FROM pg_catalog.pg_attribute
+    %% WHERE attrelid=<table_oid> AND attnum=<table_attr_number>
+    table_attr_number :: pos_integer()
 }).
 
 -record(statement, {
@@ -14,10 +30,12 @@
     parameter_info :: [epgsql_oid_db:oid_entry()]
 }).
 
+
+%% See https://www.postgresql.org/docs/current/protocol-error-fields.html
 -record(error, {
     % see client_min_messages config option
     severity :: debug | log | info | notice | warning | error | fatal | panic,
-    code :: binary(),
+    code :: binary(), % See https://www.postgresql.org/docs/current/errcodes-appendix.html
     codename :: atom(),
     message :: binary(),
     extra :: [{severity | detail | hint | position | internal_position | internal_query

include/protocol.hrl

@@ -43,6 +43,7 @@
 -define(READY_FOR_QUERY, $Z).
 -define(COPY_BOTH_RESPONSE, $W).
 -define(COPY_DATA, $d).
+-define(TERMINATE, $X).
 
 % CopyData replication messages
 -define(X_LOG_DATA, $w).

rebar.config

@@ -1,9 +1,8 @@
 %% -*- mode: erlang -*-
 
-{eunit_opts, [verbose]}.
-
 {cover_enabled, true}.
-{cover_print_enabled, true}.
+
+{edoc_opts, [{preprocess, true}]}.
 
 {profiles, [
     {test, [
@@ -27,7 +26,8 @@
      ruleset => erl_files,
      rules =>
          [{elvis_style, line_length, #{limit => 120}},
-          {elvis_style, god_modules, #{limit => 40}},
+          {elvis_style, god_modules, #{limit => 41}},
+          {elvis_style, dont_repeat_yourself, #{min_complexity => 11}},
           {elvis_style, state_record_and_type, disable} % epgsql_sock
          ]}
   ]

src/commands/epgsql_cmd_batch.erl

@@ -1,32 +1,52 @@
-%% > Bind
-%% < BindComplete
-%% > Execute
-%% < DataRow*
-%% < CommandComplete
-%% -- Repeated many times --
+%% @doc Execute multiple extended queries in a single network round-trip
+%%
+%% There are 2 kinds of interface:
+%% <ol>
+%%  <li>To execute multiple queries, each with it's own `statement()'</li>
+%%  <li>To execute multiple queries, but by binding different parameters to the
+%%  same `statement()'</li>
+%% </ol>
+%% ```
+%% > {Bind
+%% <  BindComplete
+%% >  Execute
+%% <  DataRow*
+%% <  CommandComplete}*
 %% > Sync
 %% < ReadyForQuery
+%% '''
 -module(epgsql_cmd_batch).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).
--export_type([response/0]).
+-export_type([arguments/0, response/0]).
 
--type response() :: [{ok, Count :: non_neg_integer(), Rows :: [tuple()]}
-                     | {ok, Count :: non_neg_integer()}
-                     | {ok, Rows :: [tuple()]}
-                     | {error, epgsql:query_error()}].
 
 -include("epgsql.hrl").
 -include("protocol.hrl").
 
 -record(batch,
-        {batch :: [{#statement{}, list()}],
-         decoder}).
+        {batch :: [ [epgsql:bind_param()] ] | [{#statement{}, [epgsql:bind_param()]}],
+         statement :: #statement{} | undefined,
+         decoder :: epgsql_wire:row_decoder() | undefined}).
+
+-type arguments() ::
+        {epgsql:statement(), [ [epgsql:bind_param()] ]} |
+        [{epgsql:statement(), [epgsql:bind_param()]}].
 
-init(Batch) ->
+-type response() :: [{ok, Count :: non_neg_integer(), Rows :: [tuple()]}
+                     | {ok, Count :: non_neg_integer()}
+                     | {ok, Rows :: [tuple()]}
+                     | {error, epgsql:query_error()}].
+-type state() :: #batch{}.
+
+-spec init(arguments()) -> state().
+init({#statement{} = Statement, Batch}) ->
+    #batch{statement = Statement,
+           batch = Batch};
+init(Batch) when is_list(Batch) ->
     #batch{batch = Batch}.
 
-execute(Sock, #batch{batch = Batch} = State) ->
+execute(Sock, #batch{batch = Batch, statement = undefined} = State) ->
     Codec = epgsql_sock:get_codec(Sock),
     Commands =
         lists:foldr(
@@ -34,19 +54,38 @@ execute(Sock, #batch{batch = Batch} = State) ->
                   #statement{name = StatementName,
                              columns = Columns,
                              types = Types} = Statement,
-                  TypedParameters = lists:zip(Types, Parameters),
-                  Bin1 = epgsql_wire:encode_parameters(TypedParameters, Codec),
-                  Bin2 = epgsql_wire:encode_formats(Columns),
-                  [{?BIND, [0, StatementName, 0, Bin1, Bin2]},
-                   {?EXECUTE, [0, <<0:?int32>>]} | Acc]
+                  BinFormats = epgsql_wire:encode_formats(Columns),
+                  add_command(StatementName, Types, Parameters, BinFormats, Codec, Acc)
+          end,
+          [{?SYNC, []}],
+          Batch),
+    epgsql_sock:send_multi(Sock, Commands),
+    {ok, Sock, State};
+execute(Sock, #batch{batch = Batch,
+                     statement = #statement{name = StatementName,
+                                            columns = Columns,
+                                            types = Types}} = State) ->
+    Codec = epgsql_sock:get_codec(Sock),
+    BinFormats = epgsql_wire:encode_formats(Columns),
+    %% TODO: build some kind of encoder and reuse it for each batch item
+    Commands =
+        lists:foldr(
+          fun(Parameters, Acc) ->
+                  add_command(StatementName, Types, Parameters, BinFormats, Codec, Acc)
           end,
           [{?SYNC, []}],
           Batch),
     epgsql_sock:send_multi(Sock, Commands),
     {ok, Sock, State}.
 
-handle_message(?BIND_COMPLETE, <<>>, Sock, #batch{batch = [{Stmt, _} | _]} = State) ->
-    #statement{columns = Columns} = Stmt,
+add_command(StmtName, Types, Params, BinFormats, Codec, Acc) ->
+    TypedParameters = lists:zip(Types, Params),
+    BinParams = epgsql_wire:encode_parameters(TypedParameters, Codec),
+    [{?BIND, [0, StmtName, 0, BinParams, BinFormats]},
+     {?EXECUTE, [0, <<0:?int32>>]} | Acc].
+
+handle_message(?BIND_COMPLETE, <<>>, Sock, State) ->
+    Columns = current_cols(State),
     Codec = epgsql_sock:get_codec(Sock),
     Decoder = epgsql_wire:build_decoder(Columns, Codec),
     {noaction, Sock, State#batch{decoder = Decoder}};
@@ -58,7 +97,8 @@ handle_message(?DATA_ROW, <<_Count:?int16, Bin/binary>>, Sock,
 %%     Sock1 = epgsql_sock:add_result(Sock, {complete, empty}, {ok, [], []}),
 %%     {noaction, Sock1};
 handle_message(?COMMAND_COMPLETE, Bin, Sock,
-               #batch{batch = [{#statement{columns = Columns}, _} | Batch]} = State) ->
+               #batch{batch = [_ | Batch]} = State) ->
+    Columns = current_cols(State),
     Complete = epgsql_wire:decode_complete(Bin),
     Rows = epgsql_sock:get_rows(Sock),
     Result = case Complete of
@@ -79,3 +119,14 @@ handle_message(?ERROR, Error, Sock, #batch{batch = [_ | Batch]} = State) ->
     {add_result, Result, Result, Sock, State#batch{batch = Batch}};
 handle_message(_, _, _, _) ->
     unknown.
+
+%% Helpers
+
+current_cols(Batch) ->
+    #statement{columns = Columns} = current_stmt(Batch),
+    Columns.
+
+current_stmt(#batch{batch = [{Stmt, _} | _], statement = undefined}) ->
+    Stmt;
+current_stmt(#batch{statement = #statement{} = Stmt}) ->
+    Stmt.

src/commands/epgsql_cmd_bind.erl

@@ -1,5 +1,11 @@
+%% @doc Binds placeholder parameters to prepared statement
+%%
+%% ```
 %% > Bind
 %% < BindComplete
+%% '''
+%% @see epgsql_cmd_parse
+%% @see epgsql_cmd_execute
 -module(epgsql_cmd_bind).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_close.erl

@@ -1,5 +1,9 @@
+%% @doc Closes statement / portal
+%%
+%% ```
 %% > Close
 %% < CloseComplete
+%% '''
 -module(epgsql_cmd_close).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_connect.erl

@@ -1,6 +1,8 @@
+%%% @doc Connects to the server and performs all the necessary handshakes.
+%%%
 %%% Special kind of command - it's exclusive: no other commands can run until
 %%% this one finishes.
-%%% It also uses some 'private' epgsql_sock's APIs
+%%% It also uses some "private" epgsql_sock's APIs
 %%%
 -module(epgsql_cmd_connect).
 -behaviour(epgsql_command).
@@ -47,19 +49,20 @@ init(#{host := _, username := _} = Opts) ->
 
 execute(PgSock, #connect{opts = #{host := Host} = Opts, stage = connect} = State) ->
     Timeout = maps:get(timeout, Opts, 5000),
+    Deadline = deadline(Timeout),
     Port = maps:get(port, Opts, 5432),
     SockOpts = [{active, false}, {packet, raw}, binary, {nodelay, true}, {keepalive, true}],
     case gen_tcp:connect(Host, Port, SockOpts, Timeout) of
         {ok, Sock} ->
-            client_handshake(Sock, PgSock, State);
+            client_handshake(Sock, PgSock, State, Deadline);
         {error, Reason} = Error ->
             {stop, Reason, Error, PgSock}
     end;
 execute(PgSock, #connect{stage = auth, auth_send = {PacketId, Data}} = St) ->
-    epgsql_sock:send(PgSock, PacketId, Data),
+    ok = epgsql_sock:send(PgSock, PacketId, Data),
     {ok, PgSock, St#connect{auth_send = undefined}}.
 
-client_handshake(Sock, PgSock, #connect{opts = #{username := Username} = Opts} = State) ->
+client_handshake(Sock, PgSock, #connect{opts = #{username := Username} = Opts} = State, Deadline) ->
     %% Increase the buffer size.  Following the recommendation in the inet man page:
     %%
     %%    It is recommended to have val(buffer) >=
@@ -69,7 +72,7 @@ client_handshake(Sock, PgSock, #connect{opts = #{username := Username} = Opts} =
         inet:getopts(Sock, [recbuf, sndbuf]),
     inet:setopts(Sock, [{buffer, max(RecBufSize, SndBufSize)}]),
 
-    case maybe_ssl(Sock, maps:get(ssl, Opts, false), Opts, PgSock) of
+    case maybe_ssl(Sock, maps:get(ssl, Opts, false), Opts, PgSock, Deadline) of
         {error, Reason} ->
             {stop, Reason, {error, Reason}, PgSock};
         PgSock1 ->
@@ -86,8 +89,7 @@ client_handshake(Sock, PgSock, #connect{opts = #{username := Username} = Opts} =
                          epgsql_sock:init_replication_state(PgSock1)};
                     _ -> {Opts3, PgSock1}
                 end,
-
-            epgsql_sock:send(PgSock2, [<<196608:?int32>>, Opts4, 0]),
+            ok = epgsql_sock:send(PgSock2, [<<196608:?int32>>, Opts4, 0]),
             PgSock3 = case Opts of
                           #{async := Async} ->
                               epgsql_sock:set_attr(async, Async, PgSock2);
@@ -105,7 +107,7 @@ opts_hide_password(Opts) -> Opts.
 
 
 %% @doc this function wraps plaintext password to a lambda function, so, if
-%% epgsql_sock process crashes when executing `connect` command, password will
+%% epgsql_sock process crashes when executing `connect' command, password will
 %% not appear in a crash log
 -spec hide_password(iodata()) -> fun( () -> iodata() ).
 hide_password(Password) when is_list(Password);
@@ -117,15 +119,15 @@ hide_password(PasswordFun) when is_function(PasswordFun, 0) ->
     PasswordFun.
 
 
-maybe_ssl(S, false, _, PgSock) ->
+maybe_ssl(S, false, _, PgSock, _Deadline) ->
     epgsql_sock:set_net_socket(gen_tcp, S, PgSock);
-maybe_ssl(S, Flag, Opts, PgSock) ->
+maybe_ssl(S, Flag, Opts, PgSock, Deadline) ->
     ok = gen_tcp:send(S, <<8:?int32, 80877103:?int32>>),
-    Timeout = maps:get(timeout, Opts, 5000),
-    {ok, <<Code>>} = gen_tcp:recv(S, 1, Timeout),
-    case Code of
-        $S  ->
+    Timeout0 = timeout(Deadline),
+    case gen_tcp:recv(S, 1, Timeout0) of
+        {ok, <<$S>>}  ->
             SslOpts = maps:get(ssl_opts, Opts, []),
+            Timeout = timeout(Deadline),
             case ssl:connect(S, SslOpts, Timeout) of
                 {ok, S2}        ->
                     epgsql_sock:set_net_socket(ssl, S2, PgSock);
@@ -133,13 +135,15 @@ maybe_ssl(S, Flag, Opts, PgSock) ->
                     Err = {ssl_negotiation_failed, Reason},
                     {error, Err}
             end;
-        $N ->
+        {ok, <<$N>>} ->
             case Flag of
                 true ->
                     epgsql_sock:set_net_socket(gen_tcp, S, PgSock);
                 required ->
                     {error, ssl_not_available}
-            end
+            end;
+        {error, Reason} ->
+            {error, Reason}
     end.
 
 %% Auth sub-protocol
@@ -226,7 +230,7 @@ auth_scram(_, _, _) ->
 handle_message(?AUTHENTICATION_REQUEST, <<?AUTH_OK:?int32>>, Sock, State) ->
     {noaction, Sock, State#connect{stage = initialization,
                                    auth_fun = undefined,
-                                   auth_state = undefned,
+                                   auth_state = undefined,
                                    auth_send = undefined}};
 
 handle_message(?AUTHENTICATION_REQUEST, Message, Sock, #connect{stage = Stage} = St) when Stage =/= auth ->
@@ -242,8 +246,9 @@ handle_message(?CANCELLATION_KEY, <<Pid:?int32, Key:?int32>>, Sock, _State) ->
     {noaction, epgsql_sock:set_attr(backend, {Pid, Key}, Sock)};
 
 %% ReadyForQuery
-handle_message(?READY_FOR_QUERY, _, Sock, _State) ->
-    Codec = epgsql_binary:new_codec(Sock, []),
+handle_message(?READY_FOR_QUERY, _, Sock, #connect{opts = Opts}) ->
+    CodecOpts = maps:with([nulls], Opts),
+    Codec = epgsql_binary:new_codec(Sock, CodecOpts),
     Sock1 = epgsql_sock:set_attr(codec, Codec, Sock),
     {finish, connected, connected, Sock1};
 
@@ -274,3 +279,9 @@ hex(Bin) ->
                (N) when N < 16 -> $W + N
             end,
     <<<<(HChar(H)), (HChar(L))>> || <<H:4, L:4>> <= Bin>>.
+
+deadline(Timeout) ->
+    erlang:monotonic_time(milli_seconds) + Timeout.
+
+timeout(Deadline) ->
+    erlang:max(0, Deadline - erlang:monotonic_time(milli_seconds)).

src/commands/epgsql_cmd_describe_portal.erl

@@ -1,5 +1,9 @@
-%% > Describe
+%% @doc Asks the server to provide description of portal's results columns
+%%
+%% ```
+%% > Describe(PORTAL)
 %% < RowDescription | NoData
+%% '''
 -module(epgsql_cmd_describe_portal).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).
@@ -26,12 +30,12 @@ execute(Sock, #desc_portal{name = Name} = St) ->
       ]),
     {ok, Sock, St}.
 
-handle_message(?ROW_DESCRIPTION, <<Count:?int16, Bin/binary>>, Sock, St) ->
+handle_message(?ROW_DESCRIPTION, <<Count:?int16, Bin/binary>>, Sock, _St) ->
     Codec = epgsql_sock:get_codec(Sock),
     Columns = epgsql_wire:decode_columns(Count, Bin, Codec),
-    {finish, {ok, Columns}, {columns, Columns}, St};
-handle_message(?NO_DATA, <<>>, _Sock, _State) ->
-    {finish, {ok, []}, no_data};
+    {finish, {ok, Columns}, {columns, Columns}, Sock};
+handle_message(?NO_DATA, <<>>, Sock, _State) ->
+    {finish, {ok, []}, no_data, Sock};
 handle_message(?ERROR, Error, _Sock, _State) ->
     Result = {error, Error},
     {sync_required, Result};

src/commands/epgsql_cmd_describe_statement.erl

@@ -1,7 +1,12 @@
-%% Almost the same as "parse"
-%% > Describe
+%% @doc Asks server to provide input parameter and result rows information.
+%%
+%% Almost the same as {@link epgsql_cmd_parse}.
+%%
+%% ```
+%% > Describe(STATEMENT)
 %% < ParameterDescription
 %% < RowDescription | NoData
+%% '''
 -module(epgsql_cmd_describe_statement).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_equery.erl

@@ -1,3 +1,10 @@
+%% @doc Performs 2nd stage of
+%% <a href="https://www.postgresql.org/docs/current/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY">
+%% extended query protocol.</a>
+%%
+%% Takes prepared `statement()' and bind-parameters for placeholders and produces
+%% query results.
+%% ```
 %% > Bind
 %% < BindComplete
 %% > Execute
@@ -7,6 +14,8 @@
 %% < CloseComplete
 %% > Sync
 %% < ReadyForQuery
+%% '''
+%% @see epgsql_cmd_parse
 -module(epgsql_cmd_equery).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_execute.erl

@@ -1,6 +1,14 @@
+%% @doc Executes a portal.
+%%
+%% It's possible to tell the server to only return limited number of rows by
+%% providing non-zero `MaxRows' parameter.
+%% ```
 %% > Execute
 %% < DataRow*
 %% < CommandComplete | PortalSuspended
+%% '''
+%% @see epgsql_cmd_parse
+%% @see epgsql_cmd_bind
 -module(epgsql_cmd_execute).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_parse.erl

@@ -1,8 +1,12 @@
+%% @doc Asks server to parse SQL query and send information aboud bind-parameters and result columns.
+%%
+%% ```
 %% > Parse
 %% < ParseComplete
 %% > Describe
 %% < ParameterDescription
 %% < RowDescription | NoData
+%% '''
 -module(epgsql_cmd_parse).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_prepared_query.erl

@@ -1,4 +1,7 @@
-%% Almost the same as equery, but don't execute 'CLOSE'
+%% @doc Almost the same as equery, but don't execute 'CLOSE'
+%%
+%% So, statement can be reused multiple times.
+%% ```
 %% > Bind
 %% < BindComplete
 %% > Execute
@@ -6,6 +9,7 @@
 %% < CommandComplete
 %% > Sync
 %% < ReadyForQuery
+%% '''
 -module(epgsql_cmd_prepared_query).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_squery.erl

@@ -1,4 +1,10 @@
+%% @doc Executes SQL query(es) using
+%% <a href="https://www.postgresql.org/docs/current/protocol-flow.html#id-1.10.5.7.4">
+%% simple query protocol</a>
+%%
+%% Squery can not have placeholders.
 %% Squery may contain many semicolon-separated queries
+%% ```
 %% > Query
 %% < (RowDescription?
 %% <  DataRow*
@@ -8,6 +14,7 @@
 %% > Query when len(strip(Query)) == 0
 %% < EmptyQueryResponse
 %% < ReadyForQuery
+%% '''
 -module(epgsql_cmd_squery).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_start_replication.erl

@@ -1,5 +1,10 @@
+%% @doc Requests server to start sending replication packets
+%%
+%% See {@link epgsql:connect/1} `replication' parameter.
+%% ```
 %% > SimpleQuery "START_REPLICATION ..."
 %% < CopyBothResponse | Error
+%% '''
 -module(epgsql_cmd_start_replication).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_sync.erl

@@ -1,5 +1,10 @@
+%% @doc Synchronize client and server states for multi-command combinations
+%%
+%% Should be executed if APIs start to return `{error, sync_required}'.
+%% ```
 %% > Sync
 %% < ReadyForQuery
+%% '''
 -module(epgsql_cmd_sync).
 -behaviour(epgsql_command).
 -export([init/1, execute/2, handle_message/4]).

src/commands/epgsql_cmd_update_type_cache.erl

@@ -1,4 +1,4 @@
-%% Special command. Executes Squery over pg_type table and updates codecs.
+%% @doc 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]).

src/datatypes/epgsql_codec_boolean.erl

@@ -1,8 +1,11 @@
 %%% @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
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/current/static/datatype-boolean.html]</li>
+%%%  <li>$PG$/src/backend/utils/adt/bool.c</li>
+%%% </ul>
 %%% @end
 %%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 

src/datatypes/epgsql_codec_bpchar.erl

@@ -1,9 +1,16 @@
 %%% @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
+%%% Codec for blank-padded fixed-size character type
+%%%
+%%% `CHAR' (single-byte) is represented as `byte()';
+%%% `CHARACTER(N) / CHAR(N)' as binary string
+%%%
+%%% <code>SELECT 1::char;</code> <code>SELECT 'abc'::char(10)</code>
+%%%
+%%% For 'text', 'varchar' see {@link epgsql_codec_text}.
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/10/static/datatype-character.html]</li>
+%%%  <li>$PG$/src/backend/utils/adt/varchar.c</li>
+%%% </ul>
 %%% @end
 %%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 

src/datatypes/epgsql_codec_datetime.erl

@@ -1,9 +1,16 @@
 %%% @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'
+%%%
+%%% It supports both integer and float datetime representations (see
+%%% [https://www.postgresql.org/docs/current/runtime-config-preset.html#GUC-INTEGER-DATETIMES]).
+%%% But float representation support might be eventually removed.
+%%%
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/current/static/datatype-datetime.html]</li>
+%%%  <li>$PG$/src/backend/utils/adt/timestamp.c // `timestamp', `timestamptz', `interval'</li>
+%%%  <li>$PG$/src/backend/utils/adt/datetime.c // helpers</li>
+%%%  <li>$PG$/src/backend/utils/adt/date.c // `time', `timetz', `date'</li>
+%%% </ul>
 %%% @end
 %%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 

src/datatypes/epgsql_codec_float.erl

@@ -1,7 +1,10 @@
 %%% @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
+%%%
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/current/static/datatype-numeric.html#datatype-float]</li>
+%%%  <li>$PG$/src/backend/utils/adt/float.c</li>
+%%% </ul>
 %%% @end
 %%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 

src/datatypes/epgsql_codec_geometric.erl

@@ -1,8 +1,14 @@
 %%% @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!
+%%%
+%%% Codecs for other geometric datatypes (line, box, path, polygon, circle) can
+%%% be added later.
+%%%
+%%% XXX: it's not PostGIS! For PostGIS see {@link epgsql_codec_postgis}.
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/current/static/datatype-geometric.html]</li>
+%%%  <li>$PG$/src/backend/utils/adt/geo_ops.c</li>
+%%% </ul>
 %%% @end
 %%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 %%% TODO: line, lseg, box, path, polygon, circle

src/datatypes/epgsql_codec_hstore.erl

@@ -1,10 +1,17 @@
 %%% @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/
+%%%
+%%% Hstore codec can take a jiffy-style object or map() as input.
+%%% Output format can be changed by providing `return' option. See {@link return_format()}.
+%%% Values of hstore can be `NULL'. NULL representation can be changed by providing
+%%% `nulls' option, semantics is similar to {@link epgsql:connect_opts()} `nulls' option.
+%%%
+%%% XXX: hstore is not a part of postgresql builtin datatypes, it's in contrib.
+%%% It should be enabled in postgresql by command `CREATE EXTENSION hstore'.
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/current/static/hstore.html]</li>
+%%%  <li>$PG$/contrib/hstore/</li>
+%%% </ul>
 %%% @end
 %%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 
@@ -15,43 +22,71 @@
 
 -include("protocol.hrl").
 
--export_type([data/0]).
+-export_type([data/0, options/0, return_format/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()}] }.
+-type data_in() :: { [{key_in(), binary()}] } |
+                   #{key_in() => binary() | atom()}.
+-type data_out() :: { [{Key :: binary(), Value :: binary()}] } |      % jiffy
+                    [{Key :: binary(), Value :: binary() | atom()}] | % proplist
+                    #{binary() => binary() | atom()}.                 % map
+
+-type return_format() :: map | jiffy | proplist.
+-type options() :: #{return => return_format(),
+                     nulls => [atom(), ...]}.
+
+-record(st,
+        {return :: return_format(),
+         nulls :: [atom(), ...]}).
 
 -dialyzer([{nowarn_function, [encode/3]}, no_improper_lists]).
 
-%% TODO: option for output format: proplist | jiffy-object | map
-init(_, _) -> [].
+init(Opts0, _) ->
+    Opts = epgsql:to_map(Opts0),
+    #st{return = maps:get(return, Opts, jiffy),
+        nulls = maps:get(nulls, Opts, [null, undefined])}.
 
 names() ->
     [hstore].
 
-encode({Hstore}, hstore, _) when is_list(Hstore) ->
-    Size = length(Hstore),
-    %% TODO: construct improper list when support for Erl 17 will be dropped
-    Body = [[encode_key(K), encode_value(V)]
-           || {K, V} <- Hstore],
-    [<<Size:?int32>> | Body].
+encode({KV}, hstore, #st{nulls = Nulls}) when is_list(KV) ->
+    Size = length(KV),
+    encode_kv(KV, Size, Nulls);
+encode(Map, hstore, #st{nulls = Nulls}) when is_map(Map) ->
+    Size = map_size(Map),
+    encode_kv(maps:to_list(Map), Size, Nulls).
+
+decode(<<Size:?int32, Elements/binary>>, hstore, #st{return = Return, nulls = Nulls}) ->
+    KV = do_decode(Size, Elements, hd(Nulls)),
+    case Return of
+        jiffy ->
+            {KV};
+        map ->
+            maps:from_list(KV);
+        proplist ->
+            KV
+    end.
+
+decode_text(V, _, _) -> V.
 
-decode(<<Size:?int32, Elements/binary>>, hstore, _) ->
-    {do_decode(Size, Elements)}.
+%% Internal
 
+encode_kv(KV, Size, Nulls) ->
+    %% TODO: construct improper list when support for Erl 17 will be dropped
+    Body = [[encode_key(K), encode_value(V, Nulls)]
+           || {K, V} <- KV],
+    [<<Size:?int32>> | Body].
 
 encode_key(K) ->
     encode_string(K).
 
-encode_value(null) ->
-    <<-1:?int32>>;
-encode_value(undefined) ->
-    <<-1:?int32>>;
-encode_value(V) ->
-    encode_string(V).
+encode_value(V, Nulls) ->
+    case lists:member(V, Nulls) of
+        true -> <<-1:?int32>>;
+        false -> encode_string(V)
+    end.
 
 encode_string(Str) when is_binary(Str) ->
     <<(byte_size(Str)):?int32, Str/binary>>;
@@ -66,11 +101,9 @@ encode_string(Str) when is_float(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(0, _, _) -> [];
+do_decode(N, <<KeyLen:?int32, Key:KeyLen/binary, -1:?int32, Rest/binary>>, Null) ->
+    [{Key, Null} | do_decode(N - 1, Rest, Null)];
 do_decode(N, <<KeyLen:?int32, Key:KeyLen/binary,
-               ValLen:?int32, Value:ValLen/binary, Rest/binary>>) ->
-    [{Key, Value} | do_decode(N - 1, Rest)].
-
-decode_text(V, _, _) -> V.
+               ValLen:?int32, Value:ValLen/binary, Rest/binary>>, Null) ->
+    [{Key, Value} | do_decode(N - 1, Rest, Null)].

src/datatypes/epgsql_codec_integer.erl

@@ -1,8 +1,11 @@
 %%% @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
+%%%
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/current/static/datatype-numeric.html#datatype-int]</li>
+%%%  <li>$PG$/src/backend/utils/adt/int.c</li>
+%%%  <li>$PG$/src/backend/utils/adt/int8.c</li>
+%%% </ul>
 %%% @end
 %%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 
@@ -10,16 +13,35 @@
 -behaviour(epgsql_codec).
 
 -export([init/2, names/0, encode/3, decode/3, decode_text/3]).
+-export([check_overflow_small/1, check_overflow_int/1, check_overflow_big/1]).
 
 -export_type([data/0]).
 
 %% See table 8.2
 %% https://www.postgresql.org/docs/current/static/datatype-numeric.html
+-define(SMALLINT_MAX, 16#7fff).  % 32767, (2^15 - 1)
+-define(SMALLINT_MIN, -16#8000). % -32768
+-define(INT_MAX, 16#7fffffff).  % 2147483647, (2^31 - 1)
+-define(INT_MIN, -16#80000000). % -2147483648
 -define(BIGINT_MAX, 16#7fffffffffffffff).  % 9223372036854775807, (2^63 - 1)
--define(BIGINT_MIN, -16#7fffffffffffffff). % -9223372036854775807
+-define(BIGINT_MIN, -16#8000000000000000). % -9223372036854775808
 
 -type data() :: ?BIGINT_MIN..?BIGINT_MAX.
 
+check_overflow_small(N) when N >= ?SMALLINT_MIN, N =< ?SMALLINT_MAX -> ok;
+check_overflow_small(N) ->
+    overflow(N, int2).
+
+check_overflow_int(N) when N >= ?INT_MIN, N =< ?INT_MAX -> ok;
+check_overflow_int(N) ->
+    overflow(N, int4).
+
+check_overflow_big(N) when N >= ?BIGINT_MIN, N =< ?BIGINT_MAX -> ok;
+check_overflow_big(N) ->
+    overflow(N, int8).
+
+overflow(N, Type) ->
+    error({integer_overflow, Type, N}).
 
 init(_, _) -> [].
 
@@ -27,10 +49,13 @@ names() ->
     [int2, int4, int8].
 
 encode(N, int2, _) ->
+    check_overflow_small(N),
     <<N:1/big-signed-unit:16>>;
 encode(N, int4, _) ->
+    check_overflow_int(N),
     <<N:1/big-signed-unit:32>>;
 encode(N, int8, _) ->
+    check_overflow_big(N),
     <<N:1/big-signed-unit:64>>.
 
 decode(<<N:1/big-signed-unit:16>>, int2, _)    -> N;

src/datatypes/epgsql_codec_intrange.erl

@@ -1,7 +1,12 @@
 %%% @doc
 %%% Codec for `int4range', `int8range' types.
-%%% https://www.postgresql.org/docs/current/static/rangetypes.html#rangetypes-builtin
-%%% $PG$/src/backend/utils/adt/rangetypes.c
+%%%
+%%% <ul>
+%%%   <li>[https://www.postgresql.org/docs/current/static/rangetypes.html#rangetypes-builtin]</li>
+%%%   <li>$PG$/src/backend/utils/adt/rangetypes.c</li>
+%%% </ul>
+%%% @end
+%%% @see epgsql_codec_integer
 %%% @end
 %%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 %%% TODO: universal range, based on pg_range table
@@ -16,7 +21,7 @@
 
 -export_type([data/0]).
 
--type data() :: {left(), right()}.
+-type data() :: {left(), right()} | empty.
 
 -type left() :: minus_infinity | integer().
 -type right() :: plus_infinity | integer().
@@ -27,11 +32,15 @@ init(_, _) -> [].
 names() ->
     [int4range, int8range].
 
+encode(empty, _, _) ->
+    <<1>>;
 encode(Range, int4range, _) ->
     encode_int4range(Range);
 encode(Range, int8range, _) ->
     encode_int8range(Range).
 
+decode(<<1>>, _, _) ->
+    empty;
 decode(Bin, int4range, _) ->
     decode_int4range(Bin);
 decode(Bin, int8range, _) ->
@@ -42,26 +51,34 @@ encode_int4range({minus_infinity, plus_infinity}) ->
     <<24:1/big-signed-unit:8>>;
 encode_int4range({From, plus_infinity}) ->
     FromInt = to_int(From),
+    epgsql_codec_integer:check_overflow_int(FromInt),
     <<18:1/big-signed-unit:8, 4:?int32, FromInt:?int32>>;
 encode_int4range({minus_infinity, To}) ->
     ToInt = to_int(To),
+    epgsql_codec_integer:check_overflow_int(ToInt),
     <<8:1/big-signed-unit:8, 4:?int32, ToInt:?int32>>;
 encode_int4range({From, To}) ->
     FromInt = to_int(From),
     ToInt = to_int(To),
+    epgsql_codec_integer:check_overflow_int(FromInt),
+    epgsql_codec_integer:check_overflow_int(ToInt),
     <<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),
+    epgsql_codec_integer:check_overflow_big(FromInt),
     <<18:1/big-signed-unit:8, 8:?int32, FromInt:?int64>>;
 encode_int8range({minus_infinity, To}) ->
     ToInt = to_int(To),
+    epgsql_codec_integer:check_overflow_big(ToInt),
     <<8:1/big-signed-unit:8, 8:?int32, ToInt:?int64>>;
 encode_int8range({From, To}) ->
     FromInt = to_int(From),
     ToInt = to_int(To),
+    epgsql_codec_integer:check_overflow_big(FromInt),
+    epgsql_codec_integer:check_overflow_big(ToInt),
     <<2:1/big-signed-unit:8, 8:?int32, FromInt:?int64, 8:?int32, ToInt:?int64>>.
 
 to_int(N) when is_integer(N) -> N;

src/datatypes/epgsql_codec_json.erl

@@ -1,8 +1,13 @@
 %%% @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'
+%%%
+%%% It is possible to instruct the codec to do JSON encoding/decoding to Erlang
+%%% terms by providing callback module name, see {@link json_mod()}.
+%%% <ul>
+%%%   <li>[https://www.postgresql.org/docs/current/static/datatype-json.html]</li>
+%%%   <li>$PG$/src/backend/utils/adt/json.c // `json'</li>
+%%%   <li>$PG$/src/backend/utils/adt/jsonb.c // `jsonb'</li>
+%%% </ul>
 %%% @end
 %%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 

src/datatypes/epgsql_codec_net.erl

@@ -1,9 +1,12 @@
 %%% @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.
+%%% TIP: use {@link inet:ntoa/1} and {@link inet:parse_address/1} to convert
+%%% between {@link ip()} and `string()'.
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/10/static/datatype-net-types.html]</li>
+%%%  <li>$PG$/src/backend/utils/adt/network.c</li>
+%%% </ul>
 %%% @end
 %%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 -module(epgsql_codec_net).

src/datatypes/epgsql_codec_noop.erl

@@ -1,3 +1,4 @@
+%%% @private
 %%% @doc
 %%% Dummy codec. Used internally
 %%% @end

src/datatypes/epgsql_codec_postgis.erl

@@ -1,7 +1,14 @@
 %%% @doc
 %%% Codec for `geometry' PostGIS umbrella datatype.
-%%% http://postgis.net/docs/manual-2.4/geometry.html
-%%% $POSTGIS$/postgis/lwgeom_inout.c
+%%%
+%%% XXX: PostGIS is not a Postgres's built-in datatype! It should be instaled
+%%% separately and enabled via `CREATE EXTENSION postgis'.
+%%% <ul>
+%%%  <li>[http://postgis.net/docs/manual-2.4/geometry.html]</li>
+%%%  <li>$POSTGIS$/postgis/lwgeom_inout.c</li>
+%%% </ul>
+%%% @end
+%%% @see ewkb
 %%% @end
 %%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 

src/datatypes/epgsql_codec_text.erl

@@ -1,9 +1,18 @@
 %%% @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
+%%%
+%%% If input for `text' or `varchar' is provided as a list, not binary, and it
+%%% contains not just `byte()', an attempt to perform unicode conversion will be made.
+%%%
+%%% Also, `integer()', `float()' and `atom()' are automatically converted to
+%%% strings, but this kind of conversion might be eventualy removed.
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/10/static/datatype-character.html]</li>
+%%%  <li>$PG$/src/backend/utils/adt/varchar.c</li>
+%%%  <li>$PG$/src/backend/utils/adt/varlena.c</li>
+%%% </ul>
+%%% @end
+%%% @see epgsql_codec_bpchar. epgsql_codec_bpchar - for 'char' and 'char(N)'
 %%% @end
 %%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 

src/datatypes/epgsql_codec_timerange.erl

@@ -1,7 +1,12 @@
 %%% @doc
 %%% Codec for `tsrange', `tstzrange', `daterange' types.
-%%% https://www.postgresql.org/docs/current/static/rangetypes.html#rangetypes-builtin
-%%% $PG$/src/backend/utils/adt/rangetypes.c
+%%%
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/current/static/rangetypes.html#rangetypes-builtin]</li>
+%%%  <li>$PG$/src/backend/utils/adt/rangetypes.c</li>
+%%% </ul>
+%%% @end
+%%% @see epgsql_codec_datetime
 %%% @end
 %%% Created : 16 Jul 2018 by Vladimir Sekissov <eryx67@gmail.com>
 %%% TODO: universal range, based on pg_range table
@@ -16,7 +21,10 @@
 
 -export_type([data/0]).
 
--type data() :: {epgsql_codec_datetime:data(), epgsql_codec_datetime:data()} | empty.
+-type data() :: {left(), right()} | empty.
+
+-type left() :: minus_infinity | epgsql_codec_datetime:data().
+-type right() :: plus_infinity | epgsql_codec_datetime:data().
 
 init(_, Sock) ->
     case epgsql_sock:get_parameter_internal(<<"integer_datetimes">>, Sock) of
@@ -29,6 +37,14 @@ names() ->
 
 encode(empty, _T, _CM) ->
     <<1>>;
+encode({minus_infinity, plus_infinity}, _T, _CM) ->
+    <<24:1/big-signed-unit:8>>;
+encode({From, plus_infinity}, Type, EncMod) ->
+    FromBin = encode_member(Type, From, EncMod),
+    <<18:1/big-signed-unit:8, (byte_size(FromBin)):?int32, FromBin/binary>>;
+encode({minus_infinity, To}, Type, EncMod) ->
+    ToBin = encode_member(Type, To, EncMod),
+    <<8:1/big-signed-unit:8, (byte_size(ToBin)):?int32, ToBin/binary>>;
 encode({From, To}, Type, EncMod) ->
     FromBin = encode_member(Type, From, EncMod),
     ToBin = encode_member(Type, To, EncMod),
@@ -38,11 +54,19 @@ encode({From, To}, Type, EncMod) ->
 
 decode(<<1>>, _, _) ->
     empty;
-decode(<<2:1/big-signed-unit:8,
+decode(<<Flag:1/big-signed-unit:8,
          FromLen:?int32, FromBin:FromLen/binary,
          ToLen:?int32, ToBin:ToLen/binary>>,
-       Type, EncMod) ->
-    {decode_member(Type, FromBin, EncMod), decode_member(Type, ToBin, EncMod)}.
+       Type, EncMod) when Flag =:= 0; Flag =:= 2; Flag =:= 4; Flag =:= 6 -> %% () [) (] []
+    {decode_member(Type, FromBin, EncMod), decode_member(Type, ToBin, EncMod)};
+decode(<<Flag:1/big-signed-unit:8, ToLen:?int32, ToBin:ToLen/binary>>,
+    Type, EncMod) when Flag =:= 8; Flag =:= 12 -> %% (] ()
+    {minus_infinity, decode_member(Type, ToBin, EncMod)};
+decode(<<Flag:1/big-signed-unit:8, FromLen:?int32, FromBin:FromLen/binary>>,
+    Type, EncMod) when Flag =:= 16; Flag =:= 18 -> %% [) ()
+    {decode_member(Type, FromBin, EncMod), plus_infinity};
+decode(<<24:1/big-signed-unit:8>>, _, _) ->
+    {minus_infinity, plus_infinity}.
 
 decode_text(V, _, _) -> V.
 

src/datatypes/epgsql_codec_uuid.erl

@@ -1,9 +1,12 @@
 %%% @doc
 %%% Codec for `uuid' type.
-%%% Input expected to be in hex string, eg
+%%%
+%%% Input is expected to be in hex `string()' / `binary()', eg
 %%% `<<"550e8400-e29b-41d4-a716-446655440000">>'.
-%%% https://www.postgresql.org/docs/current/static/datatype-uuid.html
-%%% $PG$/src/backend/utils/adt/uuid.c
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/current/static/datatype-uuid.html]</li>
+%%%  <li>$PG$/src/backend/utils/adt/uuid.c</li>
+%%% </ul>
 %%% @end
 %%% Created : 14 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 

src/epgsql.app.src

@@ -1,6 +1,6 @@
 {application, epgsql,
  [{description, "PostgreSQL Client"},
-  {vsn, "4.3.0"},
+  {vsn, "4.4.0"},
   {modules, []},
   {registered, []},
   {applications, [kernel,

src/epgsql.erl

@@ -1,3 +1,7 @@
+%%% @doc Synchronous interface.
+%%%
+%%% All functions block (with infinite timeout) until full result is available.
+%%% @end
 %%% Copyright (C) 2008 - Will Glozer.  All rights reserved.
 %%% Copyright (C) 2011 - Anton Lebedevich.  All rights reserved.
 
@@ -15,7 +19,7 @@
          describe/2, describe/3,
          bind/3, bind/4,
          execute/2, execute/3, execute/4,
-         execute_batch/2,
+         execute_batch/2, execute_batch/3,
          close/2, close/3,
          sync/1,
          cancel/1,
@@ -42,7 +46,7 @@
 
 -include("epgsql.hrl").
 
--type sql_query() :: iodata().
+-type sql_query() :: iodata(). % SQL query text
 -type host() :: inet:ip_address() | inet:hostname().
 -type password() :: string() | iodata() | fun( () -> iodata() ).
 -type connection() :: pid().
@@ -57,6 +61,7 @@
     {timeout,  TimeoutMs  :: timeout()}            | % default: 5000 ms
     {async,    Receiver   :: pid() | atom()}       | % process to receive LISTEN/NOTIFY msgs
     {codecs,   Codecs     :: [{epgsql_codec:codec_mod(), any()}]} |
+    {nulls,    Nulls      :: [any(), ...]} |    % terms to be used as NULL
     {replication, Replication :: string()}. % Pass "database" to connect in replication mode
 
 -type connect_opts() ::
@@ -71,10 +76,11 @@
           timeout => timeout(),
           async => pid() | atom(),
           codecs => [{epgsql_codec:codec_mod(), any()}],
+          nulls => [any(), ...],
           replication => string()}.
 
 -type connect_error() :: epgsql_cmd_connect:connect_error().
--type query_error() :: #error{}.
+-type query_error() :: #error{}.              % Error report generated by server
 
 
 -type type_name() :: atom().
@@ -88,15 +94,15 @@
 -type pg_datetime() :: epgsql_codec_datetime:pg_datetime().
 -type pg_interval() :: epgsql_codec_datetime:pg_interval().
 
-%% Deprecated
 -type bind_param() :: any().
+%% Value to be bound to placeholder (`$1', `$2' etc)
 
 -type typed_param() :: {epgsql_type(), bind_param()}.
 
 -type column() :: #column{}.
 -type statement() :: #statement{}.
 -type squery_row() :: tuple(). % tuple of binary().
--type equery_row() :: tuple(). % tuple of bind_param().
+-type equery_row() :: tuple(). % tuple of any().
 -type ok_reply(RowType) ::
         %% select
     {ok, ColumnsDescription :: [column()], RowsValues :: [RowType]} |
@@ -124,6 +130,7 @@
 %% -------------
 
 %% -- client interface --
+%% @doc connects to the server and performs all the necessary handshakes
 -spec connect(connect_opts())
         -> {ok, Connection :: connection()} | {error, Reason :: connect_error()}.
 connect(Opts) ->
@@ -138,13 +145,13 @@ connect(Host, Username, Opts) ->
 
 -spec connect(host(), string(), password(), connect_opts())
         -> {ok, Connection :: connection()} | {error, Reason :: connect_error()}.
-%% @doc connects to Postgres
-%% where
-%% `Host'     - host to connect to
-%% `Username' - username to connect as, defaults to `$USER'
-%% `Password' - optional password to authenticate with
-%% `Opts'     - proplist or map of extra options
-%% returns `{ok, Connection}' otherwise `{error, Reason}'
+%% @doc connects to the server and performs all the necessary handshakes (legacy interface)
+%% @param Host     host to connect to
+%% @param Username username to connect as, defaults to `$USER'
+%% @param Password optional password to authenticate with
+%% @param Opts     proplist or map of extra options
+%% @returns `{ok, Connection}' otherwise `{error, Reason}'
+%% @see connect/1
 connect(Host, Username, Password, Opts) ->
     {ok, C} = epgsql_sock:start_link(),
     connect(C, Host, Username, Password, Opts).
@@ -214,38 +221,37 @@ get_parameter(C, Name) ->
 set_notice_receiver(C, PidOrName) ->
     epgsql_sock:set_notice_receiver(C, PidOrName).
 
-%% @doc Returns last command status message
-%% If multiple queries were executed using `squery/2', separated by semicolon,
+%% @doc Returns last command status message.
+%% If multiple queries were executed using {@link squery/2}, separated by semicolon,
 %% only the last query's status will be available.
-%% See https://www.postgresql.org/docs/current/static/libpq-exec.html#LIBPQ-PQCMDSTATUS
+%% See [https://www.postgresql.org/docs/current/static/libpq-exec.html#LIBPQ-PQCMDSTATUS]
 -spec get_cmd_status(connection()) -> {ok, Status}
                                           when
       Status :: undefined | atom() | {atom(), integer()}.
 get_cmd_status(C) ->
     epgsql_sock:get_cmd_status(C).
 
--spec squery(connection(), sql_query()) -> epgsql_cmd_squery:response().
+-spec squery(connection(), sql_query()) -> epgsql_cmd_squery:response() | epgsql_sock:error().
 %% @doc runs simple `SqlQuery' via given `Connection'
+%% @see epgsql_cmd_squery
 squery(Connection, SqlQuery) ->
     epgsql_sock:sync_command(Connection, epgsql_cmd_squery, SqlQuery).
 
 equery(C, Sql) ->
     equery(C, Sql, []).
 
-%% TODO add fast_equery command that doesn't need parsed statement
 -spec equery(connection(), sql_query(), [bind_param()]) ->
-                    epgsql_cmd_equery:response().
+                    epgsql_cmd_equery:response() | epgsql_sock:error().
 equery(C, Sql, Parameters) ->
-    case parse(C, "", Sql, []) of
-        {ok, #statement{types = Types} = S} ->
-            TypedParameters = lists:zip(Types, Parameters),
-            epgsql_sock:sync_command(C, epgsql_cmd_equery, {S, TypedParameters});
-        Error ->
-            Error
-    end.
+    equery(C, "", Sql, Parameters).
 
+%% @doc Executes extended query
+%% @end
+%% @see epgsql_cmd_equery
+%% @end
+%% TODO add fast_equery command that doesn't need parsed statement
 -spec equery(connection(), string(), sql_query(), [bind_param()]) ->
-                    epgsql_cmd_equery:response().
+                    epgsql_cmd_equery:response() | epgsql_sock:error().
 equery(C, Name, Sql, Parameters) ->
     case parse(C, Name, Sql, []) of
         {ok, #statement{types = Types} = S} ->
@@ -255,13 +261,17 @@ equery(C, Name, Sql, Parameters) ->
             Error
     end.
 
--spec prepared_query(C::connection(), Name::string(), Parameters::[bind_param()]) ->
+%% @doc Similar to {@link equery/3}, but uses prepared statement that can be reused multiple times.
+%% @see epgsql_cmd_prepared_query
+-spec prepared_query(C::connection(), string() | statement(), Parameters::[bind_param()]) ->
                             epgsql_cmd_prepared_query:response().
-prepared_query(C, Name, Parameters) ->
+prepared_query(C, #statement{types = Types} = S, Parameters) ->
+    TypedParameters = lists:zip(Types, Parameters),
+    epgsql_sock:sync_command(C, epgsql_cmd_prepared_query, {S, TypedParameters});
+prepared_query(C, Name, Parameters) when is_list(Name) ->
     case describe(C, statement, Name) of
-        {ok, #statement{types = Types} = S} ->
-            TypedParameters = lists:zip(Types, Parameters),
-            epgsql_sock:sync_command(C, epgsql_cmd_prepared_query, {S, TypedParameters});
+        {ok, #statement{} = S} ->
+            prepared_query(C, S, Parameters);
         Error ->
             Error
     end.
@@ -308,11 +318,28 @@ execute(C, S, N) ->
 execute(C, S, PortalName, N) ->
     epgsql_sock:sync_command(C, epgsql_cmd_execute, {S, PortalName, N}).
 
+%% @doc Executes batch of `{statement(), [bind_param()]}' extended queries
+%% @see epgsql_cmd_batch
 -spec execute_batch(connection(), [{statement(), [bind_param()]}]) ->
                            epgsql_cmd_batch:response().
 execute_batch(C, Batch) ->
     epgsql_sock:sync_command(C, epgsql_cmd_batch, Batch).
 
+%% @doc Executes same statement() extended query with each parameter list of a `Batch'
+%% @see epgsql_cmd_batch
+-spec execute_batch(connection(), statement() | sql_query(), [ [bind_param()] ]) ->
+                           {[column()], epgsql_cmd_batch:response()}.
+execute_batch(C, #statement{columns = Cols} = Statement, Batch) ->
+    {Cols, epgsql_sock:sync_command(C, epgsql_cmd_batch, {Statement, Batch})};
+execute_batch(C, Sql, Batch) ->
+    case parse(C, Sql) of
+        {ok, #statement{} = S} ->
+            execute_batch(C, S, Batch);
+        Error ->
+            Error
+    end.
+
+
 %% statement/portal functions
 -spec describe(connection(), statement()) -> epgsql_cmd_describe_statement:response().
 describe(C, #statement{name = Name}) ->
@@ -335,6 +362,7 @@ describe(C, portal, Name) ->
 close(C, #statement{name = Name}) ->
     close(C, statement, Name).
 
+%% @doc close statement or portal
 -spec close(connection(), statement | portal, iodata()) -> epgsql_cmd_close:response().
 close(C, Type, Name) ->
     epgsql_sock:sync_command(C, epgsql_cmd_close, {Type, Name}).
@@ -343,6 +371,7 @@ close(C, Type, Name) ->
 sync(C) ->
     epgsql_sock:sync_command(C, epgsql_cmd_sync, []).
 
+%% @doc cancel currently executing command
 -spec cancel(connection()) -> ok.
 cancel(C) ->
     epgsql_sock:cancel(C).
@@ -358,15 +387,20 @@ with_transaction(C, F) ->
 %% @doc Execute callback function with connection in a transaction.
 %% Transaction will be rolled back in case of exception.
 %% Options (proplist or map):
-%% - reraise (true): when set to true, exception will be re-thrown, otherwise
-%%   {rollback, ErrorReason} will be returned
-%% - ensure_comitted (false): even when callback returns without exception,
+%% <dl>
+%%  <dt>reraise</dt>
+%%  <dd>when set to true, exception will be re-thrown, otherwise
+%%   `{rollback, ErrorReason}' will be returned. Default: `true'</dd>
+%%  <dt>ensure_comitted</dt>
+%%  <dd>even when callback returns without exception,
 %%   check that transaction was comitted by checking CommandComplete status
 %%   of "COMMIT" command. In case when transaction was rolled back, status will be
-%%   "rollback" instead of "commit".
-%% - begin_opts (""): append extra options to "BEGIN" command (see
+%%   "rollback" instead of "commit". Default: `false'</dd>
+%%  <dt>begin_opts</dt>
+%%  <dd>append extra options to "BEGIN" command (see
 %%   https://www.postgresql.org/docs/current/static/sql-begin.html)
-%%   Beware of SQL injections! No escaping is made on begin_opts!
+%%   Beware of SQL injections! No escaping is made on begin_opts! Default: `""'</dd>
+%% </dl>
 -spec with_transaction(
         connection(), fun((connection()) -> Reply), Opts) -> Reply | {rollback, any()} | no_return() when
       Reply :: any(),
@@ -431,18 +465,17 @@ handle_x_log_data(Mod, StartLSN, EndLSN, WALRecord, Repl) ->
     Response :: epgsql_cmd_start_replication:response(),
     Callback :: module() | pid().
 %% @doc instructs Postgres server to start streaming WAL for logical replication
-%% where
-%% `Connection'      - connection in replication mode
-%% `ReplicationSlot' - the name of the replication slot to stream changes from
-%% `Callback'        - Callback module which should have the callback functions implemented for message processing.
-%%                      or a process which should be able to receive replication messages.
-%% `CbInitState'     - Callback Module's initial state
-%% `WALPosition'     - the WAL position XXX/XXX to begin streaming at.
-%%                      "0/0" to let the server determine the start point.
-%% `PluginOpts'      - optional options passed to the slot's logical decoding plugin.
-%%                      For example: "option_name1 'value1', option_name2 'value2'"
-%% `Opts'            - options of logical replication
-%% returns `ok' otherwise `{error, Reason}'
+%% @param Connection      connection in replication mode
+%% @param ReplicationSlot the name of the replication slot to stream changes from
+%% @param Callback        Callback module which should have the callback functions implemented for message processing.
+%%                        or a process which should be able to receive replication messages.
+%% @param CbInitState     Callback Module's initial state
+%% @param WALPosition     the WAL position XXX/XXX to begin streaming at.
+%%                        "0/0" to let the server determine the start point.
+%% @param PluginOpts      optional options passed to the slot's logical decoding plugin.
+%%                        For example: "option_name1 'value1', option_name2 'value2'"
+%% @param Opts            options of logical replication
+%% @returns `ok' otherwise `{error, Reason}'
 start_replication(Connection, ReplicationSlot, Callback, CbInitState, WALPosition, PluginOpts, Opts) ->
     Command = {ReplicationSlot, Callback, CbInitState, WALPosition, PluginOpts, to_map(Opts)},
     epgsql_sock:sync_command(Connection, epgsql_cmd_start_replication, Command).

src/epgsql_binary.erl

@@ -1,9 +1,17 @@
+%%% @doc
+%%% Interface to encoder/decoder for binary postgres data representation
+%%% @end
+%%% @see epgsql_codec
+%%% @see epgsql_wire
+%%% @end
 %%% Copyright (C) 2008 - Will Glozer.  All rights reserved.
-%% XXX: maybe merge this module into epgsql_codec?
+%%% XXX: maybe merge this module into epgsql_codec?
 -module(epgsql_binary).
 
 -export([new_codec/2,
          update_codec/2,
+         null/1,
+         is_null/2,
          type_to_oid/2,
          typeinfo_to_name_array/2,
          typeinfo_to_oid_info/2,
@@ -17,10 +25,24 @@
 -export_type([codec/0, decoder/0]).
 
 -include("protocol.hrl").
+-define(DEFAULT_NULLS, [null, undefined]).
 
 -record(codec,
-        {opts = [] :: list(),                   % not used yet
+        {opts = #{} :: opts(),                   % not used yet
+         nulls = ?DEFAULT_NULLS :: nulls(),
          oid_db :: epgsql_oid_db:db()}).
+-record(array_decoder,
+        {element_decoder :: decoder(),
+         null_term :: any() }).
+-record(array_encoder,
+        {element_encoder :: epgsql_codec:codec_entry(),
+         n_dims = 0 :: non_neg_integer(),
+         lengths = [] :: [non_neg_integer()],
+         has_null = false :: boolean(),
+         codec :: codec()}).
+
+-type nulls() :: [any(), ...].
+-type opts() :: #{nulls => nulls()}.
 
 -opaque codec() :: #codec{}.
 -opaque decoder() :: {fun((binary(), epgsql:type_name(), epgsql_codec:codec_state()) -> any()),
@@ -36,7 +58,7 @@
 %% Codec is used to convert data (result rows and query parameters) between Erlang and postgresql formats
 %% It uses mappings between OID, type names and `epgsql_codec_*' modules (epgsql_oid_db)
 
--spec new_codec(epgsql_sock:pg_sock(), list()) -> codec().
+-spec new_codec(epgsql_sock:pg_sock(), opts()) -> codec().
 new_codec(PgSock, Opts) ->
     Codecs = default_codecs(),
     Oids = default_oids(),
@@ -45,7 +67,9 @@ new_codec(PgSock, Opts) ->
 new_codec(PgSock, Codecs, Oids, Opts) ->
     CodecEntries = epgsql_codec:init_mods(Codecs, PgSock),
     Types = epgsql_oid_db:join_codecs_oids(Oids, CodecEntries),
-    #codec{oid_db = epgsql_oid_db:from_list(Types), opts = Opts}.
+    #codec{oid_db = epgsql_oid_db:from_list(Types),
+           nulls = maps:get(nulls, Opts, ?DEFAULT_NULLS),
+           opts = Opts}.
 
 -spec update_codec([epgsql_oid_db:type_info()], codec()) -> codec().
 update_codec(TypeInfos, #codec{oid_db = Db} = Codec) ->
@@ -63,6 +87,16 @@ oid_to_name(Oid, Codec) ->
             end
     end.
 
+%% @doc Return the value that represents NULL (1st element of `nulls' list)
+-spec null(codec()) -> any().
+null(#codec{nulls = [Null | _]}) ->
+    Null.
+
+%% @doc Returns `true' if `Value' is a term representing `NULL'
+-spec is_null(any(), codec()) -> boolean().
+is_null(Value, #codec{nulls = Nulls}) ->
+    lists:member(Value, Nulls).
+
 -spec type_to_oid(type(), codec()) -> epgsql_oid_db:oid().
 type_to_oid({array, Name}, Codec) ->
     type_to_oid(Name, true, Codec);
@@ -117,28 +151,30 @@ decode(Bin, {Fun, TypeName, State}) ->
 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, #codec{oid_db = Db}) ->
+    {fun ?MODULE:decode_array/3, array,
+     #array_decoder{
+        element_decoder = oid_to_decoder(?RECORD_OID, binary, Codec),
+        null_term = null(Codec)}};
+oid_to_decoder(Oid, Format, #codec{oid_db = Db} = Codec) ->
     case epgsql_oid_db: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)
+            make_decoder(Type, Format, Codec)
     end.
 
--spec make_decoder(epgsql_oid_db:type_info(), binary | text) -> decoder().
-make_decoder(Type, Format) ->
+-spec make_decoder(epgsql_oid_db:type_info(), binary | text, codec()) -> decoder().
+make_decoder(Type, Format, Codec) ->
     {Name, Mod, State} = epgsql_oid_db:type_to_codec_entry(Type),
     {_Oid, Name, IsArray} = epgsql_oid_db:type_to_oid_info(Type),
-    make_decoder(Name, Mod, State, Format, IsArray).
+    make_decoder(Name, Mod, State, Codec, Format, IsArray).
 
-make_decoder(_Name, _Mod, _State, text, true) ->
+make_decoder(_Name, _Mod, _State, _Codec, text, true) ->
     %% Don't try to decode text arrays
     {fun epgsql_codec_noop:decode_text/3, undefined, []};
-make_decoder(Name, Mod, State, text, false) ->
+make_decoder(Name, Mod, State, _Codec, 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 ->
@@ -146,18 +182,18 @@ make_decoder(Name, Mod, State, text, false) ->
         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) ->
+make_decoder(Name, Mod, State, Codec, binary, true) ->
+    {fun ?MODULE:decode_array/3, array,
+     #array_decoder{
+        element_decoder = {fun Mod:decode/3, Name, State},
+        null_term = null(Codec)}};
+make_decoder(Name, Mod, State, _Codec, 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) ->
+decode_array(<<NDims:?int32, _HasNull:?int32, _Oid:?int32, Rest/binary>>, _, ArrayDecoder) ->
     %% 4b: n_dimensions;
     %% 4b: flags;
     %% 4b: Oid // should be the same as in column spec;
@@ -168,27 +204,29 @@ decode_array(<<NDims:?int32, _HasNull:?int32, _Oid:?int32, Rest/binary>>, _, Ele
     %% 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, <<>>} = decode_array1(Data, Lengths, ArrayDecoder),
     Array.
 
 decode_array1(Data, [], _)  ->
     %% zero-dimensional array
     {[], Data};
-decode_array1(Data, [Len], ElemDecoder) ->
+decode_array1(Data, [Len], ArrayDecoder) ->
     %% 1-dimensional array
-    decode_elements(Data, [], Len, ElemDecoder);
-decode_array1(Data, [Len | T], ElemDecoder) ->
+    decode_elements(Data, [], Len, ArrayDecoder);
+decode_array1(Data, [Len | T], ArrayDecoder) ->
     %% multidimensional array
-    F = fun(_N, Rest) -> decode_array1(Rest, T, ElemDecoder) end,
+    F = fun(_N, Rest) -> decode_array1(Rest, T, ArrayDecoder) end,
     lists:mapfoldl(F, Data, lists:seq(1, Len)).
 
-decode_elements(Rest, Acc, 0, _ElDec) ->
+decode_elements(Rest, Acc, 0, _ArDec) ->
     {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) ->
+decode_elements(<<-1:?int32, Rest/binary>>, Acc, N,
+                #array_decoder{null_term = Null} = ArDec) ->
+    decode_elements(Rest, [Null | Acc], N - 1, ArDec);
+decode_elements(<<Len:?int32, Value:Len/binary, Rest/binary>>, Acc, N,
+                #array_decoder{element_decoder = ElemDecoder} = ArDecoder) ->
     Value2 = decode(Value, ElemDecoder),
-    decode_elements(Rest, [Value2 | Acc], N - 1, ElemDecoder).
+    decode_elements(Rest, [Value2 | Acc], N - 1, ArDecoder).
 
 
 
@@ -199,7 +237,7 @@ decode_record(<<Size:?int32, Bin/binary>>, record, Codec) ->
 
 decode_record1(<<>>, 0, _Codec) -> [];
 decode_record1(<<_Type:?int32, -1:?int32, Rest/binary>>, Size, Codec) ->
-    [null | decode_record1(Rest, Size - 1, Codec)];
+    [null(Codec) | decode_record1(Rest, Size - 1, Codec)];
 decode_record1(<<Oid:?int32, Len:?int32, ValueBin:Len/binary, Rest/binary>>, Size, Codec) ->
     Value = decode(ValueBin, oid_to_decoder(Oid, binary, Codec)),
     [Value | decode_record1(Rest, Size - 1, Codec)].
@@ -213,44 +251,73 @@ decode_record1(<<Oid:?int32, Len:?int32, ValueBin:Len/binary, Rest/binary>>, Siz
 -spec encode(epgsql:type_name() | {array, epgsql:type_name()}, any(), codec()) -> iolist().
 encode(TypeName, Value, Codec) ->
     Type = type_to_type_info(TypeName, Codec),
-    encode_with_type(Type, Value).
+    encode_with_type(Type, Value, Codec).
 
-encode_with_type(Type, Value) ->
-    {Name, Mod, State} = epgsql_oid_db:type_to_codec_entry(Type),
+encode_with_type(Type, Value, Codec) ->
+    NameModState = epgsql_oid_db:type_to_codec_entry(Type),
     case epgsql_oid_db:type_to_oid_info(Type) of
         {_ArrayOid, _, true} ->
             %FIXME: check if this OID is the same as was returned by 'Describe'
             ElementOid = epgsql_oid_db:type_to_element_oid(Type),
-            encode_array(Value, ElementOid, {Mod, Name, State});
+            encode_array(Value, ElementOid,
+                         #array_encoder{
+                            element_encoder = NameModState,
+                            codec = Codec});
         {_Oid, _, false} ->
-            encode_value(Value, {Mod, Name, State})
+            encode_value(Value, NameModState)
     end.
 
-encode_value(Value, {Mod, Name, State}) ->
+encode_value(Value, {Name, Mod, State}) ->
     Payload = epgsql_codec:encode(Mod, 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),
+encode_array(Array, Oid, ArrayEncoder) ->
+    {Data, {NDims, Lengths, HasNull}} = encode_array_dims(Array, ArrayEncoder),
     Lens = [<<N:?int32, 1:?int32>> || N <- lists:reverse(Lengths)],
-    Hdr  = <<NDims:?int32, 0:?int32, Oid:?int32>>,
+    HasNullInt = case HasNull of
+                     true -> 1;
+                     false -> 0
+                 end,
+    Hdr  = <<NDims:?int32, HasNullInt:?int32, Oid:?int32>>,
     Payload  = [Hdr, Lens, Data],
     [<<(iolist_size(Payload)):?int32>> | Payload].
 
-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(Array, NDims, Lengths, Codec) ->
-    Lengths2 = [length(Array) | Lengths],
-    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_array_dims([], #array_encoder{n_dims = NDims,
+                                     lengths = Lengths,
+                                     has_null = HasNull}) ->
+    {[], {NDims, Lengths, HasNull}};
+encode_array_dims([H | _] = Array,
+                  #array_encoder{n_dims = NDims0,
+                                 lengths = Lengths0,
+                                 has_null = HasNull0,
+                                 codec = Codec,
+                                 element_encoder = ValueEncoder}) when not is_list(H) ->
+    F = fun(El, {Len, HasNull1}) ->
+                case is_null(El, Codec) of
+                    false ->
+                        {encode_value(El, ValueEncoder), {Len + 1, HasNull1}};
+                    true ->
+                        {<<-1:?int32>>, {Len + 1, true}}
+                end
+        end,
+    {Data, {Len, HasNull2}} = lists:mapfoldl(F, {0, HasNull0}, Array),
+    {Data, {NDims0 + 1, [Len | Lengths0], HasNull2}};
+encode_array_dims(Array, #array_encoder{lengths = Lengths0,
+                                        n_dims = NDims0,
+                                        has_null = HasNull0} = ArrayEncoder) ->
+    Lengths1 = [length(Array) | Lengths0],
+    F = fun(A2, {_NDims, _Lengths, HasNull1}) ->
+                encode_array_dims(A2, ArrayEncoder#array_encoder{
+                                   n_dims = NDims0,
+                                   has_null = HasNull1,
+                                   lengths = Lengths1})
+        end,
+    {Data, {NDims2, Lengths2, HasNull2}} =
+        lists:mapfoldl(F, {NDims0, Lengths1, HasNull0}, Array),
+    {Data, {NDims2 + 1, Lengths2, HasNull2}}.
 
 
 %% Supports

src/epgsql_codec.erl

@@ -1,9 +1,12 @@
 %%% @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
+%%% @see epgsql_binary
+%%% @end
 %%% Created : 12 Oct 2017 by Sergey Prokhorov <me@seriyps.ru>
 
 -module(epgsql_codec).

src/epgsql_command.erl

@@ -1,5 +1,5 @@
-%%% Behaviour module for epgsql_sock commands.
-%%%
+%%% @doc Behaviour module for epgsql_sock commands.
+%%% @end
 %%% Copyright (C) 2017 - Sergey Prokhorov.  All rights reserved.
 
 -module(epgsql_command).
@@ -16,6 +16,7 @@
 -type execute_return() ::
         {ok, epgsql_sock:pg_sock(), state()}
       | {stop, Reason :: any(), Response :: any(), epgsql_sock:pg_sock()}.
+
 %% Execute command. It should send commands to socket.
 %% May be called many times if 'handle_message' will return 'requeue'.
 -callback execute(epgsql_sock:pg_sock(), state()) -> execute_return().
@@ -48,6 +49,7 @@
         %% Unknown packet. Terminate `epgsql_sock' process
       | unknown.
 %% Handle incoming packet
+
 -callback handle_message(Type :: byte(), Payload :: binary() | epgsql:query_error(),
                          epgsql_sock:pg_sock(), state()) -> handle_message_return().
 

src/epgsql_errcodes.erl

@@ -1,4 +1,4 @@
-%% DO NOT EDIT - AUTOGENERATED BY ./generate_errcodes_src.sh ON 2018-02-23T11:18:01+0100
+%% DO NOT EDIT - AUTOGENERATED BY ./generate_errcodes_src.sh ON 2020-02-01T00:16:09+0100
 -module(epgsql_errcodes).
 -export([to_name/1]).
 
@@ -55,7 +55,7 @@ to_name(<<"22025">>) -> invalid_escape_sequence;
 to_name(<<"22P06">>) -> nonstandard_use_of_escape_character;
 to_name(<<"22010">>) -> invalid_indicator_parameter_value;
 to_name(<<"22023">>) -> invalid_parameter_value;
-to_name(<<"22013">>) -> invalid_preceding_following_size;
+to_name(<<"22013">>) -> invalid_preceding_or_following_size;
 to_name(<<"2201B">>) -> invalid_regular_expression;
 to_name(<<"2201W">>) -> invalid_row_count_in_limit_clause;
 to_name(<<"2201X">>) -> invalid_row_count_in_result_offset_clause;
@@ -84,6 +84,22 @@ to_name(<<"2200M">>) -> invalid_xml_document;
 to_name(<<"2200N">>) -> invalid_xml_content;
 to_name(<<"2200S">>) -> invalid_xml_comment;
 to_name(<<"2200T">>) -> invalid_xml_processing_instruction;
+to_name(<<"22030">>) -> duplicate_json_object_key_value;
+to_name(<<"22031">>) -> invalid_argument_for_json_datetime_function;
+to_name(<<"22032">>) -> invalid_json_text;
+to_name(<<"22033">>) -> invalid_sql_json_subscript;
+to_name(<<"22034">>) -> more_than_one_sql_json_item;
+to_name(<<"22035">>) -> no_sql_json_item;
+to_name(<<"22036">>) -> non_numeric_sql_json_item;
+to_name(<<"22037">>) -> non_unique_keys_in_a_json_object;
+to_name(<<"22038">>) -> singleton_sql_json_item_required;
+to_name(<<"22039">>) -> sql_json_array_not_found;
+to_name(<<"2203A">>) -> sql_json_member_not_found;
+to_name(<<"2203B">>) -> sql_json_number_not_found;
+to_name(<<"2203C">>) -> sql_json_object_not_found;
+to_name(<<"2203D">>) -> too_many_json_array_elements;
+to_name(<<"2203E">>) -> too_many_json_object_members;
+to_name(<<"2203F">>) -> sql_json_scalar_required;
 to_name(<<"23000">>) -> integrity_constraint_violation;
 to_name(<<"23001">>) -> restrict_violation;
 to_name(<<"23502">>) -> not_null_violation;
@@ -195,6 +211,7 @@ to_name(<<"55000">>) -> object_not_in_prerequisite_state;
 to_name(<<"55006">>) -> object_in_use;
 to_name(<<"55P02">>) -> cant_change_runtime_param;
 to_name(<<"55P03">>) -> lock_not_available;
+to_name(<<"55P04">>) -> unsafe_new_enum_value_usage;
 to_name(<<"57000">>) -> operator_intervention;
 to_name(<<"57014">>) -> query_canceled;
 to_name(<<"57P01">>) -> admin_shutdown;

src/epgsql_oid_db.erl

@@ -1,7 +1,7 @@
 %%% @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
+%%% Holds Oid to Type mappings (forward and reverse).
+%%% See [https://www.postgresql.org/docs/current/static/catalog-pg-type.html].
 %%% @end
 
 -module(epgsql_oid_db).
@@ -36,7 +36,7 @@
 %% pg_type Data preparation
 %%
 
-%% @doc build query to fetch OID<->type_name information from PG server
+%% @doc build query to fetch OID to type_name information from PG server
 -spec build_query([epgsql:type_name() | binary()]) -> iolist().
 build_query(TypeNames) ->
     %% TODO: lists:join/2, ERL 19+

src/epgsql_scram.erl

@@ -1,11 +1,13 @@
 %%% coding: utf-8
 %%% @doc
 %%% SCRAM--SHA-256 helper functions
-%%% See
-%%% https://www.postgresql.org/docs/current/static/sasl-authentication.html
-%%% https://en.wikipedia.org/wiki/Salted_Challenge_Response_Authentication_Mechanism
-%%% https://tools.ietf.org/html/rfc7677
-%%% https://tools.ietf.org/html/rfc5802
+%%%
+%%% <ul>
+%%%  <li>[https://www.postgresql.org/docs/current/static/sasl-authentication.html]</li>
+%%%  <li>[https://en.wikipedia.org/wiki/Salted_Challenge_Response_Authentication_Mechanism]</li>
+%%%  <li>[https://tools.ietf.org/html/rfc7677]</li>
+%%%  <li>[https://tools.ietf.org/html/rfc5802]</li>
+%%% </ul>
 %%% @end
 
 -module(epgsql_scram).

src/epgsql_sock.erl

@@ -1,25 +1,39 @@
-%%% Copyright (C) 2009 - Will Glozer.  All rights reserved.
-%%% Copyright (C) 2011 - Anton Lebedevich.  All rights reserved.
-
-%%% @doc GenServer holding all connection state (including socket).
+%%% @doc GenServer holding all the connection state (including socket).
 %%%
-%%% See https://www.postgresql.org/docs/current/static/protocol-flow.html
-%%% Commands in PostgreSQL are pipelined: you don't need to wait for reply to
-%%% be able to send next command.
+%%% See [https://www.postgresql.org/docs/current/static/protocol-flow.html]
+%%%
+%%% Commands in PostgreSQL protocol are pipelined: you don't have to wait for
+%%% reply to be able to send next command.
 %%% Commands are processed (and responses to them are generated) in FIFO order.
 %%% eg, if you execute 2 SimpleQuery: #1 and #2, first you get all response
 %%% packets for #1 and then all for #2:
+%%% ```
 %%% > SQuery #1
 %%% > SQuery #2
 %%% < RowDescription #1
-%%% < DataRow #1
+%%% < DataRow #1.1
+%%% < ...
+%%% < DataRow #1.N
 %%% < CommandComplete #1
 %%% < RowDescription #2
-%%% < DataRow #2
+%%% < DataRow #2.1
+%%% < ...
+%%% < DataRow #2.N
 %%% < CommandComplete #2
-%%%
-%%% See epgsql_cmd_connect for network connection and authentication setup
-
+%%% '''
+%%% `epgsql_sock' is capable of utilizing the pipelining feature - as soon as
+%%% it receives a new command, it sends it to the server immediately and then
+%%% it puts command's callbacks and state into internal queue of all the commands
+%%% which were sent to the server and waiting for response. So it knows in which
+%%% order it should call each pipelined command's `handle_message' callback.
+%%% But it can be easily broken if high-level command is poorly implemented or
+%%% some conflicting low-level commands (such as `parse', `bind', `execute') are
+%%% executed in a wrong order. In this case server and epgsql states become out of
+%%% sync and {@link epgsql_cmd_sync} have to be executed in order to recover.
+%%% @see epgsql_cmd_connect. epgsql_cmd_connect for network connection and authentication setup
+%%% @end
+%%% Copyright (C) 2009 - Will Glozer.  All rights reserved.
+%%% Copyright (C) 2011 - Anton Lebedevich.  All rights reserved.
 
 -module(epgsql_sock).
 
@@ -46,7 +60,7 @@
          get_parameter_internal/2,
          get_replication_state/1, set_packet_handler/2]).
 
--export_type([transport/0, pg_sock/0]).
+-export_type([transport/0, pg_sock/0, error/0]).
 
 -include("epgsql.hrl").
 -include("protocol.hrl").
@@ -59,6 +73,8 @@
 -type tcp_socket() :: port(). %gen_tcp:socket() isn't exported prior to erl 18
 -type repl_state() :: #repl{}.
 
+-type error() :: {error, sync_required | closed | sock_closed | sock_error}.
+
 -record(state, {mod :: gen_tcp | ssl | undefined,
                 sock :: tcp_socket() | ssl:sslsocket() | undefined,
                 data = <<>>,
@@ -205,6 +221,7 @@ handle_cast({{Method, From, Ref} = Transport, Command, Args}, State)
     command_new(Transport, Command, Args, State);
 
 handle_cast(stop, State) ->
+    send(State, ?TERMINATE, []),
     {stop, normal, flush_queue(State, {error, closed})};
 
 handle_cast(cancel, State = #state{backend = {Pid, Key},

src/epgsql_wire.erl

@@ -1,3 +1,9 @@
+%%% @doc
+%%% Interface to encoder/decoder for postgresql
+%%% <a href="https://www.postgresql.org/docs/current/protocol-message-formats.html">wire-protocol</a>
+%%%
+%%% See also `include/protocol.hrl'.
+%%% @end
 %%% Copyright (C) 2009 - Will Glozer.  All rights reserved.
 %%% Copyright (C) 2011 - Anton Lebedevich.  All rights reserved.
 
@@ -25,6 +31,7 @@
 
 -opaque row_decoder() :: {[epgsql_binary:decoder()], [epgsql:column()], epgsql_binary:codec()}.
 
+%% @doc tries to extract single postgresql packet from TCP stream
 -spec decode_message(binary()) -> {byte(), binary(), binary()} | binary().
 decode_message(<<Type:8, Len:?int32, Rest/binary>> = Bin) ->
     Len2 = Len - 4,
@@ -65,7 +72,7 @@ decode_fields(<<Type:8, Rest/binary>>, Acc) ->
     decode_fields(Rest2, [{Type, Str} | Acc]).
 
 %% @doc decode ErrorResponse
-%% See http://www.postgresql.org/docs/current/interactive/protocol-error-fields.html
+%% See [http://www.postgresql.org/docs/current/interactive/protocol-error-fields.html]
 -spec decode_error(binary()) -> epgsql:query_error().
 decode_error(Bin) ->
     Fields = decode_fields(Bin),
@@ -143,17 +150,17 @@ decode_data(Bin, {Decoders, _Columns, Codec}) ->
 
 decode_data(_, [], _) -> [];
 decode_data(<<-1:?int32, Rest/binary>>, [_Dec | Decs], Codec) ->
-    [null | decode_data(Rest, Decs, Codec)];
+    [epgsql_binary:null(Codec) | decode_data(Rest, Decs, Codec)];
 decode_data(<<Len:?int32, Value:Len/binary, Rest/binary>>, [Decoder | Decs], Codec) ->
     [epgsql_binary:decode(Value, Decoder)
      | decode_data(Rest, Decs, Codec)].
 
-%% @doc decode column information
+%% @doc decode RowDescription column information
 -spec decode_columns(non_neg_integer(), binary(), epgsql_binary:codec()) -> [epgsql:column()].
 decode_columns(0, _Bin, _Codec) -> [];
 decode_columns(Count, Bin, Codec) ->
     [Name, Rest] = decode_string(Bin),
-    <<_TableOid:?int32, _AttribNum:?int16, TypeOid:?int32,
+    <<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),
@@ -163,7 +170,9 @@ decode_columns(Count, Bin, Codec) ->
       oid      = TypeOid,
       size     = Size,
       modifier = Modifier,
-      format   = Format},
+      format   = Format,
+      table_oid = TableOid,
+      table_attr_number = AttribNum},
     [Desc | decode_columns(Count - 1, Rest2, Codec)].
 
 %% @doc decode ParameterDescription
@@ -175,7 +184,7 @@ decode_parameters(<<_Count:?int16, Bin/binary>>, Codec) ->
          TypeInfo -> TypeInfo
      end || <<Oid:?int32>> <= Bin].
 
-%% @doc decode command complete msg
+%% @doc decode CcommandComplete msg
 decode_complete(<<"SELECT", 0>>)        -> select;
 decode_complete(<<"SELECT", _/binary>>) -> select;
 decode_complete(<<"BEGIN", 0>>)         -> 'begin';
@@ -217,6 +226,7 @@ encode_formats([], Count, Acc) ->
 encode_formats([#column{format = Format} | T], Count, Acc) ->
     encode_formats(T, Count + 1, <<Acc/binary, Format:?int16>>).
 
+%% @doc Returns 1 if Codec knows how to decode binary format of the type provided and 0 otherwise
 format({unknown_oid, _}, _) -> 0;
 format(#column{oid = Oid}, Codec) ->
     case epgsql_binary:supports(Oid, Codec) of
@@ -244,16 +254,18 @@ encode_parameters([P | T], Count, Formats, Values, Codec) ->
       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) ->
-    {0, encode_text(Value)};
 encode_parameter({Type, Value}, Codec) ->
-    {1, epgsql_binary:encode(Type, Value, Codec)};
-encode_parameter(Value, _Codec) ->
-    {0, encode_text(Value)}.
+    case epgsql_binary:is_null(Value, Codec) of
+        false ->
+            encode_parameter(Type, Value, Codec);
+        true ->
+            {1, <<-1:?int32>>}
+    end.
+
+encode_parameter({unknown_oid, _Oid}, Value, _Codec) ->
+    {0, encode_text(Value)};
+encode_parameter(Type, Value, Codec) ->
+    {1, epgsql_binary:encode(Type, Value, Codec)}.
 
 encode_text(B) when is_binary(B)  -> encode_bin(B);
 encode_text(A) when is_atom(A)    -> encode_bin(atom_to_binary(A, utf8));

src/epgsqla.erl

@@ -1,3 +1,10 @@
+%%% @doc
+%%% Asynchronous interface.
+%%%
+%%% All the functions return `reference()' immediately. Results are delivered
+%%% asynchronously in a form of `{connection(), reference(), Result}', where
+%%% `Result' is what synchronous version of this function normally returns.
+%%% @end
 %%% Copyright (C) 2011 - Anton Lebedevich.  All rights reserved.
 
 -module(epgsqla).
@@ -15,7 +22,7 @@
          describe/2, describe/3,
          bind/3, bind/4,
          execute/2, execute/3, execute/4,
-         execute_batch/2,
+         execute_batch/2, execute_batch/3,
          close/2, close/3,
          sync/1,
          cancel/1,
@@ -123,6 +130,10 @@ execute(C, Statement, PortalName, MaxRows) ->
 execute_batch(C, Batch) ->
     cast(C, epgsql_cmd_batch, Batch).
 
+-spec execute_batch(epgsql:connection(), epgsql:statement(), [ [epgsql:bind_param()] ]) -> reference().
+execute_batch(C, #statement{} = Statement, Batch) ->
+    cast(C, epgsql_cmd_batch, {Statement, Batch}).
+
 describe(C, #statement{name = Name}) ->
     describe(C, statement, Name).
 

src/epgsqli.erl

@@ -1,3 +1,10 @@
+%%% @doc Incremental interface
+%%%
+%%% All the functions return `reference()' immediately. Each data row as well
+%%% as metadata are delivered as separate messages in a form of
+%%% `{connection(), reference(), Payload}' where `Payload' depends on command
+%%% being executed.
+%%% @end
 %%% Copyright (C) 2011 - Anton Lebedevich.  All rights reserved.
 
 -module(epgsqli).
@@ -15,7 +22,7 @@
          describe/2, describe/3,
          bind/3, bind/4,
          execute/2, execute/3, execute/4,
-         execute_batch/2,
+         execute_batch/2, execute_batch/3,
          close/2, close/3,
          sync/1,
          cancel/1]).
@@ -123,6 +130,10 @@ execute(C, Statement, PortalName, MaxRows) ->
 execute_batch(C, Batch) ->
     incremental(C, epgsql_cmd_batch, Batch).
 
+-spec execute_batch(epgsql:connection(), epgsql:statement(), [ [epgsql:bind_param()] ]) -> reference().
+execute_batch(C, #statement{} = Statement, Batch) ->
+    incremental(C, epgsql_cmd_batch, {Statement, Batch}).
+
 describe(C, #statement{name = Name}) ->
     describe(C, statement, Name).
 

src/ewkb.erl

@@ -1,5 +1,10 @@
-%% https://en.wikipedia.org/wiki/Well-known_text
-%% http://postgis.net/docs/manual-2.4/using_postgis_dbmanagement.html#EWKB_EWKT
+%% @doc
+%% Encoder/decoder for PostGIS binary data representation.
+%%
+%% <ul>
+%%  <li>[https://en.wikipedia.org/wiki/Well-known_text]</li>
+%%  <li>[http://postgis.net/docs/manual-2.4/using_postgis_dbmanagement.html#EWKB_EWKT]</li>
+%% </ul>
 -module(ewkb).
 -export([decode_geometry/1, encode_geometry/1]).
 -export_type([point_type/0, point/1, multi_point/1, line_string/1,

test/epgsql_SUITE.erl

@@ -1,6 +1,6 @@
 -module(epgsql_SUITE).
 
--include_lib("eunit/include/eunit.hrl").
+-include_lib("stdlib/include/assert.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("public_key/include/public_key.hrl").
 -include("epgsql_tests.hrl").
@@ -67,7 +67,8 @@ groups() ->
             range_type,
             range8_type,
             date_time_range_type,
-            custom_types
+            custom_types,
+            custom_null
         ]},
         {generic, [parallel], [
             with_transaction
@@ -87,6 +88,9 @@ groups() ->
         cursor,
         multiple_result,
         execute_batch,
+        execute_batch_3_named_stmt,
+        execute_batch_3_unnamed_stmt,
+        execute_batch_3_sql,
         batch_error,
         single_batch,
         extended_select,
@@ -108,6 +112,7 @@ groups() ->
         describe_with_param,
         describe_named,
         describe_error,
+        describe_portal,
         portal,
         returning,
         multiple_statement,
@@ -431,6 +436,32 @@ execute_batch(Config) ->
             Module:execute_batch(C, [{S1, [1]}, {S2, [1, 2]}])
     end).
 
+execute_batch_3_named_stmt(Config) ->
+    Module = ?config(module, Config),
+    epgsql_ct:with_connection(Config, fun(C) ->
+        {ok, Stmt} = Module:parse(C, "my_stmt", "select $1 + $2", [int4, int4]),
+        ?assertMatch(
+           {[#column{type = int4, _ = _}], [{ok, [{3}]}, {ok, [{7}]}]},
+           Module:execute_batch(C, Stmt, [[1, 2], [3, 4]]))
+    end).
+
+execute_batch_3_unnamed_stmt(Config) ->
+    Module = ?config(module, Config),
+    epgsql_ct:with_connection(Config, fun(C) ->
+        {ok, Stmt} = Module:parse(C, "select $1::integer + $2::integer"),
+        ?assertMatch(
+           {[#column{type = int4, _ = _}], [{ok, [{3}]}, {ok, [{7}]}]},
+           Module:execute_batch(C, Stmt, [[2, 1], [4, 3]]))
+    end).
+
+execute_batch_3_sql(Config) ->
+    Module = ?config(module, Config),
+    epgsql_ct:with_connection(Config, fun(C) ->
+        ?assertMatch(
+           {[#column{type = int4, _ = _}], [{ok, [{3}]}, {ok, [{7}]}]},
+           Module:execute_batch(C, "select $1::integer + $2::integer", [[1, 2], [3, 4]]))
+    end).
+
 batch_error(Config) ->
     Module = ?config(module, Config),
     epgsql_ct:with_rollback(Config, fun(C) ->
@@ -514,9 +545,9 @@ parse_column_format(Config) ->
     Module = ?config(module, Config),
     epgsql_ct:with_connection(Config, fun(C) ->
         {ok, S} = Module:parse(C, "select 1::int4, false::bool, 2.0::float4"),
-        [#column{type = int4},
-         #column{type = bool},
-         #column{type = float4}] = S#statement.columns,
+        [#column{type = int4, table_oid = 0, table_attr_number = 0},
+         #column{type = bool, table_oid = 0, table_attr_number = 0},
+         #column{type = float4, table_oid = 0, table_attr_number = 0}] = S#statement.columns,
         ok = Module:bind(C, S, []),
         {ok, [{1, false, 2.0}]} = Module:execute(C, S, 0),
         ok = Module:close(C, S),
@@ -652,6 +683,23 @@ describe_error(Config) ->
 
     end).
 
+describe_portal(Config) ->
+    Module = ?config(module, Config),
+    epgsql_ct:with_connection(Config, fun(C) ->
+        {ok, Stmt} = Module:parse(C, "my_stmt", "select * from test_table1 WHERE id = $1", []),
+        ok = Module:bind(C, Stmt, "my_portal", [1]),
+        {ok, Columns} = Module:describe(C, portal, "my_portal"),
+        ?assertMatch(
+           [#column{name = <<"id">>,
+                    type = int4},
+            #column{name = <<"value">>,
+                    type = text}],
+           Columns
+          ),
+        ok = Module:close(C, Stmt),
+        ok = Module:sync(C)
+    end).
+
 portal(Config) ->
     Module = ?config(module, Config),
     epgsql_ct:with_connection(Config, fun(C) ->
@@ -732,7 +780,25 @@ numeric_type(Config) ->
     check_type(Config, float4, "1.0", 1.0, [0.0, 1.23456, -1.23456]),
     check_type(Config, float4, "'-Infinity'", minus_infinity, [minus_infinity, plus_infinity, nan]),
     check_type(Config, float8, "1.0", 1.0, [0.0, 1.23456789012345, -1.23456789012345]),
-    check_type(Config, float8, "'nan'", nan, [minus_infinity, plus_infinity, nan]).
+    check_type(Config, float8, "'nan'", nan, [minus_infinity, plus_infinity, nan]),
+    %% Check overflow protection. Connection just crashes for now instead of silently
+    %% truncating the data. Some cleaner behaviour can be introduced later.
+    epgsql_ct:with_connection(Config, fun(C) ->
+        Module = ?config(module, Config),
+        Trap = process_flag(trap_exit, true),
+        try Module:equery(C, "SELECT $1::int2", [32768]) of
+          {error, closed} ->
+                %% epgsqla/epgsqli
+                ok
+        catch exit:Reason ->
+                %% epgsql
+                ?assertMatch({{{integer_overflow, int2, _}, _}, _}, Reason),
+                receive {'EXIT', C, _} -> ok
+                after 1000 -> error(timeout)
+                end
+        end,
+        process_flag(trap_exit, Trap)
+    end).
 
 character_type(Config) ->
     Alpha = unicode:characters_to_binary([16#03B1]),
@@ -852,6 +918,7 @@ misc_type(Config) ->
     check_type(Config, bytea, "E'\001\002'", <<1,2>>, [<<>>, <<0,128,255>>]).
 
 hstore_type(Config) ->
+    Module = ?config(module, Config),
     Values = [
         {[]},
         {[{null, null}]},
@@ -867,7 +934,42 @@ hstore_type(Config) ->
     check_type(Config, hstore, "''", {[]}, []),
     check_type(Config, hstore,
                "'a => 1, b => 2.0, c => null'",
-               {[{<<"a">>, <<"1">>}, {<<"b">>, <<"2.0">>}, {<<"c">>, null}]}, Values).
+               {[{<<"a">>, <<"1">>}, {<<"b">>, <<"2.0">>}, {<<"c">>, null}]}, Values),
+    epgsql_ct:with_connection(
+      Config,
+      fun(C) ->
+              %% Maps as input
+              [begin
+                   {ok, _, [{Res}]} = Module:equery(C, "select $1::hstore", [maps:from_list(KV)]),
+                   ?assert(compare(hstore, Res, Jiffy))
+               end || {KV} = Jiffy <- Values],
+              %% Maps as output
+              {ok, [hstore]} = epgsql:update_type_cache(
+                                 C, [{epgsql_codec_hstore, #{return => map}}]),
+              [begin
+                   {ok, _, [{Res}]} = Module:equery(C, "select $1::hstore", [maps:from_list(KV)]),
+                   HstoreMap = maps:from_list([{format_hstore_key(K), format_hstore_value(V)} || {K, V} <- KV]),
+                   ?assertEqual(HstoreMap, Res)
+               end || {KV} <- Values],
+              %% Proplist as output
+              {ok, [hstore]} = epgsql:update_type_cache(
+                                 C, [{epgsql_codec_hstore, #{return => proplist}}]),
+              [begin
+                   {ok, _, [{Res}]} = Module:equery(C, "select $1::hstore", [Jiffy]),
+                   HstoreProplist = [{format_hstore_key(K), format_hstore_value(V)} || {K, V} <- KV],
+                   ?assertEqual(lists:sort(HstoreProplist), lists:sort(Res))
+               end || {KV} = Jiffy <- Values],
+              %% Custom nulls
+              Nulls = [nil, 'NULL', aaaa],
+              {ok, [hstore]} = epgsql:update_type_cache(
+                                 C, [{epgsql_codec_hstore, #{return => map,
+                                                             nulls => Nulls}}]),
+              K = <<"k">>,
+              [begin
+                   {ok, _, [{Res}]} = Module:equery(C, "select $1::hstore", [#{K => V}]),
+                   ?assertEqual(#{K => nil}, Res)
+               end || V <- Nulls]
+      end).
 
 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}]),
@@ -884,18 +986,24 @@ array_type(Config) ->
         {ok, _, [{[1, 2]}]} = Module:equery(C, "select ($1::int[])[1:2]", [[1, 2, 3]]),
         {ok, _, [{[{1, <<"one">>}, {2, <<"two">>}]}]} =
             Module:equery(C, "select Array(select (id, value) from test_table1)", []),
-        Select = fun(Type, A) ->
+        {ok, _, [{ [[1], [null], [3], [null]] }]} =
+            Module:equery(C, "select $1::int2[]", [ [[1], [null], [3], [undefined]] ]),
+        Select = fun(Type, AIn) ->
             Query = "select $1::" ++ atom_to_list(Type) ++ "[]",
-            {ok, _Cols, [{A2}]} = Module:equery(C, Query, [A]),
-            case lists:all(fun({V, V2}) -> compare(Type, V, V2) end, lists:zip(A, A2)) of
+            {ok, _Cols, [{AOut}]} = Module:equery(C, Query, [AIn]),
+            case lists:all(fun({VIn, VOut}) ->
+                                   compare(Type, VIn, VOut)
+                           end, lists:zip(AIn, AOut)) of
                 true  -> ok;
-                false -> ?assertMatch(A, A2)
+                false -> ?assertEqual(AIn, AOut)
             end
         end,
         Select(int2,   []),
         Select(int2,   [1, 2, 3, 4]),
         Select(int2,   [[1], [2], [3], [4]]),
         Select(int2,   [[[[[[1, 2]]]]]]),
+        Select(int2,   [1, null, 3, undefined]),
+        Select(int2,   [[1], [null], [3], [null]]),
         Select(bool,   [true]),
         Select(char,   [$a, $b, $c]),
         Select(int4,   [[1, 2]]),
@@ -936,7 +1044,10 @@ record_type(Config) ->
         Select("select (1, '{2,3}'::int[])", {{1, [2, 3]}}),
 
         %% Array of records inside record
-        Select("select (0, ARRAY(select (id, value) from test_table1))", {{0,[{1,<<"one">>},{2,<<"two">>}]}})
+        Select("select (0, ARRAY(select (id, value) from test_table1))", {{0,[{1,<<"one">>},{2,<<"two">>}]}}),
+
+        %% Record with NULLs
+        Select("select (1, NULL::integer, 2)", {{1, null, 2}})
     end).
 
 custom_types(Config) ->
@@ -953,6 +1064,33 @@ custom_types(Config) ->
         ?assertMatch({ok, _, [{bar}]}, Module:equery(C, "SELECT col FROM t_foo"))
     end).
 
+custom_null(Config) ->
+    Module = ?config(module, Config),
+    epgsql_ct:with_connection(Config, fun(C) ->
+        Test3 = fun(Type, In, Out) ->
+                        Q = ["SELECT $1::", Type],
+                        {ok, _, [{Res}]} = Module:equery(C, Q, [In]),
+                        ?assertEqual(Out, Res)
+                end,
+        Test = fun(Type, In) ->
+                       Test3(Type, In, In)
+               end,
+        Test("int2", nil),
+        Test3("int2", 'NULL', nil),
+        Test("text", nil),
+        Test3("text", 'NULL', nil),
+        Test3("text", null, <<"null">>),
+        Test("int2[]", [nil, 1, nil, 2]),
+        Test3("text[]", [null, <<"ok">>], [<<"null">>, <<"ok">>]),
+        Test3("int2[]", ['NULL', 1, nil, 2], [nil, 1, nil, 2]),
+        Test("int2[]", [[nil], [1], [nil], [2]]),
+        Test3("int2[]", [['NULL'], [1], [nil], [2]], [[nil], [1], [nil], [2]]),
+        ?assertMatch(
+           {ok, _, [{ {1, nil, {2, nil, 3}} }]},
+           Module:equery(C, "SELECT (1, NULL, (2, NULL, 3))", []))
+    end,
+    [{nulls, [nil, 'NULL']}]).
+
 text_format(Config) ->
     Module = ?config(module, Config),
     epgsql_ct:with_connection(Config, fun(C) ->
@@ -1026,7 +1164,7 @@ connection_closed_by_server(Config) ->
                     {'EXIT', C2, {shutdown, #error{code = <<"57P01">>}}} ->
                         P ! ok;
                     Other ->
-                        ?debugFmt("Unexpected msg: ~p~n", [Other]),
+                        ct:pal("Unexpected msg: ~p~n", [Other]),
                         P ! error
                 end
             end)
@@ -1186,7 +1324,8 @@ range_type(Config) ->
         check_type(Config, int4range, "int4range(10, 20)", {10, 20}, [
             {1, 58}, {-1, 12}, {-985521, 5412687}, {minus_infinity, 0},
             {984655, plus_infinity}, {minus_infinity, plus_infinity}
-        ])
+        ]),
+        check_type(Config, int4range, "int4range(10, 10)", empty, [])
    end, []).
 
 range8_type(Config) ->
@@ -1195,17 +1334,26 @@ range8_type(Config) ->
             {1, 58}, {-1, 12}, {-9223372036854775808, 5412687},
             {minus_infinity, 9223372036854775807},
             {984655, plus_infinity}, {minus_infinity, plus_infinity}
-        ])
+        ]),
+        check_type(Config, int8range, "int8range(10, 10)", empty, [])
     end, []).
 
 date_time_range_type(Config) ->
     epgsql_ct:with_min_version(Config, [9, 2], fun(_C) ->
         check_type(Config, tsrange, "tsrange('2008-01-02 03:04:05', '2008-02-02 03:04:05')", {{{2008,1,2},{3,4,5.0}}, {{2008,2,2},{3,4,5.0}}}, []),
-       check_type(Config, tsrange, "tsrange('2008-01-02 03:04:05', '2008-01-02 03:04:05')", empty, []),
-
-       check_type(Config, daterange, "daterange('2008-01-02', '2008-02-02')", {{2008,1,2}, {2008, 2, 2}}, [{{-4712,1,1}, {5874897,1,1}}
-]),
-      check_type(Config, tstzrange, "tstzrange('2011-01-02 03:04:05+3', '2011-01-02 04:04:05+3')", {{{2011, 1, 2}, {0, 4, 5.0}}, {{2011, 1, 2}, {1, 4, 5.0}}}, [{{{2011, 1, 2}, {0, 4, 5.0}}, {{2011, 1, 2}, {1, 4, 5.0}}}])
+        check_type(Config, tsrange, "tsrange('2008-01-02 03:04:05', '2008-02-02 03:04:05', '[]')", {{{2008,1,2},{3,4,5.0}}, {{2008,2,2},{3,4,5.0}}}, []),
+        check_type(Config, tsrange, "tsrange('2008-01-02 03:04:05', '2008-02-02 03:04:05', '()')", {{{2008,1,2},{3,4,5.0}}, {{2008,2,2},{3,4,5.0}}}, []),
+        check_type(Config, tsrange, "tsrange('2008-01-02 03:04:05', '2008-02-02 03:04:05', '[)')", {{{2008,1,2},{3,4,5.0}}, {{2008,2,2},{3,4,5.0}}}, []),
+        check_type(Config, tsrange, "tsrange('2008-01-02 03:04:05', '2008-02-02 03:04:05', '(]')", {{{2008,1,2},{3,4,5.0}}, {{2008,2,2},{3,4,5.0}}}, []),
+        check_type(Config, tsrange, "tsrange('2008-01-02 03:04:05', '2008-01-02 03:04:05')", empty, []),
+        check_type(Config, daterange, "daterange('2008-01-02', '2008-02-02')", {{2008,1,2}, {2008, 2, 2}}, [{{-4712,1,1}, {5874897,1,1}}]),
+        check_type(Config, tstzrange, "tstzrange('2011-01-02 03:04:05+3', '2011-01-02 04:04:05+3')", {{{2011, 1, 2}, {0, 4, 5.0}}, {{2011, 1, 2}, {1, 4, 5.0}}}, [{{{2011, 1, 2}, {0, 4, 5.0}}, {{2011, 1, 2}, {1, 4, 5.0}}}]),
+        check_type(Config, tstzrange, "tstzrange('2008-01-02 03:04:05', null)", {{{2008,1,2},{3,4,5.0}}, plus_infinity}, []),
+        check_type(Config, tstzrange, "tstzrange('2008-01-02 03:04:05', null, '[]')", {{{2008,1,2},{3,4,5.0}}, plus_infinity}, []),
+        check_type(Config, tstzrange, "tstzrange('2008-01-02 03:04:05', null, '()')", {{{2008,1,2},{3,4,5.0}}, plus_infinity}, []),
+        check_type(Config, tstzrange, "tstzrange(null, '2008-01-02 03:04:05')", {minus_infinity, {{2008,1,2},{3,4,5.0}}}, []),
+        check_type(Config, tstzrange, "tstzrange(null, '2008-01-02 03:04:05', '[]')", {minus_infinity, {{2008,1,2},{3,4,5.0}}}, []),
+        check_type(Config, tstzrange, "tstzrange(null, '2008-01-02 03:04:05', '()')", {minus_infinity, {{2008,1,2},{3,4,5.0}}}, [])
 
    end, []).
 

test/epgsql_cast.erl

@@ -9,7 +9,7 @@
 -export([get_parameter/2, set_notice_receiver/2, get_cmd_status/1, squery/2, equery/2, equery/3]).
 -export([prepared_query/3]).
 -export([parse/2, parse/3, parse/4, describe/2, describe/3]).
--export([bind/3, bind/4, execute/2, execute/3, execute/4, execute_batch/2]).
+-export([bind/3, bind/4, execute/2, execute/3, execute/4, execute_batch/2, execute_batch/3]).
 -export([close/2, close/3, sync/1]).
 -export([receive_result/2, sync_on_error/2]).
 
@@ -123,6 +123,17 @@ execute_batch(C, Batch) ->
     Ref = epgsqla:execute_batch(C, Batch),
     receive_result(C, Ref).
 
+execute_batch(C, #statement{columns = Cols} = Stmt, Batch) ->
+    Ref = epgsqla:execute_batch(C, Stmt, Batch),
+    {Cols, receive_result(C, Ref)};
+execute_batch(C, Sql, Batch) ->
+    case parse(C, Sql) of
+        {ok, #statement{} = S} ->
+            execute_batch(C, S, Batch);
+        Error ->
+            Error
+    end.
+
 %% statement/portal functions
 
 describe(C, #statement{name = Name}) ->

test/epgsql_incremental.erl

@@ -9,7 +9,7 @@
 -export([get_parameter/2, set_notice_receiver/2, get_cmd_status/1, squery/2, equery/2, equery/3]).
 -export([prepared_query/3]).
 -export([parse/2, parse/3, parse/4, describe/2, describe/3]).
--export([bind/3, bind/4, execute/2, execute/3, execute/4, execute_batch/2]).
+-export([bind/3, bind/4, execute/2, execute/3, execute/4, execute_batch/2, execute_batch/3]).
 -export([close/2, close/3, sync/1]).
 
 -include("epgsql.hrl").
@@ -124,6 +124,17 @@ execute_batch(C, Batch) ->
     Ref = epgsqli:execute_batch(C, Batch),
     receive_extended_results(C, Ref, []).
 
+execute_batch(C, #statement{columns = Cols} = Stmt, Batch) ->
+    Ref = epgsqli:execute_batch(C, Stmt, Batch),
+    {Cols, receive_extended_results(C, Ref, [])};
+execute_batch(C, Sql, Batch) ->
+    case parse(C, Sql) of
+        {ok, #statement{} = S} ->
+            execute_batch(C, S, Batch);
+        Error ->
+            Error
+    end.
+
 %% statement/portal functions
 
 describe(C, #statement{name = Name}) ->
@@ -133,9 +144,9 @@ describe(C, statement, Name) ->
     Ref = epgsqli:describe(C, statement, Name),
     sync_on_error(C, receive_describe(C, Ref, #statement{name = Name}));
 
-describe(C, Type, Name) ->
-    %% TODO unknown result format of Describe portal
-    epgsqli:describe(C, Type, Name).
+describe(C, portal, Name) ->
+    Ref = epgsqli:describe(C, portal, Name),
+    sync_on_error(C, receive_describe_portal(C, Ref)).
 
 close(C, #statement{name = Name}) ->
     close(C, statement, Name).
@@ -230,6 +241,18 @@ receive_describe(C, Ref, Statement = #statement{}) ->
             {error, closed}
     end.
 
+receive_describe_portal(C, Ref) ->
+    receive
+        {C, Ref, {columns, Columns}} ->
+            {ok, Columns};
+        {C, Ref, no_data} ->
+            {ok, []};
+        {C, Ref, Error = {error, _}} ->
+            Error;
+        {'EXIT', C, _Reason} ->
+            {error, closed}
+    end.
+
 receive_atom(C, Ref, Receive, Return) ->
     receive
         {C, Ref, Receive} ->