[connector-jdbc][bugfix] fix oracle create table comment special string bug

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCatalog.java

@@ -355,11 +355,18 @@ protected String getCreateTableSql(TablePath tablePath, CatalogTable table) {
         throw new UnsupportedOperationException();
     }
 
+    protected List<String> getCreateTableSqls(TablePath tablePath, CatalogTable table) {
+        return Collections.singletonList(getCreateTableSql(tablePath, table));
+    }
+
     protected void createTableInternal(TablePath tablePath, CatalogTable table)
             throws CatalogException {
         String dbUrl = getUrlFromDatabaseName(tablePath.getDatabaseName());
         try {
-            executeInternal(dbUrl, getCreateTableSql(tablePath, table));
+            final List<String> createTableSqlList = getCreateTableSqls(tablePath, table);
+            for (String sql : createTableSqlList) {
+                executeInternal(dbUrl, sql);
+            }
         } catch (Exception e) {
             throw new CatalogException(
                     String.format("Failed creating table %s", tablePath.getFullName()), e);

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oracle/OracleCatalog.java

@@ -117,22 +117,11 @@ protected String getListDatabaseSql() {
 
     @Override
     protected String getCreateTableSql(TablePath tablePath, CatalogTable table) {
-        return new OracleCreateTableSqlBuilder(table).build(tablePath);
+        return new OracleCreateTableSqlBuilder(table).build(tablePath).get(0);
     }
 
-    @Override
-    protected void createTableInternal(TablePath tablePath, CatalogTable table)
-            throws CatalogException {
-        String dbUrl = getUrlFromDatabaseName(tablePath.getDatabaseName());
-        try {
-            String createTableSQL = getCreateTableSql(tablePath, table);
-            for (String sql : createTableSQL.split(";")) {
-                executeInternal(dbUrl, sql);
-            }
-        } catch (Exception e) {
-            // fallback to super
-            super.createTableInternal(tablePath, table);
-        }
+    protected List<String> getCreateTableSqls(TablePath tablePath, CatalogTable table) {
+        return new OracleCreateTableSqlBuilder(table).build(tablePath);
     }
 
     @Override

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oracle/OracleCreateTableSqlBuilder.java

@@ -27,6 +27,7 @@
 
 import org.apache.commons.lang3.StringUtils;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 import java.util.stream.Collectors;
@@ -45,7 +46,8 @@ public OracleCreateTableSqlBuilder(CatalogTable catalogTable) {
         this.fieldIde = catalogTable.getOptions().get("fieldIde");
     }
 
-    public String build(TablePath tablePath) {
+    public List<String> build(TablePath tablePath) {
+        List<String> sqls = new ArrayList<>();
         StringBuilder createTableSql = new StringBuilder();
         createTableSql
                 .append("CREATE TABLE ")
@@ -66,7 +68,7 @@ public String build(TablePath tablePath) {
 
         createTableSql.append(String.join(",\n", columnSqls));
         createTableSql.append("\n)");
-
+        sqls.add(createTableSql.toString());
         List<String> commentSqls =
                 columns.stream()
                         .filter(column -> StringUtils.isNotBlank(column.getComment()))
@@ -75,13 +77,8 @@ public String build(TablePath tablePath) {
                                         buildColumnCommentSql(
                                                 column, tablePath.getSchemaAndTableName("\"")))
                         .collect(Collectors.toList());
-
-        if (!commentSqls.isEmpty()) {
-            createTableSql.append(";\n");
-            createTableSql.append(String.join(";\n", commentSqls));
-        }
-
-        return createTableSql.toString();
+        sqls.addAll(commentSqls);
+        return sqls;
     }
 
     private String buildColumnSql(Column column) {
@@ -134,7 +131,7 @@ private String buildColumnCommentSql(Column column, String tableName) {
         columnCommentSql
                 .append(CatalogUtils.quoteIdentifier(column.getName(), fieldIde, "\""))
                 .append(CatalogUtils.quoteIdentifier(" IS '", fieldIde))
-                .append(column.getComment())
+                .append(column.getComment().replace("'", "''"))
                 .append("'");
         return columnCommentSql.toString();
     }

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-6/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOracleLowercaseTableIT.java

@@ -18,6 +18,7 @@
 
 package org.apache.seatunnel.connectors.seatunnel.jdbc;
 
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.oracle.OracleCatalog;
@@ -221,6 +222,8 @@ protected void initCatalog() {
     @Test
     public void testCatalog() {
         TablePath tablePathOracle = TablePath.of("XE", "TESTUSER", "E2E_TABLE_SOURCE_LOWER");
+        TablePath tablePathOracleCreateTablePath =
+                TablePath.of("XE", "TESTUSER", "E2E_TABLE_SOURCE_LOWER_AUTO");
         OracleCatalog oracleCatalog =
                 new OracleCatalog(
                         "Oracle",
@@ -230,9 +233,29 @@ public void testCatalog() {
                                 jdbcCase.getJdbcUrl().replace(HOST, dbServer.getHost())),
                         SCHEMA);
         oracleCatalog.open();
+        catalog.executeSql(
+                tablePathOracle,
+                "comment on column E2E_TABLE_SOURCE_LOWER.CHAR_10_COL is '\"#¥%……&*（）;;'',,..``````//''@特殊注释''\\''\"'");
         Assertions.assertTrue(oracleCatalog.tableExists(tablePathOracle));
+        Assertions.assertEquals(
+                oracleCatalog
+                        .getTable(tablePathOracle)
+                        .getTableSchema()
+                        .getColumns()
+                        .get(1)
+                        .getComment(),
+                "\"#¥%……&*（）;;',,..``````//'@特殊注释'\\'\"");
         oracleCatalog.truncateTable(tablePathOracle, true);
         Assertions.assertFalse(oracleCatalog.isExistsData(tablePathOracle));
+        // create table with comment
+        Assertions.assertFalse(oracleCatalog.tableExists(tablePathOracleCreateTablePath));
+        oracleCatalog.createTable(
+                tablePathOracleCreateTablePath, oracleCatalog.getTable(tablePathOracle), true);
+        Assertions.assertTrue(oracleCatalog.tableExists(tablePathOracleCreateTablePath));
+        final CatalogTable table = oracleCatalog.getTable(tablePathOracleCreateTablePath);
+        Assertions.assertEquals(
+                table.getTableSchema().getColumns().get(1).getComment(),
+                "\"#¥%……&*（）;;',,..``````//'@特殊注释'\\'\"");
         oracleCatalog.close();
     }
 }