feat: optimize Oracle mode write with multi-row MERGE

flink-connector-oceanbase/src/main/java/com/oceanbase/connector/flink/connection/OceanBaseConnectionProvider.java

@@ -109,7 +109,6 @@ protected void init() {
     private Properties initializeDefaultJdbcProperties(String jdbcUrl) {
         Properties defaultJdbcProperties = new Properties();
         defaultJdbcProperties.setProperty("useSSL", "false");
-        defaultJdbcProperties.setProperty("rewriteBatchedStatements", "true");
         defaultJdbcProperties.setProperty("initialTimeout", "2");
         defaultJdbcProperties.setProperty("autoReconnect", "true");
         defaultJdbcProperties.setProperty("maxReconnects", "3");
@@ -121,6 +120,11 @@ private Properties initializeDefaultJdbcProperties(String jdbcUrl) {
         defaultJdbcProperties.setProperty("characterEncoding", "UTF-8");
         defaultJdbcProperties.setProperty("characterSetResults", "UTF-8");
 
+        if (dialect instanceof OceanBaseMySQLDialect) {
+            defaultJdbcProperties.setProperty("allowMultiQueries", "true");
+            defaultJdbcProperties.setProperty("rewriteBatchedStatements", "true");
+        }
+
         // Avoid overwriting user's custom jdbc properties.
         List<String> jdbcUrlProperties =
                 defaultJdbcProperties.keySet().stream()

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

@@ -76,6 +76,28 @@ String getUpsertStatement(
             @Nonnull List<String> uniqueKeyFields,
             @Nullable SerializableFunction<String, String> placeholderFunc);
 
+    /**
+     * Gets the upsert statement for multiple rows
+     *
+     * @param schemaName schema name
+     * @param tableName table name
+     * @param fieldNames field names list
+     * @param uniqueKeyFields unique key field names list
+     * @param rowCount number of rows to upsert in a single statement
+     * @param placeholderFunc function used to get placeholder for the fields
+     * @return the statement string
+     */
+    default String getUpsertStatement(
+            @Nonnull String schemaName,
+            @Nonnull String tableName,
+            @Nonnull List<String> fieldNames,
+            @Nonnull List<String> uniqueKeyFields,
+            int rowCount,
+            @Nullable SerializableFunction<String, String> placeholderFunc) {
+        return getUpsertStatement(
+                schemaName, tableName, fieldNames, uniqueKeyFields, placeholderFunc);
+    }
+
     /**
      * Gets the insert statement
      *

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

@@ -26,6 +26,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 public class OceanBaseOracleDialect implements OceanBaseDialect {
 
@@ -53,11 +54,37 @@ public String getUpsertStatement(
             @Nonnull List<String> fieldNames,
             @Nonnull List<String> uniqueKeyFields,
             @Nullable SerializableFunction<String, String> placeholderFunc) {
-        String sourceFields =
+        return getUpsertStatement(
+                schemaName, tableName, fieldNames, uniqueKeyFields, 1, placeholderFunc);
+    }
+
+    @Override
+    public String getUpsertStatement(
+            @Nonnull String schemaName,
+            @Nonnull String tableName,
+            @Nonnull List<String> fieldNames,
+            @Nonnull List<String> uniqueKeyFields,
+            int rowCount,
+            @Nullable SerializableFunction<String, String> placeholderFunc) {
+        String selectFields =
                 fieldNames.stream()
                         .map(f -> getPlaceholder(f, placeholderFunc) + " AS " + quoteIdentifier(f))
                         .collect(Collectors.joining(", "));
 
+        String usingClause;
+        if (rowCount == 1) {
+            usingClause = "SELECT " + selectFields + " FROM DUAL";
+        } else {
+            usingClause =
+                    "SELECT "
+                            + selectFields
+                            + " FROM DUAL"
+                            + IntStream.range(0, rowCount - 1)
+                                    .mapToObj(
+                                            i -> " UNION ALL SELECT " + selectFields + " FROM DUAL")
+                                    .collect(Collectors.joining());
+        }
+
         String onClause =
                 uniqueKeyFields.stream()
                         .map(f -> "t." + quoteIdentifier(f) + "=s." + quoteIdentifier(f))
@@ -80,9 +107,9 @@ public String getUpsertStatement(
         return "MERGE INTO "
                 + getFullTableName(schemaName, tableName)
                 + " t "
-                + " USING (SELECT "
-                + sourceFields
-                + " FROM DUAL) s "
+                + " USING ("
+                + usingClause
+                + ") s "
                 + " ON ("
                 + onClause
                 + ") "

flink-connector-oceanbase/src/main/java/com/oceanbase/connector/flink/sink/OceanBaseRecordFlusher.java

@@ -23,6 +23,7 @@
 import com.oceanbase.connector.flink.connection.OceanBaseVersion;
 import com.oceanbase.connector.flink.dialect.OceanBaseDialect;
 import com.oceanbase.connector.flink.dialect.OceanBaseMySQLDialect;
+import com.oceanbase.connector.flink.dialect.OceanBaseOracleDialect;
 import com.oceanbase.connector.flink.table.DataChangeRecord;
 import com.oceanbase.connector.flink.table.SchemaChangeRecord;
 import com.oceanbase.connector.flink.table.TableId;
@@ -33,6 +34,8 @@
 import com.oceanbase.partition.calculator.model.TableEntry;
 import com.oceanbase.partition.calculator.model.TableEntryKey;
 
+import org.apache.flink.util.function.SerializableFunction;
+
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.slf4j.Logger;
@@ -119,6 +122,14 @@ public synchronized void flush(List<DataChangeRecord> records) throws Exception
                                 tableInfo.getPlaceholderFunc()),
                         tableInfo.getFieldNames(),
                         upsertBatch);
+            } else if (dialect instanceof OceanBaseOracleDialect) {
+                flushMultiRowUpsert(
+                        tableId.getSchemaName(),
+                        tableId.getTableName(),
+                        tableInfo.getFieldNames(),
+                        tableInfo.getKey(),
+                        tableInfo.getPlaceholderFunc(),
+                        upsertBatch);
             } else {
                 flush(
                         dialect.getUpsertStatement(
@@ -211,6 +222,50 @@ private void flush(String sql, List<String> statementFields, List<DataChangeReco
         }
     }
 
+    private void flushMultiRowUpsert(
+            String schemaName,
+            String tableName,
+            List<String> fieldNames,
+            List<String> uniqueKeyFields,
+            SerializableFunction<String, String> placeholderFunc,
+            List<DataChangeRecord> records)
+            throws Exception {
+        Map<Long, List<DataChangeRecord>> group = groupRecords(records);
+        if (group == null) {
+            return;
+        }
+        for (List<DataChangeRecord> groupRecords : group.values()) {
+            String sql =
+                    dialect.getUpsertStatement(
+                            schemaName,
+                            tableName,
+                            fieldNames,
+                            uniqueKeyFields,
+                            groupRecords.size(),
+                            placeholderFunc);
+            try (Connection connection = connectionProvider.getConnection();
+                    PreparedStatement statement = connection.prepareStatement(sql)) {
+                int fieldCount = fieldNames.size();
+                for (int rowIdx = 0; rowIdx < groupRecords.size(); rowIdx++) {
+                    DataChangeRecord record = groupRecords.get(rowIdx);
+                    for (int i = 0; i < fieldCount; i++) {
+                        statement.setObject(
+                                rowIdx * fieldCount + i + 1,
+                                record.getFieldValue(fieldNames.get(i)));
+                    }
+                }
+                statement.executeUpdate();
+            } catch (SQLException e) {
+                throw new RuntimeException(
+                        "Failed to execute multi-row upsert with sql: "
+                                + sql
+                                + ", records: "
+                                + groupRecords,
+                        e);
+            }
+        }
+    }
+
     private Map<Long, List<DataChangeRecord>> groupRecords(List<DataChangeRecord> records) {
         if (CollectionUtils.isEmpty(records)) {
             return null;