[Bug] [JDBC] schema_save_mode="CREATE_SCHEMA_WHEN_NOT_EXIST"

[privatewbluo]

Search before asking

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

What happened

首次运行能成功运行，再次运行时候，就会报底层表已经存在。

SeaTunnel Version

2.3.8，2.3.4

SeaTunnel Config

env {
  parallelism = 2
  job.mode = "BATCH"
}
source {
    Jdbc {
        url = "jdbc:mysql://localhost:3306/datahub"
        driver = "com.mysql.cj.jdbc.Driver"
        connection_check_timeout_sec = 100
        user = "datahub"
        password = "datahub"
        table_path = "datahub.metadata_index"
}
}

sink {
    Jdbc {
        # if you would use json or jsonb type insert please add jdbc url stringtype=unspecified option
        url = "jdbc:postgresql://localhost:5432/db_test"
        driver = "org.postgresql.Driver"
        user = "postgres"
        password = "postgres"
        generate_sink_sql="true"
        schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
        database = "db_test"
        table="metadata_index"
    }
}

Running Command

./bin/seatunnel.sh --config ./config/mysqlTopostgre.conf  -m local

Error Exception

Caused by: org.apache.seatunnel.api.table.catalog.exception.CatalogException: ErrorCode:[API-03], ErrorDescription:[Catalog initialize failed] - Failed creating table db_test.metadata_index
	at org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.psql.PostgresCatalog.createTableInternal(PostgresCatalog.java:185)
	at org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.AbstractJdbcCatalog.createTable(AbstractJdbcCatalog.java:425)
	at org.apache.seatunnel.connectors.seatunnel.jdbc.sink.savemode.JdbcSaveModeHandler.createTable(JdbcSaveModeHandler.java:48)
	at org.apache.seatunnel.api.sink.DefaultSaveModeHandler.createSchemaWhenNotExist(DefaultSaveModeHandler.java:115)
	at org.apache.seatunnel.api.sink.DefaultSaveModeHandler.handleSchemaSaveMode(DefaultSaveModeHandler.java:74)
	at org.apache.seatunnel.api.sink.SaveModeHandler.handleSaveMode(SaveModeHandler.java:40)
	at org.apache.seatunnel.api.sink.SaveModeExecuteWrapper.execute(SaveModeExecuteWrapper.java:36)
	at org.apache.seatunnel.engine.server.master.JobMaster.handleSaveMode(JobMaster.java:376)
	... 21 more
Caused by: org.postgresql.util.PSQLException: ERROR: relation "metadata_index" already exists
	at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2725)
	at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2412)
	at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:371)
	at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:502)
	at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:419)
	at org.postgresql.jdbc.PgPreparedStatement.executeWithFlags(PgPreparedStatement.java:194)
	at org.postgresql.jdbc.PgPreparedStatement.execute(PgPreparedStatement.java:180)
	at org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.AbstractJdbcCatalog.executeInternal(AbstractJdbcCatalog.java:620)
	at org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.psql.PostgresCatalog.createTableInternal(PostgresCatalog.java:167)
	... 28 more

	at org.apache.seatunnel.engine.server.CoordinatorService.lambda$submitJob$4(CoordinatorService.java:518)
	at org.apache.seatunnel.api.tracing.MDCRunnable.run(MDCRunnable.java:39)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:750)

Zeta or Flink or Spark Version

No response

Java or Scala Version

No response

Screenshots

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

[hailin0]

Use table="public.metadata_index"