Module kflow_postgres

Description
Data Types
Function Index
Function Details

This module inserts a batch of Erlang maps as rows in a Postgres table.

Description

This module inserts a batch of Erlang maps as rows in a Postgres table.

Input message format

   [ #{field1 => val1, field2 => val2}
   , #{field1 => val3, field2 => val4}
   ]

Configuration

Configuration is a term of config/0 type. It's a map with the following mandatory fields:

database PostgreSQL connection options, see https://github.com/epgsql/epgsql/blob/devel/src/epgsql.erl#L49
table is name of Postgres table used for storing data
fields is a list of map keys that should be inserted to the table. By default, table column with the name matching will be used

The following options are optional:

field_mappings allows to specify column names used for the keys. This value should be a map #{Key => ColumnName}
keys parameter allows to specify which columens are primary keys in the table. kflow_postgres will overwrite row with colliding keys via upsert.

This module can create table partitions automatically when partitioning` key is present in the kfnode configuration, and the value is of type `part_config/0. The following parameters can be configured:

days -- partition granularity interval in days
retention -- how many old partition should be kept before deletion
index_fields -- (optional) also create indexes for fields

Example usage

   {map, kflow_postgres, #{ database => #{ host     => "localhost"
                                         , user     => "kflow"
                                         , password => "123"
                                         , database => "postgres"
                                         , port     => 5432
                                         }
                          , table  => "foo_table"
                          , fields => [foo, bar, baz]
                          , keys   => [foo]
                          , partitioning => #{days => 10, retention => 3, index_fields => [bar]}
                          }}

Data Types

config()

config() = #{database := epgsql:connect_opts(), init_count => non_neg_integer(), fields := [field()], keys => [field()], table := list() | binary(), field_mappings => field_mappings(), partitioning => part_config()}

field()

field() = atom() | list() | binary()

field_mappings()

field_mappings() = #{field() => field()}

part_config()

part_config() = #{days := non_neg_integer(), retention := non_neg_integer(), index_fields => [field()], n_part_ahead => non_neg_integer(), n_part_behind => non_neg_integer()}