修改表字段内容

seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-mysql/src/main/java/org/apache/seatunnel/datasource/plugin/mysql/jdbc/MysqlJdbcDataSourceChannel.java

@@ -114,8 +114,8 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
             String primaryKey = getPrimaryKey(metaData, database, table);
 
             String sql =
-                    "SELECT * FROM information_schema.COLUMNS"
-                            + "WHERE "
+                    "SELECT * FROM information_schema.COLUMNS "
+                            + " WHERE "
                             + "TABLE_NAME = '"
                             + table
                             + "'  and  TABLE_SCHEMA='"
@@ -141,7 +141,7 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
                 }
 
                 tableField.setName(columnName);
-                tableField.setComment(resultSet.getString("REMARKS"));
+                tableField.setComment(resultSet.getString("COLUMN_COMMENT"));
                 Object nullable = resultSet.getObject("IS_NULLABLE");
                 tableField.setNullable(Boolean.TRUE.toString().equals(nullable.toString()));
                 tableFields.add(tableField);

seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-oracle/src/main/java/org/apache/seatunnel/datasource/plugin/oracle/jdbc/OracleDataSourceChannel.java

@@ -112,27 +112,27 @@ public class OracleDataSourceChannel implements DataSourceChannelExt {
             String primaryKey = getPrimaryKey(metaData, database, table);
 
             String sql =
-                    "SELECT\n"
-                            + "\tt.TABLE_NAME,\n"
-                            + "\tt.COLUMN_NAME,\n"
-                            + "\tt.COMMENTS,\n"
-                            + "\tt1.DATA_TYPE,\n"
-                            + "\tt1.DATA_LENGTH VAR_LEN,\n"
-                            + "\tt1.DATA_PRECISION LEN,\n"
-                            + "\tDATA_SCALE ,\n"
-                            + "\tCASE NULLABLE\n"
-                            + "\tWHEN 'N' THEN\n"
-                            + "\t\t'true'\n"
-                            + "\tELSE\n"
-                            + "\t\t'false'\n"
+                    "SELECT "
+                            + " t.TABLE_NAME,\n"
+                            + " t.COLUMN_NAME,\n"
+                            + " t.COMMENTS,\n"
+                            + " t1.DATA_TYPE,\n"
+                            + " t1.DATA_LENGTH VAR_LEN,\n"
+                            + " t1.DATA_PRECISION LEN,\n"
+                            + " DATA_SCALE ,\n"
+                            + " CASE NULLABLE\n"
+                            + " WHEN 'N' THEN\n"
+                            + "  'true'\n"
+                            + " ELSE\n"
+                            + " 'false'\n"
                             + "END  IS_NULLABLE\n"
                             + "\n"
                             + "FROM\n"
-                            + "\tall_col_comments t\n"
-                            + "\tINNER JOIN all_tab_cols t1 ON t1.TABLE_NAME = t.table_name \n"
-                            + "\tAND t1.COLUMN_NAME = t.column_name \n"
-                            + "WHERE "
-                            + "\tt.table_name = '"
+                            + "  all_col_comments t\n"
+                            + " INNER JOIN all_tab_cols t1 ON t1.TABLE_NAME = t.table_name \n"
+                            + " AND t1.COLUMN_NAME = t.column_name \n"
+                            + " WHERE "
+                            + " t.table_name = '"
                             + table
                             + "'  and  t.OWNER='"
                             + database

seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-tidb/src/main/java/org/apache/seatunnel/datasource/plugin/tidb/jdbc/TidbJdbcDataSourceChannel.java

@@ -112,7 +112,17 @@ public class TidbJdbcDataSourceChannel implements DataSourceChannelExt {
         try (Connection connection = getConnection(requestParams, database)) {
             DatabaseMetaData metaData = connection.getMetaData();
             String primaryKey = getPrimaryKey(metaData, database, table);
-            try (ResultSet resultSet = metaData.getColumns(database, null, table, null)) {
+
+            String sql =
+                    "SELECT * FROM information_schema.COLUMNS "
+                            + " WHERE "
+                            + "TABLE_NAME = '"
+                            + table
+                            + "'  and  TABLE_SCHEMA='"
+                            + database
+                            + "'";
+            Statement statement = connection.createStatement();
+            ResultSet resultSet = statement.executeQuery(sql);
             while (resultSet.next()) {
                 TableField tableField = new TableField();
                 String columnName = resultSet.getString("COLUMN_NAME");
@@ -128,12 +138,12 @@ public class TidbJdbcDataSourceChannel implements DataSourceChannelExt {
                     tableField.setScale(resultSet.getString("NUMERIC_SCALE"));
                 }
                 tableField.setName(columnName);
-                    tableField.setComment(resultSet.getString("REMARKS"));
+                tableField.setComment(resultSet.getString("COLUMN_COMMENT"));
                 Object nullable = resultSet.getObject("IS_NULLABLE");
                 tableField.setNullable(Boolean.TRUE.toString().equals(nullable.toString()));
                 tableFields.add(tableField);
             }
-            }
+
         } catch (ClassNotFoundException | SQLException e) {
             throw new DataSourcePluginException("get table fields failed", e);
         }

seatunnel-datasource/seatunnel-datasource-plugins/datasource-mysql-cdc/src/main/java/org/apache/seatunnel/datasource/plugin/cdc/mysql/MysqlCDCDataSourceChannel.java

@@ -28,14 +28,10 @@ import com.google.common.collect.Sets;
 import lombok.NonNull;
 
 import java.sql.*;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 
 public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
-
+    List<String> typeList = Arrays.asList("varchar", "char", "json");
     public static final Set<String> MYSQL_SYSTEM_DATABASES =
             Sets.newHashSet("information_schema", "mysql", "performance_schema", "sys");
 
@@ -198,7 +194,17 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
         try (Connection connection = init(requestParams); ) {
             DatabaseMetaData metaData = connection.getMetaData();
             String primaryKey = getPrimaryKey(metaData, dbName, tableName);
-            ResultSet resultSet = metaData.getColumns(dbName, null, tableName, null);
+
+            String sql =
+                    "SELECT * FROM information_schema.COLUMNS "
+                            + " WHERE "
+                            + "TABLE_NAME = '"
+                            + tableName
+                            + "'  and  TABLE_SCHEMA='"
+                            + dbName
+                            + "'";
+            Statement statement = connection.createStatement();
+            ResultSet resultSet = statement.executeQuery(sql);
             while (resultSet.next()) {
                 TableField tableField = new TableField();
                 String columnName = resultSet.getString("COLUMN_NAME");
@@ -206,12 +212,20 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
                 if (StringUtils.isNotBlank(primaryKey) && primaryKey.equals(columnName)) {
                     tableField.setPrimaryKey(true);
                 }
+                tableField.setType(resultSet.getString("DATA_TYPE"));
+                if (typeList.contains(tableField.getType().toLowerCase())) {
+                    if (tableField.getType().toLowerCase().indexOf("text") < 0) {
+                        tableField.setLen(resultSet.getString("CHARACTER_MAXIMUM_LENGTH"));
+                    }
+                } else {
+                    tableField.setLen(resultSet.getString("NUMERIC_PRECISION"));
+                    tableField.setScale(resultSet.getString("NUMERIC_SCALE"));
+                }
+
                 tableField.setName(columnName);
-                tableField.setType(resultSet.getString("TYPE_NAME"));
                 tableField.setComment(resultSet.getString("REMARKS"));
                 Object nullable = resultSet.getObject("IS_NULLABLE");
-                boolean isNullable = convertToBoolean(nullable);
-                tableField.setNullable(isNullable);
+                tableField.setNullable(Boolean.TRUE.toString().equals(nullable.toString()));
                 tableFields.add(tableField);
             }
         } catch (SQLException e) {