[bugfix] [jdbc-connector] 修复xugu新版适配问题

LeonYoah · Aug 19, 2024 · 8067670 · 8067670
1 parent bf2d5c7
commit 8067670
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 16 deletions.
diff --git a/.../src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/db2/DB2Catalog.java b/.../src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/db2/DB2Catalog.java
@@ -117,7 +117,7 @@ public List<String> listDatabases() throws CatalogException {
     }
 
     @Override
-    protected String getCreateTableSql(TablePath tablePath, CatalogTable table) {
+    protected String getCreateTableSql(TablePath tablePath, CatalogTable table, boolean createIndex) {
         return new DB2CreateTableSqlBuilder(table).build(tablePath);
     }
 

diff --git a/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/xugu/XuguCatalog.java b/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/xugu/XuguCatalog.java
@@ -24,6 +24,7 @@
 import org.apache.seatunnel.api.table.catalog.exception.CatalogException;
 import org.apache.seatunnel.api.table.catalog.exception.DatabaseNotExistException;
 import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
 import org.apache.seatunnel.common.utils.JdbcUrlUtil;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.AbstractJdbcCatalog;
 import org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.utils.CatalogUtils;
@@ -38,10 +39,14 @@
 import java.sql.DatabaseMetaData;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.seatunnel.common.exception.CommonErrorCode.UNSUPPORTED_METHOD;
 
 @Slf4j
 public class XuguCatalog extends AbstractJdbcCatalog {
@@ -148,14 +153,10 @@ protected String getListDatabaseSql() {
     }
 
     @Override
-    protected String getCreateTableSql(TablePath tablePath, CatalogTable table) {
-        return new XuguCreateTableSqlBuilder(table).build(tablePath);
+    protected String getCreateTableSql(TablePath tablePath, CatalogTable table, boolean createIndex) {
+        return new XuguCreateTableSqlBuilder(table,createIndex).build(tablePath);
     }
 
-    @Override
-    protected String getDropTableSql(TablePath tablePath) {
-        return String.format("DROP TABLE %s", tablePath.getSchemaAndTableName("\""));
-    }
 
     @Override
     protected String getCreateDatabaseSql(String databaseName) {
@@ -175,10 +176,6 @@ protected String getListTableSql(String databaseName) {
                 + "s.schema_id=t.schema_id";
     }
 
-    @Override
-    protected String getCreateTableSql(TablePath tablePath, CatalogTable table) {
-        return new XuguCreateTableSqlBuilder(table).build(tablePath);
-    }
 
     @Override
     protected String getDropTableSql(TablePath tablePath) {
@@ -261,11 +258,26 @@ && listTables(tablePath.getDatabaseName()).stream()
 
     @Override
     public boolean databaseExists(String databaseName) throws CatalogException {
-        checkArgument(StringUtils.isNotBlank(databaseName));
-        return listDatabases().stream()
-                .map(String::toUpperCase)
-                .collect(Collectors.toList())
-                .contains(databaseName.toUpperCase());
+
+        if (StringUtils.isBlank(databaseName)) {
+            return false;
+        }
+        if (SYS_DATABASES.contains(databaseName)) {
+            return false;
+        }
+        try {
+            return querySQLResultExists(
+                    getUrlFromDatabaseName(databaseName),
+                    getDatabaseWithConditionSql(databaseName.toUpperCase()));
+        } catch (SeaTunnelRuntimeException e) {
+            if (e.getSeaTunnelErrorCode().getCode().equals(UNSUPPORTED_METHOD.getCode())) {
+                log.warn(
+                        "The catalog: {} is not supported the getDatabaseWithConditionSql for databaseExists",
+                        this.catalogName);
+                return listDatabases().contains(databaseName.toUpperCase());
+            }
+            throw e;
+        }
     }
 
     private List<String> listTables() {