jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

add parallel processing for metastore

This commit is contained in:
shuaiwang999 2021-02-09 15:46:33 +08:00
parent c9fd61edb1
commit 469ac921b6
10 changed files with 845 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
hetu-metastore/src/main/java/io/hetu/core/metastore/HetuMetastoreCache.java
View File

@ -63,6 +63,17 @@ public class HetuMetastoreCache
}
}
@Override
public void createCatalogIfNotExist(CatalogEntity catalog)
{
try {
delegate.createCatalogIfNotExist(catalog);
}
finally {
catalogsCache.invalidateAll();
}
}
@Override
public void alterCatalog(String catalogName, CatalogEntity newCatalog)
{
@ -128,6 +139,17 @@ public class HetuMetastoreCache
}
}
@Override
public void createDatabaseIfNotExist(DatabaseEntity database)
{
try {
delegate.createDatabaseIfNotExist(database);
}
finally {
databasesCache.invalidate(database.getCatalogName());
}
}
@Override
public void alterDatabase(String catalogName, String databaseName, DatabaseEntity newDatabase)
{
@ -197,6 +219,18 @@ public class HetuMetastoreCache
}
}
@Override
public void createTableIfNotExist(TableEntity table)
{
try {
delegate.createTableIfNotExist(table);
}
finally {
String key = table.getCatalogName() + "." + table.getDatabaseName();
tablesCache.invalidate(key);
}
}
@Override
public void dropTable(String catalogName, String databaseName, String tableName)
{

42
hetu-metastore/src/main/java/io/hetu/core/metastore/hetufilesystem/HetuFsMetastore.java
View File

@ -168,6 +168,20 @@ public class HetuFsMetastore
});
}
@Override
public void createCatalogIfNotExist(CatalogEntity catalog)
{
try {
createCatalog(catalog);
}
catch (PrestoException e) {
Optional<CatalogEntity> existedCatalog = getCatalog(catalog.getName());
if (!(existedCatalog.isPresent() && existedCatalog.get().equals(catalog))) {
throw new PrestoException(HETU_METASTORE_CODE, e.getMessage(), e.getCause());
}
}
}
@Override
public void alterCatalog(String catalogName, CatalogEntity newCatalog)
{
@ -330,6 +344,20 @@ public class HetuFsMetastore
});
}
@Override
public void createDatabaseIfNotExist(DatabaseEntity database)
{
try {
createDatabase(database);
}
catch (PrestoException e) {
Optional<DatabaseEntity> existedDatabase = getDatabase(database.getCatalogName(), database.getName());
if (!(existedDatabase.isPresent() && existedDatabase.get().equals(database))) {
throw new PrestoException(HETU_METASTORE_CODE, e.getMessage(), e.getCause());
}
}
}
@Override
public void alterDatabase(String catalogName, String databaseName, DatabaseEntity newDatabase)
{
@ -531,6 +559,20 @@ public class HetuFsMetastore
});
}
@Override
public void createTableIfNotExist(TableEntity table)
{
try {
createTable(table);
}
catch (PrestoException e) {
Optional<TableEntity> existedTable = getTable(table.getCatalogName(), table.getDatabaseName(), table.getName());
if (!(existedTable.isPresent() && existedTable.get().equals(table))) {
throw new PrestoException(HETU_METASTORE_CODE, e.getMessage(), e.getCause());
}
}
}
@Override
public void dropTable(String catalogName, String databaseName, String tableName)
{

106
hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcBasedLock.java Normal file
View File

@ -0,0 +1,106 @@
/*
* Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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.hetu.core.metastore.jdbc;
import io.airlift.log.Logger;
import io.prestosql.spi.PrestoException;
import org.jdbi.v3.core.Jdbi;
import static io.hetu.core.metastore.jdbc.JdbcMetadataUtil.onDemand;
import static io.prestosql.spi.metastore.HetuErrorCode.HETU_METASTORE_CODE;
import static java.lang.String.format;
public class JdbcBasedLock
{
private static final Logger LOG = Logger.get(JdbcBasedLock.class);
public static final long LOCK_RETRY_COUNT = 20;
public static final long LOCK_TIMEOUT_COUNT = 10;
public static final long RETRY_INTERVAL = 500;
private static final int UNLOCK_RETRY_COUNT = 3;
private final JdbcMetadataDao dao;
public JdbcBasedLock(Jdbi jdbi)
{
this.dao = onDemand(jdbi, JdbcMetadataDao.class);
}
public void lock()
{
int count = 1;
int lockCount = 1;
Long lockId = Long.MIN_VALUE;
while (count <= LOCK_RETRY_COUNT) {
try {
dao.tryLock();
return;
}
catch (PrestoException e) {
// retry timeout
if (count == LOCK_RETRY_COUNT) {
throw new PrestoException(HETU_METASTORE_CODE,
format("After reaching the maximum %s retries, get lock failed.", LOCK_RETRY_COUNT), e);
}
LOG.debug("Failed to get lock. Will retry again in %s milliseconds. Exception: %s", RETRY_INTERVAL, e);
try {
Thread.sleep(RETRY_INTERVAL);
}
catch (InterruptedException e1) {
throw new RuntimeException(e1);
}
// check lock expired
Long id = dao.getLockId();
if (id != null && id > lockId) {
lockId = id;
lockCount = 1;
}
if (lockCount > LOCK_TIMEOUT_COUNT) {
unlock();
}
}
count++;
lockCount++;
}
}
public void unlock()
{
int count = 1;
while (count <= UNLOCK_RETRY_COUNT) {
try {
dao.releaseLock();
return;
}
catch (PrestoException e) {
if (count == UNLOCK_RETRY_COUNT) {
throw new PrestoException(HETU_METASTORE_CODE, e.getMessage(), e);
}
LOG.debug("Failed to release lock. Will retry again in %s milliseconds. Exception: %s", RETRY_INTERVAL, e);
try {
Thread.sleep(RETRY_INTERVAL);
}
catch (InterruptedException e1) {
throw new RuntimeException(e1);
}
count++;
}
}
}
}

61
hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcHetuMetastore.java
View File

@ -36,6 +36,7 @@ import static com.google.common.collect.ImmutableList.toImmutableList;
import static com.google.common.collect.MoreCollectors.toOptional;
import static io.hetu.core.metastore.jdbc.JdbcMetadataUtil.onDemand;
import static io.hetu.core.metastore.jdbc.JdbcMetadataUtil.runTransaction;
import static io.hetu.core.metastore.jdbc.JdbcMetadataUtil.runTransactionWithLock;
import static io.prestosql.spi.metastore.HetuErrorCode.HETU_METASTORE_CODE;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
@ -71,7 +72,7 @@ public class JdbcHetuMetastore
@Override
public void createCatalog(CatalogEntity catalog)
{
runTransaction(jdbi, handle -> {
runTransactionWithLock(jdbi, handle -> {
JdbcMetadataDao transactionDao = handle.attach(JdbcMetadataDao.class);
long catalogId = transactionDao.insertCatalog(catalog);
Optional<List<PropertyEntity>> properties = mapToList(catalog.getParameters());
@ -79,6 +80,20 @@ public class JdbcHetuMetastore
});
}
@Override
public void createCatalogIfNotExist(CatalogEntity catalog)
{
try {
createCatalog(catalog);
}
catch (PrestoException e) {
Optional<CatalogEntity> existedCatalog = getCatalog(catalog.getName());
if (!(existedCatalog.isPresent() && existedCatalog.get().equals(catalog))) {
throw new PrestoException(HETU_METASTORE_CODE, e.getMessage(), e.getCause());
}
}
}
@Override
public void alterCatalog(String catalogName, CatalogEntity newCatalog)
{
@ -86,7 +101,7 @@ public class JdbcHetuMetastore
throw new PrestoException(HETU_METASTORE_CODE, "Cannot alter a catalog's name");
}
runTransaction(jdbi, handle -> {
runTransactionWithLock(jdbi, handle -> {
JdbcMetadataDao transactionDao = handle.attach(JdbcMetadataDao.class);
Long catalogId = getCatalogId(transactionDao, catalogName);
@ -103,7 +118,7 @@ public class JdbcHetuMetastore
@Override
public void dropCatalog(String catalogName)
{
runTransaction(jdbi, handle -> {
runTransactionWithLock(jdbi, handle -> {
JdbcMetadataDao transactionDao = handle.attach(JdbcMetadataDao.class);
Long catalogId = getCatalogId(transactionDao, catalogName);
@ -142,7 +157,7 @@ public class JdbcHetuMetastore
@Override
public void createDatabase(DatabaseEntity database)
{
runTransaction(jdbi, handle -> {
runTransactionWithLock(jdbi, handle -> {
JdbcMetadataDao transactionDao = handle.attach(JdbcMetadataDao.class);
checkCatalogExists(transactionDao, database.getCatalogName());
long databaseId = transactionDao.insertDatabase(database);
@ -151,6 +166,20 @@ public class JdbcHetuMetastore
});
}
@Override
public void createDatabaseIfNotExist(DatabaseEntity database)
{
try {
createDatabase(database);
}
catch (PrestoException e) {
Optional<DatabaseEntity> existedDatabase = getDatabase(database.getCatalogName(), database.getName());
if (!(existedDatabase.isPresent() && existedDatabase.get().equals(database))) {
throw new PrestoException(HETU_METASTORE_CODE, e.getMessage(), e.getCause());
}
}
}
@Override
public void alterDatabase(String catalogName, String databaseName, DatabaseEntity newDatabase)
{
@ -159,7 +188,7 @@ public class JdbcHetuMetastore
format("alter database cannot cross catalog[%s,%s]", catalogName, newDatabase.getCatalogName()));
}
runTransaction(jdbi, handle -> {
runTransactionWithLock(jdbi, handle -> {
JdbcMetadataDao transactionDao = handle.attach(JdbcMetadataDao.class);
Long databaseId = transactionDao.getDatabaseId(catalogName, databaseName);
@ -175,7 +204,7 @@ public class JdbcHetuMetastore
@Override
public void dropDatabase(String catalogName, String databaseName)
{
runTransaction(jdbi, handle -> {
runTransactionWithLock(jdbi, handle -> {
JdbcMetadataDao transactionDao = handle.attach(JdbcMetadataDao.class);
Long databaseId = getDatabaseId(catalogName, databaseName, transactionDao);
transactionDao.dropDatabaseProperty(databaseId);
@ -199,7 +228,7 @@ public class JdbcHetuMetastore
@Override
public void createTable(TableEntity table)
{
runTransaction(jdbi, handle -> {
runTransactionWithLock(jdbi, handle -> {
JdbcMetadataDao transactionDao = handle.attach(JdbcMetadataDao.class);
Long databaseId = getDatabaseId(table.getCatalogName(), table.getDatabaseName(), transactionDao);
@ -212,10 +241,24 @@ public class JdbcHetuMetastore
});
}
@Override
public void createTableIfNotExist(TableEntity table)
{
try {
createTable(table);
}
catch (PrestoException e) {
Optional<TableEntity> existedTable = getTable(table.getCatalogName(), table.getDatabaseName(), table.getName());
if (!(existedTable.isPresent() && existedTable.get().equals(table))) {
throw new PrestoException(HETU_METASTORE_CODE, e.getMessage(), e.getCause());
}
}
}
@Override
public void dropTable(String catalogName, String databaseName, String tableName)
{
runTransaction(jdbi, handle -> {
runTransactionWithLock(jdbi, handle -> {
JdbcMetadataDao transactionDao = handle.attach(JdbcMetadataDao.class);
// get table id
Long tableId = transactionDao.getTableId(catalogName, databaseName, tableName);
@ -237,7 +280,7 @@ public class JdbcHetuMetastore
@Override
public void alterTable(String catalogName, String databaseName, String oldTableName, TableEntity newTable)
{
runTransaction(jdbi, handle -> {
runTransactionWithLock(jdbi, handle -> {
JdbcMetadataDao transactionDao = handle.attach(JdbcMetadataDao.class);
// get table id

23
hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataDao.java
View File

@ -444,4 +444,27 @@ public interface JdbcMetadataDao
*/
@SqlUpdate("DELETE FROM hetu_tab_cols WHERE table_id = :tableId")
int dropColumn(@Bind("tableId") long tableId);
/**
* get lock
*/
@SqlUpdate("INSERT INTO hetu_tab_lock (\n"
+ " resource, description)\n"
+ "VALUES (\n"
+ " 1, 'lock')")
void tryLock();
/**
* release lock
*/
@SqlUpdate("DELETE FROM hetu_tab_lock WHERE resource=1")
void releaseLock();
/**
* get lock id
*
* @return lock id
*/
@SqlQuery("SELECT id FROM hetu_tab_lock WHERE resource=1")
Long getLockId();
}

26
hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataUtil.java
View File

@ -129,6 +129,30 @@ public class JdbcMetadataUtil
}
}
/**
* run the transaction with lock
*
* @param jdbi jdbi
* @param callback callback
*/
public static void runTransactionWithLock(Jdbi jdbi, HandleConsumer<PrestoException> callback)
{
JdbcBasedLock jdbcLock = new JdbcBasedLock(jdbi);
try {
jdbcLock.lock();
jdbi.useTransaction(callback);
}
catch (JdbiException e) {
if (e.getCause() != null) {
throwIfInstanceOf(e.getCause(), PrestoException.class);
}
throw new PrestoException(HETU_METASTORE_CODE, "Hetu metastore operation failed.", e);
}
finally {
jdbcLock.unlock();
}
}
/**
* create all metadata tables of hetu metastore
*
@ -152,5 +176,7 @@ public class JdbcMetadataUtil
tableDao.createTableColumns();
// hetu_column_params table
tableDao.createTableColumnParameters();
// hetu_tab_lock table
tableDao.createTableLock();
}
}

12
hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/MetadataTableDao.java
View File

@ -132,4 +132,16 @@ public interface MetadataTableDao
+ " FOREIGN KEY (column_id) REFERENCES hetu_tab_cols (id)\n"
+ ")")
void createTableColumnParameters();
/**
* create table of lock
*/
@SqlUpdate("CREATE TABLE IF NOT EXISTS hetu_tab_lock (\n"
+ "id BIGINT NOT NULL AUTO_INCREMENT,\n"
+ "resource INT NOT NULL,\n"
+ "description VARCHAR(128) NOT NULL,\n"
+ "PRIMARY KEY (id),\n"
+ "UNIQUE (resource)\n"
+ ")")
void createTableLock();
}

265
hetu-metastore/src/test/java/io/hetu/core/metastore/hetufilesystem/TestHetuFsMetastore.java
View File

@ -78,6 +78,8 @@ public class TestHetuFsMetastore
private final String typeInt = "bigint";
private final String typeVarchar = "array(varchar(32))";
private boolean testResult = true;
/**
* setUp
*
@ -181,6 +183,44 @@ public class TestHetuFsMetastore
metastore.dropCatalog(vdm1.getName());
}
/**
* test create catalog parallel
*/
@Test
public void testCreateCatalogParallel()
throws InterruptedException
{
Map<String, String> properties = ImmutableMap.<String, String>builder().build();
CatalogEntity catalog = CatalogEntity.builder()
.setCatalogName("catalog100")
.setOwner("root1")
.setComment(Optional.of("create catalog parallel"))
.setParameters(properties)
.setCreateTime(System.currentTimeMillis())
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.createCatalogIfNotExist(catalog);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropCatalog(catalog.getName());
}
/**
* test drop dropCatalog
*/
@ -370,6 +410,53 @@ public class TestHetuFsMetastore
assertEquals(catalog7.get().getCreateTime(), catalog5.getCreateTime());
}
/**
* test alter catalog parallel
*/
@Test
public void testAlterCatalogParallel()
throws InterruptedException
{
Map<String, String> properties = ImmutableMap.<String, String>builder().build();
CatalogEntity catalog = CatalogEntity.builder()
.setCatalogName("catalog200")
.setOwner("root1")
.setComment(Optional.of("create catalog"))
.setParameters(properties)
.setCreateTime(System.currentTimeMillis())
.build();
metastore.createCatalog(catalog);
CatalogEntity newCatalog = CatalogEntity.builder()
.setCatalogName("catalog200")
.setOwner("root9")
.setComment(Optional.of("alter catalog parallel"))
.setParameters(properties)
.setCreateTime(System.currentTimeMillis())
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.alterCatalog(catalog.getName(), newCatalog);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropCatalog(catalog.getName());
}
/**
* test create database
*/
@ -410,6 +497,48 @@ public class TestHetuFsMetastore
}
}
/**
* test create database parallel
*/
@Test
public void testCreateDatabaseParallel()
throws InterruptedException
{
Map<String, String> properties = ImmutableMap.<String, String>builder()
.put("desc", "vschema")
.build();
DatabaseEntity database = DatabaseEntity.builder()
.setCatalogName(defaultCatalog.getName())
.setDatabaseName("database100")
.setOwner("root9")
.setComment(Optional.of("Hetu create database."))
.setCreateTime(System.currentTimeMillis())
.setParameters(properties)
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.createDatabaseIfNotExist(database);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropDatabase(database.getCatalogName(), database.getName());
}
/**
* test drop database
*/
@ -617,6 +746,58 @@ public class TestHetuFsMetastore
}
}
/**
* test alter database parallel
*/
@Test
public void testAlterDatabaseParallel()
throws InterruptedException
{
Map<String, String> properties = ImmutableMap.<String, String>builder()
.put("desc", "vschema")
.build();
DatabaseEntity database = DatabaseEntity.builder()
.setCatalogName(defaultCatalog.getName())
.setDatabaseName("database200")
.setOwner("root9")
.setComment(Optional.of("Hetu create database."))
.setCreateTime(System.currentTimeMillis())
.setParameters(properties)
.build();
metastore.createDatabase(database);
DatabaseEntity newDatabase = DatabaseEntity.builder()
.setCatalogName(defaultCatalog.getName())
.setDatabaseName("database200")
.setOwner("root10")
.setComment(Optional.of("alter database parallel"))
.setCreateTime(System.currentTimeMillis())
.setParameters(properties)
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.alterDatabase(database.getCatalogName(), database.getName(), newDatabase);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropDatabase(database.getCatalogName(), database.getName());
}
/**
* test get table
*/
@ -780,6 +961,44 @@ public class TestHetuFsMetastore
}
}
/**
* test create table parallel
*/
@Test
public void testCreateTableParallel()
throws InterruptedException
{
String tableName = "table100";
SchemaTableName schemaTableName = new SchemaTableName(defaultDatabase.getName(), tableName);
TableEntity tableEntity = TableEntity.builder()
.setCatalogName(defaultDatabase.getCatalogName())
.setDatabaseName(defaultDatabase.getName())
.setTableName(schemaTableName.getTableName())
.setTableType(TableEntityType.TABLE.toString())
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.createTableIfNotExist(tableEntity);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropTable(tableEntity.getCatalogName(), tableEntity.getDatabaseName(), tableName);
}
/**
* test drop table
*/
@ -864,6 +1083,52 @@ public class TestHetuFsMetastore
assertTableEquals(newTable1.get(), newTable);
}
/**
* test alter table parallel
*/
@Test
public void testAlterTableParallel()
throws InterruptedException
{
String tableName = "table200";
TableEntity tableEntity1 = TableEntity.builder()
.setCatalogName(defaultDatabase.getCatalogName())
.setDatabaseName(defaultDatabase.getName())
.setTableName(tableName)
.setTableType(TableEntityType.TABLE.toString())
.build();
metastore.createTable(tableEntity1);
TableEntity tableEntity2 = TableEntity.builder()
.setCatalogName(defaultDatabase.getCatalogName())
.setDatabaseName(defaultDatabase.getName())
.setTableName(tableName)
.setTableType(TableEntityType.TABLE.toString())
.setComment("alter table parallel")
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.alterTable(defaultDatabase.getCatalogName(), defaultDatabase.getName(), tableName, tableEntity2);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropTable(defaultDatabase.getCatalogName(), defaultDatabase.getName(), tableName);
}
/**
* test alter view
*/

264
hetu-metastore/src/test/java/io/hetu/core/metastore/jdbc/mysql/TestJdbcHetuMetastore.java
View File

@ -78,6 +78,8 @@ public class TestJdbcHetuMetastore
private static String user;
private static String password;
private boolean testResult = true;
static {
MetastoreUtFileLoader metastoreUtFileLoader = MetastoreUtFileLoader.getInstance();
@ -186,6 +188,44 @@ public class TestJdbcHetuMetastore
metastore.dropCatalog(vdm1.getName());
}
/**
* test create catalog parallel
*/
@Test
public void testCreateCatalogParallel()
throws InterruptedException
{
Map<String, String> properties = ImmutableMap.<String, String>builder().build();
CatalogEntity catalog = CatalogEntity.builder()
.setCatalogName("catalog100")
.setOwner("root1")
.setComment(Optional.of("create catalog parallel"))
.setParameters(properties)
.setCreateTime(System.currentTimeMillis())
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.createCatalogIfNotExist(catalog);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropCatalog(catalog.getName());
}
/**
* test drop dropCatalog
*/
@ -365,6 +405,53 @@ public class TestJdbcHetuMetastore
assertEquals(catalog7.get().getCreateTime(), catalog5.getCreateTime());
}
/**
* test alter catalog parallel
*/
@Test
public void testAlterCatalogParallel()
throws InterruptedException
{
Map<String, String> properties = ImmutableMap.<String, String>builder().build();
CatalogEntity catalog = CatalogEntity.builder()
.setCatalogName("catalog100")
.setOwner("root1")
.setComment(Optional.of("create catalog"))
.setParameters(properties)
.setCreateTime(System.currentTimeMillis())
.build();
metastore.createCatalog(catalog);
CatalogEntity newCatalog = CatalogEntity.builder()
.setCatalogName("catalog100")
.setOwner("root9")
.setComment(Optional.of("alter catalog parallel"))
.setParameters(properties)
.setCreateTime(System.currentTimeMillis())
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.alterCatalog(catalog.getName(), newCatalog);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropCatalog(catalog.getName());
}
/**
* test create database
*/
@ -404,6 +491,48 @@ public class TestJdbcHetuMetastore
}
}
/**
* test create database parallel
*/
@Test
public void testCreateDatabaseParallel()
throws InterruptedException
{
Map<String, String> properties = ImmutableMap.<String, String>builder()
.put("desc", "vschema")
.build();
DatabaseEntity database = DatabaseEntity.builder()
.setCatalogName(defaultCatalog.getName())
.setDatabaseName("database100")
.setOwner("root9")
.setComment(Optional.of("Hetu create database."))
.setCreateTime(System.currentTimeMillis())
.setParameters(properties)
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.createDatabaseIfNotExist(database);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropDatabase(database.getCatalogName(), database.getName());
}
/**
* test drop database
*/
@ -574,6 +703,58 @@ public class TestJdbcHetuMetastore
}
}
/**
* test alter database parallel
*/
@Test
public void testAlterDatabaseParallel()
throws InterruptedException
{
Map<String, String> properties = ImmutableMap.<String, String>builder()
.put("desc", "vschema")
.build();
DatabaseEntity database = DatabaseEntity.builder()
.setCatalogName(defaultCatalog.getName())
.setDatabaseName("database200")
.setOwner("root9")
.setComment(Optional.of("Hetu create database."))
.setCreateTime(System.currentTimeMillis())
.setParameters(properties)
.build();
metastore.createDatabase(database);
DatabaseEntity newDatabase = DatabaseEntity.builder()
.setCatalogName(defaultCatalog.getName())
.setDatabaseName("database200")
.setOwner("root10")
.setComment(Optional.of("alter database parallel"))
.setCreateTime(System.currentTimeMillis())
.setParameters(properties)
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.alterDatabase(database.getCatalogName(), database.getName(), newDatabase);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropDatabase(database.getCatalogName(), database.getName());
}
/**
* test get table
*/
@ -736,6 +917,43 @@ public class TestJdbcHetuMetastore
}
}
/**
* test create table parallel
*/
@Test
public void testCreateTableParallel()
throws InterruptedException
{
String tableName = "table100";
TableEntity tableEntity = TableEntity.builder()
.setCatalogName(defaultDatabase.getCatalogName())
.setDatabaseName(defaultDatabase.getName())
.setTableName(tableName)
.setTableType(TableEntityType.TABLE.toString())
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.createTableIfNotExist(tableEntity);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropTable(tableEntity.getCatalogName(), tableEntity.getDatabaseName(), tableName);
}
/**
* test drop table
*/
@ -818,6 +1036,52 @@ public class TestJdbcHetuMetastore
assertTableEquals(newTable1.get(), newTable);
}
/**
* test alter table parallel
*/
@Test
public void testAlterTableParallel()
throws InterruptedException
{
String tableName = "table200";
TableEntity tableEntity1 = TableEntity.builder()
.setCatalogName(defaultDatabase.getCatalogName())
.setDatabaseName(defaultDatabase.getName())
.setTableName(tableName)
.setTableType(TableEntityType.TABLE.toString())
.build();
metastore.createTable(tableEntity1);
TableEntity tableEntity2 = TableEntity.builder()
.setCatalogName(defaultDatabase.getCatalogName())
.setDatabaseName(defaultDatabase.getName())
.setTableName(tableName)
.setTableType(TableEntityType.TABLE.toString())
.setComment("alter table parallel")
.build();
testResult = true;
Thread[] threads = new Thread[5];
for (int i = 0; i < 5; i++) {
threads[i] = new Thread(() -> {
try {
metastore.alterTable(defaultDatabase.getCatalogName(), defaultDatabase.getName(), tableName, tableEntity2);
}
catch (PrestoException e) {
testResult = false;
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertTrue(testResult);
metastore.dropTable(defaultDatabase.getCatalogName(), defaultDatabase.getName(), tableName);
}
/**
* test alter view
*/

21
presto-spi/src/main/java/io/prestosql/spi/metastore/HetuMetastore.java
View File

@ -35,6 +35,13 @@ public interface HetuMetastore
*/
void createCatalog(CatalogEntity catalog);
/**
* create catalog if not exist
*
* @param catalog catalog
*/
void createCatalogIfNotExist(CatalogEntity catalog);
/**
* alter the catalog entity in hetu metastore,
* Currently only the owner,type,comment and parameters of the database can be changed.
@ -73,6 +80,13 @@ public interface HetuMetastore
*/
void createDatabase(DatabaseEntity database);
/**
* create database if not exist
*
* @param database database
*/
void createDatabaseIfNotExist(DatabaseEntity database);
/**
* alter the database entity in hetu metastore,
* Currently only the name,owner,comment and parameters of the database can be changed.
@ -115,6 +129,13 @@ public interface HetuMetastore
*/
void createTable(TableEntity table);
/**
* create table if not exist
*
* @param table table
*/
void createTableIfNotExist(TableEntity table);
/**
* drop table
*