Skip to content

Commit 849d748

Browse files
hailin0hailin0
authored
[Improve][Jdbc] Add quote identifier for sql (#6669)
1 parent 19888e7 commit 849d748

File tree

11 files changed

+350
-7
lines changed

11 files changed

+350
-7
lines changed

Diff for: ‎seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlUtils.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -434,7 +434,7 @@ private static void addPrimaryKeyColumnsToCondition(
434434
SeaTunnelRowType rowType, StringBuilder sql, String predicate) {
435435
for (Iterator<String> fieldNamesIt = Arrays.stream(rowType.getFieldNames()).iterator();
436436
fieldNamesIt.hasNext(); ) {
437-
sql.append(fieldNamesIt.next()).append(predicate);
437+
sql.append(quote(fieldNamesIt.next())).append(predicate);
438438
if (fieldNamesIt.hasNext()) {
439439
sql.append(" AND ");
440440
}

Diff for: ‎seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlUtilsTest.java

+72
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils;
19+
20+
import org.apache.seatunnel.api.table.type.BasicType;
21+
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
22+
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
23+
24+
import org.junit.jupiter.api.Assertions;
25+
import org.junit.jupiter.api.Test;
26+
27+
import io.debezium.relational.TableId;
28+
29+
public class MySqlUtilsTest {
30+
31+
@Test
32+
public void testSplitScanQuery() {
33+
String splitScanSQL =
34+
MySqlUtils.buildSplitScanQuery(
35+
TableId.parse("db1.table1"),
36+
new SeaTunnelRowType(
37+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
38+
false,
39+
false);
40+
Assertions.assertEquals(
41+
"SELECT * FROM `db1`.`table1` WHERE `id` >= ? AND NOT (`id` = ?) AND `id` <= ?",
42+
splitScanSQL);
43+
44+
splitScanSQL =
45+
MySqlUtils.buildSplitScanQuery(
46+
TableId.parse("db1.table1"),
47+
new SeaTunnelRowType(
48+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
49+
true,
50+
true);
51+
Assertions.assertEquals("SELECT * FROM `db1`.`table1`", splitScanSQL);
52+
53+
splitScanSQL =
54+
MySqlUtils.buildSplitScanQuery(
55+
TableId.parse("db1.table1"),
56+
new SeaTunnelRowType(
57+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
58+
true,
59+
false);
60+
Assertions.assertEquals(
61+
"SELECT * FROM `db1`.`table1` WHERE `id` <= ? AND NOT (`id` = ?)", splitScanSQL);
62+
63+
splitScanSQL =
64+
MySqlUtils.buildSplitScanQuery(
65+
TableId.parse("db1.table1"),
66+
new SeaTunnelRowType(
67+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
68+
false,
69+
true);
70+
Assertions.assertEquals("SELECT * FROM `db1`.`table1` WHERE `id` >= ?", splitScanSQL);
71+
}
72+
}

Diff for: ‎seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtils.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -439,7 +439,7 @@ private static void addPrimaryKeyColumnsToCondition(
439439
SeaTunnelRowType rowType, StringBuilder sql, String predicate) {
440440
for (Iterator<String> fieldNamesIt = Arrays.stream(rowType.getFieldNames()).iterator();
441441
fieldNamesIt.hasNext(); ) {
442-
sql.append(fieldNamesIt.next()).append(predicate);
442+
sql.append(quote(fieldNamesIt.next())).append(predicate);
443443
if (fieldNamesIt.hasNext()) {
444444
sql.append(" AND ");
445445
}

Diff for: ‎seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/utils/OracleUtilsTest.java

+73
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.cdc.oracle.utils;
19+
20+
import org.apache.seatunnel.api.table.type.BasicType;
21+
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
22+
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
23+
24+
import org.junit.jupiter.api.Assertions;
25+
import org.junit.jupiter.api.Test;
26+
27+
import io.debezium.relational.TableId;
28+
29+
public class OracleUtilsTest {
30+
@Test
31+
public void testSplitScanQuery() {
32+
String splitScanSQL =
33+
OracleUtils.buildSplitScanQuery(
34+
TableId.parse("db1.schema1.table1"),
35+
new SeaTunnelRowType(
36+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
37+
false,
38+
false);
39+
Assertions.assertEquals(
40+
"SELECT * FROM \"schema1\".\"table1\" WHERE \"id\" >= ? AND NOT (\"id\" = ?) AND \"id\" <= ?",
41+
splitScanSQL);
42+
43+
splitScanSQL =
44+
OracleUtils.buildSplitScanQuery(
45+
TableId.parse("db1.schema1.table1"),
46+
new SeaTunnelRowType(
47+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
48+
true,
49+
true);
50+
Assertions.assertEquals("SELECT * FROM \"schema1\".\"table1\"", splitScanSQL);
51+
52+
splitScanSQL =
53+
OracleUtils.buildSplitScanQuery(
54+
TableId.parse("db1.schema1.table1"),
55+
new SeaTunnelRowType(
56+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
57+
true,
58+
false);
59+
Assertions.assertEquals(
60+
"SELECT * FROM \"schema1\".\"table1\" WHERE \"id\" <= ? AND NOT (\"id\" = ?)",
61+
splitScanSQL);
62+
63+
splitScanSQL =
64+
OracleUtils.buildSplitScanQuery(
65+
TableId.parse("db1.schema1.table1"),
66+
new SeaTunnelRowType(
67+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
68+
false,
69+
true);
70+
Assertions.assertEquals(
71+
"SELECT * FROM \"schema1\".\"table1\" WHERE \"id\" >= ?", splitScanSQL);
72+
}
73+
}

Diff for: ‎seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresUtils.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -444,7 +444,7 @@ private static void addPrimaryKeyColumnsToCondition(
444444
SeaTunnelRowType rowType, StringBuilder sql, String predicate) {
445445
for (Iterator<String> fieldNamesIt = Arrays.stream(rowType.getFieldNames()).iterator();
446446
fieldNamesIt.hasNext(); ) {
447-
sql.append(fieldNamesIt.next()).append(predicate);
447+
sql.append(quote(fieldNamesIt.next())).append(predicate);
448448
if (fieldNamesIt.hasNext()) {
449449
sql.append(" AND ");
450450
}

Diff for: ‎seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/utils/PostgresUtilsTest.java

+73
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.cdc.postgres.utils;
19+
20+
import org.apache.seatunnel.api.table.type.BasicType;
21+
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
22+
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
23+
24+
import org.junit.jupiter.api.Assertions;
25+
import org.junit.jupiter.api.Test;
26+
27+
import io.debezium.relational.TableId;
28+
29+
public class PostgresUtilsTest {
30+
@Test
31+
public void testSplitScanQuery() {
32+
String splitScanSQL =
33+
PostgresUtils.buildSplitScanQuery(
34+
TableId.parse("db1.schema1.table1"),
35+
new SeaTunnelRowType(
36+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
37+
false,
38+
false);
39+
Assertions.assertEquals(
40+
"SELECT * FROM \"schema1\".\"table1\" WHERE \"id\" >= ? AND NOT (\"id\" = ?) AND \"id\" <= ?",
41+
splitScanSQL);
42+
43+
splitScanSQL =
44+
PostgresUtils.buildSplitScanQuery(
45+
TableId.parse("db1.schema1.table1"),
46+
new SeaTunnelRowType(
47+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
48+
true,
49+
true);
50+
Assertions.assertEquals("SELECT * FROM \"schema1\".\"table1\"", splitScanSQL);
51+
52+
splitScanSQL =
53+
PostgresUtils.buildSplitScanQuery(
54+
TableId.parse("db1.schema1.table1"),
55+
new SeaTunnelRowType(
56+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
57+
true,
58+
false);
59+
Assertions.assertEquals(
60+
"SELECT * FROM \"schema1\".\"table1\" WHERE \"id\" <= ? AND NOT (\"id\" = ?)",
61+
splitScanSQL);
62+
63+
splitScanSQL =
64+
PostgresUtils.buildSplitScanQuery(
65+
TableId.parse("db1.schema1.table1"),
66+
new SeaTunnelRowType(
67+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
68+
false,
69+
true);
70+
Assertions.assertEquals(
71+
"SELECT * FROM \"schema1\".\"table1\" WHERE \"id\" >= ?", splitScanSQL);
72+
}
73+
}

Diff for: ‎seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/utils/SqlServerUtils.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -467,7 +467,7 @@ private static void addPrimaryKeyColumnsToCondition(
467467
SeaTunnelRowType rowType, StringBuilder sql, String predicate) {
468468
for (Iterator<String> fieldNamesIt = Arrays.stream(rowType.getFieldNames()).iterator();
469469
fieldNamesIt.hasNext(); ) {
470-
sql.append(fieldNamesIt.next()).append(predicate);
470+
sql.append(quote(fieldNamesIt.next())).append(predicate);
471471
if (fieldNamesIt.hasNext()) {
472472
sql.append(" AND ");
473473
}

Diff for: ‎seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/utils/SqlServerUtilsTest.java

+72
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source.utils;
19+
20+
import org.apache.seatunnel.api.table.type.BasicType;
21+
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
22+
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
23+
24+
import org.junit.jupiter.api.Assertions;
25+
import org.junit.jupiter.api.Test;
26+
27+
import io.debezium.relational.TableId;
28+
29+
public class SqlServerUtilsTest {
30+
@Test
31+
public void testSplitScanQuery() {
32+
String splitScanSQL =
33+
SqlServerUtils.buildSplitScanQuery(
34+
TableId.parse("db1.schema1.table1"),
35+
new SeaTunnelRowType(
36+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
37+
false,
38+
false);
39+
Assertions.assertEquals(
40+
"SELECT * FROM [schema1].[table1] WHERE [id] >= ? AND NOT ([id] = ?) AND [id] <= ?",
41+
splitScanSQL);
42+
43+
splitScanSQL =
44+
SqlServerUtils.buildSplitScanQuery(
45+
TableId.parse("db1.schema1.table1"),
46+
new SeaTunnelRowType(
47+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
48+
true,
49+
true);
50+
Assertions.assertEquals("SELECT * FROM [schema1].[table1]", splitScanSQL);
51+
52+
splitScanSQL =
53+
SqlServerUtils.buildSplitScanQuery(
54+
TableId.parse("db1.schema1.table1"),
55+
new SeaTunnelRowType(
56+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
57+
true,
58+
false);
59+
Assertions.assertEquals(
60+
"SELECT * FROM [schema1].[table1] WHERE [id] <= ? AND NOT ([id] = ?)",
61+
splitScanSQL);
62+
63+
splitScanSQL =
64+
SqlServerUtils.buildSplitScanQuery(
65+
TableId.parse("db1.schema1.table1"),
66+
new SeaTunnelRowType(
67+
new String[] {"id"}, new SeaTunnelDataType[] {BasicType.LONG_TYPE}),
68+
false,
69+
true);
70+
Assertions.assertEquals("SELECT * FROM [schema1].[table1] WHERE [id] >= ?", splitScanSQL);
71+
}
72+
}

Diff for: ‎seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psql/PostgresDialect.java

+5
Original file line numberDiff line numberDiff line change
@@ -137,6 +137,11 @@ public String quoteIdentifier(String identifier) {
137137
return "\"" + getFieldIde(identifier, fieldIde) + "\"";
138138
}
139139

140+
@Override
141+
public String tableIdentifier(TablePath tablePath) {
142+
return tablePath.getFullNameWithQuoted("\"");
143+
}
144+
140145
@Override
141146
public String quoteDatabaseIdentifier(String identifier) {
142147
return "\"" + identifier + "\"";

Diff for: ‎seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java

+6-3
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818
package org.apache.seatunnel.connectors.seatunnel.jdbc.source;
1919

20+
import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting;
21+
2022
import org.apache.seatunnel.api.table.catalog.TablePath;
2123
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
2224
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
@@ -449,7 +451,8 @@ private int objectCompare(Object obj1, Object obj2) {
449451
return ObjectUtils.compare(obj1, obj2);
450452
}
451453

452-
private String createDynamicSplitQuerySQL(JdbcSourceSplit split) {
454+
@VisibleForTesting
455+
String createDynamicSplitQuerySQL(JdbcSourceSplit split) {
453456
SeaTunnelRowType rowType =
454457
new SeaTunnelRowType(
455458
new String[] {split.getSplitKeyName()},
@@ -499,11 +502,11 @@ private String createDynamicSplitQuerySQL(JdbcSourceSplit split) {
499502
return sql.toString();
500503
}
501504

502-
private static void addKeyColumnsToCondition(
505+
private void addKeyColumnsToCondition(
503506
SeaTunnelRowType rowType, StringBuilder sql, String predicate) {
504507
for (Iterator<String> fieldNamesIt = Arrays.stream(rowType.getFieldNames()).iterator();
505508
fieldNamesIt.hasNext(); ) {
506-
sql.append(fieldNamesIt.next()).append(predicate);
509+
sql.append(jdbcDialect.quoteIdentifier(fieldNamesIt.next())).append(predicate);
507510
if (fieldNamesIt.hasNext()) {
508511
sql.append(" AND ");
509512
}

0 commit comments

Comments
 (0)