fix: use INSERT IGNORE for primary key only table (#66)

flink-connector-oceanbase/src/main/java/com/oceanbase/connector/flink/dialect/OceanBaseMySQLDialect.java

@@ -18,6 +18,8 @@
 
 import org.apache.flink.util.function.SerializableFunction;
 
+import org.apache.commons.lang3.StringUtils;
+
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
@@ -45,9 +47,15 @@ public String getUpsertStatement(
  .filter(f -> !uniqueKeyFields.contains(f))
  .map(f -> quoteIdentifier(f) + "=VALUES(" + quoteIdentifier(f) + ")")
  .collect(Collectors.joining(", "));
- return getInsertIntoStatement(schemaName, tableName, fieldNames, placeholderFunc)
- + " ON DUPLICATE KEY UPDATE "
- + updateClause;
+ String insertIntoStatement =
+ getInsertIntoStatement(schemaName, tableName, fieldNames, placeholderFunc);
+ if (StringUtils.isNotEmpty(updateClause)) {
+ // ON DUPLICATE KEY UPDATE
+ return insertIntoStatement + " ON DUPLICATE KEY UPDATE " + updateClause;
+ } else {
+ // INSERT IGNORE
+ return StringUtils.replace(insertIntoStatement, "INSERT", "INSERT IGNORE", 1);
+ }
  }
 
  @Override

flink-connector-oceanbase/src/test/java/com/oceanbase/connector/flink/dialect/OceanBaseMySQLDialectTest.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2024 OceanBase.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.oceanbase.connector.flink.dialect;
+
+import org.apache.flink.util.function.SerializableFunction;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+class OceanBaseMySQLDialectTest {
+
+ @Test
+ void getUpsertStatementUpdate() {
+ OceanBaseMySQLDialect dialect = new OceanBaseMySQLDialect();
+ String upsertStatement =
+ dialect.getUpsertStatement(
+ "sche1",
+ "tb1",
+ Stream.of("id", "name").collect(Collectors.toList()),
+ Stream.of("id").collect(Collectors.toList()),
+ (SerializableFunction<String, String>) s -> "?");
+ Assertions.assertEquals(
+ "INSERT INTO `sche1`.`tb1`(`id`, `name`) VALUES (?, ?) ON DUPLICATE KEY UPDATE `name`=VALUES(`name`)",
+ upsertStatement);
+ }
+
+ @Test
+ void getUpsertStatementIgnore() {
+ OceanBaseMySQLDialect dialect = new OceanBaseMySQLDialect();
+ String upsertStatement =
+ dialect.getUpsertStatement(
+ "sche1",
+ "tb1",
+ Stream.of("id").collect(Collectors.toList()),
+ Stream.of("id").collect(Collectors.toList()),
+ (SerializableFunction<String, String>) s -> "?");
+ Assertions.assertEquals(
+ "INSERT IGNORE INTO `sche1`.`tb1`(`id`) VALUES (?)", upsertStatement);
+ }
+}