Skip to content

Commit 56110bf

Browse files
hailin0hailin0
authored
[Improve][Jdbc] Support upsert for opengauss (#8627)
1 parent d6d3000 commit 56110bf

File tree

3 files changed

+137
-0
lines changed

3 files changed

+137
-0
lines changed

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

+49
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
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.jdbc.internal.dialect.opengauss;
19+
20+
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.psql.PostgresDialect;
21+
22+
import java.util.Arrays;
23+
import java.util.Optional;
24+
import java.util.stream.Collectors;
25+
26+
public class OpenGaussDialect extends PostgresDialect {
27+
28+
@Override
29+
public Optional<String> getUpsertStatement(
30+
String database, String tableName, String[] fieldNames, String[] uniqueKeyFields) {
31+
String updateClause =
32+
Arrays.stream(fieldNames)
33+
.filter(fieldName -> !Arrays.asList(uniqueKeyFields).contains(fieldName))
34+
.map(
35+
fieldName ->
36+
quoteIdentifier(fieldName)
37+
+ "=EXCLUDED."
38+
+ quoteIdentifier(fieldName))
39+
.collect(Collectors.joining(", "));
40+
if (updateClause.isEmpty()) {
41+
return Optional.empty();
42+
}
43+
String upsertSQL =
44+
String.format(
45+
"%s ON DUPLICATE KEY UPDATE %s",
46+
getInsertIntoStatement(database, tableName, fieldNames), updateClause);
47+
return Optional.of(upsertSQL);
48+
}
49+
}

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

+13
Original file line numberDiff line numberDiff line change
@@ -16,16 +16,29 @@
1616
*/
1717
package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.opengauss;
1818

19+
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
1920
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectFactory;
2021
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.psql.PostgresDialectFactory;
2122

2223
import com.google.auto.service.AutoService;
2324

25+
import javax.annotation.Nonnull;
26+
2427
@AutoService(JdbcDialectFactory.class)
2528
public class OpenGaussDialectFactory extends PostgresDialectFactory {
2629

2730
@Override
2831
public boolean acceptsURL(String url) {
2932
return url.startsWith("jdbc:opengauss:");
3033
}
34+
35+
@Override
36+
public JdbcDialect create() {
37+
return new OpenGaussDialect();
38+
}
39+
40+
@Override
41+
public JdbcDialect create(@Nonnull String compatibleMode, String fieldIde) {
42+
return new OpenGaussDialect();
43+
}
3144
}

Diff for: ‎seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/opengauss/OpenGaussDialectTest.java

+75
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
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.jdbc.internal.dialect.opengauss;
19+
20+
import org.junit.jupiter.api.Test;
21+
22+
import java.util.Optional;
23+
24+
import static org.junit.jupiter.api.Assertions.assertEquals;
25+
import static org.junit.jupiter.api.Assertions.assertFalse;
26+
import static org.junit.jupiter.api.Assertions.assertTrue;
27+
28+
public class OpenGaussDialectTest {
29+
30+
@Test
31+
void returnsUpsertStatementWhenUpdateClauseIsNotEmpty() {
32+
OpenGaussDialect dialect = new OpenGaussDialect();
33+
String[] fieldNames = {"id", "name", "age"};
34+
String[] uniqueKeyFields = {"id"};
35+
Optional<String> upsertStatement =
36+
dialect.getUpsertStatement("test_db", "test_table", fieldNames, uniqueKeyFields);
37+
assertTrue(upsertStatement.isPresent());
38+
assertEquals(
39+
"INSERT INTO \"test_db\".\"test_table\" (\"id\", \"name\", \"age\") VALUES (:id, :name, :age) ON DUPLICATE KEY UPDATE \"name\"=EXCLUDED.\"name\", \"age\"=EXCLUDED.\"age\"",
40+
upsertStatement.get());
41+
}
42+
43+
@Test
44+
void returnsEmptyWhenUpdateClauseIsEmpty() {
45+
OpenGaussDialect dialect = new OpenGaussDialect();
46+
String[] fieldNames = {"id"};
47+
String[] uniqueKeyFields = {"id"};
48+
Optional<String> upsertStatement =
49+
dialect.getUpsertStatement("test_db", "test_table", fieldNames, uniqueKeyFields);
50+
assertFalse(upsertStatement.isPresent());
51+
}
52+
53+
@Test
54+
void handlesEmptyFieldNames() {
55+
OpenGaussDialect dialect = new OpenGaussDialect();
56+
String[] fieldNames = {};
57+
String[] uniqueKeyFields = {"id"};
58+
Optional<String> upsertStatement =
59+
dialect.getUpsertStatement("test_db", "test_table", fieldNames, uniqueKeyFields);
60+
assertFalse(upsertStatement.isPresent());
61+
}
62+
63+
@Test
64+
void handlesEmptyUniqueKeyFields() {
65+
OpenGaussDialect dialect = new OpenGaussDialect();
66+
String[] fieldNames = {"id", "name", "age"};
67+
String[] uniqueKeyFields = {};
68+
Optional<String> upsertStatement =
69+
dialect.getUpsertStatement("test_db", "test_table", fieldNames, uniqueKeyFields);
70+
assertTrue(upsertStatement.isPresent());
71+
assertEquals(
72+
"INSERT INTO \"test_db\".\"test_table\" (\"id\", \"name\", \"age\") VALUES (:id, :name, :age) ON DUPLICATE KEY UPDATE \"id\"=EXCLUDED.\"id\", \"name\"=EXCLUDED.\"name\", \"age\"=EXCLUDED.\"age\"",
73+
upsertStatement.get());
74+
}
75+
}

0 commit comments

Comments
 (0)