[Bug] [Doris] sync from mysql to doris create table error

[zmlgit]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

there's 1500+ tables in my source database，doris create some table error，perhaps my source table does not contains primary key？ 有一千五百多张表在我的源数据库里，在Doris 创建表的时候就报错了，可能是我的源表没有主键，生成Doris建表语句的时候没考虑到这点，导致出现错误

SeaTunnel Version

2.3.4

SeaTunnel Config

sink {
Doris {
     base-url = "jdbc:mysql://localhost:9030"                        
     username = "root"                        
     password = "****"                        
     database = "***"                        
     enable_upsert_delete = true                        
     batch_max_rows = 8192                        
     batch_max_bytes = 10485760                        
     fenodes = "127.0.0.1:8030"                        
     sink.enable-2pc = "true"                        
     sink.enable-delete = "true"                        
     doris.config {                                
         format = "json"                                
         read_json_by_line = "true"                        
     }        
}
}

Running Command

sh bin/seatunnel.sh --config job/mysql.streaming.config -m local

Error Exception

Exception in thread "main" org.apache.seatunnel.core.starter.exception.CommandExecuteException: SeaTunnel job executed failed
        at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:199)
        at org.apache.seatunnel.core.starter.SeaTunnel.run(SeaTunnel.java:40)
        at org.apache.seatunnel.core.starter.seatunnel.SeaTunnelClient.main(SeaTunnelClient.java:34)
Caused by: org.apache.seatunnel.common.exception.SeaTunnelRuntimeException: ErrorCode:[API-09], ErrorDescription:[Handle save mode failed]
        at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.handleSaveMode(MultipleTableJobConfigParser.java:669)
        at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.createSinkAction(MultipleTableJobConfigParser.java:655)
        at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parseSink(MultipleTableJobConfigParser.java:567)
        at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parse(MultipleTableJobConfigParser.java:193)
        at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.getLogicalDag(ClientJobExecutionEnvironment.java:88)
        at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.execute(ClientJobExecutionEnvironment.java:161)
        at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:146)
        ... 2 more
Caused by: org.apache.seatunnel.api.table.catalog.exception.CatalogException: ErrorCode:[API-03], ErrorDescription:[Catalog initialize failed] - create table statement execute failed
        at org.apache.seatunnel.connectors.doris.catalog.DorisCatalog.createTable(DorisCatalog.java:286)
        at org.apache.seatunnel.api.sink.DefaultSaveModeHandler.createTable(DefaultSaveModeHandler.java:142)
        at org.apache.seatunnel.api.sink.DefaultSaveModeHandler.createSchemaWhenNotExist(DefaultSaveModeHandler.java:101)
        at org.apache.seatunnel.api.sink.DefaultSaveModeHandler.handleSchemaSaveMode(DefaultSaveModeHandler.java:62)
        at org.apache.seatunnel.api.sink.SaveModeHandler.handleSaveMode(SaveModeHandler.java:27)
        at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.handleSaveMode(MultipleTableJobConfigParser.java:667)
        ... 8 more
Caused by: java.sql.SQLException: errCode = 2, detailMessage = Syntax error in line 21:
 UNIQUE KEY ()
             ^
Encountered: )
Expected: IDENTIFIER

        at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:129)
        at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122)
        at com.mysql.cj.jdbc.StatementImpl.executeInternal(StatementImpl.java:763)
        at com.mysql.cj.jdbc.StatementImpl.execute(StatementImpl.java:648)
        at org.apache.seatunnel.connectors.doris.catalog.DorisCatalog.createTable(DorisCatalog.java:284)
        ... 13 more

Zeta or Flink or Spark Version

No response

Java or Scala Version

No response

Screenshots

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[dailai]

You can use this property（save_mode_create_template） to override table creation statements. If your source table does not have a primary KEY, it is recommended to use DUPLICATE KEY. See: https://doris.apache.org/zh-CN/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-TABLE#keys_type