Skip to content

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

添加tablename查询

上级 12c34977
隐藏空白字符变更
内嵌 并排
正在显示 包含 90 行增加 和 99 行删除
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-mysql/src/main/java/org/apache/seatunnel/datasource/plugin/mysql/jdbc/MysqlJdbcDataSourceChannel.java
浏览文件 @ 3474b595
...@@ -213,7 +213,8 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt { ...@@ -213,7 +213,8 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
List<Map> tableNames = new ArrayList<>(); List<Map> tableNames = new ArrayList<>();
try (Connection connection = getConnection(requestParams)) { try (Connection connection = getConnection(requestParams)) {
StringBuilder sql =new StringBuilder( "SELECT * FROM information_schema.tables WHERE 1=1 "); StringBuilder sql =
new StringBuilder("SELECT * FROM information_schema.tables WHERE 1=1 ");
if (StringUtils.isNotBlank(database)) { if (StringUtils.isNotBlank(database)) {
...@@ -225,7 +226,7 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt { ...@@ -225,7 +226,7 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
sql.append(" and TABLE_NAME='" + tableName + "'"); sql.append(" and TABLE_NAME='" + tableName + "'");
} }
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
// statement.execute("use " + database); // statement.execute("use " + database);
ResultSet resultSet = statement.executeQuery(sql.toString()); ResultSet resultSet = statement.executeQuery(sql.toString());
while (resultSet.next()) { while (resultSet.next()) {
...@@ -272,7 +273,7 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt { ...@@ -272,7 +273,7 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
+ database + database
+ "'"; + "'";
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
// statement.execute("use " + database); // statement.execute("use " + database);
ResultSet resultSet = statement.executeQuery(sql); ResultSet resultSet = statement.executeQuery(sql);
ResultSetMetaData metaData = resultSet.getMetaData(); ResultSetMetaData metaData = resultSet.getMetaData();
// 字段的个数 // 字段的个数
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-oracle/src/main/java/org/apache/seatunnel/datasource/plugin/oracle/jdbc/OracleDataSourceChannel.java
浏览文件 @ 3474b595
...@@ -54,10 +54,10 @@ public class OracleDataSourceChannel implements DataSourceChannelExt { ...@@ -54,10 +54,10 @@ public class OracleDataSourceChannel implements DataSourceChannelExt {
Map<String, String> option) { Map<String, String> option) {
List<String> tableNames = new ArrayList<>(); List<String> tableNames = new ArrayList<>();
try (Connection connection = getConnection(requestParams); try (Connection connection = getConnection(requestParams);
ResultSet resultSet = ResultSet resultSet =
connection connection
.getMetaData() .getMetaData()
.getTables(database, null, null, new String[]{"TABLE"});) { .getTables(database, null, null, new String[] {"TABLE"}); ) {
while (resultSet.next()) { while (resultSet.next()) {
String tableName = resultSet.getString("TABLE_NAME"); String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) { if (StringUtils.isNotBlank(tableName)) {
...@@ -75,8 +75,8 @@ public class OracleDataSourceChannel implements DataSourceChannelExt { ...@@ -75,8 +75,8 @@ public class OracleDataSourceChannel implements DataSourceChannelExt {
@NonNull String pluginName, @NonNull Map<String, String> requestParams) { @NonNull String pluginName, @NonNull Map<String, String> requestParams) {
List<String> dbNames = new ArrayList<>(); List<String> dbNames = new ArrayList<>();
try (Connection connection = getConnection(requestParams); try (Connection connection = getConnection(requestParams);
PreparedStatement statement = connection.prepareStatement("SHOW DATABASES;"); PreparedStatement statement = connection.prepareStatement("SHOW DATABASES;");
ResultSet re = statement.executeQuery()) { ResultSet re = statement.executeQuery()) {
// filter system databases // filter system databases
while (re.next()) { while (re.next()) {
String dbName = re.getString("database"); String dbName = re.getString("database");
...@@ -236,7 +236,7 @@ public class OracleDataSourceChannel implements DataSourceChannelExt { ...@@ -236,7 +236,7 @@ public class OracleDataSourceChannel implements DataSourceChannelExt {
} }
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
// statement.execute("use " + database); // statement.execute("use " + database);
ResultSet resultSet = statement.executeQuery(sql.toString()); ResultSet resultSet = statement.executeQuery(sql.toString());
ResultSetMetaData metaData = resultSet.getMetaData(); ResultSetMetaData metaData = resultSet.getMetaData();
while (resultSet.next()) { while (resultSet.next()) {
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-postgresql/src/main/java/org/apache/seatunnel/datasource/plugin/postgresql/jdbc/PostgresqlDataSourceChannel.java
浏览文件 @ 3474b595
...@@ -56,13 +56,13 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt { ...@@ -56,13 +56,13 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt {
String query = "SELECT table_schema, table_name FROM information_schema.tables"; String query = "SELECT table_schema, table_name FROM information_schema.tables";
try (Connection connection = getConnection(requestParams, database)) { try (Connection connection = getConnection(requestParams, database)) {
try (Statement statement = connection.createStatement(); try (Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(query)) { ResultSet resultSet = statement.executeQuery(query)) {
while (resultSet.next()) { while (resultSet.next()) {
String schemaName = resultSet.getString("table_schema"); String schemaName = resultSet.getString("table_schema");
String tableName = resultSet.getString("table_name"); String tableName = resultSet.getString("table_name");
if (StringUtils.isNotBlank(schemaName) if (StringUtils.isNotBlank(schemaName)
&& !PostgresqlDataSourceConfig.POSTGRESQL_SYSTEM_DATABASES.contains( && !PostgresqlDataSourceConfig.POSTGRESQL_SYSTEM_DATABASES.contains(
schemaName)) { schemaName)) {
tableNames.add(schemaName + "." + tableName); tableNames.add(schemaName + "." + tableName);
} }
} }
...@@ -78,14 +78,14 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt { ...@@ -78,14 +78,14 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt {
@NonNull String pluginName, @NonNull Map<String, String> requestParams) { @NonNull String pluginName, @NonNull Map<String, String> requestParams) {
List<String> dbNames = new ArrayList<>(); List<String> dbNames = new ArrayList<>();
try (Connection connection = getConnection(requestParams); try (Connection connection = getConnection(requestParams);
PreparedStatement statement = PreparedStatement statement =
connection.prepareStatement("select datname from pg_database;"); connection.prepareStatement("select datname from pg_database;");
ResultSet re = statement.executeQuery()) { ResultSet re = statement.executeQuery()) {
while (re.next()) { while (re.next()) {
String dbName = re.getString("datname"); String dbName = re.getString("datname");
if (StringUtils.isNotBlank(dbName) if (StringUtils.isNotBlank(dbName)
&& !PostgresqlDataSourceConfig.POSTGRESQL_SYSTEM_DATABASES.contains( && !PostgresqlDataSourceConfig.POSTGRESQL_SYSTEM_DATABASES.contains(
dbName)) { dbName)) {
dbNames.add(dbName); dbNames.add(dbName);
} }
} }
...@@ -112,7 +112,7 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt { ...@@ -112,7 +112,7 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt {
@NonNull String database, @NonNull String database,
@NonNull String table) { @NonNull String table) {
List<TableField> tableFields = new ArrayList<>(); List<TableField> tableFields = new ArrayList<>();
try (Connection connection = getConnection(requestParams, database);) { try (Connection connection = getConnection(requestParams, database); ) {
DatabaseMetaData metaData = connection.getMetaData(); DatabaseMetaData metaData = connection.getMetaData();
String primaryKey = getPrimaryKey(metaData, database, table); String primaryKey = getPrimaryKey(metaData, database, table);
String[] split = table.split("\\."); String[] split = table.split("\\.");
...@@ -203,7 +203,9 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt { ...@@ -203,7 +203,9 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt {
Map<String, String> options) { Map<String, String> options) {
List<Map> tableNames = new ArrayList<>(); List<Map> tableNames = new ArrayList<>();
try (Connection connection = getConnection(requestParams)) { try (Connection connection = getConnection(requestParams)) {
StringBuilder sql = new StringBuilder("SELECT table_schema, table_name TABLE_NAME, table_comment TABLE_COMMENT FROM information_schema.tables where 1=1 "); StringBuilder sql =
new StringBuilder(
"SELECT table_schema, table_name TABLE_NAME, table_comment TABLE_COMMENT FROM information_schema.tables where 1=1 ");
if (StringUtils.isNotBlank(database)) { if (StringUtils.isNotBlank(database)) {
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-sqlserver/src/main/java/org/apache/seatunnel/datasource/plugin/sqlserver/jdbc/SqlServerDataSourceChannel.java
浏览文件 @ 3474b595
...@@ -228,13 +228,14 @@ public class SqlServerDataSourceChannel implements DataSourceChannelExt { ...@@ -228,13 +228,14 @@ public class SqlServerDataSourceChannel implements DataSourceChannelExt {
List<Map> tableNames = new ArrayList<>(); List<Map> tableNames = new ArrayList<>();
try (Connection connection = getConnection(pluginName, requestParams)) { try (Connection connection = getConnection(pluginName, requestParams)) {
ResultSet resultSet=null; ResultSet resultSet = null;
if (StringUtils.isNotBlank(database)) { if (StringUtils.isNotBlank(database)) {
StringBuilder sql =
StringBuilder sql =new StringBuilder( "select ROW_NUMBER() OVER (ORDER BY a.name) AS No, a.name TABLE_NAME ,g.class_desc,g.major_id,g.class,g.value TABLE_COMMENT from " new StringBuilder(
+ " sys.tables a left join sys.extended_properties g on (a.object_id = g.major_id AND g.minor_id = 0) where 1=1 "); "select ROW_NUMBER() OVER (ORDER BY a.name) AS No, a.name TABLE_NAME ,g.class_desc,g.major_id,g.class,g.value TABLE_COMMENT from "
+ " sys.tables a left join sys.extended_properties g on (a.object_id = g.major_id AND g.minor_id = 0) where 1=1 ");
if (StringUtils.isNotBlank(tableName)) { if (StringUtils.isNotBlank(tableName)) {
sql.append(" and a.name='" + tableName + "'"); sql.append(" and a.name='" + tableName + "'");
...@@ -242,58 +243,52 @@ public class SqlServerDataSourceChannel implements DataSourceChannelExt { ...@@ -242,58 +243,52 @@ public class SqlServerDataSourceChannel implements DataSourceChannelExt {
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
statement.execute("use " + database); statement.execute("use " + database);
resultSet = statement.executeQuery(sql.toString()); resultSet = statement.executeQuery(sql.toString());
fillTableNames(tableNames, resultSet); fillTableNames(tableNames, resultSet);
} else { } else {
String sqlDatabases = "SELECT * FROM sys.sysdatabases"; String sqlDatabases = "SELECT * FROM sys.sysdatabases";
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
resultSet = statement.executeQuery(sqlDatabases.toString()); resultSet = statement.executeQuery(sqlDatabases.toString());
while (resultSet.next()) { while (resultSet.next()) {
// ResultSetMetaData metaData = resultSet.getMetaData();
// ResultSetMetaData metaData = resultSet.getMetaData(); // // 获取表列数
// // 获取表列数 // int columnCount = metaData.getColumnCount();
// int columnCount = metaData.getColumnCount(); // for (int i = 1; i <= columnCount; i++) {
// for (int i = 1; i <= columnCount; i++) { // tables.put(metaData.getColumnName(i),
// tables.put(metaData.getColumnName(i), resultSet.getObject(i)); // resultSet.getObject(i));
// } // }
String databaseName = resultSet.getString("name"); String databaseName = resultSet.getString("name");
if (StringUtils.isNotBlank(databaseName) && !Arrays.asList("master","tempdb","model","msdb").contains(databaseName)) { if (StringUtils.isNotBlank(databaseName)
StringBuilder sql =new StringBuilder( "select ROW_NUMBER() OVER (ORDER BY a.name) AS No, a.name TABLE_NAME ,g.class_desc,g.major_id,g.class,g.value TABLE_COMMENT from " && !Arrays.asList("master", "tempdb", "model", "msdb")
+ " sys.tables a left join sys.extended_properties g on (a.object_id = g.major_id AND g.minor_id = 0) where 1=1 "); .contains(databaseName)) {
StringBuilder sql =
new StringBuilder(
"select ROW_NUMBER() OVER (ORDER BY a.name) AS No, a.name TABLE_NAME ,g.class_desc,g.major_id,g.class,g.value TABLE_COMMENT from "
+ " sys.tables a left join sys.extended_properties g on (a.object_id = g.major_id AND g.minor_id = 0) where 1=1 ");
if (StringUtils.isNotBlank(tableName)) { if (StringUtils.isNotBlank(tableName)) {
sql.append(" and a.name='" + tableName + "'"); sql.append(" and a.name='" + tableName + "'");
} }
statement.execute("use " + database); statement.execute("use " + database);
resultSet = statement.executeQuery(sql.toString()); resultSet = statement.executeQuery(sql.toString());
fillTableNames(tableNames, resultSet); fillTableNames(tableNames, resultSet);
} }
} }
} }
return tableNames; return tableNames;
} catch (SQLException e) { } catch (SQLException e) {
throw new DataSourcePluginException("get table names failed", e); throw new DataSourcePluginException("get table names failed", e);
} }
} }
private static void fillTableNames(List<Map> tableNames, ResultSet resultSet) throws SQLException { private static void fillTableNames(List<Map> tableNames, ResultSet resultSet)
if(Objects.nonNull(resultSet)) { throws SQLException {
if (Objects.nonNull(resultSet)) {
while (resultSet.next()) { while (resultSet.next()) {
Map tables = new HashMap(); Map tables = new HashMap();
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-tidb/src/main/java/org/apache/seatunnel/datasource/plugin/tidb/jdbc/TidbJdbcDataSourceChannel.java
浏览文件 @ 3474b595
...@@ -211,7 +211,8 @@ public class TidbJdbcDataSourceChannel implements DataSourceChannelExt { ...@@ -211,7 +211,8 @@ public class TidbJdbcDataSourceChannel implements DataSourceChannelExt {
List<Map> tableNames = new ArrayList<>(); List<Map> tableNames = new ArrayList<>();
try (Connection connection = getConnection(requestParams)) { try (Connection connection = getConnection(requestParams)) {
StringBuilder sql =new StringBuilder( "SELECT * FROM information_schema.tables WHERE 1=1 "); StringBuilder sql =
new StringBuilder("SELECT * FROM information_schema.tables WHERE 1=1 ");
if (StringUtils.isNotBlank(database)) { if (StringUtils.isNotBlank(database)) {
...@@ -223,7 +224,7 @@ public class TidbJdbcDataSourceChannel implements DataSourceChannelExt { ...@@ -223,7 +224,7 @@ public class TidbJdbcDataSourceChannel implements DataSourceChannelExt {
sql.append(" and TABLE_NAME='" + tableName + "'"); sql.append(" and TABLE_NAME='" + tableName + "'");
} }
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
// statement.execute("use " + database); // statement.execute("use " + database);
ResultSet resultSet = statement.executeQuery(sql.toString()); ResultSet resultSet = statement.executeQuery(sql.toString());
while (resultSet.next()) { while (resultSet.next()) {
Map tables = new HashMap(); Map tables = new HashMap();
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-mysql-cdc/src/main/java/org/apache/seatunnel/datasource/plugin/cdc/mysql/MysqlCDCDataSourceChannel.java
浏览文件 @ 3474b595
...@@ -27,13 +27,10 @@ import org.apache.commons.lang3.StringUtils; ...@@ -27,13 +27,10 @@ import org.apache.commons.lang3.StringUtils;
import com.google.common.collect.Sets; import com.google.common.collect.Sets;
import lombok.NonNull; import lombok.NonNull;
import org.apache.seatunnel.datasource.plugin.api.utils.JdbcUtils;
import java.sql.*; import java.sql.*;
import java.util.*; import java.util.*;
import static com.google.common.base.Preconditions.checkNotNull;
public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
List<String> typeList = Arrays.asList("varchar", "char", "json"); List<String> typeList = Arrays.asList("varchar", "char", "json");
public static final Set<String> MYSQL_SYSTEM_DATABASES = public static final Set<String> MYSQL_SYSTEM_DATABASES =
...@@ -100,9 +97,9 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -100,9 +97,9 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
@SuppressWarnings("checkstyle:MagicNumber") @SuppressWarnings("checkstyle:MagicNumber")
protected boolean checkJdbcConnectivity(Map<String, String> requestParams) { protected boolean checkJdbcConnectivity(Map<String, String> requestParams) {
try (Connection connection = init(requestParams); try (Connection connection = init(requestParams);
Statement statement = connection.createStatement()) { Statement statement = connection.createStatement()) {
try (ResultSet resultSet = statement.executeQuery("SHOW MASTER STATUS");) { try (ResultSet resultSet = statement.executeQuery("SHOW MASTER STATUS"); ) {
if (resultSet.next()) { if (resultSet.next()) {
String binlogFile = resultSet.getString("File"); String binlogFile = resultSet.getString("File");
if (StringUtils.isBlank(binlogFile)) { if (StringUtils.isBlank(binlogFile)) {
...@@ -114,7 +111,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -114,7 +111,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
} }
try (ResultSet resultSet = try (ResultSet resultSet =
statement.executeQuery("SHOW VARIABLES LIKE 'binlog_format'")) { statement.executeQuery("SHOW VARIABLES LIKE 'binlog_format'")) {
if (resultSet.next()) { if (resultSet.next()) {
String binlogFormat = resultSet.getString("Value"); String binlogFormat = resultSet.getString("Value");
if (!"ROW".equalsIgnoreCase(binlogFormat)) { if (!"ROW".equalsIgnoreCase(binlogFormat)) {
...@@ -126,7 +123,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -126,7 +123,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
} }
try (ResultSet resultSet = try (ResultSet resultSet =
statement.executeQuery("SHOW VARIABLES LIKE 'binlog_row_image'")) { statement.executeQuery("SHOW VARIABLES LIKE 'binlog_row_image'")) {
if (resultSet.next()) { if (resultSet.next()) {
String binlogRowImage = resultSet.getString("Value"); String binlogRowImage = resultSet.getString("Value");
if (!"FULL".equalsIgnoreCase(binlogRowImage)) { if (!"FULL".equalsIgnoreCase(binlogRowImage)) {
...@@ -160,8 +157,8 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -160,8 +157,8 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
protected List<String> getDataBaseNames(Map<String, String> requestParams) throws SQLException { protected List<String> getDataBaseNames(Map<String, String> requestParams) throws SQLException {
List<String> dbNames = new ArrayList<>(); List<String> dbNames = new ArrayList<>();
try (Connection connection = init(requestParams); try (Connection connection = init(requestParams);
PreparedStatement statement = connection.prepareStatement("SHOW DATABASES;"); PreparedStatement statement = connection.prepareStatement("SHOW DATABASES;");
ResultSet re = statement.executeQuery()) { ResultSet re = statement.executeQuery()) {
// filter system databases // filter system databases
while (re.next()) { while (re.next()) {
String dbName = re.getString("database"); String dbName = re.getString("database");
...@@ -176,10 +173,10 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -176,10 +173,10 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
protected List<String> getTableNames(Map<String, String> requestParams, String dbName) { protected List<String> getTableNames(Map<String, String> requestParams, String dbName) {
List<String> tableNames = new ArrayList<>(); List<String> tableNames = new ArrayList<>();
try (Connection connection = init(requestParams); try (Connection connection = init(requestParams);
ResultSet resultSet = ResultSet resultSet =
connection connection
.getMetaData() .getMetaData()
.getTables(dbName, null, null, new String[]{"TABLE"})) { .getTables(dbName, null, null, new String[] {"TABLE"})) {
while (resultSet.next()) { while (resultSet.next()) {
String tableName = resultSet.getString("TABLE_NAME"); String tableName = resultSet.getString("TABLE_NAME");
if (StringUtils.isNotBlank(tableName)) { if (StringUtils.isNotBlank(tableName)) {
...@@ -195,7 +192,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -195,7 +192,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
protected List<TableField> getTableFields( protected List<TableField> getTableFields(
Map<String, String> requestParams, String dbName, String tableName) { Map<String, String> requestParams, String dbName, String tableName) {
List<TableField> tableFields = new ArrayList<>(); List<TableField> tableFields = new ArrayList<>();
try (Connection connection = init(requestParams);) { try (Connection connection = init(requestParams); ) {
DatabaseMetaData metaData = connection.getMetaData(); DatabaseMetaData metaData = connection.getMetaData();
String primaryKey = getPrimaryKey(metaData, dbName, tableName); String primaryKey = getPrimaryKey(metaData, dbName, tableName);
...@@ -272,9 +269,10 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -272,9 +269,10 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
Map<String, String> options) { Map<String, String> options) {
List<Map> tableNames = new ArrayList<>(); List<Map> tableNames = new ArrayList<>();
try (Connection connection =init(requestParams)) { try (Connection connection = init(requestParams)) {
StringBuilder sql = new StringBuilder("SELECT * FROM information_schema.tables WHERE 1=1 "); StringBuilder sql =
new StringBuilder("SELECT * FROM information_schema.tables WHERE 1=1 ");
if (StringUtils.isNotBlank(database)) { if (StringUtils.isNotBlank(database)) {
...@@ -286,7 +284,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -286,7 +284,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
sql.append(" and TABLE_NAME='" + tableName + "'"); sql.append(" and TABLE_NAME='" + tableName + "'");
} }
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
// statement.execute("use " + database); // statement.execute("use " + database);
ResultSet resultSet = statement.executeQuery(sql.toString()); ResultSet resultSet = statement.executeQuery(sql.toString());
while (resultSet.next()) { while (resultSet.next()) {
...@@ -325,7 +323,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -325,7 +323,7 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
Map<String, String> options, Map<String, String> options,
String tableName) { String tableName) {
try (Connection connection = init(requestParams);) { try (Connection connection = init(requestParams); ) {
String sql = String sql =
"select * from information_schema.tables where table_name ='" "select * from information_schema.tables where table_name ='"
+ tableName + tableName
......
seatunnel-datasource/seatunnel-datasource-plugins/datasource-sqlserver-cdc/src/main/java/org/apache/seatunnel/datasource/plugin/cdc/sqlserver/SqlServerCDCDataSourceChannel.java
浏览文件 @ 3474b595
...@@ -233,13 +233,14 @@ public class SqlServerCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -233,13 +233,14 @@ public class SqlServerCDCDataSourceChannel implements DataSourceChannelExt {
List<Map> tableNames = new ArrayList<>(); List<Map> tableNames = new ArrayList<>();
try (Connection connection = getConnection(pluginName, requestParams)) { try (Connection connection = getConnection(pluginName, requestParams)) {
ResultSet resultSet=null; ResultSet resultSet = null;
if (StringUtils.isNotBlank(database)) { if (StringUtils.isNotBlank(database)) {
StringBuilder sql =
StringBuilder sql =new StringBuilder( "select ROW_NUMBER() OVER (ORDER BY a.name) AS No, a.name TABLE_NAME ,g.class_desc,g.major_id,g.class,g.value TABLE_COMMENT from " new StringBuilder(
+ " sys.tables a left join sys.extended_properties g on (a.object_id = g.major_id AND g.minor_id = 0) where 1=1 "); "select ROW_NUMBER() OVER (ORDER BY a.name) AS No, a.name TABLE_NAME ,g.class_desc,g.major_id,g.class,g.value TABLE_COMMENT from "
+ " sys.tables a left join sys.extended_properties g on (a.object_id = g.major_id AND g.minor_id = 0) where 1=1 ");
if (StringUtils.isNotBlank(tableName)) { if (StringUtils.isNotBlank(tableName)) {
sql.append(" and a.name='" + tableName + "'"); sql.append(" and a.name='" + tableName + "'");
...@@ -247,58 +248,52 @@ public class SqlServerCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -247,58 +248,52 @@ public class SqlServerCDCDataSourceChannel implements DataSourceChannelExt {
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
statement.execute("use " + database); statement.execute("use " + database);
resultSet = statement.executeQuery(sql.toString()); resultSet = statement.executeQuery(sql.toString());
fillTableNames(tableNames, resultSet); fillTableNames(tableNames, resultSet);
} else { } else {
String sqlDatabases = "SELECT * FROM sys.sysdatabases"; String sqlDatabases = "SELECT * FROM sys.sysdatabases";
Statement statement = connection.createStatement(); Statement statement = connection.createStatement();
resultSet = statement.executeQuery(sqlDatabases.toString()); resultSet = statement.executeQuery(sqlDatabases.toString());
while (resultSet.next()) { while (resultSet.next()) {
// ResultSetMetaData metaData = resultSet.getMetaData();
// ResultSetMetaData metaData = resultSet.getMetaData(); // // 获取表列数
// // 获取表列数 // int columnCount = metaData.getColumnCount();
// int columnCount = metaData.getColumnCount(); // for (int i = 1; i <= columnCount; i++) {
// for (int i = 1; i <= columnCount; i++) { // tables.put(metaData.getColumnName(i),
// tables.put(metaData.getColumnName(i), resultSet.getObject(i)); // resultSet.getObject(i));
// } // }
String databaseName = resultSet.getString("name"); String databaseName = resultSet.getString("name");
if (StringUtils.isNotBlank(databaseName) && !Arrays.asList("master","tempdb","model","msdb").contains(databaseName)) { if (StringUtils.isNotBlank(databaseName)
StringBuilder sql =new StringBuilder( "select ROW_NUMBER() OVER (ORDER BY a.name) AS No, a.name TABLE_NAME ,g.class_desc,g.major_id,g.class,g.value TABLE_COMMENT from " && !Arrays.asList("master", "tempdb", "model", "msdb")
+ " sys.tables a left join sys.extended_properties g on (a.object_id = g.major_id AND g.minor_id = 0) where 1=1 "); .contains(databaseName)) {
StringBuilder sql =
new StringBuilder(
"select ROW_NUMBER() OVER (ORDER BY a.name) AS No, a.name TABLE_NAME ,g.class_desc,g.major_id,g.class,g.value TABLE_COMMENT from "
+ " sys.tables a left join sys.extended_properties g on (a.object_id = g.major_id AND g.minor_id = 0) where 1=1 ");
if (StringUtils.isNotBlank(tableName)) { if (StringUtils.isNotBlank(tableName)) {
sql.append(" and a.name='" + tableName + "'"); sql.append(" and a.name='" + tableName + "'");
} }
statement.execute("use " + database); statement.execute("use " + database);
resultSet = statement.executeQuery(sql.toString()); resultSet = statement.executeQuery(sql.toString());
fillTableNames(tableNames, resultSet); fillTableNames(tableNames, resultSet);
} }
} }
} }
return tableNames; return tableNames;
} catch (SQLException e) { } catch (SQLException e) {
throw new DataSourcePluginException("get table names failed", e); throw new DataSourcePluginException("get table names failed", e);
} }
} }
private static void fillTableNames(List<Map> tableNames, ResultSet resultSet) throws SQLException { private static void fillTableNames(List<Map> tableNames, ResultSet resultSet)
if(Objects.nonNull(resultSet)) { throws SQLException {
if (Objects.nonNull(resultSet)) {
while (resultSet.next()) { while (resultSet.next()) {
Map tables = new HashMap(); Map tables = new HashMap();
...@@ -324,7 +319,6 @@ public class SqlServerCDCDataSourceChannel implements DataSourceChannelExt { ...@@ -324,7 +319,6 @@ public class SqlServerCDCDataSourceChannel implements DataSourceChannelExt {
} }
} }
@Override @Override
public Map getTableName( public Map getTableName(
@NonNull String pluginName, @NonNull String pluginName,
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论