Skip to content

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

添加表信息

上级 858e3ad7
隐藏空白字符变更
内嵌 并排
正在显示 包含 360 行增加 和 0 行删除
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-mysql/src/main/java/org/apache/seatunnel/datasource/plugin/mysql/jdbc/MysqlJdbcDataSourceChannel.java
浏览文件 @ 4f1e6809
......@@ -23,6 +23,7 @@ import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
import org.apache.seatunnel.datasource.plugin.api.model.TableField;
import org.apache.seatunnel.datasource.plugin.api.utils.JdbcUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import lombok.NonNull;
......@@ -242,4 +243,54 @@ public class MysqlJdbcDataSourceChannel implements DataSourceChannelExt {
throw new DataSourcePluginException("get table names failed", e);
}
}
@Override
public Map getTableName(
@NonNull String pluginName,
Map<String, String> requestParams,
String database,
Map<String, String> options,
String tableName) {
try (Connection connection = getConnection(requestParams, database)) {
String sql =
"select * from information_schema.tables where table_name ='"
+ tableName
+ "' and table_schema ='"
+ database
+ "'";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
ResultSetMetaData metaData = resultSet.getMetaData();
// 字段的个数
int count = metaData.getColumnCount();
// 创建List集合
List<Map> list = new ArrayList<>();
// 处理结果集
while (resultSet.next()) {
// 创建Map集合
Map map = new HashMap<>();
// 根据字段的个数, 循环
for (int i = 0; i < count; i++) {
// 获取字段的名字
String columnName = metaData.getColumnName(i + 1);
// 获取字段对应的值
Object object = resultSet.getObject(columnName);
// 将字段名和字段值, 存入map集合中
map.put(columnName, object);
}
// 将map集合添加到List集合中
list.add(map);
}
if (CollectionUtils.isNotEmpty(list)) {
return list.get(0);
}
} catch (SQLException e) {
throw new DataSourcePluginException("get table fields failed", e);
} catch (ClassNotFoundException e) {
throw new DataSourcePluginException("get table fields failed", e);
}
return null;
}
}
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-oracle/src/main/java/org/apache/seatunnel/datasource/plugin/oracle/jdbc/OracleDataSourceChannel.java
浏览文件 @ 4f1e6809
......@@ -23,6 +23,7 @@ import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
import org.apache.seatunnel.datasource.plugin.api.model.TableField;
import org.apache.seatunnel.datasource.plugin.api.utils.JdbcUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import lombok.NonNull;
......@@ -250,4 +251,54 @@ public class OracleDataSourceChannel implements DataSourceChannelExt {
throw new DataSourcePluginException("get table names failed", e);
}
}
@Override
public Map getTableName(
@NonNull String pluginName,
Map<String, String> requestParams,
String database,
Map<String, String> options,
String tableName) {
try (Connection connection = getConnection(requestParams, database)) {
String sql =
"sselect * from all_tables where table_name='"
+ tableName
+ "' and OWNER ='"
+ database
+ "'";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
ResultSetMetaData metaData = resultSet.getMetaData();
// 字段的个数
int count = metaData.getColumnCount();
// 创建List集合
List<Map> list = new ArrayList<>();
// 处理结果集
while (resultSet.next()) {
// 创建Map集合
Map map = new HashMap<>();
// 根据字段的个数, 循环
for (int i = 0; i < count; i++) {
// 获取字段的名字
String columnName = metaData.getColumnName(i + 1);
// 获取字段对应的值
Object object = resultSet.getObject(columnName);
// 将字段名和字段值, 存入map集合中
map.put(columnName, object);
}
// 将map集合添加到List集合中
list.add(map);
}
if (CollectionUtils.isNotEmpty(list)) {
return list.get(0);
}
} catch (SQLException e) {
throw new DataSourcePluginException("get table fields failed", e);
} catch (ClassNotFoundException e) {
throw new DataSourcePluginException("get table fields failed", e);
}
return null;
}
}
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-postgresql/src/main/java/org/apache/seatunnel/datasource/plugin/postgresql/jdbc/PostgresqlDataSourceChannel.java
浏览文件 @ 4f1e6809
......@@ -23,6 +23,7 @@ import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
import org.apache.seatunnel.datasource.plugin.api.model.TableField;
import org.apache.seatunnel.datasource.plugin.api.utils.JdbcUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import lombok.NonNull;
......@@ -232,4 +233,54 @@ public class PostgresqlDataSourceChannel implements DataSourceChannelExt {
throw new DataSourcePluginException("get table names failed", e);
}
}
@Override
public Map getTableName(
@NonNull String pluginName,
Map<String, String> requestParams,
String database,
Map<String, String> options,
String tableName) {
try (Connection connection = getConnection(requestParams, database)) {
String sql =
"select * from information_schema.tables where table_name ='"
+ tableName
+ "' and table_schema ='"
+ database
+ "'";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
ResultSetMetaData metaData = resultSet.getMetaData();
// 字段的个数
int count = metaData.getColumnCount();
// 创建List集合
List<Map> list = new ArrayList<>();
// 处理结果集
while (resultSet.next()) {
// 创建Map集合
Map map = new HashMap<>();
// 根据字段的个数, 循环
for (int i = 0; i < count; i++) {
// 获取字段的名字
String columnName = metaData.getColumnName(i + 1);
// 获取字段对应的值
Object object = resultSet.getObject(columnName);
// 将字段名和字段值, 存入map集合中
map.put(columnName, object);
}
// 将map集合添加到List集合中
list.add(map);
}
if (CollectionUtils.isNotEmpty(list)) {
return list.get(0);
}
} catch (SQLException e) {
throw new DataSourcePluginException("get table fields failed", e);
} catch (ClassNotFoundException e) {
throw new DataSourcePluginException("get table fields failed", e);
}
return null;
}
}
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-sqlserver/src/main/java/org/apache/seatunnel/datasource/plugin/sqlserver/jdbc/SqlServerDataSourceChannel.java
浏览文件 @ 4f1e6809
......@@ -23,6 +23,7 @@ import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
import org.apache.seatunnel.datasource.plugin.api.model.TableField;
import org.apache.seatunnel.datasource.plugin.api.utils.JdbcUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import lombok.NonNull;
......@@ -253,4 +254,54 @@ public class SqlServerDataSourceChannel implements DataSourceChannelExt {
throw new DataSourcePluginException("get table names failed", e);
}
}
@Override
public Map getTableName(
@NonNull String pluginName,
Map<String, String> requestParams,
String database,
Map<String, String> options,
String tableName) {
try (Connection connection = getConnection(requestParams, database)) {
String sql =
"SELECT * FROM sys.all_objects WHERE object_id = OBJECT_ID(N'["
+ database
+ "].[dbo].["
+ tableName
+ "]') AND type IN ('U')";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
ResultSetMetaData metaData = resultSet.getMetaData();
// 字段的个数
int count = metaData.getColumnCount();
// 创建List集合
List<Map> list = new ArrayList<>();
// 处理结果集
while (resultSet.next()) {
// 创建Map集合
Map map = new HashMap<>();
// 根据字段的个数, 循环
for (int i = 0; i < count; i++) {
// 获取字段的名字
String columnName = metaData.getColumnName(i + 1);
// 获取字段对应的值
Object object = resultSet.getObject(columnName);
// 将字段名和字段值, 存入map集合中
map.put(columnName, object);
}
// 将map集合添加到List集合中
list.add(map);
}
if (CollectionUtils.isNotEmpty(list)) {
return list.get(0);
}
} catch (SQLException e) {
throw new DataSourcePluginException("get table fields failed", e);
} catch (ClassNotFoundException e) {
throw new DataSourcePluginException("get table fields failed", e);
}
return null;
}
}
seatunnel-datasource/seatunnel-datasource-plugins/datasource-jdbc-tidb/src/main/java/org/apache/seatunnel/datasource/plugin/tidb/jdbc/TidbJdbcDataSourceChannel.java
浏览文件 @ 4f1e6809
......@@ -23,6 +23,7 @@ import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
import org.apache.seatunnel.datasource.plugin.api.model.TableField;
import org.apache.seatunnel.datasource.plugin.api.utils.JdbcUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import lombok.NonNull;
......@@ -238,4 +239,54 @@ public class TidbJdbcDataSourceChannel implements DataSourceChannelExt {
throw new DataSourcePluginException("get table names failed", e);
}
}
@Override
public Map getTableName(
@NonNull String pluginName,
Map<String, String> requestParams,
String database,
Map<String, String> options,
String tableName) {
try (Connection connection = getConnection(requestParams, database)) {
String sql =
"select * from information_schema.tables where table_name ='"
+ tableName
+ "' and table_schema ='"
+ database
+ "'";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
ResultSetMetaData metaData = resultSet.getMetaData();
// 字段的个数
int count = metaData.getColumnCount();
// 创建List集合
List<Map> list = new ArrayList<>();
// 处理结果集
while (resultSet.next()) {
// 创建Map集合
Map map = new HashMap<>();
// 根据字段的个数, 循环
for (int i = 0; i < count; i++) {
// 获取字段的名字
String columnName = metaData.getColumnName(i + 1);
// 获取字段对应的值
Object object = resultSet.getObject(columnName);
// 将字段名和字段值, 存入map集合中
map.put(columnName, object);
}
// 将map集合添加到List集合中
list.add(map);
}
if (CollectionUtils.isNotEmpty(list)) {
return list.get(0);
}
} catch (SQLException e) {
throw new DataSourcePluginException("get table fields failed", e);
} catch (ClassNotFoundException e) {
throw new DataSourcePluginException("get table fields failed", e);
}
return null;
}
}
seatunnel-datasource/seatunnel-datasource-plugins/datasource-mysql-cdc/src/main/java/org/apache/seatunnel/datasource/plugin/cdc/mysql/MysqlCDCDataSourceChannel.java
浏览文件 @ 4f1e6809
......@@ -22,6 +22,7 @@ import org.apache.seatunnel.datasource.plugin.api.DataSourceChannelExt;
import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
import org.apache.seatunnel.datasource.plugin.api.model.TableField;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import com.google.common.collect.Sets;
......@@ -297,4 +298,52 @@ public class MysqlCDCDataSourceChannel implements DataSourceChannelExt {
throw new DataSourcePluginException("get table names failed", e);
}
}
@Override
public Map getTableName(
@NonNull String pluginName,
Map<String, String> requestParams,
String database,
Map<String, String> options,
String tableName) {
try (Connection connection = init(requestParams); ) {
String sql =
"select * from information_schema.tables where table_name ='"
+ tableName
+ "' and table_schema ='"
+ database
+ "'";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
ResultSetMetaData metaData = resultSet.getMetaData();
// 字段的个数
int count = metaData.getColumnCount();
// 创建List集合
List<Map> list = new ArrayList<>();
// 处理结果集
while (resultSet.next()) {
// 创建Map集合
Map map = new HashMap<>();
// 根据字段的个数, 循环
for (int i = 0; i < count; i++) {
// 获取字段的名字
String columnName = metaData.getColumnName(i + 1);
// 获取字段对应的值
Object object = resultSet.getObject(columnName);
// 将字段名和字段值, 存入map集合中
map.put(columnName, object);
}
// 将map集合添加到List集合中
list.add(map);
}
if (CollectionUtils.isNotEmpty(list)) {
return list.get(0);
}
} catch (SQLException e) {
throw new DataSourcePluginException("get table fields failed", e);
}
return null;
}
}
seatunnel-datasource/seatunnel-datasource-plugins/datasource-plugins-api/src/main/java/org/apache/seatunnel/datasource/plugin/api/DataSourceChannelExt.java
浏览文件 @ 4f1e6809
......@@ -12,4 +12,11 @@ public interface DataSourceChannelExt extends DataSourceChannel {
Map<String, String> requestParams,
String database,
Map<String, String> options);
Map getTableName(
@NonNull String pluginName,
Map<String, String> requestParams,
String database,
Map<String, String> options,
String tableName);
}
seatunnel-datasource/seatunnel-datasource-plugins/datasource-sqlserver-cdc/src/main/java/org/apache/seatunnel/datasource/plugin/cdc/sqlserver/SqlServerCDCDataSourceChannel.java
浏览文件 @ 4f1e6809
......@@ -22,6 +22,7 @@ import org.apache.seatunnel.datasource.plugin.api.DataSourceChannelExt;
import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
import org.apache.seatunnel.datasource.plugin.api.model.TableField;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
......@@ -265,4 +266,52 @@ public class SqlServerCDCDataSourceChannel implements DataSourceChannelExt {
throw new DataSourcePluginException("get table names failed", e);
}
}
@Override
public Map getTableName(
@NonNull String pluginName,
Map<String, String> requestParams,
String database,
Map<String, String> options,
String tableName) {
try (Connection connection = getConnection(pluginName, requestParams)) {
String sql =
"SELECT * FROM sys.all_objects WHERE object_id = OBJECT_ID(N'["
+ database
+ "].[dbo].["
+ tableName
+ "]') AND type IN ('U')";
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
ResultSetMetaData metaData = resultSet.getMetaData();
// 字段的个数
int count = metaData.getColumnCount();
// 创建List集合
List<Map> list = new ArrayList<>();
// 处理结果集
while (resultSet.next()) {
// 创建Map集合
Map map = new HashMap<>();
// 根据字段的个数, 循环
for (int i = 0; i < count; i++) {
// 获取字段的名字
String columnName = metaData.getColumnName(i + 1);
// 获取字段对应的值
Object object = resultSet.getObject(columnName);
// 将字段名和字段值, 存入map集合中
map.put(columnName, object);
}
// 将map集合添加到List集合中
list.add(map);
}
if (CollectionUtils.isNotEmpty(list)) {
return list.get(0);
}
} catch (SQLException e) {
throw new DataSourcePluginException("get table fields failed", e);
}
return null;
}
}
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论