Skip to content

S
seatunnel-web
提交 d542c81e authored 作者: 宋勇's avatar 宋勇
浏览文件
操作

修改表字段内容

上级 9150e0e4
隐藏空白字符变更
内嵌 并排
正在显示 包含 189 行增加 和 171 行删除
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-mysql/src/main/java/org/apache/seatunnel/datasource/plugin/mysql/jdbc/MysqlJdbcDataSourceChannel.java
浏览文件 @ d542c81e
......@@ -55,10 +55,10 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
Map<String, String> option) {
List<String> tableNames = new ArrayList<>();
try (Connection connection = getConnection(requestParams);
ResultSet resultSet =
connection
.getMetaData()
.getTables(database, null, null, new String[]{"TABLE"})) {
ResultSet resultSet =
connection
.getMetaData()
.getTables(database, null, null, new String[] {"TABLE"})) {
while (resultSet.next()) {
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
......@@ -76,8 +76,8 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
@NonNull String pluginName, @NonNull Map<String, String> requestParams) {
List<String> dbNames = new ArrayList<>();
try (Connection connection = getConnection(requestParams);
PreparedStatement statement = connection.prepareStatement("SHOW DATABASES;");
ResultSet re = statement.executeQuery()) {
PreparedStatement statement = connection.prepareStatement("SHOW DATABASES;");
ResultSet re = statement.executeQuery()) {
// filter system databases
while (re.next()) {
String dbName = re.getString("database");
......@@ -113,30 +113,40 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
DatabaseMetaData metaData = connection.getMetaData();
String primaryKey = getPrimaryKey(metaData, database, table);
try (ResultSet resultSet = metaData.getColumns(database, null, table, null)) {
while (resultSet.next()) {
TableField tableField = new TableField();
String columnName = resultSet.getString("COLUMN_NAME");
tableField.setPrimaryKey(false);
if (StringUtils.isNotBlank(primaryKey) && primaryKey.equals(columnName)) {
tableField.setPrimaryKey(true);
}
tableField.setType(resultSet.getString("TYPE_NAME"));
if (typeList.contains(tableField.getType().toLowerCase())) {
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");
tableField.setPrimaryKey(false);
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.setComment(resultSet.getString("REMARKS"));
Object nullable = resultSet.getObject("IS_NULLABLE");
tableField.setNullable(Boolean.TRUE.toString().equals(nullable.toString()));
tableFields.add(tableField);
} else {
tableField.setLen(resultSet.getString("NUMERIC_PRECISION"));
tableField.setScale(resultSet.getString("NUMERIC_SCALE"));
}
tableField.setName(columnName);
tableField.setComment(resultSet.getString("REMARKS"));
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);
}
......@@ -196,13 +206,11 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
List<Map> tableNames = new ArrayList<>();
try (Connection connection = getConnection(requestParams)) {
String sql =
"SELECT * FROM information_schema.tables WHERE table_schema='" + database + "'" ;
"SELECT * FROM information_schema.tables WHERE table_schema='" + database + "'";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()) {
Map tables = new HashMap();
......@@ -210,19 +218,20 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
// 获取表列数
int columnCount = metaData.getColumnCount();
for (int i = 1; i <= columnCount; i++) {
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableComment = resultSet.getString("TABLE_COMMENT");
String tableComment = resultSet.getString("TABLE_COMMENT");
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tableNames.add(tables);
}
return tableNames;
} catch (ClassNotFoundException | SQLException e) {
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-oracle/src/main/java/org/apache/seatunnel/datasource/plugin/oracle/jdbc/OracleDataSourceChannel.java
浏览文件 @ d542c81e
......@@ -33,7 +33,8 @@ import java.util.*;
import static com.google.common.base.Preconditions.checkNotNull;
public class OracleDataSourceChannel implements DataSourceChannelExt {
List<String> typeList = Arrays.asList("varchar","varchar2", "char", "json");
List<String> typeList = Arrays.asList("varchar", "varchar2", "char", "json");
@Override
public OptionRule getDataSourceOptions(@NonNull String pluginName) {
return OracleDataSourceConfig.OPTION_RULE;
......@@ -110,54 +111,56 @@ public class OracleDataSourceChannel implements DataSourceChannelExt {
DatabaseMetaData metaData = connection.getMetaData();
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" +
"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 = '"+table+"' and t.OWNER='"+database+"'";
"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"
+ "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 = '"
+ table
+ "' and t.OWNER='"
+ database
+ "'";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()) {
TableField tableField = new TableField();
String columnName = resultSet.getString("COLUMN_NAME");
tableField.setPrimaryKey(false);
if (StringUtils.isNotBlank(primaryKey) && primaryKey.equals(columnName)) {
tableField.setPrimaryKey(true);
}
while (resultSet.next()) {
TableField tableField = new TableField();
String columnName = resultSet.getString("COLUMN_NAME");
tableField.setPrimaryKey(false);
if (StringUtils.isNotBlank(primaryKey) && primaryKey.equals(columnName)) {
tableField.setPrimaryKey(true);
}
if (typeList.contains(tableField.getType().toLowerCase())) {
tableField.setLen(resultSet.getString("VAR_LEN"));
} else {
tableField.setLen(resultSet.getString("LEN"));
tableField.setScale(resultSet.getString("DATA_SCALE"));
}
tableField.setName(columnName);
tableField.setType(resultSet.getString("TYPE_NAME"));
tableField.setComment(resultSet.getString("REMARKS"));
Object nullable = resultSet.getObject("IS_NULLABLE");
tableField.setNullable(Boolean.TRUE.toString().equals(nullable.toString()));
tableFields.add(tableField);
if (typeList.contains(tableField.getType().toLowerCase())) {
tableField.setLen(resultSet.getString("VAR_LEN"));
} else {
tableField.setLen(resultSet.getString("LEN"));
tableField.setScale(resultSet.getString("DATA_SCALE"));
}
tableField.setName(columnName);
tableField.setType(resultSet.getString("TYPE_NAME"));
tableField.setComment(resultSet.getString("REMARKS"));
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);
......@@ -222,19 +225,20 @@ public class OracleDataSourceChannel implements DataSourceChannelExt {
// 获取表列数
int columnCount = metaData.getColumnCount();
for (int i = 1; i <= columnCount; i++) {
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableComment = resultSet.getString("TABLE_COMMENT");
String tableComment = resultSet.getString("TABLE_COMMENT");
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tableNames.add(tables);
}
return tableNames;
} catch (ClassNotFoundException | SQLException e) {
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-postgresql/src/main/java/org/apache/seatunnel/datasource/plugin/postgresql/jdbc/PostgresqlDataSourceChannel.java
浏览文件 @ d542c81e
......@@ -34,6 +34,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
public class PostgresqlDataSourceChannel implements DataSourceChannelExt {
List<String> typeList = Arrays.asList("varchar", "char", "json");
@Override
public OptionRule getDataSourceOptions(@NonNull String pluginName) {
return PostgresqlDataSourceConfig.OPTION_RULE;
......@@ -199,7 +200,6 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt {
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()) {
Map tables = new HashMap();
......@@ -207,19 +207,20 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt {
// 获取表列数
int columnCount = metaData.getColumnCount();
for (int i = 1; i <= columnCount; i++) {
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableComment = resultSet.getString("TABLE_COMMENT");
String tableComment = resultSet.getString("TABLE_COMMENT");
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tableNames.add(tables);
}
return tableNames;
} catch (ClassNotFoundException | SQLException e) {
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-sqlserver/src/main/java/org/apache/seatunnel/datasource/plugin/sqlserver/jdbc/SqlServerDataSourceChannel.java
浏览文件 @ d542c81e
......@@ -116,49 +116,51 @@ public class SqlServerDataSourceChannel implements DataSourceChannelExt {
DatabaseMetaData metaData = connection.getMetaData();
String primaryKey = getPrimaryKey(metaData, database, table);
String sql =
"SELECT\n" +
"\tA.name AS TABLE_NAME,\n" +
"\tf.value AS TABLE_COMMENT,\n" +
"\tB.name AS COLUMN_NAME,\n" +
"\tB.max_length AS COLUMN_LENGTH,\n" +
"\tB.scale AS COLUMN_SCALE,\n" +
"\tC.value AS REMARKS,\n" +
"\tD.name AS TYPE_NAME,\n" +
"CASE\n" +
"\t\tA.type \n" +
"\t\tWHEN 'u' THEN\n" +
"\t\t'用户定义表' \n" +
"\tEND AS Type,\n" +
"\tB.is_nullable AS IS_NULLABLE \n" +
"FROM\n" +
"\tsys.tables AS A\n" +
"\tINNER JOIN sys.columns AS B ON B.object_id = A.object_id\n" +
"\tLEFT JOIN sys.extended_properties AS C ON C.major_id = B.object_id \n" +
"\tAND C.minor_id = B.column_id\n" +
"\tLEFT JOIN sys.types AS D ON D.system_type_id = B.system_type_id\n" +
"\tleft JOIN sys.extended_properties AS f ON A.object_id = f.major_id \n" +
"\tAND f.minor_id = 0 \n" +
"WHERE\n" +
"\tA.name = '"+table+"'";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()) {
TableField tableField = new TableField();
String columnName = resultSet.getString("COLUMN_NAME");
tableField.setPrimaryKey(false);
if (StringUtils.isNotBlank(primaryKey) && primaryKey.equals(columnName)) {
tableField.setPrimaryKey(true);
}
tableField.setLen(resultSet.getString("COLUMN_LENGTH"));
tableField.setScale(resultSet.getString("COLUMN_SCALE"));
tableField.setName(columnName);
tableField.setType(resultSet.getString("TYPE_NAME"));
tableField.setComment(resultSet.getString("REMARKS"));
Object nullable = resultSet.getObject("IS_NULLABLE");
tableField.setNullable(Boolean.TRUE.toString().equals(nullable.toString()));
tableFields.add(tableField);
String sql =
"SELECT\n"
+ "\tA.name AS TABLE_NAME,\n"
+ "\tf.value AS TABLE_COMMENT,\n"
+ "\tB.name AS COLUMN_NAME,\n"
+ "\tB.max_length AS COLUMN_LENGTH,\n"
+ "\tB.scale AS COLUMN_SCALE,\n"
+ "\tC.value AS REMARKS,\n"
+ "\tD.name AS TYPE_NAME,\n"
+ "CASE\n"
+ "\t\tA.type \n"
+ "\t\tWHEN 'u' THEN\n"
+ "\t\t'用户定义表' \n"
+ "\tEND AS Type,\n"
+ "\tB.is_nullable AS IS_NULLABLE \n"
+ "FROM\n"
+ "\tsys.tables AS A\n"
+ "\tINNER JOIN sys.columns AS B ON B.object_id = A.object_id\n"
+ "\tLEFT JOIN sys.extended_properties AS C ON C.major_id = B.object_id \n"
+ "\tAND C.minor_id = B.column_id\n"
+ "\tLEFT JOIN sys.types AS D ON D.system_type_id = B.system_type_id\n"
+ "\tleft JOIN sys.extended_properties AS f ON A.object_id = f.major_id \n"
+ "\tAND f.minor_id = 0 \n"
+ "WHERE\n"
+ "\tA.name = '"
+ table
+ "'";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()) {
TableField tableField = new TableField();
String columnName = resultSet.getString("COLUMN_NAME");
tableField.setPrimaryKey(false);
if (StringUtils.isNotBlank(primaryKey) && primaryKey.equals(columnName)) {
tableField.setPrimaryKey(true);
}
tableField.setLen(resultSet.getString("COLUMN_LENGTH"));
tableField.setScale(resultSet.getString("COLUMN_SCALE"));
tableField.setName(columnName);
tableField.setType(resultSet.getString("TYPE_NAME"));
tableField.setComment(resultSet.getString("REMARKS"));
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);
......@@ -224,19 +226,20 @@ public class SqlServerDataSourceChannel implements DataSourceChannelExt {
// 获取表列数
int columnCount = metaData.getColumnCount();
for (int i = 1; i <= columnCount; i++) {
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableComment = resultSet.getString("TABLE_COMMENT");
String tableComment = resultSet.getString("TABLE_COMMENT");
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tableNames.add(tables);
}
return tableNames;
} catch (SQLException e) {
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-tidb/src/main/java/org/apache/seatunnel/datasource/plugin/tidb/jdbc/TidbJdbcDataSourceChannel.java
浏览文件 @ d542c81e
......@@ -36,6 +36,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
public class TidbJdbcDataSourceChannel implements DataSourceChannelExt {
List<String> typeList = Arrays.asList("varchar", "char", "json");
@Override
public OptionRule getDataSourceOptions(@NonNull String pluginName) {
return TidbDataSourceConfig.OPTION_RULE;
......@@ -202,19 +203,20 @@ public class TidbJdbcDataSourceChannel implements DataSourceChannelExt {
// 获取表列数
int columnCount = metaData.getColumnCount();
for (int i = 1; i <= columnCount; i++) {
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableComment = resultSet.getString("TABLE_COMMENT");
String tableComment = resultSet.getString("TABLE_COMMENT");
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tableNames.add(tables);
}
return tableNames;
} catch (ClassNotFoundException | SQLException e) {
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-mysql-cdc/src/main/java/org/apache/seatunnel/datasource/plugin/cdc/mysql/MysqlCDCDataSourceChannel.java
浏览文件 @ d542c81e
......@@ -263,19 +263,20 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
// 获取表列数
int columnCount = metaData.getColumnCount();
for (int i = 1; i <= columnCount; i++) {
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) {
tables.put("TABLE_NAME", tableName);
}
String tableComment = resultSet.getString("TABLE_COMMENT");
String tableComment = resultSet.getString("TABLE_COMMENT");
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tables.put(metaData.getColumnName(i), resultSet.getObject(i));
if (StringUtils.isNotBlank(tableComment)) {
tables.put("TABLE_COMMENT", tableComment);
}
tableNames.add(tables);
}
return tableNames;
} catch (SQLException e) {
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-plugins-api/src/main/java/org/apache/seatunnel/datasource/plugin/api/model/TableField.java
浏览文件 @ d542c81e
......@@ -42,9 +42,7 @@ public class TableField {
private String outputDataType;
private String len;
private String scale;
}
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论