[fix](cdc)resolve the issue of automatic table creation failure in SQ…

…L Server due to the presence of the `systranschemas` (#302)
apache · Jan 24, 2024 · 480a2aa · 480a2aa
1 parent 1a9e7bc
commit 480a2aa
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 7 deletions.
diff --git a/...s-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java b/...s-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java
@@ -230,7 +230,6 @@ public DataStreamSource<String> buildCdcSource(StreamExecutionEnvironment env) {
 
     @Override
     public String getTableListPrefix() {
-        String schemaName = config.get(OracleSourceOptions.SCHEMA_NAME);
-        return schemaName;
+        return config.get(OracleSourceOptions.SCHEMA_NAME);
     }
 }
diff --git a/...ector/src/main/java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerDatabaseSync.java b/...ector/src/main/java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerDatabaseSync.java
@@ -99,7 +99,7 @@ public List<SourceSchema> getSchemaList() throws Exception {
         try (Connection conn = getConnection()) {
             DatabaseMetaData metaData = conn.getMetaData();
             try (ResultSet tables =
-                    metaData.getTables(databaseName, null, "%", new String[] {"TABLE"})) {
+                    metaData.getTables(databaseName, schemaName, "%", new String[] {"TABLE"})) {
                 while (tables.next()) {
                     String tableName = tables.getString("TABLE_NAME");
                     String tableComment = tables.getString("REMARKS");
@@ -108,7 +108,7 @@ public List<SourceSchema> getSchemaList() throws Exception {
                     }
                     SourceSchema sourceSchema =
                             new SqlServerSchema(
-                                    metaData, databaseName, null, tableName, tableComment);
+                                    metaData, databaseName, schemaName, tableName, tableComment);
                     sourceSchema.setModel(
                             !sourceSchema.primaryKeys.isEmpty()
                                     ? DataModel.UNIQUE
@@ -129,7 +129,7 @@ public DataStreamSource<String> buildCdcSource(StreamExecutionEnvironment env) {
 
         String tableName = config.get(JdbcSourceOptions.TABLE_NAME);
         String hostname = config.get(JdbcSourceOptions.HOSTNAME);
-        Integer port = config.getInteger(PORT, 1433);
+        int port = config.getInteger(PORT, 1433);
         String username = config.get(JdbcSourceOptions.USERNAME);
         String password = config.get(JdbcSourceOptions.PASSWORD);
 
@@ -204,7 +204,6 @@ public DataStreamSource<String> buildCdcSource(StreamExecutionEnvironment env) {
 
     @Override
     public String getTableListPrefix() {
-        String schemaName = config.get(JdbcSourceOptions.SCHEMA_NAME);
-        return schemaName;
+        return config.get(JdbcSourceOptions.SCHEMA_NAME);
     }
 }
diff --git a/...ris-connector/src/main/java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerType.java b/...ris-connector/src/main/java/org/apache/doris/flink/tools/cdc/sqlserver/SqlServerType.java
@@ -102,6 +102,8 @@ public static String toDorisType(String originSqlServerType, Integer precision,
             case DATETIMEOFFSET:
             case TIMESTAMP:
             case UNIQUEIDENTIFIER:
+            case BINARY:
+            case VARBINARY:
                 return DorisType.STRING;
             default:
                 throw new UnsupportedOperationException(