diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/ForCachingHiveMetastoreTableRefresh.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/ForCachingHiveMetastoreTableRefresh.java new file mode 100644 index 000000000..b43bd0f12 --- /dev/null +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/ForCachingHiveMetastoreTableRefresh.java @@ -0,0 +1,32 @@ +/* + * Copyright (C) 2018-2021. 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.prestosql.plugin.hive; + +import javax.inject.Qualifier; + +import java.lang.annotation.Retention; +import java.lang.annotation.Target; + +import static java.lang.annotation.ElementType.FIELD; +import static java.lang.annotation.ElementType.METHOD; +import static java.lang.annotation.ElementType.PARAMETER; +import static java.lang.annotation.RetentionPolicy.RUNTIME; + +@Retention(RUNTIME) +@Target({FIELD, PARAMETER, METHOD}) +@Qualifier +public @interface ForCachingHiveMetastoreTableRefresh +{ +} diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveConfig.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveConfig.java index 7d0dda07a..9c7068f50 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveConfig.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveConfig.java @@ -502,7 +502,7 @@ public class HiveConfig return this; } - @Min(1) + @Min(10) public int getMaxMetastoreRefreshThreads() { return maxMetastoreRefreshThreads; diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveModule.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveModule.java index 0d4ee6695..ed892c38e 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveModule.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveModule.java @@ -187,7 +187,17 @@ public class HiveModule { return new BoundedExecutor( newCachedThreadPool(daemonThreadsNamed("hive-metastore-" + catalogName + "-%s")), - hiveConfig.getMaxMetastoreRefreshThreads()); + (int) Math.max(hiveConfig.getMaxMetastoreRefreshThreads() * 0.9, 9)); + } + + @ForCachingHiveMetastoreTableRefresh + @Singleton + @Provides + public Executor createCachingHiveMetastoreTableRefreshExecutor(HiveCatalogName catalogName, HiveConfig hiveConfig) + { + return new BoundedExecutor( + newCachedThreadPool(daemonThreadsNamed("hive-metastore-refresh-" + catalogName + "-%s")), + (int) Math.max(hiveConfig.getMaxMetastoreRefreshThreads() * 0.1, 1)); } @Singleton diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/CachingHiveMetastore.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/CachingHiveMetastore.java index 01a6d6bdd..8297a2485 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/CachingHiveMetastore.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/CachingHiveMetastore.java @@ -24,6 +24,7 @@ import com.google.common.util.concurrent.UncheckedExecutionException; import io.airlift.log.Logger; import io.airlift.units.Duration; import io.prestosql.plugin.hive.ForCachingHiveMetastore; +import io.prestosql.plugin.hive.ForCachingHiveMetastoreTableRefresh; import io.prestosql.plugin.hive.HiveBasicStatistics; import io.prestosql.plugin.hive.HiveConfig; import io.prestosql.plugin.hive.HiveErrorCode; @@ -113,11 +114,16 @@ public class CachingHiveMetastore private final boolean dontVerifyCacheEntry; @Inject - public CachingHiveMetastore(@ForCachingHiveMetastore HiveMetastore delegate, @ForCachingHiveMetastore Executor executor, HiveConfig hiveConfig, NodeManager nodeManager) + public CachingHiveMetastore(@ForCachingHiveMetastore HiveMetastore delegate, + @ForCachingHiveMetastore Executor executor, + @ForCachingHiveMetastoreTableRefresh Executor tableRefreshExecutor, + HiveConfig hiveConfig, + NodeManager nodeManager) { this( delegate, executor, + tableRefreshExecutor, hiveConfig.getMetastoreCacheTtl(), hiveConfig.getMetastoreRefreshInterval(), hiveConfig.getMetastoreDBCacheTtl(), @@ -126,14 +132,14 @@ public class CachingHiveMetastore !(nodeManager.getCurrentNode().isCoordinator() || hiveConfig.getWorkerMetaStoreCacheEnabled())); } - public CachingHiveMetastore(HiveMetastore delegate, Executor executor, Duration cacheTtl, Duration refreshInterval, + public CachingHiveMetastore(HiveMetastore delegate, Executor executor, Executor tableRefreshExecutor, Duration cacheTtl, Duration refreshInterval, Duration tableCacheTtl, Duration tableRefreshInterval, long maximumSize, boolean skipCache) { this( delegate, executor, - OptionalLong.of(cacheTtl.toMillis()), + tableRefreshExecutor, OptionalLong.of(cacheTtl.toMillis()), refreshInterval.toMillis() >= cacheTtl.toMillis() ? OptionalLong.empty() : OptionalLong.of(refreshInterval.toMillis()), OptionalLong.of(tableCacheTtl.toMillis()), tableRefreshInterval.toMillis() >= tableCacheTtl.toMillis() ? OptionalLong.empty() : OptionalLong.of(tableRefreshInterval.toMillis()), @@ -147,6 +153,7 @@ public class CachingHiveMetastore return new CachingHiveMetastore( delegate, newDirectExecutorService(), + newDirectExecutorService(), OptionalLong.empty(), OptionalLong.empty(), OptionalLong.empty(), @@ -155,7 +162,7 @@ public class CachingHiveMetastore false || delegate instanceof CachingHiveMetastore); } - private CachingHiveMetastore(HiveMetastore delegate, Executor executor, + private CachingHiveMetastore(HiveMetastore delegate, Executor executor, Executor tableRefreshExecutor, OptionalLong expiresAfterWriteMillisTable, OptionalLong refreshMillsTable, OptionalLong expiresAfterWriteMillisDB, OptionalLong refreshMillsDB, long maximumSize, boolean skipCache) @@ -197,10 +204,10 @@ public class CachingHiveMetastore .build(asyncReloading(CacheLoader.from(this::loadAllViews), executor)); tableCache = newCacheBuilder(tableCacheTtl, tableRefreshTtl, maximumSize) - .build(asyncReloading(CacheLoader.from(this::loadTable), executor)); + .build(asyncReloading(CacheLoader.from(this::loadTable), tableRefreshExecutor)); partitionNamesCache = newCacheBuilder(tableCacheTtl, tableRefreshTtl, maximumSize) - .build(asyncReloading(CacheLoader.from(this::loadPartitionNames), executor)); + .build(asyncReloading(CacheLoader.from(this::loadPartitionNames), tableRefreshExecutor)); tableStatisticsCache = newCacheBuilder(expiresAfterWriteMillisTable, refreshMillsTable, maximumSize) .build(asyncReloading(new CacheLoader, WithValidation>() diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java index 64f6dd57c..917aa2da5 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java @@ -584,6 +584,7 @@ public abstract class AbstractTestHive protected ConnectorPageSourceProvider pageSourceProvider; protected ConnectorPageSinkProvider pageSinkProvider; protected ExecutorService executor; + protected ExecutorService executorRefresh; private ScheduledExecutorService heartbeatService; private ScheduledExecutorService vacuumExecutorService; @@ -593,6 +594,7 @@ public abstract class AbstractTestHive public void setupClass() { executor = newCachedThreadPool(daemonThreadsNamed("hive-%s")); + executorRefresh = newCachedThreadPool(daemonThreadsNamed("hive-refresh-%s")); heartbeatService = newScheduledThreadPool(1); vacuumExecutorService = newScheduledThreadPool(1); hiveMetastoreClientService = newScheduledThreadPool(1); @@ -714,7 +716,7 @@ public abstract class AbstractTestHive HiveMetastore metastore = new CachingHiveMetastore( new BridgingHiveMetastore(new ThriftHiveMetastore(metastoreLocator, new ThriftHiveMetastoreConfig())), executor, - Duration.valueOf("1m"), + executorRefresh, Duration.valueOf("1m"), Duration.valueOf("15s"), Duration.valueOf("1m"), Duration.valueOf("15s"), diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java index e297f3f36..555aa9930 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java @@ -74,6 +74,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; import java.util.UUID; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; @@ -173,6 +174,7 @@ public abstract class AbstractTestHiveFileSystem MetastoreLocator metastoreLocator = new TestingMetastoreLocator(config, host, port); ExecutorService executor = newCachedThreadPool(daemonThreadsNamed("hive-%s")); + ExecutorService executorRefresh = newCachedThreadPool(daemonThreadsNamed("hive-refresh-%s")); HivePartitionManager hivePartitionManager = new HivePartitionManager(TYPE_MANAGER, config); HdfsConfiguration hdfsConfiguration = hdfsConfigurationProvider.apply(config); @@ -181,6 +183,7 @@ public abstract class AbstractTestHiveFileSystem metastoreClient = new TestingHiveMetastore( new BridgingHiveMetastore(new ThriftHiveMetastore(metastoreLocator, new ThriftHiveMetastoreConfig())), executor, + executorRefresh, config, getBasePath(), hdfsEnvironment); @@ -472,9 +475,9 @@ public abstract class AbstractTestHiveFileSystem private final Path basePath; private final HdfsEnvironment hdfsEnvironment; - public TestingHiveMetastore(HiveMetastore delegate, ExecutorService executor, HiveConfig hiveConfig, Path basePath, HdfsEnvironment hdfsEnvironment) + public TestingHiveMetastore(HiveMetastore delegate, ExecutorService executor, Executor executorRefresh, HiveConfig hiveConfig, Path basePath, HdfsEnvironment hdfsEnvironment) { - super(delegate, executor, hiveConfig, new TestingNodeManager("fake-environment")); + super(delegate, executor, executorRefresh, hiveConfig, new TestingNodeManager("fake-environment")); this.basePath = basePath; this.hdfsEnvironment = hdfsEnvironment; } diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveWriterFactory.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveWriterFactory.java index 7667c518b..e0a74c2f6 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveWriterFactory.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveWriterFactory.java @@ -67,17 +67,19 @@ public class TestHiveWriterFactory { private ThriftMetastoreClient mockClient; protected ExecutorService executor; + protected ExecutorService executorRefresh; protected HiveMetastore metastore; private void setUp() { mockClient = new MockThriftMetastoreClient(); executor = newCachedThreadPool(daemonThreadsNamed("hive-%s")); + executorRefresh = newCachedThreadPool(daemonThreadsNamed("hive-refresh-%s")); MetastoreLocator metastoreLocator = new MockMetastoreLocator(mockClient); metastore = new CachingHiveMetastore( new BridgingHiveMetastore(new ThriftHiveMetastore(metastoreLocator, new ThriftHiveMetastoreConfig())), executor, - Duration.valueOf("1m"), + executorRefresh, Duration.valueOf("1m"), Duration.valueOf("15s"), Duration.valueOf("1m"), Duration.valueOf("15s"), diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/metastore/TestCachingHiveMetastore.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/metastore/TestCachingHiveMetastore.java index 70d0df611..e122b8c7b 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/metastore/TestCachingHiveMetastore.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/metastore/TestCachingHiveMetastore.java @@ -56,11 +56,12 @@ public class TestCachingHiveMetastore mockClient = new MockThriftMetastoreClient(); MetastoreLocator metastoreLocator = new MockMetastoreLocator(mockClient); ListeningExecutorService executor = listeningDecorator(newCachedThreadPool(daemonThreadsNamed("test-%s"))); + ListeningExecutorService executorRefresh = listeningDecorator(newCachedThreadPool(daemonThreadsNamed("test-%s"))); ThriftHiveMetastore thriftHiveMetastore = new ThriftHiveMetastore(metastoreLocator, new ThriftHiveMetastoreConfig()); metastore = new CachingHiveMetastore( new BridgingHiveMetastore(thriftHiveMetastore), executor, - new Duration(5, TimeUnit.MINUTES), + executorRefresh, new Duration(5, TimeUnit.MINUTES), new Duration(1, TimeUnit.MINUTES), new Duration(5, TimeUnit.MINUTES), new Duration(1, TimeUnit.MINUTES),