Postgres Connector

Postgres connector allows reading snapshot data and incremental data from Postgres database and provides end-to-end full-database data synchronization capabilities. This document describes how to set up the Postgres connector.

Example

An example of the pipeline for reading data from Postgres and sink to Fluss can be defined as follows:

source:
   type: postgres
   name: Postgres Source
   hostname: 127.0.0.1
   port: 5432
   username: admin
   password: pass
   # make sure all the tables share same database.
   tables: adb.\.*.\.*
   decoding.plugin.name:  pgoutput
   slot.name: pgtest
   schema-change.enabled: true

sink:
  type: fluss
  name: Fluss Sink
  bootstrap.servers: localhost:9123
  # Security-related properties for the Fluss client
  properties.client.security.protocol: sasl
  properties.client.security.sasl.mechanism: PLAIN
  properties.client.security.sasl.username: developer
  properties.client.security.sasl.password: developer-pass

pipeline:
   name: Postgres to Fluss Pipeline
   parallelism: 4
   schema.change.behavior: lenient

Connector Options

Option	Required	Default	Type	Description
hostname	required	(none)	String	IP address or hostname of the Postgres database server.
port	optional	5432	Integer	Integer port number of the Postgres database server.
username	required	(none)	String	Name of the Postgres database to use when connecting to the Postgres database server.
password	required	(none)	String	Password to use when connecting to the Postgres database server.
tables	required	(none)	String	Table name of the Postgres database to monitor. The table-name also supports regular expressions to monitor multiple tables that satisfy the regular expressions. All the tables are required to share same database. It is important to note that the dot (.) is treated as a delimiter for database, schema and table names. If there is a need to use a dot (.) in a regular expression to match any character, it is necessary to escape the dot with a backslash. for example: bdb.user_schema_[0-9].user_table_[0-9]+, bdb.schema_\..order_\.
slot.name	required	(none)	String	The name of the PostgreSQL logical decoding slot that was created for streaming changes from a particular plug-in for a particular database/schema. The server uses this slot to stream events to the connector that you are configuring. Slot names must conform to PostgreSQL replication slot naming rules, which state: "Each replication slot has a name, which can contain lower-case letters, numbers, and the underscore character."
decoding.plugin.name	optional	(none)	String	The name of the Postgres logical decoding plug-in installed on the server. Supported values are decoderbufs and pgoutput.
tables.exclude	optional	(none)	String	Table name of the Postgres database to exclude, parameter will have an exclusion effect after the tables parameter. The table-name also supports regular expressions to exclude multiple tables that satisfy the regular expressions. The usage is the same as the tables parameter
server-time-zone	optional	(none)	String	The session time zone in the database server, for example: "Asia/Shanghai".It controls how TIMESTAMP values in Postgres are converted to strings.For more information, please refer to here.If not set, the system default time zone (ZoneId.systemDefault()) will be used to determine the server time zone.
scan.incremental.snapshot.chunk.size	optional	8096	Integer	The chunk size (number of rows) of table snapshot, captured tables are split into multiple chunks when read the snapshot of table.
scan.snapshot.fetch.size	optional	1024	Integer	The maximum fetch size for per poll when read table snapshot.
scan.startup.mode	optional	initial	String	Optional startup mode for Postgres CDC consumer, valid enumerations are "initial", "latest-offset", "committed-offset" or "snapshot".
scan.incremental.close-idle-reader.enabled	optional	false	Boolean	Whether to close idle readers at the end of the snapshot phase. The flink version is required to be greater than or equal to 1.14 when 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' is set to true. If the flink version is greater than or equal to 1.15, the default value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has been changed to true, so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
scan.lsn-commit.checkpoints-num-delay	optional	3	Integer	The number of checkpoint delays before starting to commit the LSN offsets. The checkpoint LSN offsets will be committed in rolling fashion, the earliest checkpoint identifier will be committed first from the delayed checkpoints.
connect.timeout	optional	30s	Duration	The maximum time that the connector should wait after trying to connect to the Postgres database server before timing out. This value cannot be less than 250ms.
connect.max-retries	optional	3	Integer	The max retry times that the connector should retry to build Postgres database server connection.
connection.pool.size	optional	20	Integer	The connection pool size.
jdbc.properties.*	optional	20	String	Option to pass custom JDBC URL properties. User can pass custom properties like 'jdbc.properties.useSSL' = 'false'.
heartbeat.interval	optional	30s	Duration	The interval of sending heartbeat event for tracing the latest available wal log offsets.
debezium.*	optional	(none)	String	Pass-through Debezium's properties to Debezium Embedded Engine which is used to capture data changes from postgres server. For example: `'debezium.snapshot.mode' = 'never'`. See more about the Debezium's Postgres Connector properties
chunk-meta.group.size	optional	1000	String	The group size of chunk meta, if the meta size exceeds the group size, the meta will be divided into multiple groups.
metadata.list	optional	(none)	String	List of readable metadata from SourceRecord to be passed to downstream and could be used in transform module, split by `,`. Available readable metadata are: op_ts, table_name, database_name, schema_name. See Supported Metadata Columns for more details.
scan.incremental.snapshot.unbounded-chunk-first.enabled	optional	false	String	Whether to assign the unbounded chunks first during snapshot reading phase. This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk. Experimental option, defaults to false.
table-id.include-database	optional	false	Boolean	Whether to include database in the generated Table ID. If set to true, the Table ID will be in the format (database, schema, table). If set to false, the Table ID will be in the format (schema, table). Defaults to false.
scan.newly-added-table.enabled	optional	false	Boolean	Whether to scan the newly added tables or not. Defaults to false. This option only takes effect when restoring from a savepoint or checkpoint, and enables the existing SnapshotSplitAssigner#captureNewlyAddedTables() code path to discover tables that match the source `tables` pattern but were not part of the captured set at savepoint time.
scan.pre-epoch-timestamp.wall-clock-conversion.enabled	optional	false	Boolean	Whether to convert the values of PostgreSQL "timestamp without time zone" columns which are before 1970-01-01 (the epoch) by keeping the date and time (wall clock) stored in the database. If enabled, such a value is not changed by the time zone of the JVM running the job, which matters in time zones which have historical offsets: with the JVM time zone set to Asia/Shanghai, the stored value "1900-01-01 00:00:00.123" would otherwise be read as "1900-01-01 00:05:43.123". If disabled (default), the previous conversion behavior is kept. Values which are not before 1970-01-01 are converted in the same way whatever the value of this option is, and this option never changes the data type of a column.
schema-change.enabled	optional	false	Boolean	Whether to enable schema change inference for the Postgres source. When enabled, the connector infers schema change events (add column, drop column, rename column, alter column type) by comparing pgoutput Relation messages against the cached schema. Requires `decoding.plugin.name` to be set to `pgoutput`. Defaults to false.

Note:

The tables configuration option specifies the tables to be captured by Postgres CDC, in the format db.schema1.table1,db.schema2.table2. All db values must be the same, as the PostgreSQL connection URL requires a single database name. Currently, CDC only supports connecting to one database.

Startup Reading Position

The config option scan.startup.mode specifies the startup mode for PostgreSQL CDC consumer. The valid enumerations are:

initial (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the replication slot.
latest-offset: Never to perform snapshot on the monitored database tables upon first startup, just read from the end of the replication which means only have the changes since the connector was started.
committed-offset: Skip snapshot phase and start reading events from a confirmed_flush_lsn offset of replication slot.
snapshot: Only the snapshot phase is performed and exits after the snapshot phase reading is completed.

Available Source metrics

Metrics can help understand the progress of assignments, and the following are the supported Flink metrics:

Group	Name	Type	Description
namespace.schema.table	isSnapshotting	Gauge	Weather the table is snapshotting or not
namespace.schema.table	isStreamReading	Gauge	Weather the table is stream reading or not
namespace.schema.table	numTablesSnapshotted	Gauge	The number of tables that have been snapshotted
namespace.schema.table	numTablesRemaining	Gauge	The number of tables that have not been snapshotted
namespace.schema.table	numSnapshotSplitsProcessed	Gauge	The number of splits that is being processed
namespace.schema.table	numSnapshotSplitsRemaining	Gauge	The number of splits that have not been processed
namespace.schema.table	numSnapshotSplitsFinished	Gauge	The number of splits that have been processed
namespace.schema.table	snapshotStartTime	Gauge	The time when the snapshot started
namespace.schema.table	snapshotEndTime	Gauge	The time when the snapshot ended

Notice:

The group name is namespace.schema.table, where namespace is the actual database name, schema is the actual schema name, and table is the actual table name.

Supported Metadata Columns

PostgreSQL CDC connector supports reading metadata columns from source records. These metadata columns can be used in transform operations or passed to downstream sinks.

Note: Some metadata information is also available through Transform expressions (e.g., __namespace_name__, __schema_name__, __table_name__). The key differences are:

op_ts: Only available via metadata.list - provides the actual operation timestamp from the database.
table_name, database_name, schema_name: Can be obtained via either metadata.list or Transform expressions. Using metadata.list allows you to pass these values directly to downstream sinks without writing transform rules, which is simpler for basic use cases.

To enable metadata columns, configure the metadata.list option with a comma-separated list of metadata column names:

source:
  type: postgres
  # ... other configurations
  metadata.list: op_ts,table_name,database_name,schema_name

The following metadata columns are supported:

Metadata Column	Data Type	Description
op_ts	BIGINT NOT NULL	The timestamp (in milliseconds since epoch) when the change event occurred in the database. For snapshot records, this value is 0.
table_name	STRING NOT NULL	The name of the table that contains the changed row. Alternative: use `__table_name__` in Transform expressions.
database_name	STRING NOT NULL	The name of the database that contains the changed row. Alternative: use `__namespace_name__` in Transform expressions.
schema_name	STRING NOT NULL	The name of the schema that contains the changed row. This is specific to PostgreSQL. Alternative: use `__schema_name__` in Transform expressions.

Example Usage:

source:
  type: postgres
  hostname: localhost
  port: 5432
  username: postgres
  password: postgres
  tables: mydb.public.orders
  slot.name: flink_slot
  metadata.list: op_ts,table_name,schema_name

transform:
  - source-table: mydb.public.orders
    projection: order_id, customer_id, op_ts, table_name, schema_name
    description: Include metadata columns in output

Data Type Mapping

PostgreSQL type	CDC type
BOOLEAN BIT(1)	BOOLEAN
BIT( > 1)	BYTES
SMALLINT INT2 SMALLSERIAL SERIAL2	SMALLINT
INTEGER SERIAL	INT
BIGINT BIGSERIAL OID	BIGINT
REAL FLOAT4	FLOAT
NUMERIC	DECIMAL(38, 0)
DOUBLE PRECISION FLOAT8	DOUBLE
CHAR[(M)] VARCHAR[(M)] CHARACTER[(M)] BPCHAR[(M)] CHARACTER VARYING[(M)]	STRING
TIMESTAMPTZ TIMESTAMP WITH TIME ZONE	ZonedTimestampType
INTERVAL [P]	BIGINT
INTERVAL [P]	STRING(when `debezium.interval.handling.mode` is set to string)
BYTEA	BYTES or STRING (when `debezium.binary.handling.mode` is set to base64 or base64-url-safe or hex)
JSON JSONB XML UUID POINT LTREE CITEXT INET INT4RANGE INT8RANGE NUMRANGE TSRANGE DATERANGE ENUM	STRING

Note: Since the Debezium version does not support multidimensional arrays, only PostgreSQL single-dimensional arrays are currently supported, such as :'ARRAY[1,2,3]', 'int[]'.

Temporal types Mapping

Other than PostgreSQL’s TIMESTAMPTZ data types, which contain time zone information, how temporal types are mapped depends on the value of debezium.time.precision.mode connector configuration property. The following sections describe these mappings:

debezium.time.precision.mode=adaptive
debezium.time.precision.mode=adaptive_time_microseconds
debezium.time.precision.mode=connect

Note: Due to the current CDC limitation, the precision for the TIME type is fixed at 3. Regardless of whether

debezium.time.precision.mode is set to adaptive, adaptive_time_microseconds, or connect, the TIME type will be converted to TIME(3).


debezium.time.precision.mode=adaptive
When the debezium.time.precision.mode property is set to the default value adaptive_time_microseconds, the precision of TIME is 3, and the precision of TIMESTAMP is 6.


    
      
        PostgreSQL type
        CDC type
      
    
    
       
        
          DATE
        DATE
      
      
        
          TIME([P])
        
        TIME(3)
      
      
        
          TIMESTAMP([P])
        
        TIMESTAMP([P])
      
    

debezium.time.precision.mode=adaptive_time_microseconds
When the debezium.time.precision.mode property is set to the value adaptive_time_microseconds, the precision of TIME is 3, and the precision of TIMESTAMP is 6.


    
      
        PostgreSQL type
        CDC type
      
    
    
       
        
          DATE
        DATE
      
      
        
          TIME([P])
        
        TIME(3)
      
      
        
          TIMESTAMP([P])
        
        TIMESTAMP([P])
      
    

debezium.time.precision.mode=connect
When the debezium.time.precision.mode property is set to the default value connect, both TIME and TIMESTAMP have a precision of 3.


    
      
        PostgreSQL type
        CDC type
      
    
    
       
        
          DATE
        DATE
      
      
        
          TIME([P])
        
        TIME(3)
      
      
        
          TIMESTAMP([P])
        
        TIMESTAMP(3)
      
    

Decimal types Mapping
The setting of the PostgreSQL connector configuration property debezium.decimal.handling.mode determines how the connector maps decimal types.
When the debezium.decimal.handling.mode property is set to precise, the connector uses the Kafka Connect org.apache.kafka.connect.data.Decimal logical type for all DECIMAL, NUMERIC and MONEY columns. This is the default mode.


    
      
        PostgreSQL type
        CDC type
      
    
    
       
        
          NUMERIC[(M[,D])]
        DECIMAL[(M[,D])]
      
      
        
          NUMERIC
        DECIMAL(38,0)
      
      
        
          DECIMAL[(M[,D])]
        DECIMAL[(M[,D])]
      
      
        
          DECIMAL
        DECIMAL(38,0)
      
      
        
          MONEY[(M[,D])]
        DECIMAL(38,digits)(The scale schema parameter contains an integer representing how many digits the decimal point was shifted. The scale schema parameter is determined by the money.fraction.digits connector configuration property.)
      
    

When the debezium.decimal.handling.mode property is set to double, the connector represents all DECIMAL, NUMERIC and MONEY values as Java double values and encodes them as shown in the following table.


    
      
        PostgreSQL type
        CDC type
      
    
    
       
        
          NUMERIC[(M[,D])]
        DOUBLE
      
      
        
          DECIMAL[(M[,D])]
        DOUBLE
      
      
        
          MONEY[(M[,D])]
        DOUBLE
      
    

The last possible setting for the debezium.decimal.handling.mode configuration property is string. In this case, the connector represents DECIMAL, NUMERIC and MONEY values as their formatted string representation, and encodes them as shown in the following table.


    
      
        PostgreSQL type
        CDC type
      
    
    
       
        
          NUMERIC[(M[,D])]
        STRING
      
      
        
          DECIMAL[(M[,D])]
        STRING
      
      
        
          MONEY[(M[,D])]
        STRING
      
    

PostgreSQL supports NaN (not a number) as a special value to be stored in DECIMAL/NUMERIC values when the setting of debezium.decimal.handling.mode is string or double. In this case, the connector encodes NaN as either Double.NaN or the string constant NAN.
HSTORE type Mapping
The setting of the PostgreSQL connector configuration property debezium.hstore.handling.mode determines how the connector maps HSTORE values.
When the debezium.hstore.handling.mode property is set to json (the default), the connector represents HSTORE values as string representations of JSON values and encodes them as shown in the following table. When the debezium.hstore.handling.mode property is set to map, the connector uses the MAP schema type for HSTORE values.


    
      
        PostgreSQL type
        CDC type
      
    
    
       
        
         HSTORE
        STRING(`debezium.hstore.handling.mode`=`string`)
      
      
        
         HSTORE
        MAP(`debezium.hstore.handling.mode`=`map`)
      
    

Network address types Mapping
PostgreSQL has data types that can store IPv4, IPv6, and MAC addresses. It is better to use these types instead of plain text types to store network addresses. Network address types offer input error checking and specialized operators and functions.


    
      
        PostgreSQL type
        CDC type
      
    
    
       
        
         INET
        STRING
      
      
        
         CIDR
        STRING
      
      
        
         MACADDR
        STRING
      
      
        
         MACADDR8
        STRING
       
    

PostGIS Types Mapping
PostgreSQL supports spatial data types through the PostGIS extension:
    GEOMETRY(POINT, xx): Represents a point in a Cartesian coordinate system, with EPSG:xx defining the coordinate system. It is suitable for local planar calculations. 
    GEOGRAPHY(MULTILINESTRING): Stores multiple line strings in latitude and longitude, based on a spherical model. It is suitable for global-scale spatial analysis.

The former is used for small-area planar data, while the latter is used for large-area data requiring consideration of Earth's curvature.


    
      
        Spatial data in Postgres
        Json String converted in Flink
      
    
    
      
        GEOMETRY(POINT, xx)
        {"coordinates":"[[174.9479, -36.7208]]","type":"Point","srid":3187}"
      
      
        GEOGRAPHY(MULTILINESTRING)
        {"coordinates":"[[169.1321, -44.7032],[167.8974, -44.6414]]","type":"MultiLineString","srid":4326}

PostgreSQL type	CDC type
NUMERIC[(M[,D])]	DECIMAL[(M[,D])]
NUMERIC	DECIMAL(38,0)
DECIMAL[(M[,D])]	DECIMAL[(M[,D])]
DECIMAL	DECIMAL(38,0)
MONEY[(M[,D])]	DECIMAL(38,digits)(The scale schema parameter contains an integer representing how many digits the decimal point was shifted. The scale schema parameter is determined by the money.fraction.digits connector configuration property.)

PostgreSQL type	CDC type
HSTORE	STRING(`debezium.hstore.handling.mode`=`string`)
HSTORE	MAP(`debezium.hstore.handling.mode`=`map`)

Spatial data in Postgres	Json String converted in Flink
GEOMETRY(POINT, xx)	{"coordinates":"[[174.9479, -36.7208]]","type":"Point","srid":3187}"
GEOGRAPHY(MULTILINESTRING)	{"coordinates":"[[169.1321, -44.7032],[167.8974, -44.6414]]","type":"MultiLineString","srid":4326}

Example​

Connector Options​

Startup Reading Position​

Available Source metrics​

Supported Metadata Columns​

Data Type Mapping​

Temporal types Mapping​

Decimal types Mapping​

HSTORE type Mapping​

Network address types Mapping​

PostGIS Types Mapping​

Example

Connector Options

Startup Reading Position

Available Source metrics

Supported Metadata Columns

Data Type Mapping

Temporal types Mapping

Decimal types Mapping

HSTORE type Mapping

Network address types Mapping

PostGIS Types Mapping