This is an automated email from the ASF dual-hosted git repository.
xxyu pushed a commit to branch kylin5
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit ca4f1d6000220ca8fa63ac4b252881aa517e4429
Author: Jiawei Li <1019037...@qq.com>
AuthorDate: Tue Jan 3 19:01:23 2023 +0800
KYLIN-5445 set epoch_target as primary key of epoch table
* KYLIN-5445 set epoch_target as primary key of epoch table
* KYLIN-5445 minor fix default value
* KYLIN-5445 minor add ut
* KYLIN-5445 minor fix sonar
---
.../persistence/metadata/JdbcEpochStore.java | 32 ++++++++++++++++++++--
.../common/persistence/metadata/jdbc/JdbcUtil.java | 24 ++++++++++++++++
.../resources/metadata-jdbc-default.properties | 4 +--
.../src/main/resources/metadata-jdbc-h2.properties | 4 +--
.../main/resources/metadata-jdbc-mysql.properties | 4 +--
.../resources/metadata-jdbc-postgresql.properties | 4 +--
.../metadata/epochstore/JdbcEpochStoreTest.java | 15 ++++++++++
7 files changed, 76 insertions(+), 11 deletions(-)
diff --git
a/src/core-common/src/main/java/org/apache/kylin/common/persistence/metadata/JdbcEpochStore.java
b/src/core-common/src/main/java/org/apache/kylin/common/persistence/metadata/JdbcEpochStore.java
index bb444822ac..7faa05f117 100644
---
a/src/core-common/src/main/java/org/apache/kylin/common/persistence/metadata/JdbcEpochStore.java
+++
b/src/core-common/src/main/java/org/apache/kylin/common/persistence/metadata/JdbcEpochStore.java
@@ -19,11 +19,13 @@ package org.apache.kylin.common.persistence.metadata;
import static
org.apache.kylin.common.exception.CommonErrorCode.FAILED_UPDATE_METADATA;
import static
org.apache.kylin.common.persistence.metadata.jdbc.JdbcUtil.datasourceParameters;
+import static
org.apache.kylin.common.persistence.metadata.jdbc.JdbcUtil.isPrimaryKeyExists;
import static
org.apache.kylin.common.persistence.metadata.jdbc.JdbcUtil.isTableExists;
import static
org.apache.kylin.common.persistence.metadata.jdbc.JdbcUtil.withTransaction;
import static
org.apache.kylin.common.persistence.metadata.jdbc.JdbcUtil.withTransactionTimeout;
import java.io.InputStream;
+import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
@@ -32,6 +34,8 @@ import java.util.Locale;
import java.util.Objects;
import java.util.Properties;
+import javax.sql.DataSource;
+
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.dbcp2.BasicDataSource;
import org.apache.kylin.common.KylinConfig;
@@ -62,6 +66,8 @@ public class JdbcEpochStore extends EpochStore {
static final String MAINTENANCE_MODE_REASON = "maintenance_mode_reason";
static final String MVCC = "mvcc";
+ static final String ADD_PRIMARY_KEY_SQL = "alter table %s ADD PRIMARY
KEY(" + EPOCH_TARGET + ")";
+
static final String INSERT_SQL = "insert into %s (" +
Joiner.on(",").join(EPOCH_ID, EPOCH_TARGET,
CURRENT_EPOCH_OWNER, LAST_EPOCH_RENEW_TIME, SERVER_MODE,
MAINTENANCE_MODE_REASON, MVCC)
+ ") values (?, ?, ?, ?, ?, ?, ?)";
@@ -101,12 +107,32 @@ public class JdbcEpochStore extends EpochStore {
public static String getEpochSql(String sql, String tableName) {
return String.format(Locale.ROOT, sql, tableName, EPOCH_ID,
EPOCH_TARGET, CURRENT_EPOCH_OWNER,
- LAST_EPOCH_RENEW_TIME, SERVER_MODE, MAINTENANCE_MODE_REASON,
MVCC, tableName, EPOCH_TARGET,
- EPOCH_TARGET);
+ LAST_EPOCH_RENEW_TIME, SERVER_MODE, MAINTENANCE_MODE_REASON,
MVCC, EPOCH_TARGET);
+ }
+
+ public static String getAddPrimarykeySql(String tableName) {
+ return String.format(Locale.ROOT, ADD_PRIMARY_KEY_SQL, tableName);
+
}
+ private Connection getConnection(JdbcTemplate jdbcTemplate) throws
SQLException {
+ DataSource dataSource = jdbcTemplate.getDataSource();
+ if (dataSource == null) {
+ return null;
+ }
+ return dataSource.getConnection();
+ }
+
+ @Override
public void createIfNotExist() throws Exception {
- if (isTableExists(jdbcTemplate.getDataSource().getConnection(),
table)) {
+ if (isTableExists(getConnection(jdbcTemplate), table)) {
+ if (!isPrimaryKeyExists(getConnection(jdbcTemplate), table)) {
+ withTransaction(transactionManager, () -> {
+ jdbcTemplate.execute(getAddPrimarykeySql(table));
+ return 1;
+ });
+ log.info("Succeed to add table primary key: {}", table);
+ }
return;
}
String fileName = "metadata-jdbc-default.properties";
diff --git
a/src/core-common/src/main/java/org/apache/kylin/common/persistence/metadata/jdbc/JdbcUtil.java
b/src/core-common/src/main/java/org/apache/kylin/common/persistence/metadata/jdbc/JdbcUtil.java
index 8d4d54873c..a4bc8e4221 100644
---
a/src/core-common/src/main/java/org/apache/kylin/common/persistence/metadata/jdbc/JdbcUtil.java
+++
b/src/core-common/src/main/java/org/apache/kylin/common/persistence/metadata/jdbc/JdbcUtil.java
@@ -118,6 +118,30 @@ public class JdbcUtil {
return false;
}
+ public static boolean isPrimaryKeyExists(Connection conn, String table)
throws SQLException {
+ return isPrimaryKeyExists(conn, table, table.toUpperCase(Locale.ROOT),
table.toLowerCase(Locale.ROOT));
+ }
+
+ private static boolean isPrimaryKeyExists(Connection conn, String...
tables) throws SQLException {
+ try {
+ for (String table : tables) {
+ val resultSet =
conn.getMetaData().getPrimaryKeys(conn.getCatalog(), conn.getSchema(), table);
+ if (resultSet.next()) {
+ return true;
+ }
+ }
+
+ return false;
+ } catch (Exception e) {
+ logger.error("Fail to know if table {} primary key exists",
tables, e);
+ } finally {
+ if (!conn.isClosed()) {
+ conn.close();
+ }
+ }
+ return true;
+ }
+
public static boolean isIndexExists(Connection conn, String table, String
index) throws SQLException {
return isIndexExists(conn, index, table,
table.toUpperCase(Locale.ROOT), table.toLowerCase(Locale.ROOT));
}
diff --git
a/src/core-common/src/main/resources/metadata-jdbc-default.properties
b/src/core-common/src/main/resources/metadata-jdbc-default.properties
index 35879f9e34..09f0bfea78 100644
--- a/src/core-common/src/main/resources/metadata-jdbc-default.properties
+++ b/src/core-common/src/main/resources/metadata-jdbc-default.properties
@@ -36,7 +36,7 @@ create.auditlog.store.table=create table if not exists %s ( \
create.epoch.store.table=create table if not exists %s ( \
%s int null, \
- %s varchar(255) null, \
+ %s varchar(255), \
%s varchar(2000) null, \
%s bigint null, \
%s varchar(10) null, \
@@ -45,5 +45,5 @@ create.epoch.store.table=create table if not exists %s ( \
`reserved_field_1` VARCHAR(50), \
`reserved_field_2` longblob, \
`reserved_field_3` longblob, \
- constraint %s_%s_uindex unique (%s) \
+ primary key(%s) \
);
\ No newline at end of file
diff --git a/src/core-common/src/main/resources/metadata-jdbc-h2.properties
b/src/core-common/src/main/resources/metadata-jdbc-h2.properties
index 45ad513c27..77328950c1 100644
--- a/src/core-common/src/main/resources/metadata-jdbc-h2.properties
+++ b/src/core-common/src/main/resources/metadata-jdbc-h2.properties
@@ -102,7 +102,7 @@ create.rawrecommendation.store.index=
create.epoch.store.table=create table if not exists %s ( \
%s int null, \
- %s varchar(255) null, \
+ %s varchar(255), \
%s varchar(2000) null, \
%s bigint null, \
%s varchar(10) null, \
@@ -111,7 +111,7 @@ create.epoch.store.table=create table if not exists %s ( \
`reserved_field_1` VARCHAR(50), \
`reserved_field_2` longblob, \
`reserved_field_3` longblob, \
- constraint %s_%s_uindex unique (%s) \
+ primary key(%s) \
);
#### JDBC STREAMING JOB STATS STORE
diff --git a/src/core-common/src/main/resources/metadata-jdbc-mysql.properties
b/src/core-common/src/main/resources/metadata-jdbc-mysql.properties
index 2f855a8eb6..6a2df4c3b1 100644
--- a/src/core-common/src/main/resources/metadata-jdbc-mysql.properties
+++ b/src/core-common/src/main/resources/metadata-jdbc-mysql.properties
@@ -174,7 +174,7 @@ create.rawrecommendation.store.index=ALTER TABLE %s ADD
UNIQUE %s_idx (project,
create.epoch.store.table=create table if not exists %s ( \
%s int null, \
- %s varchar(255) null, \
+ %s varchar(255), \
%s varchar(2000) null, \
%s bigint null, \
%s varchar(10) null, \
@@ -183,7 +183,7 @@ create.epoch.store.table=create table if not exists %s ( \
`reserved_field_1` VARCHAR(50), \
`reserved_field_2` longblob, \
`reserved_field_3` longblob, \
- constraint %s_%s_uindex unique (%s) \
+ primary key(%s) \
) ENGINE=INNODB DEFAULT CHARSET=utf8;
### jdbc distributed lock
diff --git
a/src/core-common/src/main/resources/metadata-jdbc-postgresql.properties
b/src/core-common/src/main/resources/metadata-jdbc-postgresql.properties
index a8b8839fc3..a26f852134 100644
--- a/src/core-common/src/main/resources/metadata-jdbc-postgresql.properties
+++ b/src/core-common/src/main/resources/metadata-jdbc-postgresql.properties
@@ -173,7 +173,7 @@ create.rawrecommendation.store.index=CREATE UNIQUE INDEX
%s_idx ON %s using btre
create.epoch.store.table=create table if not exists %s ( \
%s int null, \
- %s varchar(255) null, \
+ %s varchar(255), \
%s varchar(2000) null, \
%s bigint null, \
%s varchar(10) null, \
@@ -182,7 +182,7 @@ create.epoch.store.table=create table if not exists %s ( \
reserved_field_1 VARCHAR(50), \
reserved_field_2 bytea, \
reserved_field_3 bytea, \
- constraint %s_%s_uindex unique (%s) \
+ primary key(%s) \
);
### jdbc distributed lock
diff --git
a/src/core-common/src/test/java/org/apache/kylin/common/persistence/metadata/epochstore/JdbcEpochStoreTest.java
b/src/core-common/src/test/java/org/apache/kylin/common/persistence/metadata/epochstore/JdbcEpochStoreTest.java
index d51c15ec66..5b30affd7a 100644
---
a/src/core-common/src/test/java/org/apache/kylin/common/persistence/metadata/epochstore/JdbcEpochStoreTest.java
+++
b/src/core-common/src/test/java/org/apache/kylin/common/persistence/metadata/epochstore/JdbcEpochStoreTest.java
@@ -18,9 +18,12 @@
package org.apache.kylin.common.persistence.metadata.epochstore;
import static
org.apache.kylin.common.persistence.metadata.jdbc.JdbcUtil.datasourceParameters;
+import static
org.apache.kylin.common.persistence.metadata.jdbc.JdbcUtil.isPrimaryKeyExists;
import static org.apache.kylin.common.util.TestUtils.getTestConfig;
import static org.awaitility.Awaitility.await;
+import java.sql.Connection;
+import java.util.Locale;
import java.util.concurrent.TimeUnit;
import org.apache.commons.dbcp2.BasicDataSourceFactory;
@@ -59,6 +62,18 @@ public final class JdbcEpochStoreTest extends
AbstractEpochStoreTest {
return new JdbcTemplate(dataSource);
}
+ @Test
+ void testAddPrimaryKey() throws Exception {
+ val jdbcTemplate = getJdbcTemplate();
+ String table = getTestConfig().getMetadataUrl().getIdentifier() +
"_epoch";
+ jdbcTemplate.execute(String.format(Locale.ROOT, "alter table %s drop
primary key", table));
+ Connection conn = jdbcTemplate.getDataSource().getConnection();
+ assert !isPrimaryKeyExists(conn, table);
+ epochStore = getEpochStore();
+ conn = getJdbcTemplate().getDataSource().getConnection();
+ assert isPrimaryKeyExists(conn, table);
+ }
+
@Test
void testExecuteWithTransaction_RollBack() {
