feat: mysql-oceanbase performance

xbsura · xbsura · commit e92bfe8fa442 · 2025-03-28T17:53:42.000+08:00
diff --git a/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/PostgresSqlMaker.java b/connectors-common/postgres-core/src/main/java/io/tapdata/connector/postgres/PostgresSqlMaker.java
@@ -38,6 +38,9 @@ protected void buildNullDefinition(StringBuilder builder, TapField tapField) {
     }
 
     protected void buildDefaultDefinition(StringBuilder builder, TapField tapField) {
+        if (EmptyKit.isNotNull(dbVersion) && Integer.parseInt(dbVersion) < 100000 && EmptyKit.isNotBlank(tapField.getSequenceName())) {
+            return;
+        }
         if (EmptyKit.isNotNull(tapField.getDefaultValue())) {
             builder.append("DEFAULT").append(' ');
             if (EmptyKit.isNotNull(tapField.getDefaultFunction())) {
@@ -62,9 +65,17 @@ protected void buildAutoIncDefinition(StringBuilder builder, TapField tapField)
         } else {
             startValue = 1;
         }
+        long incrementValue;
+        if (EmptyKit.isNotNull(tapField.getAutoIncrementValue())) {
+            incrementValue = tapField.getAutoIncrementValue();
+        } else {
+            incrementValue = 1;
+        }
         builder.append("GENERATED BY DEFAULT AS IDENTITY (START WITH ")
                 .append(startValue)
-                .append(" INCREMENT BY 1 CACHE ")
+                .append(" INCREMENT BY ")
+                .append(incrementValue)
+                .append(" CACHE ")
                 .append(autoIncCacheValue)
                 .append(") ");
     }
diff --git a/connectors-common/sql-core/src/main/java/io/tapdata/common/CommonDbConfig.java b/connectors-common/sql-core/src/main/java/io/tapdata/common/CommonDbConfig.java
@@ -47,6 +47,7 @@ public class CommonDbConfig implements Serializable {
     private long autoIncJumpValue = 1000000L;
     private long autoIncCacheValue = 100;
     private Boolean applyDefault = false;
+    private Boolean applyForeignKey = false;
     private Integer writeThreadSize = 15;
     protected String timezone = "+00:00";
     protected ZoneId zoneId;
@@ -322,6 +323,14 @@ public void setApplyDefault(Boolean applyDefault) {
         this.applyDefault = applyDefault;
     }
 
+    public Boolean getApplyForeignKey() {
+        return applyForeignKey;
+    }
+
+    public void setApplyForeignKey(Boolean applyForeignKey) {
+        this.applyForeignKey = applyForeignKey;
+    }
+
     public Integer getWriteThreadSize() {
         return writeThreadSize;
     }
diff --git a/connectors-common/sql-core/src/main/java/io/tapdata/common/CommonDbConnector.java b/connectors-common/sql-core/src/main/java/io/tapdata/common/CommonDbConnector.java
@@ -450,6 +450,9 @@ protected void createIndex(TapConnectorContext connectorContext, TapTable tapTab
     }
 
     protected void createConstraint(TapConnectorContext connectorContext, TapTable tapTable, TapCreateConstraintEvent createConstraintEvent, boolean create) {
+        if (!Boolean.TRUE.equals(commonDbConfig.getApplyForeignKey())) {
+            return;
+        }
         List<TapConstraint> constraintList = createConstraintEvent.getConstraintList();
         if (EmptyKit.isNotEmpty(constraintList)) {
             List<String> constraintSqlList = new ArrayList<>();
diff --git a/connectors-common/sql-core/src/main/java/io/tapdata/common/CommonSqlMaker.java b/connectors-common/sql-core/src/main/java/io/tapdata/common/CommonSqlMaker.java
@@ -37,6 +37,7 @@ public class CommonSqlMaker {
     protected long autoIncCacheValue = 1;
     protected Boolean applyDefault = false;
     protected String schema;
+    protected String dbVersion;
 
     public CommonSqlMaker() {
 
@@ -71,6 +72,11 @@ public <T extends CommonSqlMaker> T schema(String schema) {
         return (T) this;
     }
 
+    public <T extends CommonSqlMaker> T dbVersion(String dbVersion) {
+        this.dbVersion = dbVersion;
+        return (T) this;
+    }
+
     public char getEscapeChar() {
         return escapeChar;
     }
diff --git a/connectors-common/sql-core/src/main/java/io/tapdata/common/dml/NormalRecordWriter.java b/connectors-common/sql-core/src/main/java/io/tapdata/common/dml/NormalRecordWriter.java
@@ -42,6 +42,7 @@ public class NormalRecordWriter {
     protected Map<String, Object> autoIncMap = new HashMap<>();
     protected boolean largeSql = false;
     protected CommonDbConfig commonDbConfig;
+    protected boolean needCloseIdentity = false;
 
     public NormalRecordWriter(JdbcContext jdbcContext, TapTable tapTable) throws SQLException {
         this.commonDbConfig = jdbcContext.getConfig();
@@ -147,6 +148,9 @@ public void write(List<TapRecordEvent> tapRecordEvents, Consumer<WriteListResult
             updateRecorder.releaseResource();
             deleteRecorder.releaseResource();
             if (!isTransaction) {
+                if (needCloseIdentity) {
+                    openIdentity();
+                }
                 connection.close();
             }
             writeListResultConsumer.accept(listResult
@@ -217,9 +221,23 @@ public void closeIdentity() throws SQLException {
                 statement.execute(sql);
             }
         }
+        needCloseIdentity = true;
     }
 
     protected String getIdentitySql() {
         return null;
     }
+
+    public void openIdentity() throws SQLException {
+        String sql = getOpenIdentitySql();
+        if (EmptyKit.isNotBlank(sql)) {
+            try (Statement statement = connection.createStatement()) {
+                statement.execute(sql);
+            }
+        }
+    }
+
+    protected String getOpenIdentitySql() {
+        return null;
+    }
 }
diff --git a/connectors/greenplum-connector/src/main/java/io/tapdata/connector/greenplum/GreenplumWriteRecorder.java b/connectors/greenplum-connector/src/main/java/io/tapdata/connector/greenplum/GreenplumWriteRecorder.java
@@ -37,7 +37,7 @@ protected void upsert(Map<String, Object> after, WriteListResult<TapRecordEvent>
             }
             try (ResultSet resultSet = queryStatement.executeQuery()) {
                 if (resultSet.next()) {
-                    justUpdate(after, DbKit.getBeforeForUpdate(after, new HashMap<>(), allColumn, uniqueCondition), listResult);
+                    justUpdate(DbKit.getAfterForUpdate(after, new HashMap<>(), allColumn, uniqueCondition), DbKit.getBeforeForUpdate(after, new HashMap<>(), allColumn, uniqueCondition), listResult);
                 } else {
                     justInsert(after);
                 }
diff --git a/connectors/mysql-connector/src/main/java/io/tapdata/connector/mysql/MysqlConnector.java b/connectors/mysql-connector/src/main/java/io/tapdata/connector/mysql/MysqlConnector.java
@@ -971,6 +971,15 @@ protected void createConstraint(TapConnectorContext connectorContext, TapTable t
                             } catch (Exception e1) {
                                 exception.addException(c, sql, e1);
                             }
+                        } else if (e instanceof SQLException && (((SQLException) e).getErrorCode() == 1826 || ((SQLException) e).getErrorCode() == 1823)) {
+                            String rename = c.getName() + "_" + UUID.randomUUID().toString().replaceAll("-", "").substring(28);
+                            c.setName(rename);
+                            sql = getCreateConstraintSql(tapTable, c);
+                            try {
+                                jdbcContext.execute(sql);
+                            } catch (Exception e1) {
+                                exception.addException(c, sql, e1);
+                            }
                         } else {
                             exception.addException(c, sql, e);
                         }
diff --git a/connectors/oceanbase-mysql-connector/src/main/java/io/tapdata/oceanbase/OceanbaseJdbcContext.java b/connectors/oceanbase-mysql-connector/src/main/java/io/tapdata/oceanbase/OceanbaseJdbcContext.java
@@ -0,0 +1,42 @@
+package io.tapdata.oceanbase;
+
+import com.oceanbase.jdbc.OceanBaseStatement;
+import com.zaxxer.hikari.pool.HikariProxyStatement;
+import io.tapdata.common.CommonDbConfig;
+import io.tapdata.common.ResultSetConsumer;
+import io.tapdata.connector.mysql.MysqlJdbcContextV2;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+public class OceanbaseJdbcContext extends MysqlJdbcContextV2 {
+
+    public OceanbaseJdbcContext(CommonDbConfig config) {
+        super(config);
+    }
+
+    public void queryWithStream(String sql, ResultSetConsumer resultSetConsumer) throws Throwable {
+        try (
+                Connection connection = getConnection();
+                Statement statement = connection.createStatement()
+        ) {
+            if (statement instanceof HikariProxyStatement) {
+                OceanBaseStatement statementImpl = statement.unwrap(OceanBaseStatement.class);
+                if (null != statementImpl) {
+                    statementImpl.setFetchSize(Integer.MIN_VALUE);
+                }
+            }
+            try (
+                    ResultSet resultSet = statement.executeQuery(sql)
+            ) {
+                if (null != resultSet) {
+                    resultSetConsumer.accept(resultSet);
+                }
+            }
+        } catch (SQLException e) {
+            throw new Exception("Execute steaming query failed, sql: " + sql + ", code: " + e.getSQLState() + "(" + e.getErrorCode() + "), error: " + e.getMessage(), e);
+        }
+    }
+}
diff --git a/connectors/oceanbase-mysql-connector/src/main/java/io/tapdata/oceanbase/cdc/OceanbaseReader.java b/connectors/oceanbase-mysql-connector/src/main/java/io/tapdata/oceanbase/cdc/OceanbaseReader.java
@@ -77,13 +77,19 @@ public void start(BooleanSupplier isAlive) throws Throwable {
             @Override
             public void notify(LogMessage message) {
                 try {
+                    String op = message.getOpt().name();
                     if (!tableList.contains(message.getTableName())) {
-                        return;
+                        switch (op) {
+                            case "INSERT":
+                            case "UPDATE":
+                            case "DELETE":
+                                return;
+                        }
                     }
                     Map<String, Object> after = DataMap.create();
                     Map<String, Object> before = DataMap.create();
                     analyzeMessage(message, after, before);
-                    switch (message.getOpt().name()) {
+                    switch (op) {
                         case "INSERT":
                             eventList.get().add(new TapInsertRecordEvent().init().table(message.getTableName()).after(after).referenceTime(Long.parseLong(message.getTimestamp()) * 1000));
                             break;
diff --git a/connectors/oceanbase-mysql-connector/src/main/java/io/tapdata/oceanbase/connector/OceanbaseConnector.java b/connectors/oceanbase-mysql-connector/src/main/java/io/tapdata/oceanbase/connector/OceanbaseConnector.java
@@ -4,7 +4,6 @@
 import io.tapdata.common.SqlExecuteCommandFunction;
 import io.tapdata.connector.mysql.MysqlConnector;
 import io.tapdata.connector.mysql.MysqlExceptionCollector;
-import io.tapdata.connector.mysql.MysqlJdbcContextV2;
 import io.tapdata.connector.mysql.ddl.sqlmaker.MysqlDDLSqlGenerator;
 import io.tapdata.connector.mysql.dml.MysqlRecordWriter;
 import io.tapdata.entity.codec.TapCodecsRegistry;
@@ -21,6 +20,7 @@
 import io.tapdata.kit.DbKit;
 import io.tapdata.kit.EmptyKit;
 import io.tapdata.kit.StringKit;
+import io.tapdata.oceanbase.OceanbaseJdbcContext;
 import io.tapdata.oceanbase.OceanbaseTest;
 import io.tapdata.oceanbase.bean.OceanbaseConfig;
 import io.tapdata.oceanbase.cdc.OceanbaseReader;
@@ -195,7 +195,7 @@ public void onStart(TapConnectionContext tapConnectionContext) throws Throwable
 
         mysqlConfig = new OceanbaseConfig().load(tapConnectionContext.getConnectionConfig());
         mysqlConfig.load(tapConnectionContext.getNodeConfig());
-        mysqlJdbcContext = new MysqlJdbcContextV2(mysqlConfig);
+        mysqlJdbcContext = new OceanbaseJdbcContext(mysqlConfig);
         commonDbConfig = mysqlConfig;
 
         jdbcContext = mysqlJdbcContext;
diff --git a/connectors/postgres-connector/src/main/java/io/tapdata/connector/postgres/PostgresConnector.java b/connectors/postgres-connector/src/main/java/io/tapdata/connector/postgres/PostgresConnector.java
@@ -356,7 +356,11 @@ private void initConnection(TapConnectionContext connectionContext) {
             postgresConfig.load(tapConnectorContext.getNodeConfig());
         });
         postgresVersion = postgresJdbcContext.queryVersion();
-        commonSqlMaker = new PostgresSqlMaker().schema(postgresConfig.getSchema()).closeNotNull(postgresConfig.getCloseNotNull()).autoIncCacheValue(postgresConfig.getAutoIncCacheValue());
+        commonSqlMaker = new PostgresSqlMaker()
+                .dbVersion(postgresVersion)
+                .schema(postgresConfig.getSchema())
+                .closeNotNull(postgresConfig.getCloseNotNull())
+                .autoIncCacheValue(postgresConfig.getAutoIncCacheValue());
         if (Boolean.TRUE.equals(postgresConfig.getCreateAutoInc()) && Integer.parseInt(postgresVersion) > 100000) {
             commonSqlMaker.createAutoInc(true);
         }

Original file line number	Diff line number	Diff line change
`@@ -450,6 +450,9 @@ protected void createIndex(TapConnectorContext connectorContext, TapTable tapTab`
`450`	`450`	`}`
`451`	`451`
`452`	`452`	`protected void createConstraint(TapConnectorContext connectorContext, TapTable tapTable, TapCreateConstraintEvent createConstraintEvent, boolean create) {`
	`453`	`+ if (!Boolean.TRUE.equals(commonDbConfig.getApplyForeignKey())) {`
	`454`	`+ return;`
	`455`	`+ }`
`453`	`456`	`List<TapConstraint> constraintList = createConstraintEvent.getConstraintList();`
`454`	`457`	`if (EmptyKit.isNotEmpty(constraintList)) {`
`455`	`458`	`List<String> constraintSqlList = new ArrayList<>();`
Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ protected void upsert(Map<String, Object> after, WriteListResult<TapRecordEvent>`
`37`	`37`	`}`
`38`	`38`	`try (ResultSet resultSet = queryStatement.executeQuery()) {`
`39`	`39`	`if (resultSet.next()) {`
`40`		`- justUpdate(after, DbKit.getBeforeForUpdate(after, new HashMap<>(), allColumn, uniqueCondition), listResult);`
	`40`	`+ justUpdate(DbKit.getAfterForUpdate(after, new HashMap<>(), allColumn, uniqueCondition), DbKit.getBeforeForUpdate(after, new HashMap<>(), allColumn, uniqueCondition), listResult);`
`41`	`41`	`} else {`
`42`	`42`	`justInsert(after);`
`43`	`43`	`}`