[
https://issues.apache.org/jira/browse/FLINK-39009?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Gaurav Miglani updated FLINK-39009:
-----------------------------------
Description:
*Summary:* Support PostgreSQL UNNEST bulk insert in the JDBC Table/SQL sink
h3. Problem
The JDBC Table/SQL sink writes each buffered batch with JDBC statement
batching. On PostgreSQL that is either one INSERT execution per row, or (with
{{{}reWriteBatchedInserts=true{}}}) a multi-row VALUES statement whose text and
parameter count change with the batch size. In streaming jobs the batch size
varies constantly, so PostgreSQL sees many distinct statements to parse and
plan.
h3. Proposal
Add an opt-in sink option {{sink.bulk-insert.enabled}} (default {{{}false{}}}).
When enabled, the sink writes each batch as a single statement that binds one
array per column, e.g. on PostgreSQL:
{code:sql}
INSERT INTO t (id, name) SELECT * FROM UNNEST(?::INTEGER[], ?::VARCHAR[]) AS
u(id, name)
[ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name]
{code}
Statement text and parameter count no longer depend on the batch size.
h3. Public API
* New Table/SQL option {{{}sink.bulk-insert.enabled{}}}.
* New {{@PublicEvolving}} capability interface {{{}JdbcBulkInsertDialect
extends JdbcDialect{}}}, implemented by {{{}PostgresDialect{}}}.
* The sink fails when it is created if the option is on and the dialect does
not implement the interface (no silent fallback).
* Table/SQL only. {{JdbcExecutionOptions}} and the DataStream {{JdbcSink}} are
unchanged.
h3. Scope
Append-only and upsert (CDC) writes. DELETEs keep the existing path. Supported
column types: BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE,
DECIMAL, CHAR, VARCHAR, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ, VARBINARY.
h3. Links
* FLIP: _..._
* Prototype PR: [https://github.com/apache/flink-connector-jdbc/pull/188]
* Prior art: Debezium JDBC sink,
[https://github.com/debezium/debezium/pull/7005]
was:
*Summary:* Support PostgreSQL UNNEST bulk insert in the JDBC Table/SQL sink
h3. Problem
The JDBC Table/SQL sink writes each buffered batch with JDBC statement
batching. On PostgreSQL that is either one INSERT execution per row, or (with
\{{reWriteBatchedInserts=true}}) a multi-row VALUES statement whose text and
parameter count change with the batch size. In streaming jobs the batch size
varies constantly, so PostgreSQL sees many distinct statements to parse and
plan.
h3. Proposal
Add an opt-in sink option \{{sink.bulk-insert.enabled}} (default \{{false}}).
When enabled, the sink writes each batch as a single statement that binds one
array per column, e.g. on PostgreSQL:
{code:sql}
INSERT INTO t (id, name) SELECT * FROM UNNEST(?::INTEGER[], ?::VARCHAR[]) AS
u(id, name)
[ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name]
{code}
Statement text and parameter count no longer depend on the batch size.
h3. Public API
* New Table/SQL option \{{sink.bulk-insert.enabled}}.
* New \{{@PublicEvolving}} capability interface \{{JdbcBulkInsertDialect
extends JdbcDialect}}, implemented by \{{PostgresDialect}}.
* The sink fails when it is created if the option is on and the dialect does
not implement the interface (no silent fallback).
* Table/SQL only. \{{JdbcExecutionOptions}} and the DataStream \{{JdbcSink}}
are unchanged.
h3. Scope
Append-only and upsert (CDC) writes. DELETEs keep the existing path. Supported
column types: BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE,
DECIMAL, CHAR, VARCHAR, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ, VARBINARY.
h3. Links
* FLIP: _link once created_
* Prototype PR: https://github.com/apache/flink-connector-jdbc/pull/188
* Prior art: Debezium JDBC sink, https://github.com/debezium/debezium/pull/7005
> flink postgres unnest insert way
> --------------------------------
>
> Key: FLINK-39009
> URL: https://issues.apache.org/jira/browse/FLINK-39009
> Project: Flink
> Issue Type: New Feature
> Components: Connectors / JDBC
> Reporter: Gaurav Miglani
> Priority: Major
> Labels: pull-request-available
>
> *Summary:* Support PostgreSQL UNNEST bulk insert in the JDBC Table/SQL sink
> h3. Problem
> The JDBC Table/SQL sink writes each buffered batch with JDBC statement
> batching. On PostgreSQL that is either one INSERT execution per row, or (with
> {{{}reWriteBatchedInserts=true{}}}) a multi-row VALUES statement whose text
> and parameter count change with the batch size. In streaming jobs the batch
> size varies constantly, so PostgreSQL sees many distinct statements to parse
> and plan.
> h3. Proposal
> Add an opt-in sink option {{sink.bulk-insert.enabled}} (default
> {{{}false{}}}). When enabled, the sink writes each batch as a single
> statement that binds one array per column, e.g. on PostgreSQL:
> {code:sql}
> INSERT INTO t (id, name) SELECT * FROM UNNEST(?::INTEGER[], ?::VARCHAR[]) AS
> u(id, name)
> [ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name]
> {code}
> Statement text and parameter count no longer depend on the batch size.
> h3. Public API
> * New Table/SQL option {{{}sink.bulk-insert.enabled{}}}.
> * New {{@PublicEvolving}} capability interface {{{}JdbcBulkInsertDialect
> extends JdbcDialect{}}}, implemented by {{{}PostgresDialect{}}}.
> * The sink fails when it is created if the option is on and the dialect does
> not implement the interface (no silent fallback).
> * Table/SQL only. {{JdbcExecutionOptions}} and the DataStream {{JdbcSink}}
> are unchanged.
> h3. Scope
> Append-only and upsert (CDC) writes. DELETEs keep the existing path.
> Supported column types: BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT,
> DOUBLE, DECIMAL, CHAR, VARCHAR, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ,
> VARBINARY.
> h3. Links
> * FLIP: _..._
> * Prototype PR: [https://github.com/apache/flink-connector-jdbc/pull/188]
> * Prior art: Debezium JDBC sink,
> [https://github.com/debezium/debezium/pull/7005]
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
