fix(jdbc-sink): add schema support for postgres sink #10576

StrikeW · 2023-06-27T03:32:07Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Add a optional schema.name field to the WITH clause to allow user to specify schema for pg sink

CREATE SINK s2_postgres
FROM
    lineitem_rw WITH (
        connector = 'jdbc',
        jdbc.url = 'jdbc:postgresql://localhost:5432/postgres?user=postgres&password=xxxx',
        table.name = 't_lineitem',
        schema.name='biz',
        type='upsert'
);

Fix #10418

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

Documentation

My PR contains user-facing changes.

Click here for Documentation

Types of user-facing changes

Please keep the types that apply to your changes, and remove the others.

Connector (sources & sinks)
SQL commands, functions, and operators

Release note

Add a optional schema.name field to the WITH clause to allow user to specify schema for Postgres sink

codecov · 2023-06-27T10:06:44Z

Codecov Report

Merging #10576 (ea1f24c) into main (2100dc2) will not change coverage.
The diff coverage is n/a.

@@           Coverage Diff           @@
##             main   #10576   +/-   ##
=======================================
  Coverage   70.14%   70.14%           
=======================================
  Files        1276     1276           
  Lines      219818   219818           
=======================================
  Hits       154200   154200           
  Misses      65618    65618

Flag	Coverage Δ
rust	`70.14% <ø> (ø)`

Flags with carried forward coverage won't be shown. Click here to find out more.

see 2 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

wenym1

Rest LGTM.

java/connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/JDBCSink.java

...nnector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/JDBCSinkConfig.java

github-actions

license-eye has totally checked 3692 files.

Valid	Invalid	Ignored	Fixed
1668	1	2023	0

Click to see the invalid file list

java/connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/SchemaTableName.java

...r-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/SchemaTableName.java

github-actions

license-eye has totally checked 3695 files.

Valid	Invalid	Ignored	Fixed
1669	2	2024	0

Click to see the invalid file list

java/connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/PostgresSchemaTableName.java
java/connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/SchemaTableName.java

...isingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/PostgresSchemaTableName.java

wenym1 · 2023-06-29T03:24:13Z

...r-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/SchemaTableName.java

+        return tableName;
+    }
+
+    public String getNormalizedTableName() {


I think this method should be inside the JdbcDialect, and has the signature of getNormalizedTableName(SchemaTableName schemaTableName), since how to concatenate the schema name and table name is a database-specific behavior. This class should be simply a POJO that contains that raw input from user.

wenym1 · 2023-06-29T03:28:49Z

...isingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/PostgresSchemaTableName.java

+package com.risingwave.connector.jdbc;
+
+public class PostgresSchemaTableName extends SchemaTableName {
+    public PostgresSchemaTableName(String schemaName, String tableName) {


We already have a createSchemaTable in JdbcDialect. We can set the default schema name there. This class is unnecessary.

wenym1 · 2023-06-29T03:34:48Z

...ector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/JdbcDialect.java

        String columns =
                fieldNames.stream().map(this::quoteIdentifier).collect(Collectors.joining(", "));
        String placeholders = fieldNames.stream().map(f -> "?").collect(Collectors.joining(", "));
        return "INSERT INTO "
-                + quoteIdentifier(tableName)
+                + quoteIdentifier(tableName.getNormalizedTableName())


It should be "INSERT INTO " + getNormalizedTableName() .... The current code cannot handle databases that quote the schema name and table name separatedly (i.e. `schema_name`.`table_name`).

wenym1

LGTM. Thanks for the PR!

Co-authored-by: lmatz <lmatz823@gmail.com>

StrikeW added 3 commits June 27, 2023 00:00

add schema support for pg sink

c5e3a93

fmt

4bea879

minor

934e302

github-actions bot added the type/fix Bug fix label Jun 27, 2023

StrikeW added 2 commits June 27, 2023 11:32

minor

04157e1

add test cases

b548b41

StrikeW marked this pull request as ready for review June 27, 2023 04:10

StrikeW marked this pull request as draft June 27, 2023 09:18

fix

2ec50cd

StrikeW marked this pull request as ready for review June 27, 2023 09:48

StrikeW requested review from liurenjie1024, tabVersion, wenym1 and xx01cyx June 27, 2023 09:51

StrikeW added the user-facing-changes Contains changes that are visible to users label Jun 28, 2023

tabVersion approved these changes Jun 28, 2023

View reviewed changes

wenym1 reviewed Jun 28, 2023

View reviewed changes

java/connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/JDBCSink.java Outdated Show resolved Hide resolved

...nnector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/JDBCSinkConfig.java Outdated Show resolved Hide resolved

liurenjie1024 approved these changes Jun 28, 2023

View reviewed changes

StrikeW added 3 commits June 28, 2023 18:31

fix comment

737d468

Merge remote-tracking branch 'origin/main' into siyuan/jdbc-sink-fix

6968fac

minor

a521a8f

StrikeW enabled auto-merge June 28, 2023 11:51

wenym1 disabled auto-merge June 28, 2023 11:55

fix

9f896dc

StrikeW requested a review from wenym1 June 28, 2023 16:36

github-actions bot reviewed Jun 28, 2023

View reviewed changes

...r-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/SchemaTableName.java Show resolved Hide resolved

introduce SchemaTableName

1f1dc0d

StrikeW force-pushed the siyuan/jdbc-sink-fix branch from e5bf443 to 1f1dc0d Compare June 29, 2023 01:57

github-actions bot reviewed Jun 29, 2023

View reviewed changes

...isingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/PostgresSchemaTableName.java Outdated Show resolved Hide resolved

StrikeW added 2 commits June 29, 2023 10:05

license

d3ae8ec

typo

15d899d

wenym1 reviewed Jun 29, 2023

View reviewed changes

StrikeW added 2 commits June 29, 2023 12:07

update

7c41a70

typo

084dde5

wenym1 approved these changes Jun 29, 2023

View reviewed changes

fmt

ea1f24c

StrikeW added this pull request to the merge queue Jun 29, 2023

lmatz added the need-cherry-pick-v0.19 label Jun 29, 2023

Merged via the queue into main with commit 8ef743f Jun 29, 2023
34 of 35 checks passed

StrikeW deleted the siyuan/jdbc-sink-fix branch June 29, 2023 04:57

StrikeW mentioned this pull request Jun 29, 2023

cherrypick fix(jdbc-sink): add schema support for postgres sink (#10576) to branch v0.19.0-rc #10644

Closed

StrikeW added a commit that referenced this pull request Jun 29, 2023

fix(jdbc-sink): add schema support for postgres sink (#10576)

29f38a4

lmatz added a commit that referenced this pull request Jun 30, 2023

fix(jdbc-sink): cherry-pick #10326 and #10576 to v0.19.0-rc (#10651)

2dfba7f

Co-authored-by: lmatz <lmatz823@gmail.com>

emile-00 mentioned this pull request Jul 11, 2023

Document schema.name parameter for PostgreSQL sink risingwavelabs/risingwave-docs#1031

Closed

emile-00 added the 📖✓ Covered or will be covered in the user docs. label Jul 11, 2023

emile-00 mentioned this pull request Jul 14, 2023

Add schema.name parameter for Postgres sink risingwavelabs/risingwave-docs#1054

Merged

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix(jdbc-sink): add schema support for postgres sink #10576

fix(jdbc-sink): add schema support for postgres sink #10576

StrikeW commented Jun 27, 2023 •

edited

Loading

codecov bot commented Jun 27, 2023 •

edited

Loading

wenym1 left a comment

github-actions bot left a comment

github-actions bot left a comment

wenym1 Jun 29, 2023

wenym1 Jun 29, 2023

wenym1 Jun 29, 2023

wenym1 left a comment

fix(jdbc-sink): add schema support for postgres sink #10576

fix(jdbc-sink): add schema support for postgres sink #10576

Conversation

StrikeW commented Jun 27, 2023 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Types of user-facing changes

Release note

codecov bot commented Jun 27, 2023 • edited Loading

Codecov Report

wenym1 left a comment

Choose a reason for hiding this comment

github-actions bot left a comment

Choose a reason for hiding this comment

github-actions bot left a comment

Choose a reason for hiding this comment

wenym1 Jun 29, 2023

Choose a reason for hiding this comment

wenym1 Jun 29, 2023

Choose a reason for hiding this comment

wenym1 Jun 29, 2023

Choose a reason for hiding this comment

wenym1 left a comment

Choose a reason for hiding this comment

StrikeW commented Jun 27, 2023 •

edited

Loading

codecov bot commented Jun 27, 2023 •

edited

Loading