Hi all, I am trying to upsert data into a Postgres JDBC sink. The sink:

CREATE TABLE group (
            group_id INT NOT NULL,
            representative INT,
            score INT,
            PRIMARY KEY (group_id) NOT ENFORCED 
        ) WITH (
            'connector' = 'jdbc',
            'url' = 'jdbc:<postgresql://host.docker.internal:5432/doc_data>',
            'table-name' = 'xml.t_group',
            'username' = 'root',
            'password' = 'root',
            'driver' = 'org.postgresql.Driver'
        );

The insert statement:

INSERT INTO group (group_id, representative, score) SELECT * FROM tmp_docs
        ON CONFLICT (group_id) DO UPDATE SET
            representative = CASE WHEN group.score >= tmp_docs.score THEN group.representative ELSE tmp_docs.representative;

According to the JDBC docs this is how to do it in Postgres https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/jdbc/ But instead of it working, I get the following error message:

org.apache.flink.table.api.SqlParserException: SQL parse failed. Incorrect syntax near the keyword 'ON'

Or does it mean that Flink uses this logic internally, and it it not possible to customize the

ON CONFLICT

clause?

You will still need to use Flink SQL to insert into the table. Under the hood that will be translated into the Postgres dialect. But IIRC there’s indeed nog support for ON CONFLICT in the Postgres dialect. Not 100% sure though

Hmm… that is unfortunate. Thanks anyways!

^ Yes if you’re using the DataStream API

I am not using it currently, but I could modify the code to work with it I guess. I will try, and see if it works out.

Yep

I have tried running it with Postgres CDC as source table and JDBC stream as sink, but I get the following error:

pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Table sink '*anonymous_datastream_sink$1*' doesn't support consuming update and delete changes which is produced by node TableSourceScan

Similar setup works well for me, but I’m not using the Python API… Could be something Python-specific?

No I think I have it working now. There is a method

<http://table_env.to|table_env.to>_changelog_stream

which converts the changelog to a data stream. It seems to be working now!