From 469ac921b6e8ef2f5ce144b52956ab0aaa5ff0d4 Mon Sep 17 00:00:00 2001 From: shuaiwang999 Date: Tue, 9 Feb 2021 15:46:33 +0800 Subject: [PATCH] add parallel processing for metastore --- .../core/metastore/HetuMetastoreCache.java | 34 +++ .../hetufilesystem/HetuFsMetastore.java | 42 +++ .../core/metastore/jdbc/JdbcBasedLock.java | 106 +++++++ .../metastore/jdbc/JdbcHetuMetastore.java | 61 +++- .../core/metastore/jdbc/JdbcMetadataDao.java | 23 ++ .../core/metastore/jdbc/JdbcMetadataUtil.java | 26 ++ .../core/metastore/jdbc/MetadataTableDao.java | 12 + .../hetufilesystem/TestHetuFsMetastore.java | 265 ++++++++++++++++++ .../jdbc/mysql/TestJdbcHetuMetastore.java | 264 +++++++++++++++++ .../spi/metastore/HetuMetastore.java | 21 ++ 10 files changed, 845 insertions(+), 9 deletions(-) create mode 100644 hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcBasedLock.java diff --git a/hetu-metastore/src/main/java/io/hetu/core/metastore/HetuMetastoreCache.java b/hetu-metastore/src/main/java/io/hetu/core/metastore/HetuMetastoreCache.java index c28a643a9..dda2bf879 100644 --- a/hetu-metastore/src/main/java/io/hetu/core/metastore/HetuMetastoreCache.java +++ b/hetu-metastore/src/main/java/io/hetu/core/metastore/HetuMetastoreCache.java @@ -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) { diff --git a/hetu-metastore/src/main/java/io/hetu/core/metastore/hetufilesystem/HetuFsMetastore.java b/hetu-metastore/src/main/java/io/hetu/core/metastore/hetufilesystem/HetuFsMetastore.java index a4d199b91..ff2216d48 100644 --- a/hetu-metastore/src/main/java/io/hetu/core/metastore/hetufilesystem/HetuFsMetastore.java +++ b/hetu-metastore/src/main/java/io/hetu/core/metastore/hetufilesystem/HetuFsMetastore.java @@ -168,6 +168,20 @@ public class HetuFsMetastore }); } + @Override + public void createCatalogIfNotExist(CatalogEntity catalog) + { + try { + createCatalog(catalog); + } + catch (PrestoException e) { + Optional 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 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 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) { diff --git a/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcBasedLock.java b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcBasedLock.java new file mode 100644 index 000000000..28b2dab4c --- /dev/null +++ b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcBasedLock.java @@ -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++; + } + } + } +} diff --git a/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcHetuMetastore.java b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcHetuMetastore.java index 97870fc0d..d10e95039 100644 --- a/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcHetuMetastore.java +++ b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcHetuMetastore.java @@ -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> properties = mapToList(catalog.getParameters()); @@ -79,6 +80,20 @@ public class JdbcHetuMetastore }); } + @Override + public void createCatalogIfNotExist(CatalogEntity catalog) + { + try { + createCatalog(catalog); + } + catch (PrestoException e) { + Optional 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 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 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 diff --git a/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataDao.java b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataDao.java index bb489448c..802d8d397 100644 --- a/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataDao.java +++ b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataDao.java @@ -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(); } diff --git a/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataUtil.java b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataUtil.java index 904f74e3f..16180df14 100644 --- a/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataUtil.java +++ b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/JdbcMetadataUtil.java @@ -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 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(); } } diff --git a/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/MetadataTableDao.java b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/MetadataTableDao.java index 6264cbafa..a7b849a16 100644 --- a/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/MetadataTableDao.java +++ b/hetu-metastore/src/main/java/io/hetu/core/metastore/jdbc/MetadataTableDao.java @@ -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(); } diff --git a/hetu-metastore/src/test/java/io/hetu/core/metastore/hetufilesystem/TestHetuFsMetastore.java b/hetu-metastore/src/test/java/io/hetu/core/metastore/hetufilesystem/TestHetuFsMetastore.java index f3f37b542..762e91807 100644 --- a/hetu-metastore/src/test/java/io/hetu/core/metastore/hetufilesystem/TestHetuFsMetastore.java +++ b/hetu-metastore/src/test/java/io/hetu/core/metastore/hetufilesystem/TestHetuFsMetastore.java @@ -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 properties = ImmutableMap.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 properties = ImmutableMap.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 properties = ImmutableMap.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 properties = ImmutableMap.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 */ diff --git a/hetu-metastore/src/test/java/io/hetu/core/metastore/jdbc/mysql/TestJdbcHetuMetastore.java b/hetu-metastore/src/test/java/io/hetu/core/metastore/jdbc/mysql/TestJdbcHetuMetastore.java index f2f8bd924..1d76f4807 100644 --- a/hetu-metastore/src/test/java/io/hetu/core/metastore/jdbc/mysql/TestJdbcHetuMetastore.java +++ b/hetu-metastore/src/test/java/io/hetu/core/metastore/jdbc/mysql/TestJdbcHetuMetastore.java @@ -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 properties = ImmutableMap.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 properties = ImmutableMap.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 properties = ImmutableMap.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 properties = ImmutableMap.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 */ diff --git a/presto-spi/src/main/java/io/prestosql/spi/metastore/HetuMetastore.java b/presto-spi/src/main/java/io/prestosql/spi/metastore/HetuMetastore.java index f5069590a..c499631b0 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/metastore/HetuMetastore.java +++ b/presto-spi/src/main/java/io/prestosql/spi/metastore/HetuMetastore.java @@ -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 *