[FLINK-34722] Support conditional upserts with Postgres JDBC sink

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: jdbc-3.1.2
• Component/s: Connectors / JDBC
• Assignee: Unassigned
• Reporter: Pietro
• Created: 18/Mar/24 16:58
• Updated: 22/Mar/24 07:59

Description

The default Postgres dialect used by the JDBC sink for PostgreSQL DBs does not support custom WHERE conditions inside upsert statements at the moment.

Indeed, upsert statements returned by the getUpsertStatement() method are limited to:

ON CONFLICT (col1, ..., colN) 
DO UPDATE SET (col1=EXCLUDED.col1, ..., colN=EXCLUDED.colN)

PostgreSQL allows a finer-grained control of upsert statements by specifying a WHERE statement (see ON CONFLICT Clause), for instance:

ON CONFLICT (col1, ..., colN) 
DO UPDATE SET (col1=EXCLUDED.col1, ..., colN=EXCLUDED.colN) 
WHERE colN < EXCLUDED.colN

 
This could be useful in many use cases, for instance, in a CDC scenario where a batch reconciliation process has written records in the destination, which now face the risk of being overwritten by late arriving, stale, records in the streaming pipeline (adding a condition on the operation timestamp could protect from these events). 

My proposal is to extend the AbstractPostgresCompatibleDialect functionalities by making the upsert query support WHERE statements provided by users.

I'm thinking of two possible approaches, but I'd love to hear your opinion on this:

1. provide the statement through options of the JDBC sink connector.
2. allow users to plug custom dialects without them having to rewrite the whole JDBC sink (about this I'll open a separate issue soon) 

Thanks for your consideration