diff --git a/datavines-connector/datavines-connector-api/src/main/java/io/datavines/connector/api/Dialect.java b/datavines-connector/datavines-connector-api/src/main/java/io/datavines/connector/api/Dialect.java index 860580ca..01774bb9 100644 --- a/datavines-connector/datavines-connector-api/src/main/java/io/datavines/connector/api/Dialect.java +++ b/datavines-connector/datavines-connector-api/src/main/java/io/datavines/connector/api/Dialect.java @@ -100,11 +100,11 @@ default String getQuoteIdentifier() { } default String getTableExistsQuery(String table) { - return String.format("SELECT * FROM %s WHERE 1=0", table); + return String.format("SELECT 1 FROM %s WHERE 1=0", table); } default String getSchemaQuery(String table) { - return String.format("SELECT * FROM %s WHERE 1=0", table); + return String.format("SELECT 1 FROM %s WHERE 1=0", table); } default String getCountQuery(String table) { diff --git a/datavines-connector/datavines-connector-plugins/datavines-connector-hive/src/main/java/io/datavines/connector/plugin/HiveParameterConverter.java b/datavines-connector/datavines-connector-plugins/datavines-connector-hive/src/main/java/io/datavines/connector/plugin/HiveParameterConverter.java index 2dc99a80..7c6f415a 100644 --- a/datavines-connector/datavines-connector-plugins/datavines-connector-hive/src/main/java/io/datavines/connector/plugin/HiveParameterConverter.java +++ b/datavines-connector/datavines-connector-plugins/datavines-connector-hive/src/main/java/io/datavines/connector/plugin/HiveParameterConverter.java @@ -26,7 +26,6 @@ public class HiveParameterConverter extends JdbcParameterConverter { @Override protected String getUrl(Map parameter) { - StringBuilder address = new StringBuilder(); address.append("jdbc:hive2://"); Object port = parameter.get(PORT); diff --git a/datavines-connector/datavines-connector-plugins/datavines-connector-impala/src/main/java/io/datavines/connector/plugin/ImpalaDataSourceInfo.java b/datavines-connector/datavines-connector-plugins/datavines-connector-impala/src/main/java/io/datavines/connector/plugin/ImpalaDataSourceInfo.java index 04bcdfeb..32fe8d25 100644 --- a/datavines-connector/datavines-connector-plugins/datavines-connector-impala/src/main/java/io/datavines/connector/plugin/ImpalaDataSourceInfo.java +++ b/datavines-connector/datavines-connector-plugins/datavines-connector-impala/src/main/java/io/datavines/connector/plugin/ImpalaDataSourceInfo.java @@ -58,9 +58,4 @@ protected String getSeparator() { return ";"; } - @Override - public String getJdbcUrl() { - return super.getJdbcUrl(); - } - } diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-api/src/main/java/io/datavines/engine/flink/api/stream/FlinkStreamSink.java b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-api/src/main/java/io/datavines/engine/flink/api/stream/FlinkStreamSink.java index df59f4f0..825900aa 100644 --- a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-api/src/main/java/io/datavines/engine/flink/api/stream/FlinkStreamSink.java +++ b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-api/src/main/java/io/datavines/engine/flink/api/stream/FlinkStreamSink.java @@ -22,9 +22,6 @@ import io.datavines.engine.flink.api.FlinkRuntimeEnvironment; public interface FlinkStreamSink extends Component { - - /** - * 输出数据流 - */ - void output(DataStream dataStream, FlinkRuntimeEnvironment environment); + + void output(DataStream dataStream, FlinkRuntimeEnvironment environment) throws Exception; } diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-config/src/main/java/io/datavines/engine/flink/config/BaseFlinkConfigurationBuilder.java b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-config/src/main/java/io/datavines/engine/flink/config/BaseFlinkConfigurationBuilder.java index 4685b8e7..7937463b 100644 --- a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-config/src/main/java/io/datavines/engine/flink/config/BaseFlinkConfigurationBuilder.java +++ b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-config/src/main/java/io/datavines/engine/flink/config/BaseFlinkConfigurationBuilder.java @@ -117,13 +117,13 @@ protected List getSourceConfigs() throws DataVinesException { String tableAlias = getTableAlias(metricInputParameter.get(DATABASE), metricInputParameter.get(SCHEMA), metricInputParameter.get(TABLE), "1"); connectorParameterMap.put(OUTPUT_TABLE, outputTable); connectorParameterMap.put(DRIVER, connectorFactory.getDialect().getDriver()); + connectorParameterMap.put(SRC_CONNECTOR_TYPE, connectorParameter.getType()); table2OutputTable.put(table, outputTable); metricInputParameter.put(TABLE, outputTable); metricInputParameter.put(TABLE_ALIAS, tableAlias); metricInputParameter.put(COLUMN, metricInputParameter.get(COLUMN)); metricInputParameter.put(SRC_CONNECTOR_TYPE, connectorParameter.getType()); -// metricInputParameter.put(ENGINE_TYPE, SPARK); String connectorUUID = connectorFactory.getConnectorParameterConverter().getConnectorUUID(connectorParameterMap); @@ -174,12 +174,12 @@ protected List getSourceConfigs() throws DataVinesException { String tableAlias = getTableAlias(metricInputParameter.get(DATABASE), metricInputParameter.get(SCHEMA), metricInputParameter.get(TABLE), "1"); connectorParameterMap.put(OUTPUT_TABLE, outputTable); connectorParameterMap.put(DRIVER, connectorFactory.getDialect().getDriver()); + connectorParameterMap.put(SRC_CONNECTOR_TYPE, connectorParameter.getType()); metricInputParameter.put(TABLE, outputTable); metricInputParameter.put(TABLE_ALIAS, tableAlias); metricInputParameter.put(COLUMN, metricInputParameter.get(COLUMN)); metricInputParameter.put(SRC_CONNECTOR_TYPE, connectorParameter.getType()); -// metricInputParameter.put(ENGINE_TYPE, SPARK); String connectorUUID = connectorFactory.getConnectorParameterConverter().getConnectorUUID(connectorParameterMap); diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/pom.xml b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/pom.xml index 82bb0638..e2440950 100644 --- a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/pom.xml +++ b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/pom.xml @@ -57,5 +57,11 @@ flink-table-planner_${scala.binary.version} + + io.datavines + datavines-engine-common + ${project.version} + + diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/sink/JdbcSink.java b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/sink/JdbcSink.java index d3433c57..7ba1d1fb 100644 --- a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/sink/JdbcSink.java +++ b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/sink/JdbcSink.java @@ -16,6 +16,9 @@ */ package io.datavines.engine.flink.jdbc.sink; +import io.datavines.common.utils.StringUtils; +import io.datavines.common.utils.ThreadUtils; +import io.datavines.engine.common.utils.ParserUtils; import io.datavines.engine.flink.api.entity.FLinkColumnInfo; import io.datavines.engine.flink.jdbc.utils.FlinkTableUtils; import lombok.extern.slf4j.Slf4j; @@ -30,9 +33,14 @@ import io.datavines.engine.flink.api.FlinkRuntimeEnvironment; import io.datavines.engine.flink.api.stream.FlinkStreamSink; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.SQLException; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.Properties; import java.util.stream.Collectors; import static io.datavines.common.ConfigConstants.*; @@ -46,7 +54,7 @@ public class JdbcSink implements FlinkStreamSink { private final List columns = new ArrayList<>(); @Override - public void output(DataStream dataStream, FlinkRuntimeEnvironment environment) { + public void output(DataStream dataStream, FlinkRuntimeEnvironment environment) throws Exception{ String sql = config.getString(SQL).replace("\\n", " ").replaceAll("/", ""); Table table = environment.getTableEnv().sqlQuery(sql); ResolvedSchema schema = table.getResolvedSchema(); @@ -56,9 +64,10 @@ public void output(DataStream dataStream, FlinkRuntimeEnvironment environme columnInfo.setDataType(column.getDataType().getLogicalType().asSerializableString()); columns.add(columnInfo); }); + checkTableNotExistAndCreate(); - String createTableSql = FlinkTableUtils.generateCreateTableStatement(config.getString(OUTPUT_TABLE), config.getString(TABLE), columns, config); + String createTableSql = FlinkTableUtils.generateCreateTableStatement(config.getString(DATABASE), config.getString(OUTPUT_TABLE), config.getString(TABLE), columns, config); log.info("sink create table sql: {}", createTableSql); environment.getTableEnv().executeSql(createTableSql); table.executeInsert(config.getString(OUTPUT_TABLE)); @@ -101,8 +110,58 @@ public CheckResult checkConfig() { public void prepare(RuntimeEnvironment env) throws Exception { } - private void checkTableNotExistAndCreate() { - // Check if the table exists - // If not, create the table + private void checkTableNotExistAndCreate() throws Exception { + String jdbcUrl = config.getString(URL); + String user = config.getString(USER); + String password = config.getString(PASSWORD); + String query = String.format("SELECT * FROM %s WHERE 1=0", config.getString(DATABASE) + "." + config.getString(TABLE)); + + Properties properties = new Properties(); + properties.setProperty(USER, user); + if (!StringUtils.isEmptyOrNullStr(password)) { + properties.setProperty(PASSWORD, ParserUtils.decode(password)); + } + + String[] url2Array = jdbcUrl.split("\\?"); + String url = url2Array[0]; + if (url2Array.length > 1) { + String[] keyArray = url2Array[1].split("&"); + for (String prop : keyArray) { + String[] values = prop.split("="); + properties.setProperty(values[0], values[1]); + } + } + + boolean tableExists = false; + Connection conn = DriverManager.getConnection(url, properties); + int retryTimes = 3; + while (retryTimes > 0 && !tableExists) { + if (conn != null) { + PreparedStatement statement = null; + try { + statement = conn.prepareStatement(query); + statement.setQueryTimeout(30000); + statement.execute(); + tableExists = true; + } catch (SQLException e) { + log.error("tableExists error : ", e); + retryTimes--; + ThreadUtils.sleep(2000); + } finally { + if (statement != null) { + try { + statement.close(); + } catch (SQLException e){ + log.error("close statement error : ", e); + } + } + } + } + } + + if (!tableExists) { + String createTableSql = FlinkTableUtils.generateCreateTableStatement(config.getString(DATABASE), config.getString(TABLE), columns); + conn.prepareStatement(createTableSql).execute(); + } } } diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/source/JdbcSource.java b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/source/JdbcSource.java index 49ef5669..ceb04e99 100644 --- a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/source/JdbcSource.java +++ b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/source/JdbcSource.java @@ -17,6 +17,7 @@ package io.datavines.engine.flink.jdbc.source; import io.datavines.common.utils.StringUtils; +import io.datavines.engine.common.utils.ParserUtils; import io.datavines.engine.flink.api.entity.FLinkColumnInfo; import io.datavines.engine.flink.jdbc.utils.FlinkTableUtils; import lombok.extern.slf4j.Slf4j; @@ -46,7 +47,7 @@ public class JdbcSource implements FlinkStreamSource { @Override public DataStream getData(FlinkRuntimeEnvironment environment) throws Exception { getRowTypeInfo(config.getString(URL), config.getString(USER), config.getString(PASSWORD), "select * from " + config.getString(TABLE)); - String createTableSql = FlinkTableUtils.generateCreateTableStatement(config.getString(OUTPUT_TABLE), config.getString(TABLE), columns, config); + String createTableSql = FlinkTableUtils.generateCreateTableStatement(config.getString(DATABASE), config.getString(OUTPUT_TABLE), config.getString(TABLE), columns, config); log.info("source create table sql: {}", createTableSql); environment.getTableEnv().executeSql(createTableSql); return null; @@ -56,7 +57,7 @@ private void getRowTypeInfo(String jdbcUrl, String user, String password, String Properties properties = new Properties(); properties.setProperty(USER, user); if (!StringUtils.isEmptyOrNullStr(password)) { - properties.setProperty(PASSWORD, password); + properties.setProperty(PASSWORD, ParserUtils.decode(password)); } String[] url2Array = jdbcUrl.split("\\?"); diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/utils/DatabaseUrlReplacer.java b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/utils/DatabaseUrlReplacer.java new file mode 100644 index 00000000..496fa440 --- /dev/null +++ b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/utils/DatabaseUrlReplacer.java @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.datavines.engine.flink.jdbc.utils; + +import java.util.regex.*; + +public class DatabaseUrlReplacer { + + public static String replaceDatabase(String url, String newDb) throws IllegalArgumentException { + if (url.startsWith("jdbc:mysql:")) { + return processMysql(url, newDb); + } else if (url.startsWith("jdbc:postgresql:")) { + return processPostgresql(url, newDb); + } else if (url.startsWith("jdbc:sqlserver:")) { + return processSqlServer(url, newDb); + } else if (url.startsWith("jdbc:oracle:")) { + return processOracle(url, newDb); + } else if (url.startsWith("jdbc:clickhouse:")) { + return processClickhouse(url, newDb); + } else if (url.startsWith("jdbc:trino:")) { + return processTrino(url, newDb); + } else if (url.startsWith("jdbc:presto:")) { + return processPresto(url, newDb); + } else if (url.startsWith("jdbc:hive2:")) { + return processHive(url, newDb); + } else if (url.startsWith("jdbc:dm:")) { + return processDm(url, newDb); + } else { + return processMysql(url, newDb); + } + } + + private static String processMysql(String url, String newDb) { + Pattern pattern = Pattern.compile("^(jdbc:mysql://[^/]+)(/?[^?]*)?(\\?.*)?$"); + Matcher matcher = pattern.matcher(url); + if (matcher.find()) { + String protocol = matcher.group(1); + String path = matcher.group(2) != null ? matcher.group(2) : ""; + String query = matcher.group(3) != null ? matcher.group(3) : ""; + path = "/" + newDb; + return protocol + path + query; + } + return url; + } + + private static String processPostgresql(String url, String newDb) { + return processMysql(url, newDb); + } + + private static String processSqlServer(String url, String newDb) { + String[] parts = url.split(";", 2); + String hostPart = parts[0]; + String params = parts.length > 1 ? parts[1] : ""; + String[] paramsArray = params.split(";"); + StringBuilder newParams = new StringBuilder(); + for (String param : paramsArray) { + if (param.contains("=")) { + String[] paramParts = param.split("=", 2); + if (paramParts[0].equalsIgnoreCase("databaseName")) { + newParams.append("databaseName=").append(newDb); + } else { + newParams.append(param); + } + } + } + return hostPart + ";" + newParams; + } + + private static String processOracle(String url, String newDb) { + Pattern sidPattern = Pattern.compile("(jdbc:oracle:thin:@[^:]+:\\d+:)(\\w+)"); + Matcher sidMatcher = sidPattern.matcher(url); + if (sidMatcher.find()) { + return sidMatcher.group(1) + newDb; + } + Pattern servicePattern = Pattern.compile("(jdbc:oracle:thin:@//[^/]+/)(\\w+)"); + Matcher serviceMatcher = servicePattern.matcher(url); + if (serviceMatcher.find()) { + return serviceMatcher.group(1) + newDb; + } + return url; + } + + private static String processClickhouse(String url, String newDb) { + Pattern pattern = Pattern.compile("(jdbc:clickhouse://[^/]+)(/[^?]*)?(\\?.*)?"); + Matcher matcher = pattern.matcher(url); + if (matcher.find()) { + String protocol = matcher.group(1); + String path = matcher.group(2) != null ? matcher.group(2) : ""; + String query = matcher.group(3) != null ? matcher.group(3) : ""; + path = "/" + newDb; + query = query.replaceAll("([?&])database=[^&]*", ""); + return protocol + path + query; + } + return url; + } + + private static String processTrino(String url, String newDb) { + return url.replaceAll("(jdbc:trino://[^/]+(/[^/]+)*/)([^/?#]+)", "$1" + newDb); + } + + private static String processPresto(String url, String newDb) { + return url.replaceAll("(jdbc:presto://[^/]+(/[^/]+)*/)([^/?#]+)", "$1" + newDb); + } + + private static String processHive(String url, String newDb) { + return url.replaceAll("(jdbc:hive2://[^/]+/)([^/?]*)", "$1" + newDb); + } + + private static String processDm(String url, String newDb) { + String[] parts = url.split(";", 2); + String hostPart = parts[0]; + String params = parts.length > 1 ? parts[1] : ""; + String[] paramsArray = params.split("\\?"); + StringBuilder newParams = new StringBuilder(); + for (String param : paramsArray) { + if (param.contains("=")) { + String[] paramParts = param.split("=", 2); + if (paramParts[0].equalsIgnoreCase("schema")) { + newParams.append("schema=").append(newDb); + } else { + newParams.append(param); + } + } + } + return hostPart + ";" + newParams; + } +} diff --git a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/utils/FlinkTableUtils.java b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/utils/FlinkTableUtils.java index cc70e6ef..8ab8e1ae 100644 --- a/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/utils/FlinkTableUtils.java +++ b/datavines-engine/datavines-engine-plugins/datavines-engine-flink/datavines-engine-flink-connector-jdbc/src/main/java/io/datavines/engine/flink/jdbc/utils/FlinkTableUtils.java @@ -18,13 +18,14 @@ import io.datavines.common.config.Config; import io.datavines.common.utils.StringUtils; +import io.datavines.engine.common.utils.ParserUtils; import io.datavines.engine.flink.api.entity.FLinkColumnInfo; import java.util.List; public class FlinkTableUtils { - public static String generateCreateTableStatement(String outputTable, String tableName, List columns, Config config) { + public static String generateCreateTableStatement(String database, String outputTable, String tableName, List columns, Config config) { StringBuilder createTableSql = new StringBuilder(); createTableSql.append("CREATE TABLE ").append("`").append(outputTable).append("`").append(" (\n"); @@ -32,14 +33,16 @@ public static String generateCreateTableStatement(String outputTable, String tab createTableSql.append(" `").append(column.getColumn()).append("` ").append(mapDataType(column.getDataType())).append(",\n"); } - // Remove the last comma and newline + String url = config.getString("url"); + url = DatabaseUrlReplacer.replaceDatabase(url, database); + createTableSql.setLength(createTableSql.length() - 2); createTableSql.append("\n) WITH (\n"); createTableSql.append(" 'connector' = 'jdbc',\n"); - createTableSql.append(" 'url' = '").append(config.getString("url")).append("',\n"); + createTableSql.append(" 'url' = '").append(url).append("',\n"); createTableSql.append(" 'table-name' = '").append(tableName).append("',\n"); if (!StringUtils.isEmptyOrNullStr(config.getString("password"))) { - createTableSql.append(" 'password' = '").append(config.getString("password")).append("',\n"); + createTableSql.append(" 'password' = '").append(ParserUtils.decode(config.getString("password"))).append("',\n"); } createTableSql.append(" 'username' = '").append(config.getString("user")).append("'\n"); createTableSql.append(")"); @@ -47,6 +50,21 @@ public static String generateCreateTableStatement(String outputTable, String tab return createTableSql.toString(); } + public static String generateCreateTableStatement(String database, String outputTable, List columns) { + StringBuilder createTableSql = new StringBuilder(); + createTableSql.append("CREATE TABLE ").append(database).append(".").append(outputTable).append(" (\n"); + + for (FLinkColumnInfo column : columns) { + createTableSql.append(" `").append(column.getColumn()).append("` ").append(mapDataTypeToSource(column.getDataType())).append(",\n"); + } + + // Remove the last comma and newline + createTableSql.setLength(createTableSql.length() - 2); + createTableSql.append("\n)"); + + return createTableSql.toString(); + } + private static String mapDataType(String flinkType) { switch (flinkType.toUpperCase().split(" ")[0].split("\\(")[0]) { case "VARCHAR": @@ -85,4 +103,41 @@ private static String mapDataType(String flinkType) { return "STRING"; } } + + private static String mapDataTypeToSource(String flinkType) { + switch (flinkType.toUpperCase().split(" ")[0].split("\\(")[0]) { + case "STRING": + return "TEXT"; + case "INT": + case "INTEGER": + case "SMALLINT": + case "TINYINT": + return "INT"; + case "BIGINT": + return "BIGINT"; + case "DOUBLE": + case "FLOAT": + case "REAL": + return "DOUBLE"; + case "DECIMAL": + case "NUMERIC": + return "DECIMAL"; + case "DATE": + return "DATE"; + case "TIME": + return "TIME"; + case "DATETIME": + case "TIMESTAMP": + return "DATETIME"; + case "BOOLEAN": + case "BIT": + return "BOOLEAN"; + case "BINARY": + case "VARBINARY": + case "LONGVARBINARY": + return "BYTES"; + default: + return "TEXT"; + } + } } diff --git a/scripts/sql/datavines-mysql.sql b/scripts/sql/datavines-mysql.sql index 2faf8d61..a820c458 100644 --- a/scripts/sql/datavines-mysql.sql +++ b/scripts/sql/datavines-mysql.sql @@ -885,6 +885,7 @@ INSERT INTO `dv_config` VALUES ('27', '-1', 'spark.engine.parameter.driver.memor INSERT INTO `dv_config` VALUES ('28', '-1', 'spark.engine.parameter.executor.cores', '1', '1', '1', '2023-09-05 21:02:38', '1', '2023-09-05 21:02:38'); INSERT INTO `dv_config` VALUES ('29', '-1', 'spark.engine.parameter.executor.memory', '512M', '1', '1', '2023-09-05 21:02:38', '1', '2023-09-05 21:02:38'); INSERT INTO `dv_config` VALUES ('30', '-1', 'datavines.fqdn', 'http://127.0.0.1:5600', '1', '1', '2024-05-21 15:15:38', '1', '2024-05-21 15:15:38'); +INSERT INTO `dv_config` VALUES ('31', '-1', 'data.quality.flink.jar.name', '/libs/datavines-engine-flink-core-1.0.0-SNAPSHOT.jar', '1', '1', '2025-02-02 11:43:04', '1', '2025-02-02 11:43:04'); INSERT INTO `dv_user` (`id`, `username`, `password`, `email`, `phone`, `admin`) VALUES ('1', 'admin', '$2a$10$9ZcicUYFl/.knBi9SE53U.Nml8bfNeArxr35HQshxXzimbA6Ipgqq', 'admin@gmail.com', NULL, '0'); INSERT INTO `dv_workspace` (`id`, `name`, `create_by`, `update_by`) VALUES ('1', "admin\'s default", '1', '1');