Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(jdbc-sink): fix primary key bindings for the DELETE statement #16770

Merged
merged 7 commits into from
May 16, 2024
Merged

fix(jdbc-sink): fix primary key bindings for the DELETE statement #16770

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
e22b358
fix
StrikeW May 15, 2024
0f9e657
fix primary key
StrikeW May 15, 2024
b0a2292
remove mili seconds part from temporal column
StrikeW May 15, 2024
3887108
minor
StrikeW May 15, 2024
4ae596b
minor
StrikeW May 15, 2024
e3c95fc
refine
StrikeW May 16, 2024
def7292
Merge remote-tracking branch 'origin/main' into siyuan/fix-jdbc-sink
StrikeW May 16, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions e2e_test/sink/remote/jdbc.check.pg.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ select * from t1_uuid;
221 74605c5a-a7bb-4b3b-8742-2a12e9709dea hello world


query T
query TIT
select * from sk_t1_uuid
----
21189447-8736-44bd-b254-26b5dec91da9
21189447-8736-44bd-b254-26b5dec91da9 2 bb
8 changes: 4 additions & 4 deletions e2e_test/sink/remote/jdbc.load.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,22 +166,22 @@ INSERT INTO t1_uuid values (221, '74605c5a-a7bb-4b3b-8742-2a12e9709dea', 'hello


statement ok
CREATE TABLE t1_test_uuid_delete (id varchar, primary key(id));
CREATE TABLE t1_test_uuid_delete (id varchar, v1 int, v2 varchar, primary key(id, v2));

statement ok
INSERT INTO t1_test_uuid_delete VALUES ('fb48ecc1-917f-4f4b-ab6d-d8e37809caf8'), ('21189447-8736-44bd-b254-26b5dec91da9');
INSERT INTO t1_test_uuid_delete VALUES ('fb48ecc1-917f-4f4b-ab6d-d8e37809caf8', 1, 'aa'), ('21189447-8736-44bd-b254-26b5dec91da9', 2, 'bb');

statement ok
CREATE SINK sk_t1_uuid FROM t1_test_uuid_delete WITH (
connector='jdbc',
jdbc.url='jdbc:postgresql://db:5432/test?user=test&password=connector',
table.name='sk_t1_uuid',
primary_key='id',
primary_key='id, v2',
type='upsert'
);

statement ok
DELETE FROM t1_test_uuid_delete WHERE ID='fb48ecc1-917f-4f4b-ab6d-d8e37809caf8';
DELETE FROM t1_test_uuid_delete WHERE id='fb48ecc1-917f-4f4b-ab6d-d8e37809caf8' AND v2='aa';


statement ok
Expand Down
2 changes: 1 addition & 1 deletion e2e_test/sink/remote/pg_create_table.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,4 +84,4 @@ CREATE TABLE biz.t2 (
"aBc" INTEGER PRIMARY KEY
);

CREATE TABLE sk_t1_uuid (id uuid, primary key(id));
CREATE TABLE sk_t1_uuid (id uuid, v1 int, v2 varchar, primary key(id, v2));
34 changes: 6 additions & 28 deletions .../connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/JDBCSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,9 +53,8 @@ public JDBCSink(JDBCSinkConfig config, TableSchema tableSchema) {
this.config = config;
try {
conn = JdbcUtils.getConnection(config.getJdbcUrl());
// Retrieve primary keys and column type mappings from the database
this.pkColumnNames =
getPkColumnNames(conn, config.getTableName(), config.getSchemaName());
// Table schema has been validated before, so we get the PK from it directly
this.pkColumnNames = tableSchema.getPrimaryKeys();
// column name -> java.sql.Types
Map<String, Integer> columnTypeMapping =
getColumnTypeMapping(conn, config.getTableName(), config.getSchemaName());
Expand All @@ -72,9 +71,10 @@ public JDBCSink(JDBCSinkConfig config, TableSchema tableSchema) {
.collect(Collectors.toList());

LOG.info(
"schema = {}, table = {}, columnSqlTypes = {}, pkIndices = {}",
"schema = {}, table = {}, tableSchema = {}, columnSqlTypes = {}, pkIndices = {}",
config.getSchemaName(),
config.getTableName(),
tableSchema,
columnSqlTypes,
pkIndices);

Expand Down Expand Up @@ -125,28 +125,6 @@ private static Map<String, Integer> getColumnTypeMapping(
return columnTypeMap;
}

private static List<String> getPkColumnNames(
Connection conn, String tableName, String schemaName) {
List<String> pkColumnNames = new ArrayList<>();
try {
var pks = conn.getMetaData().getPrimaryKeys(null, schemaName, tableName);
while (pks.next()) {
pkColumnNames.add(pks.getString(JDBC_COLUMN_NAME_KEY));
}
} catch (SQLException e) {
throw Status.INTERNAL
.withDescription(
String.format(ERROR_REPORT_TEMPLATE, e.getSQLState(), e.getMessage()))
.asRuntimeException();
}
LOG.info(
"schema = {}, table = {}: detected pk column = {}",
schemaName,
tableName,
pkColumnNames);
return pkColumnNames;
}

@Override
public boolean write(Iterable<SinkRow> rows) {
final int maxRetryCount = 4;
Expand Down Expand Up @@ -311,7 +289,7 @@ public void prepareDelete(SinkRow row) {
.asRuntimeException();
}
try {
jdbcDialect.bindDeleteStatement(deleteStatement, row);
jdbcDialect.bindDeleteStatement(deleteStatement, tableSchema, row);
deleteStatement.addBatch();
} catch (SQLException e) {
throw Status.INTERNAL
Expand Down Expand Up @@ -362,7 +340,7 @@ private void executeStatement(PreparedStatement stmt) throws SQLException {
if (stmt == null) {
return;
}
LOG.debug("Executing statement: {}", stmt);
LOG.info("Executing statement: {}", stmt);
stmt.executeBatch();
stmt.clearParameters();
}
Expand Down
3 changes: 2 additions & 1 deletion ...or-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/JdbcDialect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,5 +112,6 @@ void bindInsertIntoStatement(
throws SQLException;

/** Bind the values of primary key fields to the {@code DELETE} statement. */
void bindDeleteStatement(PreparedStatement stmt, SinkRow row) throws SQLException;
void bindDeleteStatement(PreparedStatement stmt, TableSchema tableSchema, SinkRow row)
throws SQLException;
}
34 changes: 27 additions & 7 deletions ...r-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/MySqlDialect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,25 +19,27 @@
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;

public class MySqlDialect implements JdbcDialect {

private final int[] columnSqlTypes;
private final int[] pkIndices;
private final int[] pkColumnSqlTypes;

public MySqlDialect(List<Integer> columnSqlTypes, List<Integer> pkIndices) {
this.columnSqlTypes = columnSqlTypes.stream().mapToInt(i -> i).toArray();
var columnSqlTypesArr = columnSqlTypes.stream().mapToInt(i -> i).toArray();
this.pkIndices = pkIndices.stream().mapToInt(i -> i).toArray();

// derive sql types for pk columns
var pkColumnSqlTypes = new int[pkIndices.size()];
for (int i = 0; i < pkIndices.size(); i++) {
pkColumnSqlTypes[i] = this.columnSqlTypes[this.pkIndices[i]];
pkColumnSqlTypes[i] = columnSqlTypesArr[this.pkIndices[i]];
}
this.pkColumnSqlTypes = pkColumnSqlTypes;
}
Expand Down Expand Up @@ -118,12 +120,30 @@ public void bindInsertIntoStatement(
}

@Override
public void bindDeleteStatement(PreparedStatement stmt, SinkRow row) throws SQLException {
public void bindDeleteStatement(PreparedStatement stmt, TableSchema tableSchema, SinkRow row)
throws SQLException {
// set the values of primary key fields
int placeholderIdx = 1;
for (int idx : pkIndices) {
Object pkField = row.get(idx);
stmt.setObject(placeholderIdx++, pkField, pkColumnSqlTypes[idx]);
var columnDescs = tableSchema.getColumnDescs();
for (int i = 0; i < pkIndices.length; ++i) {
int pkIdx = pkIndices[i];
Object pkField = row.get(pkIdx);
// remove the milliseconds part from the timestamp/timestamptz
switch (columnDescs.get(pkIdx).getDataType().getTypeName()) {
case TIMESTAMP:
LocalDateTime ldt = (LocalDateTime) pkField;
pkField =
LocalDateTime.ofEpochSecond(
ldt.toEpochSecond(ZoneOffset.UTC), 0, ZoneOffset.UTC);
break;
case TIMESTAMPTZ:
OffsetDateTime odt = (OffsetDateTime) pkField;
pkField = LocalDateTime.ofEpochSecond(odt.toEpochSecond(), 0, ZoneOffset.UTC);
break;
Copy link
Contributor

@xiangjinwu xiangjinwu May 16, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • Both LocalDateTime and OffsetDateTime have a method withNano which floor to whole seconds. Shorter than current implementation.
  • It seems TIMESTAMPTZ branch gets OffsetDateTime out but puts LocalDateTime. Is it intended or accidental?
  • Could you elaborate why it is necessary to remove the subsecond part? Maybe it only works when the MySQL table is declared with DATETIME and TIMESTAMP, but would fail with DATETIME(6) or TIMESTAMP(6). Specially, PostgreSQL (and RisingWave) default precision is 6 (i.e. timestamptz is short for timestamptz(6)) while MySQL default precision is 0 (i.e. datetime is short for datetime(0)).
    • This may imply that we need to convert from 6 to whatever precision defined in downstream MySQL. (demo) Are we able to get that info from the downstream column?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems TIMESTAMPTZ branch gets OffsetDateTime out but puts LocalDateTime. Is it intended or accidental?

Intended. since downstream timestamp doesn't store timezone, so I convert to LocalDateTime

default:
break;
}
stmt.setObject(placeholderIdx++, pkField, pkColumnSqlTypes[i]);
}
}
}
17 changes: 5 additions & 12 deletions ...ode/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/jdbc/PostgresDialect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,18 +31,10 @@ public class PostgresDialect implements JdbcDialect {

private final int[] columnSqlTypes;
private final int[] pkIndices;
private final int[] pkColumnSqlTypes;

public PostgresDialect(List<Integer> columnSqlTypes, List<Integer> pkIndices) {
this.columnSqlTypes = columnSqlTypes.stream().mapToInt(i -> i).toArray();
this.pkIndices = pkIndices.stream().mapToInt(i -> i).toArray();

// derive sql types for pk columns
var pkColumnSqlTypes = new int[pkIndices.size()];
for (int i = 0; i < pkIndices.size(); i++) {
pkColumnSqlTypes[i] = this.columnSqlTypes[this.pkIndices[i]];
}
this.pkColumnSqlTypes = pkColumnSqlTypes;
}

private static final HashMap<TypeName, String> RW_TYPE_TO_JDBC_TYPE_NAME;
Expand Down Expand Up @@ -166,12 +158,13 @@ public void bindInsertIntoStatement(
}

@Override
public void bindDeleteStatement(PreparedStatement stmt, SinkRow row) throws SQLException {
public void bindDeleteStatement(PreparedStatement stmt, TableSchema tableSchema, SinkRow row)
throws SQLException {
// set the values of primary key fields
int placeholderIdx = 1;
for (int idx : pkIndices) {
Object pkField = row.get(idx);
stmt.setObject(placeholderIdx++, pkField, pkColumnSqlTypes[idx]);
for (int pkIdx : pkIndices) {
Object pkField = row.get(pkIdx);
stmt.setObject(placeholderIdx++, pkField, columnSqlTypes[pkIdx]);
}
}
}
Loading